# sage_client.py import requests from typing import Dict, List, Optional from config.config import settings import logging logger = logging.getLogger(__name__) class SageGatewayClient: def __init__( self, gateway_url: Optional[str] = None, gateway_token: Optional[str] = None, gateway_id: Optional[str] = None, ): self.url = (gateway_url or settings.sage_gateway_url).rstrip("/") self.token = gateway_token or settings.sage_gateway_token self.gateway_id = gateway_id self.headers = { "X-Sage-Token": self.token, "Content-Type": "application/json", } self.timeout = 30 @classmethod def from_context( cls, url: str, token: str, gateway_id: Optional[str] = None ) -> "SageGatewayClient": return cls(gateway_url=url, gateway_token=token, gateway_id=gateway_id) def _post(self, endpoint: str, data: dict = None, retries: int = 3) -> dict: 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: 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) def lister_clients(self, filtre: str = "") -> List[Dict]: return self._post("/sage/clients/list", {"filtre": filtre}).get("data", []) def lire_client(self, code: str) -> Optional[Dict]: return self._post("/sage/clients/get", {"code": code}).get("data") def creer_client(self, client_data: Dict) -> Dict: return self._post("/sage/clients/create", client_data).get("data", {}) def modifier_client(self, code: str, client_data: Dict) -> Dict: return self._post( "/sage/clients/update", {"code": code, "client_data": client_data} ).get("data", {}) def lister_articles(self, filtre: str = "") -> List[Dict]: return self._post("/sage/articles/list", {"filtre": filtre}).get("data", []) def lire_article(self, ref: str) -> Optional[Dict]: return self._post("/sage/articles/get", {"code": ref}).get("data") def creer_article(self, article_data: Dict) -> Dict: return self._post("/sage/articles/create", article_data).get("data", {}) def modifier_article(self, reference: str, article_data: Dict) -> Dict: return self._post( "/sage/articles/update", {"reference": reference, "article_data": article_data}, ).get("data", {}) def creer_devis(self, devis_data: Dict) -> Dict: return self._post("/sage/devis/create", devis_data).get("data", {}) def lire_devis(self, numero: str) -> Optional[Dict]: 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]: 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 modifier_devis(self, numero: str, devis_data: Dict) -> Dict: return self._post( "/sage/devis/update", {"numero": numero, "devis_data": devis_data} ).get("data", {}) def lister_commandes( self, limit: int = 100, statut: Optional[int] = None ) -> List[Dict]: payload = {"limit": limit} if statut is not None: payload["statut"] = statut return self._post("/sage/commandes/list", payload).get("data", []) def creer_commande(self, commande_data: Dict) -> Dict: return self._post("/sage/commandes/create", commande_data).get("data", {}) def modifier_commande(self, numero: str, commande_data: Dict) -> Dict: return self._post( "/sage/commandes/update", {"numero": numero, "commande_data": commande_data} ).get("data", {}) def lister_factures( self, limit: int = 100, statut: Optional[int] = None ) -> List[Dict]: payload = {"limit": limit} if statut is not None: payload["statut"] = statut return self._post("/sage/factures/list", payload).get("data", []) def creer_facture(self, facture_data: Dict) -> Dict: return self._post("/sage/factures/create", facture_data).get("data", {}) def modifier_facture(self, numero: str, facture_data: Dict) -> Dict: return self._post( "/sage/factures/update", {"numero": numero, "facture_data": facture_data} ).get("data", {}) def lister_livraisons( self, limit: int = 100, statut: Optional[int] = None ) -> List[Dict]: payload = {"limit": limit} if statut is not None: payload["statut"] = statut return self._post("/sage/livraisons/list", payload).get("data", []) def creer_livraison(self, livraison_data: Dict) -> Dict: return self._post("/sage/livraisons/create", livraison_data).get("data", {}) def modifier_livraison(self, numero: str, livraison_data: Dict) -> Dict: return self._post( "/sage/livraisons/update", {"numero": numero, "livraison_data": livraison_data}, ).get("data", {}) def lister_avoirs( self, limit: int = 100, statut: Optional[int] = None ) -> List[Dict]: payload = {"limit": limit} if statut is not None: payload["statut"] = statut return self._post("/sage/avoirs/list", payload).get("data", []) def creer_avoir(self, avoir_data: Dict) -> Dict: return self._post("/sage/avoirs/create", avoir_data).get("data", {}) def modifier_avoir(self, numero: str, avoir_data: Dict) -> Dict: return self._post( "/sage/avoirs/update", {"numero": numero, "avoir_data": avoir_data} ).get("data", {}) def lire_document(self, numero: str, type_doc: int) -> Optional[Dict]: return self._post( "/sage/documents/get", {"numero": numero, "type_doc": type_doc} ).get("data") def changer_statut_document( self, document_type_code: int, numero: str, nouveau_statut: int ) -> Dict: try: r = requests.post( f"{self.url}/sage/document/statut", params={ "numero": numero, "type_doc": document_type_code, "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 def transformer_document( self, numero_source: str, type_source: int, type_cible: int ) -> Dict: 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: 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) def mettre_a_jour_derniere_relance(self, doc_id: str, type_doc: int) -> bool: resp = self._post( "/sage/documents/derniere-relance", {"doc_id": doc_id, "type_doc": type_doc} ) return resp.get("success", False) def generer_pdf_document(self, doc_id: str, type_doc: int) -> bytes: try: logger.info(f"Demande génération PDF: doc_id={doc_id}, type={type_doc}") 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() if not response_data.get("success"): error_msg = response_data.get("error", "Erreur inconnue") raise RuntimeError(f"Gateway a retourné une erreur: {error_msg}") pdf_base64 = response_data.get("data", {}).get("pdf_base64", "") if not pdf_base64: raise ValueError( f"PDF vide retourné par la gateway pour {doc_id} (type {type_doc})" ) pdf_bytes = base64.b64decode(pdf_base64) logger.info(f"PDF décodé: {len(pdf_bytes)} octets") return pdf_bytes except requests.exceptions.Timeout: logger.error(f"Timeout génération PDF pour {doc_id}") raise RuntimeError( f"Timeout lors de la génération du PDF (>60s). " f"Le document {doc_id} est peut-être trop volumineux." ) except requests.exceptions.RequestException as e: logger.error(f"Erreur HTTP génération PDF: {e}") raise RuntimeError(f"Erreur de communication avec la gateway: {str(e)}") except Exception as e: logger.error(f"Erreur génération PDF: {e}", exc_info=True) raise def lister_prospects(self, filtre: str = "") -> List[Dict]: return self._post("/sage/prospects/list", {"filtre": filtre}).get("data", []) def lire_prospect(self, code: str) -> Optional[Dict]: return self._post("/sage/prospects/get", {"code": code}).get("data") def lister_fournisseurs(self, filtre: str = "") -> List[Dict]: return self._post("/sage/fournisseurs/list", {"filtre": filtre}).get("data", []) def lire_fournisseur(self, code: str) -> Optional[Dict]: return self._post("/sage/fournisseurs/get", {"code": code}).get("data") def creer_fournisseur(self, fournisseur_data: Dict) -> Dict: return self._post("/sage/fournisseurs/create", fournisseur_data).get("data", {}) def modifier_fournisseur(self, code: str, fournisseur_data: Dict) -> Dict: return self._post( "/sage/fournisseurs/update", {"code": code, "fournisseur_data": fournisseur_data}, ).get("data", {}) def lister_tiers( self, type_tiers: Optional[str] = None, filtre: str = "" ) -> List[Dict]: return self._post( "/sage/tiers/list", {"type_tiers": type_tiers, "filtre": filtre} ).get("data", []) def lire_tiers(self, code: str) -> Optional[Dict]: return self._post("/sage/tiers/get", {"code": code}).get("data") def lire_contact_client(self, code_client: str) -> Optional[Dict]: return self._post("/sage/contact/read", {"code": code_client}).get("data") def creer_contact(self, contact_data: Dict) -> Dict: return self._post("/sage/contacts/create", contact_data) def lister_contacts(self, numero: str) -> List[Dict]: return self._post("/sage/contacts/list", {"numero": numero}).get("data", []) def obtenir_contact(self, numero: str, contact_numero: int) -> Dict: result = self._post( "/sage/contacts/get", {"numero": numero, "contact_numero": contact_numero} ) return result.get("data") if result.get("success") else None def modifier_contact(self, numero: str, contact_numero: int, updates: Dict) -> Dict: return self._post( "/sage/contacts/update", {"numero": numero, "contact_numero": contact_numero, "updates": updates}, ) def supprimer_contact(self, numero: str, contact_numero: int) -> Dict: return self._post( "/sage/contacts/delete", {"numero": numero, "contact_numero": contact_numero}, ) def definir_contact_defaut(self, numero: str, contact_numero: int) -> Dict: return self._post( "/sage/contacts/set-default", {"numero": numero, "contact_numero": contact_numero}, ) def lister_familles(self, filtre: str = "") -> List[Dict]: return self._get("/sage/familles", params={"filtre": filtre}).get("data", []) def lire_famille(self, code: str) -> Optional[Dict]: try: response = self._get(f"/sage/familles/{code}") return response.get("data") except Exception as e: logger.error(f"Erreur lecture famille {code}: {e}") return None def creer_famille(self, famille_data: Dict) -> Dict: return self._post("/sage/familles/create", famille_data).get("data", {}) def get_stats_familles(self) -> Dict: return self._get("/sage/familles/stats").get("data", {}) def creer_entree_stock(self, entree_data: Dict) -> Dict: return self._post("/sage/stock/entree", entree_data).get("data", {}) def creer_sortie_stock(self, sortie_data: Dict) -> Dict: return self._post("/sage/stock/sortie", sortie_data).get("data", {}) def lire_mouvement_stock(self, numero: str) -> Optional[Dict]: try: response = self._get(f"/sage/stock/mouvement/{numero}") return response.get("data") except Exception as e: logger.error(f"Erreur lecture mouvement {numero}: {e}") return None def lire_remise_max_client(self, code_client: str) -> float: result = self._post("/sage/client/remise-max", {"code": code_client}) return result.get("data", {}).get("remise_max", 10.0) def lister_collaborateurs( self, filtre: Optional[str] = None, actifs_seulement: bool = True ) -> List[Dict]: """Liste tous les collaborateurs""" return self._post( "/sage/collaborateurs/list", { "filtre": filtre or "", "actifs_seulement": actifs_seulement, }, ).get("data", []) def lire_collaborateur(self, numero: int) -> Optional[Dict]: """Lit un collaborateur par numéro""" return self._post("/sage/collaborateurs/get", {"numero": numero}).get("data") def creer_collaborateur(self, data: Dict) -> Optional[Dict]: """Crée un nouveau collaborateur""" return self._post("/sage/collaborateurs/create", data).get("data") def modifier_collaborateur(self, numero: int, data: Dict) -> Optional[Dict]: """Modifie un collaborateur existant""" return self._post( "/sage/collaborateurs/update", {"numero": numero, **data} ).get("data") def lire_informations_societe(self) -> Optional[Dict]: """Lit les informations de la société depuis P_DOSSIER""" return self._get("/sage/societe/info").get("data") def valider_facture(self, numero_facture: str) -> dict: """Valide une facture""" return self._post(f"/sage/factures/{numero_facture}/valider", {}).get( "data", {} ) def devalider_facture(self, numero_facture: str) -> dict: """Dévalide une facture""" return self._post(f"/sage/factures/{numero_facture}/devalider", {}).get( "data", {} ) def get_statut_validation(self, numero_facture: str) -> dict: """Récupère le statut de validation""" return self._get(f"/sage/factures/{numero_facture}/statut-validation").get( "data", {} ) def refresh_cache(self) -> Dict: return self._post("/sage/cache/refresh") def get_cache_info(self) -> Dict: return self._get("/sage/cache/info").get("data", {}) def health(self) -> dict: try: r = requests.get(f"{self.url}/health", timeout=5) return r.json() except Exception: return {"status": "down"} sage_client = SageGatewayClient()