""" DOOBAT - Service Odoo Connexion XML-RPC à Odoo Online avec cache, retry et pagination. Copyright 2026 Vincent Gallet - Vincesoft """ import logging import re import time import unicodedata import xmlrpc.client from configparser import ConfigParser from datetime import datetime, timedelta from functools import wraps from typing import Any, Optional logger = logging.getLogger("doobat.odoo") # ── Décorateur retry ─────────────────────────────────────────────────────────── def retry(max_attempts: int = 3, delay: float = 1.0, exceptions=(Exception,)): """Décorateur pour retry automatique sur les appels Odoo.""" def decorator(func): @wraps(func) def wrapper(*args, **kwargs): last_error = None for attempt in range(1, max_attempts + 1): try: return func(*args, **kwargs) except exceptions as e: last_error = e if attempt < max_attempts: logger.warning( f"Tentative {attempt}/{max_attempts} échouée pour {func.__name__} : {e}. " f"Retry dans {delay}s." ) time.sleep(delay) else: logger.error(f"Toutes les tentatives épuisées pour {func.__name__} : {e}") raise last_error return wrapper return decorator # ── Cache léger en mémoire ───────────────────────────────────────────────────── class SimpleCache: """Cache TTL simple pour réduire les appels Odoo.""" def __init__(self, ttl_seconds: int = 300): self._store: dict[str, tuple[Any, float]] = {} self._ttl = ttl_seconds def get(self, key: str) -> Optional[Any]: if key in self._store: value, expires_at = self._store[key] if time.time() < expires_at: return value del self._store[key] return None def set(self, key: str, value: Any) -> None: self._store[key] = (value, time.time() + self._ttl) def invalidate(self, key: str) -> None: self._store.pop(key, None) def clear(self) -> None: self._store.clear() # ── Service Odoo ─────────────────────────────────────────────────────────────── class OdooService: """ Service principal pour interagir avec Odoo Online via XML-RPC. Gère authentification, cache, retry et pagination. """ def __init__(self, config: ConfigParser): self._url = config.get("ODOO", "url").rstrip("/") self._db = config.get("ODOO", "db") self._username = config.get("ODOO", "username") self._password = config.get("ODOO", "password") self._timeout = config.getint("ODOO", "timeout", fallback=30) self._max_retries = config.getint("ODOO", "max_retries", fallback=3) phone_fields_raw = config.get("ODOO", "phone_fields", fallback="phone,mobile") self._phone_fields = [ field.strip() for field in phone_fields_raw.split(",") if field.strip() ] or ["phone", "mobile"] self._partner_fields: Optional[set[str]] = None self._uid: Optional[int] = None self._cache = SimpleCache(ttl_seconds=300) # Proxies XML-RPC transport = xmlrpc.client.Transport() self._common = xmlrpc.client.ServerProxy( f"{self._url}/xmlrpc/2/common", transport=transport, allow_none=True, ) self._models = xmlrpc.client.ServerProxy( f"{self._url}/xmlrpc/2/object", allow_none=True, ) logger.info(f"OdooService initialisé pour {self._url} / db={self._db}") # ── Authentification ─────────────────────────────────────────────────────── @retry(max_attempts=3, delay=2.0, exceptions=(Exception,)) def authenticate(self) -> int: """Authentifie sur Odoo et retourne l'UID utilisateur.""" if self._uid: return self._uid uid = self._common.authenticate( self._db, self._username, self._password, {} ) if not uid: raise ConnectionError( f"Authentification Odoo échouée pour {self._username}@{self._db}" ) self._uid = uid logger.info(f"Authentification Odoo réussie, UID={uid}") return uid def _execute(self, model: str, method: str, domain: list, **kwargs) -> Any: """Exécute un appel XML-RPC Odoo avec gestion d'erreurs.""" uid = self.authenticate() return self._models.execute_kw( self._db, uid, self._password, model, method, [domain], kwargs, ) # ── Recherche partenaire ─────────────────────────────────────────────────── def _get_partner_fields(self) -> set[str]: """Retourne les champs disponibles sur res.partner.""" if self._partner_fields is not None: return self._partner_fields uid = self.authenticate() fields = self._models.execute_kw( self._db, uid, self._password, "res.partner", "fields_get", [], {"attributes": ["string"]}, ) self._partner_fields = set(fields.keys()) return self._partner_fields @staticmethod def clean_phone(phone: str) -> str: """Normalise un numéro de téléphone : retire espaces, tirets, points.""" if not phone: return "" cleaned = re.sub(r"[\s\-\.\(\)]", "", phone.strip()) # Convertir format local FR en international if cleaned.startswith("0") and len(cleaned) == 10: cleaned = "+33" + cleaned[1:] return cleaned @classmethod def phones_match(cls, expected: str, actual: str) -> bool: """Compare deux numeros apres normalisation.""" expected_clean = cls.clean_phone(expected) actual_clean = cls.clean_phone(str(actual or "")) if not expected_clean or not actual_clean: return False return ( expected_clean == actual_clean or expected_clean[-9:] == actual_clean[-9:] ) @staticmethod def phone_display_variants(cleaned: str) -> list[str]: """Construit des variantes courantes de formatage pour Odoo.""" variants = [cleaned] if cleaned.startswith("+33") and len(cleaned) == 12: national = "0" + cleaned[3:] groups = [national[i:i + 2] for i in range(0, len(national), 2)] variants.extend([ national, " ".join(groups), ".".join(groups), "-".join(groups), "+33 " + " ".join(groups)[1:], "+33 " + " ".join(groups[1:]), ]) return list(dict.fromkeys(variants)) def find_partner_by_phone(self, phone: str, refresh: bool = False) -> Optional[dict]: """ Recherche un partenaire Odoo par numéro de téléphone ou mobile. Retourne le premier partenaire trouvé ou None. """ cleaned = self.clean_phone(phone) cache_key = f"partner_phone_{cleaned}" cached = self._cache.get(cache_key) if cached is not None and not refresh: return cached # Rechercher sur phone ET mobile (différents formats) available_fields = self._get_partner_fields() phone_fields = [ field for field in self._phone_fields if field in available_fields ] missing_fields = [ field for field in self._phone_fields if field not in available_fields ] if missing_fields: logger.warning( "Champs telephone Odoo ignores car absents de res.partner : %s", ", ".join(missing_fields), ) if not phone_fields: logger.error( "Aucun champ telephone configure n'existe dans res.partner : %s", ", ".join(self._phone_fields), ) return None variants = self.phone_display_variants(cleaned) conditions = [] for field in phone_fields: conditions.append([field, "in", variants]) conditions.append([field, "ilike", cleaned[-9:]]) conditions.append([field, "ilike", cleaned[-4:]]) conditions.append([field, "ilike", cleaned[-2:]]) domain = self._or_domain(conditions) fields = [ field for field in [ "id", "name", "phone", "email", "x_studio_titre_bot", "user_id", "active", ] if field in available_fields ] for field in phone_fields: if field not in fields: fields.append(field) try: results = self._execute("res.partner", "search_read", domain, fields=fields, limit=20) partner = None for candidate in results: for field in phone_fields: if self.phones_match(cleaned, candidate.get(field, "")): partner = candidate break if partner: break self._cache.set(cache_key, partner) if partner: logger.info(f"Partenaire trouvé par téléphone : {partner['name']} (ID={partner['id']})") else: logger.info(f"Aucun partenaire trouvé pour téléphone : {cleaned}") return partner except Exception as e: logger.error(f"Erreur recherche partenaire par téléphone {cleaned} : {e}") return None @staticmethod def _or_domain(conditions: list[list]) -> list: """Construit un domaine Odoo en OR a partir d'une liste de conditions.""" if not conditions: return [] if len(conditions) == 1: return conditions[0] return ["|"] * (len(conditions) - 1) + conditions @staticmethod def _or_leaves(conditions: list[list]) -> list: """Comme _or_domain mais retourne toujours une liste de feuilles, utilisable directement dans un domaine combine par AND (append/+ ).""" if not conditions: return [] if len(conditions) == 1: return [conditions[0]] return ["|"] * (len(conditions) - 1) + conditions @staticmethod def _strip_accents(value: str) -> str: """Retire les accents d'une chaine (ex: 'décapant' -> 'decapant').""" return "".join( char for char in unicodedata.normalize("NFKD", str(value or "")) if not unicodedata.combining(char) ) @staticmethod def _add_singular_variants(values: list[str]) -> list[str]: """Ajoute la forme singuliere (sans 's' final) de chaque terme, pour qu'une recherche au pluriel ('boites') trouve aussi les articles nommes au singulier ('Boite de ...').""" result = list(values) for term in values: if term.endswith("s") and len(term) > 3: singular = term[:-1] if not any(singular.casefold() == existing.casefold() for existing in result): result.append(singular) return result @classmethod def _search_terms(cls, value: str) -> list[str]: """Retourne le terme original et ses variantes (sans accent, singulier), pour que les recherches ilike traitent les lettres accentuees, la casse et le pluriel comme equivalents.""" value = (value or "").strip() if not value: return [] terms = [value] stripped = cls._strip_accents(value) if stripped.casefold() != value.casefold(): terms.append(stripped) return cls._add_singular_variants(terms) @classmethod def _accent_insensitive_key(cls, value: str) -> str: """Clef de comparaison ignorant accents et casse.""" return cls._strip_accents(str(value or "")).casefold() def find_partner_by_id(self, partner_id: int, refresh: bool = False) -> Optional[dict]: """Retourne un partenaire Odoo par son ID.""" cache_key = f"partner_id_{partner_id}" cached = self._cache.get(cache_key) if cached is not None and not refresh: return cached available_fields = self._get_partner_fields() fields = [ field for field in [ "id", "name", "phone", "email", "x_studio_titre_bot", "user_id", "active", ] if field in available_fields ] for field in self._phone_fields: if field in available_fields and field not in fields: fields.append(field) try: results = self._execute( "res.partner", "search_read", [["id", "=", partner_id], ["active", "=", True]], fields=fields, limit=1, ) partner = results[0] if results else None self._cache.set(cache_key, partner) return partner except Exception as e: logger.error(f"Erreur recherche partenaire ID={partner_id} : {e}") return None def get_partner_phone(self, partner: dict) -> str: """Retourne le premier numero configure sur une fiche partenaire.""" for field in self._phone_fields: phone = str(partner.get(field, "") or "").strip() if phone: return phone return "" def find_partner_by_name(self, name: str) -> Optional[dict]: """Recherche un partenaire Odoo par nom (ilike).""" cache_key = f"partner_name_{name.lower()}" cached = self._cache.get(cache_key) if cached is not None: return cached fields = ["id", "name", "user_id", "x_studio_titre_bot"] try: results = self._execute( "res.partner", "search_read", [["name", "ilike", name], ["active", "=", True]], fields=fields, limit=5, ) partner = results[0] if results else None self._cache.set(cache_key, partner) return partner except Exception as e: logger.error(f"Erreur recherche partenaire par nom '{name}' : {e}") return None def find_supplier_by_name(self, name: str) -> Optional[dict]: """Recherche une fiche fournisseur active par nom.""" try: results = self._execute( "res.partner", "search_read", [ ["name", "ilike", name], ["supplier_rank", ">", 0], ["active", "=", True], ], fields=["id", "name", "supplier_rank"], order="supplier_rank desc, name", limit=5, ) return results[0] if results else None except Exception as e: logger.error(f"Erreur recherche fournisseur par nom '{name}' : {e}") return None def search_document_contact_candidates(self, name: str, limit: int = 5) -> list[dict]: """Recherche des contacts possibles pour une demande de documents. Retourne des candidats typés client/fournisseur afin que le webhook puisse inferer le bon flux ou proposer un choix a l'utilisateur. """ clean_name = (name or "").strip() if not clean_name: return [] candidates: list[dict] = [] try: name_conditions = [ ["name", "ilike", term] for term in self._search_terms(clean_name) ] client_results = self._execute( "res.partner", "search_read", self._or_leaves(name_conditions) + [ ["customer_rank", ">", 0], ["active", "=", True], ], fields=["id", "name", "user_id", "customer_rank", "supplier_rank"], order="customer_rank desc, name", limit=limit, ) for partner in client_results: candidates.append({ **partner, "document_kind": "client", "rank": partner.get("customer_rank", 0) or 0, }) supplier_results = self._execute( "res.partner", "search_read", self._or_leaves(name_conditions) + [ ["supplier_rank", ">", 0], ["active", "=", True], ], fields=["id", "name", "user_id", "customer_rank", "supplier_rank"], order="supplier_rank desc, name", limit=limit, ) for partner in supplier_results: candidates.append({ **partner, "document_kind": "supplier", "rank": partner.get("supplier_rank", 0) or 0, }) if not candidates: words = [ word for word in re.split(r"\W+", clean_name) if len(word) >= 3 ][:4] for word in list(words): stripped = self._strip_accents(word) if stripped.casefold() != word.casefold() and not any( stripped.casefold() == existing.casefold() for existing in words ): words.append(stripped) if words: word_conditions = [["name", "ilike", word] for word in words] generic_domain = ( word_conditions if len(word_conditions) == 1 else self._or_domain(word_conditions) ) generic_domain.append(["active", "=", True]) generic_results = self._execute( "res.partner", "search_read", generic_domain, fields=["id", "name", "user_id", "customer_rank", "supplier_rank"], order="name", limit=limit, ) for partner in generic_results: customer_rank = partner.get("customer_rank", 0) or 0 supplier_rank = partner.get("supplier_rank", 0) or 0 document_kind = ( "supplier" if supplier_rank > customer_rank else "client" ) candidates.append({ **partner, "document_kind": document_kind, "rank": max(customer_rank, supplier_rank), }) except Exception as e: logger.error(f"Erreur recherche candidats documents '{clean_name}' : {e}") return [] deduped: dict[tuple[int, str], dict] = {} for candidate in candidates: key = (candidate["id"], candidate["document_kind"]) deduped[key] = candidate # Fusionner les contacts dont le nom ne diffère que par les accents : # ce sont la même proposition pour l'utilisateur (accents = mêmes lettres). name_deduped: dict[tuple[str, str], dict] = {} for candidate in deduped.values(): key = ( self._accent_insensitive_key(candidate.get("name", "")), candidate["document_kind"], ) current = name_deduped.get(key) if not current or (candidate.get("rank", 0) or 0) > (current.get("rank", 0) or 0): name_deduped[key] = candidate normalized = self._accent_insensitive_key(clean_name) return sorted( name_deduped.values(), key=lambda c: ( self._accent_insensitive_key(c.get("name", "")) != normalized, -int(c.get("rank", 0) or 0), str(c.get("name", "")).casefold(), ), )[:limit] # ── Vérifications commerciales ───────────────────────────────────────────── @staticmethod def is_commercial(partner: dict) -> bool: """Retourne True si le partenaire a le rôle Commercial.""" titre = partner.get("x_studio_titre_bot", "") return str(titre).strip().lower() == "commercial" def check_client_assigned_to_commercial( self, client_id: int, commercial_user_id: int ) -> bool: """ Vérifie que le client est bien affecté au commercial via res.partner.user_id. """ cache_key = f"assigned_{client_id}_{commercial_user_id}" cached = self._cache.get(cache_key) if cached is not None: return cached try: results = self._execute( "res.partner", "search_read", [["id", "=", client_id], ["user_id", "=", commercial_user_id]], fields=["id", "name", "user_id"], limit=1, ) result = bool(results) self._cache.set(cache_key, result) logger.info( f"Client {client_id} affecté au commercial {commercial_user_id} : {result}" ) return result except Exception as e: logger.error(f"Erreur vérification affectation client : {e}") return False def get_commercial_of_client(self, client_id: int) -> Optional[str]: """Retourne le nom du commercial affecté à un client.""" try: results = self._execute( "res.partner", "search_read", [["id", "=", client_id]], fields=["user_id"], limit=1, ) if results and results[0].get("user_id"): return results[0]["user_id"][1] # [id, name] return None except Exception as e: logger.error(f"Erreur récupération commercial du client {client_id} : {e}") return None # ── Articles et tarif ────────────────────────────────────────────────────── def get_all_products(self, limit: int = 500, offset: int = 0) -> list[dict]: """ Retourne tous les articles actifs Odoo avec prix et catégorie. Pagination intégrée. """ cache_key = f"products_{limit}_{offset}" cached = self._cache.get(cache_key) if cached is not None: return cached fields = [ "id", "name", "default_code", "list_price", "categ_id", "active", "type", "description_sale", ] try: results = self._execute( "product.template", "search_read", [["active", "=", True], ["sale_ok", "=", True]], fields=fields, limit=limit, offset=offset, order="categ_id, name", ) self._cache.set(cache_key, results) logger.info(f"Articles récupérés : {len(results)} (offset={offset})") return results except Exception as e: logger.error(f"Erreur récupération articles : {e}") return [] def get_products_paginated(self, page_size: int = 200) -> list[dict]: """Récupère tous les articles par pages successives.""" all_products = [] offset = 0 while True: batch = self.get_all_products(limit=page_size, offset=offset) if not batch: break all_products.extend(batch) if len(batch) < page_size: break offset += page_size logger.info(f"Total articles récupérés : {len(all_products)}") return all_products def search_product_by_name(self, article_name: str) -> list[dict]: """Recherche des articles par nom (ilike).""" fields = [ "id", "name", "default_code", "list_price", "categ_id", "type", ] try: conditions = [] for term in self._search_terms(article_name): conditions.append(["name", "ilike", term]) conditions.append(["default_code", "ilike", term]) results = self._execute( "product.template", "search_read", [["active", "=", True]] + self._or_leaves(conditions), fields=fields, limit=10, ) logger.info(f"Recherche article '{article_name}' : {len(results)} résultat(s)") return results except Exception as e: logger.error(f"Erreur recherche article '{article_name}' : {e}") return [] def search_similar_stock_products(self, product_name: str, limit: int = 12) -> list[dict]: """Recherche les articles de stock ressemblants a une demande libre.""" clean_name = (product_name or "").strip() if not clean_name: return [] words = [ word for word in re.split(r"\W+", clean_name) if len(word) >= 3 ][:5] fields = [ "id", "name", "default_code", "list_price", "categ_id", "qty_available", "virtual_available", ] searches = [clean_name] + [word for word in words if word.casefold() != clean_name.casefold()] # Ajouter les variantes sans accent : "é" et "e" doivent être traités # comme la même lettre lors de la recherche. for term in list(searches): stripped = self._strip_accents(term) if stripped.casefold() != term.casefold() and not any( stripped.casefold() == existing.casefold() for existing in searches ): searches.append(stripped) # Ajouter les formes singulieres : "boites" doit aussi trouver # les articles nommes au singulier ("Boite de ..."). searches = self._add_singular_variants(searches) candidates: dict[int, dict] = {} try: for index, term in enumerate(searches): results = self._execute( "product.product", "search_read", [ ["active", "=", True], "|", ["name", "ilike", term], ["default_code", "ilike", term], ], fields=fields, limit=limit, order="name", ) for product in results: current = candidates.get(product["id"]) score = self._score_product_candidate(product, clean_name, words, index) if not current or score > current.get("_match_score", 0): candidates[product["id"]] = { **product, "_match_score": score, } except Exception as e: logger.error(f"Erreur recherche articles similaires '{clean_name}' : {e}") return [] # Fusionner les articles dont le nom ne diffère que par les accents : # ce sont la même proposition pour l'utilisateur, on ne garde que la # mieux notée pour éviter de poser une question inutile. deduped: dict[tuple[str, str], dict] = {} for product in candidates.values(): key = ( self._accent_insensitive_key(product.get("name", "")), str(product.get("default_code") or "").casefold(), ) current = deduped.get(key) if not current or product.get("_match_score", 0) > current.get("_match_score", 0): deduped[key] = product return sorted( deduped.values(), key=lambda p: ( -float(p.get("_match_score", 0)), str(p.get("name", "")).casefold(), ), )[:limit] @classmethod def _score_product_candidate(cls, product: dict, query: str, words: list[str], search_index: int) -> float: """Score simple pour presenter d'abord les articles les plus proches. Les accents sont ignorés : 'é' et 'e' comptent comme la même lettre.""" label = cls._accent_insensitive_key(" ".join([ str(product.get("name", "")), str(product.get("default_code", "") or ""), ])) normalized_query = cls._accent_insensitive_key(query) score = max(0, 20 - search_index) if normalized_query and normalized_query in label: score += 50 score += sum(10 for word in words if cls._accent_insensitive_key(word) in label) if cls._accent_insensitive_key(product.get("default_code", "") or "") == normalized_query: score += 100 return score # ── Stock ────────────────────────────────────────────────────────────────── def get_stock_for_product(self, product_name: str) -> list[dict]: """ Retourne les informations de stock pour un article donné. Combine product.product et stock.quant. """ cache_key = f"stock_{product_name.lower()}" cached = self._cache.get(cache_key) if cached is not None: return cached try: # Rechercher les variantes de produit (accents ignorés) conditions = [] for term in self._search_terms(product_name): conditions.append(["name", "ilike", term]) conditions.append(["default_code", "ilike", term]) products = self._execute( "product.product", "search_read", [["active", "=", True]] + self._or_leaves(conditions), fields=[ "id", "name", "default_code", "list_price", "categ_id", "qty_available", "virtual_available", ], limit=20, ) if not products: logger.info(f"Aucun produit trouvé pour '{product_name}'") return [] # Enrichir avec les emplacements de stock product_ids = [p["id"] for p in products] quants = self._execute( "stock.quant", "search_read", [ ["product_id", "in", product_ids], ["location_id.usage", "=", "internal"], ], fields=["product_id", "location_id", "quantity", "reserved_quantity"], limit=500, ) # Grouper les quants par produit quants_by_product: dict[int, list] = {} for q in quants: pid = q["product_id"][0] quants_by_product.setdefault(pid, []).append(q) for p in products: p["stock_details"] = quants_by_product.get(p["id"], []) self._cache.set(cache_key, products) logger.info(f"Stock récupéré pour '{product_name}' : {len(products)} produit(s)") return products except Exception as e: logger.error(f"Erreur récupération stock '{product_name}' : {e}") return [] # ── Documents client ─────────────────────────────────────────────────────── def get_stock_products( self, category_name: str = "", page_size: int = 500, ) -> list[dict]: """ Retourne l'etat de stock global, avec filtre optionnel par categorie. Les sous-categories Odoo sont incluses lorsque la categorie est trouvee. """ category_label = category_name.strip() cache_key = f"stock_products_{category_label.lower()}_{page_size}" cached = self._cache.get(cache_key) if cached is not None: return cached try: domain = [["active", "=", True]] if category_label: category_conditions = [ ["name", "ilike", term] for term in self._search_terms(category_label) ] categories = self._execute( "product.category", "search_read", self._or_leaves(category_conditions), fields=["id", "name", "complete_name"], limit=20, ) if not categories: logger.info(f"Aucune categorie trouvee pour '{category_label}'") self._cache.set(cache_key, []) return [] category_ids = [c["id"] for c in categories] domain.append(["categ_id", "child_of", category_ids]) fields = [ "id", "name", "default_code", "list_price", "categ_id", "qty_available", "virtual_available", ] products = [] offset = 0 while True: batch = self._execute( "product.product", "search_read", domain, fields=fields, limit=page_size, offset=offset, order="categ_id, name", ) if not batch: break products.extend(batch) if len(batch) < page_size: break offset += page_size self._cache.set(cache_key, products) if category_label: logger.info( f"Stock categorie '{category_label}' : {len(products)} produit(s)" ) else: logger.info(f"Stock general : {len(products)} produit(s)") return products except Exception as e: logger.error(f"Erreur recuperation stock global/categorie : {e}") return [] def get_client_sale_orders(self, client_id: int, months: int = 12) -> list[dict]: """Retourne les commandes de vente d'un client sur les N derniers mois.""" date_from = (datetime.now() - timedelta(days=months * 30)).strftime("%Y-%m-%d") try: results = self._execute( "sale.order", "search_read", [ ["partner_id", "=", client_id], ["date_order", ">=", date_from], ["state", "not in", ["cancel"]], ], fields=[ "name", "date_order", "state", "amount_total", "currency_id", "partner_id", "user_id", ], order="date_order desc", limit=100, ) logger.info(f"Commandes client {client_id} : {len(results)} résultat(s)") return results except Exception as e: logger.error(f"Erreur commandes client {client_id} : {e}") return [] def get_client_quotations(self, client_id: int, months: int = 12) -> list[dict]: """Retourne les devis d'un client sur les N derniers mois.""" date_from = (datetime.now() - timedelta(days=months * 30)).strftime("%Y-%m-%d") try: results = self._execute( "sale.order", "search_read", [ ["partner_id", "=", client_id], ["date_order", ">=", date_from], ["state", "in", ["draft", "sent"]], ], fields=[ "name", "date_order", "state", "amount_total", "currency_id", "partner_id", ], order="date_order desc", limit=50, ) logger.info(f"Devis client {client_id} : {len(results)} résultat(s)") return results except Exception as e: logger.error(f"Erreur devis client {client_id} : {e}") return [] def get_client_invoices(self, client_id: int, months: int = 12) -> list[dict]: """Retourne les factures d'un client sur les N derniers mois.""" date_from = (datetime.now() - timedelta(days=months * 30)).strftime("%Y-%m-%d") try: results = self._execute( "account.move", "search_read", [ ["partner_id", "=", client_id], ["move_type", "in", ["out_invoice", "out_refund"]], ["invoice_date", ">=", date_from], ["state", "not in", ["cancel"]], ], fields=[ "name", "invoice_date", "invoice_date_due", "state", "move_type", "amount_total", "amount_residual", "currency_id", "partner_id", "payment_state", ], order="invoice_date desc", limit=100, ) logger.info(f"Factures client {client_id} : {len(results)} résultat(s)") return results except Exception as e: logger.error(f"Erreur factures client {client_id} : {e}") return [] def get_full_client_documents( self, client_id: int, months: int = 12 ) -> dict[str, list]: """ Récupère tous les documents d'un client (devis, commandes, factures) sur les N derniers mois. """ return { "quotations": self.get_client_quotations(client_id, months), "orders": self.get_client_sale_orders(client_id, months), "invoices": self.get_client_invoices(client_id, months), } def get_supplier_purchase_orders( self, supplier_id: Optional[int] = None, months: int = 12 ) -> list[dict]: """Retourne les commandes d'achat d'un fournisseur.""" date_from = (datetime.now() - timedelta(days=months * 30)).strftime("%Y-%m-%d") try: domain = [ ["date_order", ">=", date_from], ["state", "not in", ["cancel"]], ] if supplier_id: domain.insert(0, ["partner_id", "child_of", supplier_id]) return self._execute( "purchase.order", "search_read", domain, fields=[ "name", "date_order", "state", "amount_total", "currency_id", "partner_id", "user_id", ], order="date_order desc", limit=500, ) except Exception as e: logger.error(f"Erreur commandes fournisseur {supplier_id} : {e}") return [] def get_supplier_bills( self, supplier_id: Optional[int] = None, months: int = 12 ) -> list[dict]: """Retourne les factures et avoirs d'un fournisseur.""" date_from = (datetime.now() - timedelta(days=months * 30)).strftime("%Y-%m-%d") try: domain = [ ["move_type", "in", ["in_invoice", "in_refund"]], ["invoice_date", ">=", date_from], ["state", "not in", ["cancel"]], ] if supplier_id: domain.insert(0, ["commercial_partner_id", "=", supplier_id]) return self._execute( "account.move", "search_read", domain, fields=[ "name", "invoice_date", "invoice_date_due", "state", "move_type", "amount_total", "amount_residual", "currency_id", "partner_id", "payment_state", ], order="invoice_date desc", limit=500, ) except Exception as e: logger.error(f"Erreur factures fournisseur {supplier_id} : {e}") return [] def get_full_supplier_documents( self, supplier_id: Optional[int] = None, months: int = 12 ) -> dict[str, list]: """Recupere les commandes d'achat et factures d'un fournisseur.""" return { "quotations": [], "orders": self.get_supplier_purchase_orders(supplier_id, months), "invoices": self.get_supplier_bills(supplier_id, months), } # ── Utilitaires ──────────────────────────────────────────────────────────── def get_sale_order_states_fr(self) -> dict[str, str]: """Traduit les statuts de commande Odoo en français.""" return { "draft": "Devis", "sent": "Devis envoyé", "sale": "Commande confirmée", "done": "Verrouillée", "cancel": "Annulée", } def get_invoice_states_fr(self) -> dict[str, str]: """Traduit les statuts de facture Odoo en français.""" return { "draft": "Brouillon", "posted": "Publiée", "cancel": "Annulée", } def get_payment_states_fr(self) -> dict[str, str]: """Traduit les statuts de paiement en français.""" return { "not_paid": "Non payée", "in_payment": "En cours de paiement", "paid": "Payée", "partial": "Partiellement payée", "reversed": "Remboursée", }