450 lines
17 KiB
Python
450 lines
17 KiB
Python
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()
|