522 lines
19 KiB
Python
522 lines
19 KiB
Python
# 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 regler_facture(
|
|
self,
|
|
numero_facture: str,
|
|
montant: float,
|
|
mode_reglement: int = 2,
|
|
date_reglement: str = None,
|
|
reference: str = "",
|
|
libelle: str = "",
|
|
code_journal: str = "BEU",
|
|
) -> dict:
|
|
"""Règle une facture"""
|
|
payload = {
|
|
"montant": montant,
|
|
"mode_reglement": mode_reglement,
|
|
"reference": reference,
|
|
"libelle": libelle,
|
|
"code_journal": code_journal,
|
|
}
|
|
if date_reglement:
|
|
payload["date_reglement"] = date_reglement
|
|
|
|
return self._post(f"/sage/factures/{numero_facture}/regler", payload).get(
|
|
"data", {}
|
|
)
|
|
|
|
def regler_factures_client(
|
|
self,
|
|
client_code: str,
|
|
montant_total: float,
|
|
mode_reglement: int = 2,
|
|
date_reglement: str = None,
|
|
reference: str = "",
|
|
libelle: str = "",
|
|
code_journal: str = "BEU",
|
|
numeros_factures: list = None,
|
|
) -> dict:
|
|
"""Règle plusieurs factures d'un client"""
|
|
payload = {
|
|
"client_code": client_code,
|
|
"montant_total": montant_total,
|
|
"mode_reglement": mode_reglement,
|
|
"reference": reference,
|
|
"libelle": libelle,
|
|
"code_journal": code_journal,
|
|
}
|
|
if date_reglement:
|
|
payload["date_reglement"] = date_reglement
|
|
if numeros_factures:
|
|
payload["numeros_factures"] = numeros_factures
|
|
|
|
return self._post("/sage/reglements/multiple", payload).get("data", {})
|
|
|
|
def get_reglements_facture(self, numero_facture: str) -> dict:
|
|
"""Récupère les règlements d'une facture"""
|
|
return self._get(f"/sage/factures/{numero_facture}/reglements").get("data", {})
|
|
|
|
def get_reglements_client(
|
|
self,
|
|
client_code: str,
|
|
date_debut: str = None,
|
|
date_fin: str = None,
|
|
inclure_soldees: bool = True,
|
|
) -> dict:
|
|
"""Récupère les règlements d'un client"""
|
|
params = {"inclure_soldees": inclure_soldees}
|
|
if date_debut:
|
|
params["date_debut"] = date_debut
|
|
if date_fin:
|
|
params["date_fin"] = date_fin
|
|
|
|
return self._get(f"/sage/clients/{client_code}/reglements", params=params).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()
|