import requests from typing import Dict, List, Optional from config import settings import logging logger = logging.getLogger(__name__) class SageGatewayClient: """ Client HTTP pour communiquer avec la gateway Sage Windows """ def __init__(self): self.url = settings.sage_gateway_url.rstrip("/") self.headers = { "X-Sage-Token": settings.sage_gateway_token, "Content-Type": "application/json", } self.timeout = 30 def _post(self, endpoint: str, data: dict = None, retries: int = 3) -> dict: """POST avec retry automatique""" import time for attempt in range(retries): try: r = requests.post( f"{self.url}{endpoint}", json=data or {}, headers=self.headers, timeout=self.timeout, ) r.raise_for_status() return r.json() except requests.exceptions.RequestException as e: if attempt == retries - 1: logger.error( f"❌ Échec après {retries} tentatives sur {endpoint}: {e}" ) raise time.sleep(2**attempt) def _get(self, endpoint: str, params: dict = None, retries: int = 3) -> dict: """GET avec retry automatique""" import time for attempt in range(retries): try: r = requests.get( f"{self.url}{endpoint}", params=params or {}, headers=self.headers, timeout=self.timeout, ) r.raise_for_status() return r.json() except requests.exceptions.RequestException as e: if attempt == retries - 1: logger.error( f"❌ Échec GET après {retries} tentatives sur {endpoint}: {e}" ) raise time.sleep(2**attempt) # ===================================================== # CLIENTS # ===================================================== def lister_clients(self, filtre: str = "") -> List[Dict]: """Liste tous les clients avec filtre optionnel""" return self._post("/sage/clients/list", {"filtre": filtre}).get("data", []) def lire_client(self, code: str) -> Optional[Dict]: """Lecture d'un client par code""" return self._post("/sage/clients/get", {"code": code}).get("data") # ===================================================== # ARTICLES # ===================================================== def lister_articles(self, filtre: str = "") -> List[Dict]: """Liste tous les articles avec filtre optionnel""" return self._post("/sage/articles/list", {"filtre": filtre}).get("data", []) def lire_article(self, ref: str) -> Optional[Dict]: """Lecture d'un article par référence""" return self._post("/sage/articles/get", {"code": ref}).get("data") # ===================================================== # DEVIS (US-A1) # ===================================================== def creer_devis(self, devis_data: Dict) -> Dict: """Création d'un devis""" return self._post("/sage/devis/create", devis_data).get("data", {}) def lire_devis(self, numero: str) -> Optional[Dict]: """Lecture d'un devis""" return self._post("/sage/devis/get", {"code": numero}).get("data") def lister_devis( self, limit: int = 100, statut: Optional[int] = None, inclure_lignes: bool = True, ) -> List[Dict]: """ ✅ Liste tous les devis avec filtres """ payload = {"limit": limit, "inclure_lignes": inclure_lignes} if statut is not None: payload["statut"] = statut return self._post("/sage/devis/list", payload).get("data", []) def changer_statut_devis(self, numero: str, nouveau_statut: int) -> Dict: """ ✅ CORRECTION: Utilise query params au lieu du body """ try: r = requests.post( f"{self.url}/sage/devis/statut", params={ "numero": numero, "nouveau_statut": nouveau_statut, }, headers=self.headers, timeout=self.timeout, ) r.raise_for_status() return r.json().get("data", {}) except requests.exceptions.RequestException as e: logger.error(f"❌ Erreur changement statut: {e}") raise # ===================================================== # DOCUMENTS GÉNÉRIQUES # ===================================================== def lire_document(self, numero: str, type_doc: int) -> Optional[Dict]: """Lecture d'un document générique""" return self._post( "/sage/documents/get", {"numero": numero, "type_doc": type_doc} ).get("data") def transformer_document( self, numero_source: str, type_source: int, type_cible: int ) -> Dict: """ ✅ CORRECTION: Utilise query params pour la transformation """ try: r = requests.post( f"{self.url}/sage/documents/transform", params={ "numero_source": numero_source, "type_source": type_source, "type_cible": type_cible, }, headers=self.headers, timeout=60, ) r.raise_for_status() return r.json().get("data", {}) except requests.exceptions.RequestException as e: logger.error(f"❌ Erreur transformation: {e}") raise def mettre_a_jour_champ_libre( self, doc_id: str, type_doc: int, nom_champ: str, valeur: str ) -> bool: """Mise à jour d'un champ libre""" resp = self._post( "/sage/documents/champ-libre", { "doc_id": doc_id, "type_doc": type_doc, "nom_champ": nom_champ, "valeur": valeur, }, ) return resp.get("success", False) # ===================================================== # COMMANDES (US-A2) # ===================================================== def lister_commandes( self, limit: int = 100, statut: Optional[int] = None ) -> List[Dict]: """ Utilise l'endpoint /sage/commandes/list qui filtre déjà sur type 10 """ payload = {"limit": limit} if statut is not None: payload["statut"] = statut return self._post("/sage/commandes/list", payload).get("data", []) def lister_factures( self, limit: int = 100, statut: Optional[int] = None ) -> List[Dict]: """ ✅ Liste toutes les factures Utilise l'endpoint /sage/factures/list qui filtre déjà sur type 60 """ payload = {"limit": limit} if statut is not None: payload["statut"] = statut return self._post("/sage/factures/list", payload).get("data", []) def mettre_a_jour_derniere_relance(self, doc_id: str, type_doc: int) -> bool: """Met à jour le champ 'Dernière relance' d'une facture""" resp = self._post( "/sage/documents/derniere-relance", {"doc_id": doc_id, "type_doc": type_doc} ) return resp.get("success", False) # ===================================================== # CONTACTS (US-A6) # ===================================================== def lire_contact_client(self, code_client: str) -> Optional[Dict]: """Lecture du contact principal d'un client""" return self._post("/sage/contact/read", {"code": code_client}).get("data") # ===================================================== # REMISES (US-A5) # ===================================================== def lire_remise_max_client(self, code_client: str) -> float: """Récupère la remise max autorisée pour un client""" result = self._post("/sage/client/remise-max", {"code": code_client}) return result.get("data", {}).get("remise_max", 10.0) # ===================================================== # GÉNÉRATION PDF (pour email_queue) # ===================================================== def generer_pdf_document(self, doc_id: str, type_doc: int) -> bytes: """Génère le PDF d'un document via la gateway Windows""" try: r = requests.post( f"{self.url}/sage/documents/generate-pdf", json={"doc_id": doc_id, "type_doc": type_doc}, headers=self.headers, timeout=60, ) r.raise_for_status() import base64 response_data = r.json() pdf_base64 = response_data.get("data", {}).get("pdf_base64", "") if not pdf_base64: raise ValueError("PDF vide retourné par la gateway") return base64.b64decode(pdf_base64) except Exception as e: logger.error(f"Erreur génération PDF: {e}") raise # ===================================================== # PROSPECTS # ===================================================== def lister_prospects(self, filtre: str = "") -> List[Dict]: """Liste tous les prospects avec filtre optionnel""" return self._post("/sage/prospects/list", {"filtre": filtre}).get("data", []) def lire_prospect(self, code: str) -> Optional[Dict]: """Lecture d'un prospect par code""" return self._post("/sage/prospects/get", {"code": code}).get("data") # ===================================================== # FOURNISSEURS # ===================================================== def lister_fournisseurs(self, filtre: str = "") -> List[Dict]: """Liste tous les fournisseurs avec filtre optionnel""" return self._post("/sage/fournisseurs/list", {"filtre": filtre}).get("data", []) def lire_fournisseur(self, code: str) -> Optional[Dict]: """Lecture d'un fournisseur par code""" return self._post("/sage/fournisseurs/get", {"code": code}).get("data") def creer_fournisseur(self, fournisseur_data: Dict) -> Dict: """ Envoie la requête de création de fournisseur à la gateway Windows. Args: fournisseur_data: Dict contenant intitule, compte_collectif, etc. Returns: Fournisseur créé avec son numéro définitif """ return self._post("/sage/fournisseurs/create", fournisseur_data).get("data", {}) def modifier_fournisseur(self, code: str, fournisseur_data: Dict) -> Dict: """ ✏️ Modification d'un fournisseur existant Args: code: Code du fournisseur à modifier fournisseur_data: Dictionnaire contenant les champs à modifier (seuls les champs présents seront mis à jour) Returns: Fournisseur modifié """ return self._post("/sage/fournisseurs/update", { "code": code, "fournisseur_data": fournisseur_data }).get("data", {}) # ===================================================== # AVOIRS # ===================================================== def lister_avoirs( self, limit: int = 100, statut: Optional[int] = None ) -> List[Dict]: """Liste tous les avoirs""" payload = {"limit": limit} if statut is not None: payload["statut"] = statut return self._post("/sage/avoirs/list", payload).get("data", []) def lire_avoir(self, numero: str) -> Optional[Dict]: """Lecture d'un avoir avec ses lignes""" return self._post("/sage/avoirs/get", {"code": numero}).get("data") # ===================================================== # LIVRAISONS # ===================================================== def lister_livraisons( self, limit: int = 100, statut: Optional[int] = None ) -> List[Dict]: """Liste tous les bons de livraison""" payload = {"limit": limit} if statut is not None: payload["statut"] = statut return self._post("/sage/livraisons/list", payload).get("data", []) def lire_livraison(self, numero: str) -> Optional[Dict]: """Lecture d'une livraison avec ses lignes""" return self._post("/sage/livraisons/get", {"code": numero}).get("data") # ===================================================== # CACHE (ADMIN) # ===================================================== def refresh_cache(self) -> Dict: """Force le rafraîchissement du cache Windows""" return self._post("/sage/cache/refresh") def get_cache_info(self) -> Dict: """Récupère les infos du cache Windows""" return self._get("/sage/cache/info").get("data", {}) # ===================================================== # HEALTH # ===================================================== def health(self) -> dict: """Health check de la gateway Windows""" try: r = requests.get(f"{self.url}/health", timeout=5) return r.json() except: return {"status": "down"} def creer_client(self, client_data: Dict) -> Dict: """ Envoie la requête de création de client à la gateway Windows. :param client_data: Dict contenant intitule, compte_collectif, etc. """ # On appelle la route définie dans main.py return self._post("/sage/clients/create", client_data).get("data", {}) def modifier_client(self, code: str, client_data: Dict) -> Dict: """ ✏️ Modification d'un client existant Args: code: Code du client à modifier client_data: Dictionnaire contenant les champs à modifier (seuls les champs présents seront mis à jour) Returns: Client modifié """ return self._post("/sage/clients/update", { "code": code, "client_data": client_data }).get("data", {}) def modifier_devis(self, numero: str, devis_data: Dict) -> Dict: """ ✏️ Modification d'un devis existant Args: numero: Numéro du devis à modifier devis_data: Dictionnaire contenant les champs à modifier: - date_devis (str, optional): Nouvelle date - lignes (List, optional): Nouvelles lignes - statut (int, optional): Nouveau statut Returns: Devis modifié avec totaux recalculés """ return self._post("/sage/devis/update", { "numero": numero, "devis_data": devis_data }).get("data", {}) def creer_commande(self, commande_data: Dict) -> Dict: """ ➕ Création d'une nouvelle commande (Bon de commande) Args: commande_data: Dictionnaire contenant: - client_id (str): Code du client - date_commande (str, optional): Date au format ISO - reference (str, optional): Référence externe - lignes (List[Dict]): Liste des lignes avec: - article_code (str) - quantite (float) - prix_unitaire_ht (float, optional) - remise_pourcentage (float, optional) Returns: Commande créée avec son numéro et ses totaux """ return self._post("/sage/commandes/create", commande_data).get("data", {}) def modifier_commande(self, numero: str, commande_data: Dict) -> Dict: """ ✏️ Modification d'une commande existante Args: numero: Numéro de la commande à modifier commande_data: Dictionnaire contenant les champs à modifier: - date_commande (str, optional): Nouvelle date - lignes (List, optional): Nouvelles lignes - statut (int, optional): Nouveau statut - reference (str, optional): Nouvelle référence Returns: Commande modifiée avec totaux recalculés """ return self._post("/sage/commandes/update", { "numero": numero, "commande_data": commande_data }).get("data", {}) # Instance globale sage_client = SageGatewayClient()