feat: Add BaremeRemiseResponse model, expand SageGatewayClient with methods for document listing, status updates, discount retrieval, and PDF generation, and ignore .db files.

This commit is contained in:
Fanilo-Nantenaina 2025-11-27 13:08:44 +03:00
parent 636e2a96a7
commit df6e09af07
3 changed files with 95 additions and 18 deletions

2
.gitignore vendored
View file

@ -36,3 +36,5 @@ htmlcov/
*~ *~
.build/ .build/
dist/ dist/
*.db

8
api.py
View file

@ -131,6 +131,14 @@ class RelanceDevisRequest(BaseModel):
message_personnalise: Optional[str] = None message_personnalise: Optional[str] = None
class BaremeRemiseResponse(BaseModel):
client_id: str
remise_max_autorisee: float
remise_demandee: float
autorisee: bool
message: str
# ===================================================== # =====================================================
# SERVICES EXTERNES (Universign) # SERVICES EXTERNES (Universign)
# ===================================================== # =====================================================

View file

@ -9,6 +9,7 @@ logger = logging.getLogger(__name__)
class SageGatewayClient: class SageGatewayClient:
""" """
Client HTTP pour communiquer avec la gateway Sage Windows Client HTTP pour communiquer avec la gateway Sage Windows
VERSION COMPLÈTE avec toutes les routes nécessaires
""" """
def __init__(self): def __init__(self):
@ -63,46 +64,59 @@ class SageGatewayClient:
raise raise
time.sleep(2**attempt) time.sleep(2**attempt)
# === CLIENTS === # =====================================================
# CLIENTS
# =====================================================
def lister_clients(self, filtre: str = "") -> List[Dict]: 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", []) return self._post("/sage/clients/list", {"filtre": filtre}).get("data", [])
def lire_client(self, code: str) -> Optional[Dict]: def lire_client(self, code: str) -> Optional[Dict]:
"""Lecture d'un client par code"""
return self._post("/sage/clients/get", {"code": code}).get("data") return self._post("/sage/clients/get", {"code": code}).get("data")
# === ARTICLES === # =====================================================
# ARTICLES
# =====================================================
def lister_articles(self, filtre: str = "") -> List[Dict]: 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", []) return self._post("/sage/articles/list", {"filtre": filtre}).get("data", [])
def lire_article(self, ref: str) -> Optional[Dict]: 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") return self._post("/sage/articles/get", {"code": ref}).get("data")
# === DEVIS === # =====================================================
# DEVIS (US-A1)
# =====================================================
def creer_devis(self, devis_data: Dict) -> Dict: def creer_devis(self, devis_data: Dict) -> Dict:
"""Création d'un devis"""
return self._post("/sage/devis/create", devis_data).get("data", {}) return self._post("/sage/devis/create", devis_data).get("data", {})
def lire_devis(self, numero: str) -> Optional[Dict]: def lire_devis(self, numero: str) -> Optional[Dict]:
"""Lecture d'un devis"""
return self._post("/sage/devis/get", {"code": numero}).get("data") return self._post("/sage/devis/get", {"code": numero}).get("data")
# 🆕 US-A1: Lister devis
def lister_devis( def lister_devis(
self, limit: int = 100, statut: Optional[int] = None self, limit: int = 100, statut: Optional[int] = None
) -> List[Dict]: ) -> List[Dict]:
"""Liste tous les devis avec filtres""" """✅ NOUVEAU: Liste tous les devis avec filtres"""
payload = {"limit": limit} payload = {"limit": limit}
if statut is not None: if statut is not None:
payload["statut"] = statut payload["statut"] = statut
return self._post("/sage/devis/list", payload).get("data", []) return self._post("/sage/devis/list", payload).get("data", [])
# 🆕 US-A1: Changer statut devis
def changer_statut_devis(self, numero: str, nouveau_statut: int) -> Dict: def changer_statut_devis(self, numero: str, nouveau_statut: int) -> Dict:
"""Change le statut d'un devis""" """✅ NOUVEAU: Change le statut d'un devis"""
return self._post( return self._post(
"/sage/devis/statut", {"numero": numero, "nouveau_statut": nouveau_statut} "/sage/devis/statut", {"numero": numero, "nouveau_statut": nouveau_statut}
).get("data", {}) ).get("data", {})
# === DOCUMENTS === # =====================================================
# DOCUMENTS GÉNÉRIQUES
# =====================================================
def lire_document(self, numero: str, type_doc: int) -> Optional[Dict]: def lire_document(self, numero: str, type_doc: int) -> Optional[Dict]:
"""Lecture d'un document générique"""
return self._post( return self._post(
"/sage/documents/get", {"numero": numero, "type_doc": type_doc} "/sage/documents/get", {"numero": numero, "type_doc": type_doc}
).get("data") ).get("data")
@ -110,6 +124,7 @@ class SageGatewayClient:
def transformer_document( def transformer_document(
self, numero_source: str, type_source: int, type_cible: int self, numero_source: str, type_source: int, type_cible: int
) -> Dict: ) -> Dict:
"""Transformation de document (devis → commande → facture)"""
return self._post( return self._post(
"/sage/documents/transform", "/sage/documents/transform",
{ {
@ -122,6 +137,7 @@ class SageGatewayClient:
def mettre_a_jour_champ_libre( def mettre_a_jour_champ_libre(
self, doc_id: str, type_doc: int, nom_champ: str, valeur: str self, doc_id: str, type_doc: int, nom_champ: str, valeur: str
) -> bool: ) -> bool:
"""Mise à jour d'un champ libre"""
resp = self._post( resp = self._post(
"/sage/documents/champ-libre", "/sage/documents/champ-libre",
{ {
@ -133,46 +149,97 @@ class SageGatewayClient:
) )
return resp.get("success", False) return resp.get("success", False)
# 🆕 US-A2: Lister commandes # =====================================================
# COMMANDES (US-A2)
# =====================================================
def lister_commandes( def lister_commandes(
self, limit: int = 100, statut: Optional[int] = None self, limit: int = 100, statut: Optional[int] = None
) -> List[Dict]: ) -> List[Dict]:
"""Liste toutes les commandes""" """✅ NOUVEAU: Liste toutes les commandes"""
payload = {"limit": limit} payload = {"limit": limit}
if statut is not None: if statut is not None:
payload["statut"] = statut payload["statut"] = statut
return self._post("/sage/commandes/list", payload).get("data", []) return self._post("/sage/commandes/list", payload).get("data", [])
# 🆕 US-A7: Lister factures # =====================================================
# FACTURES (US-A7)
# =====================================================
def lister_factures( def lister_factures(
self, limit: int = 100, statut: Optional[int] = None self, limit: int = 100, statut: Optional[int] = None
) -> List[Dict]: ) -> List[Dict]:
"""Liste toutes les factures""" """✅ NOUVEAU: Liste toutes les factures"""
payload = {"limit": limit} payload = {"limit": limit}
if statut is not None: if statut is not None:
payload["statut"] = statut payload["statut"] = statut
return self._post("/sage/factures/list", payload).get("data", []) return self._post("/sage/factures/list", payload).get("data", [])
# === CONTACTS === def mettre_a_jour_derniere_relance(self, doc_id: str, type_doc: int) -> bool:
"""✅ NOUVEAU: 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]: 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") return self._post("/sage/contact/read", {"code": code_client}).get("data")
# 🆕 US-A5: Lire remise max client # =====================================================
# REMISES (US-A5)
# =====================================================
def lire_remise_max_client(self, code_client: str) -> float: def lire_remise_max_client(self, code_client: str) -> float:
"""Récupère la remise max autorisée pour un client""" """✅ NOUVEAU: Récupère la remise max autorisée pour un client"""
result = self._post("/sage/client/remise-max", {"code": code_client}) result = self._post("/sage/client/remise-max", {"code": code_client})
return result.get("data", {}).get("remise_max", 10.0) return result.get("data", {}).get("remise_max", 10.0)
# === CACHE === # =====================================================
# GÉNÉRATION PDF (pour email_queue)
# =====================================================
def generer_pdf_document(self, doc_id: str, type_doc: int) -> bytes:
"""✅ NOUVEAU: 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, # Timeout plus long pour génération PDF
)
r.raise_for_status()
# Le PDF est retourné en base64 dans la réponse JSON
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
# =====================================================
# CACHE (ADMIN)
# =====================================================
def refresh_cache(self) -> Dict: def refresh_cache(self) -> Dict:
"""Force le rafraîchissement du cache Windows"""
return self._post("/sage/cache/refresh") return self._post("/sage/cache/refresh")
def get_cache_info(self) -> Dict: def get_cache_info(self) -> Dict:
"""Récupère les infos du cache""" """Récupère les infos du cache Windows"""
return self._get("/sage/cache/info").get("data", {}) return self._get("/sage/cache/info").get("data", {})
# === HEALTH === # =====================================================
# HEALTH
# =====================================================
def health(self) -> dict: def health(self) -> dict:
"""Health check de la gateway Windows"""
try: try:
r = requests.get(f"{self.url}/health", timeout=5) r = requests.get(f"{self.url}/health", timeout=5)
return r.json() return r.json()