fanilo/Sage100-ws
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: add PDF generation endpoint and optimize list endpoints

Browse source
This commit is contained in:
Fanilo-Nantenaina 2025-12-08 17:39:51 +03:00
parent ec5a0f0089
commit 51f49298c2
1 changed files with 142 additions and 380 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

484
main.py
View file

@ -185,6 +185,11 @@ class FactureUpdateGatewayRequest(BaseModel):
numero: str
facture_data: Dict
class PDFGenerationRequest(BaseModel):
"""Modèle pour génération PDF"""
doc_id: str = Field(..., description="Numéro du document")
type_doc: int = Field(..., ge=0, le=60, description="Type de document Sage")
# =====================================================
# SÉCURITÉ
# =====================================================
@ -378,8 +383,14 @@ def creer_devis(req: DevisRequest):
@app.post("/sage/devis/get", dependencies=[Depends(verify_token)])
def lire_devis(req: CodeRequest):
"""Lecture d'un devis"""
"""
📄 Lecture d'un devis AVEC ses lignes (lecture Sage directe)
⚠️ Plus lent que /list car charge les lignes depuis Sage
💡 Utiliser /list pour afficher une table rapide
"""
try:
# ✅ Lecture complète depuis Sage (avec lignes)
devis = sage.lire_devis(req.code)
if not devis:
raise HTTPException(404, f"Devis {req.code} non trouvé")
@ -393,195 +404,35 @@ def lire_devis(req: CodeRequest):
@app.post("/sage/devis/list", dependencies=[Depends(verify_token)])
def devis_list(
limit: int = 100, statut: Optional[int] = None, inclure_lignes: bool = Query(True)
limit: int = Query(100, description="Nombre max de devis"),
statut: Optional[int] = Query(None, description="Filtrer par statut"),
filtre: str = Query("", description="Filtre texte (numero, client)")
):
"""
📋 Liste tous les devis avec filtres optionnels
📋 Liste rapide des devis depuis le CACHE (sans lignes)
Args:
limit: Nombre max de devis à retourner
statut: Filtre par statut (optionnel)
inclure_lignes: Si True, charge les lignes de chaque devis (par défaut: True)
✅ AMÉLIORATION: Charge maintenant les lignes de chaque devis
⚡ ULTRA-RAPIDE: Utilise le cache mémoire au lieu de scanner Sage
💡 Pour les détails avec lignes, utiliser GET /sage/devis/get
"""
try:
if not sage or not sage.cial:
raise HTTPException(503, "Service Sage indisponible")
# ✅ Récupération depuis le cache (instantané)
devis_list = sage.lister_tous_devis_cache(filtre)
with sage._com_context(), sage._lock_com:
factory = sage.cial.FactoryDocumentVente
devis_list = []
index = 1
max_iterations = limit * 3
erreurs_consecutives = 0
max_erreurs = 50
# Filtrer par statut si demandé
if statut is not None:
devis_list = [d for d in devis_list if d.get("statut") == statut]
logger.info(
f"🔍 Recherche devis (limit={limit}, statut={statut}, inclure_lignes={inclure_lignes})"
)
# Limiter le nombre de résultats
devis_list = devis_list[:limit]
while (
len(devis_list) < limit
and index < max_iterations
and erreurs_consecutives < max_erreurs
):
try:
persist = factory.List(index)
if persist is None:
logger.debug(f"Fin de liste à l'index {index}")
break
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
doc.Read()
# Filtrer uniquement devis (type 0)
doc_type = getattr(doc, "DO_Type", -1)
if doc_type != 0:
index += 1
continue
doc_statut = getattr(doc, "DO_Statut", 0)
# Filtre statut
if statut is not None and doc_statut != statut:
index += 1
continue
# ✅ Charger client via .Client
client_code = ""
client_intitule = ""
try:
client_obj = getattr(doc, "Client", None)
if client_obj:
client_obj.Read()
client_code = getattr(client_obj, "CT_Num", "").strip()
client_intitule = getattr(
client_obj, "CT_Intitule", ""
).strip()
logger.debug(
f"✅ Client: {client_code} - {client_intitule}"
)
except Exception as e:
logger.debug(f"Erreur chargement client: {e}")
# Fallback sur cache si code disponible
if client_code:
client_cache = sage.lire_client(client_code)
if client_cache:
client_intitule = client_cache.get("intitule", "")
# ✅✅ NOUVEAU: Charger les lignes si demandé
lignes = []
if inclure_lignes:
try:
factory_lignes = getattr(doc, "FactoryDocumentLigne", None)
if not factory_lignes:
factory_lignes = getattr(
doc, "FactoryDocumentVenteLigne", None
)
if factory_lignes:
ligne_index = 1
while ligne_index <= 100: # Max 100 lignes par devis
try:
ligne_persist = factory_lignes.List(ligne_index)
if ligne_persist is None:
break
ligne = win32com.client.CastTo(
ligne_persist, "IBODocumentLigne3"
)
ligne.Read()
# Charger référence article
article_ref = ""
try:
article_ref = getattr(
ligne, "AR_Ref", ""
).strip()
if not article_ref:
article_obj = getattr(
ligne, "Article", None
)
if article_obj:
article_obj.Read()
article_ref = getattr(
article_obj, "AR_Ref", ""
).strip()
except:
pass
lignes.append(
{
"article": article_ref,
"designation": getattr(
ligne, "DL_Design", ""
),
"quantite": float(
getattr(ligne, "DL_Qte", 0.0)
),
"prix_unitaire": float(
getattr(
ligne, "DL_PrixUnitaire", 0.0
)
),
"montant_ht": float(
getattr(ligne, "DL_MontantHT", 0.0)
),
}
)
ligne_index += 1
except Exception as e:
logger.debug(f"Erreur ligne {ligne_index}: {e}")
break
except Exception as e:
logger.debug(f"Erreur chargement lignes: {e}")
devis_list.append(
{
"numero": getattr(doc, "DO_Piece", ""),
"date": str(getattr(doc, "DO_Date", "")),
"client_code": client_code,
"client_intitule": client_intitule,
"total_ht": float(getattr(doc, "DO_TotalHT", 0.0)),
"total_ttc": float(getattr(doc, "DO_TotalTTC", 0.0)),
"statut": doc_statut,
"lignes": lignes, # ✅ Lignes incluses
}
)
erreurs_consecutives = 0
index += 1
except Exception as e:
erreurs_consecutives += 1
logger.debug(f"⚠️ Erreur index {index}: {e}")
index += 1
if erreurs_consecutives >= max_erreurs:
logger.warning(
f"⚠️ Arrêt après {max_erreurs} erreurs consécutives"
)
break
nb_avec_client = sum(1 for d in devis_list if d["client_intitule"])
nb_avec_lignes = sum(1 for d in devis_list if d.get("lignes"))
logger.info(
f"✅ {len(devis_list)} devis retournés "
f"({nb_avec_client} avec client, {nb_avec_lignes} avec lignes)"
)
logger.info(f"✅ {len(devis_list)} devis retournés depuis le cache")
return {"success": True, "data": devis_list}
except HTTPException:
raise
except Exception as e:
logger.error(f"❌ Erreur liste devis: {e}", exc_info=True)
raise HTTPException(500, str(e))
@app.post("/sage/devis/statut", dependencies=[Depends(verify_token)])
def changer_statut_devis_endpoint(numero: str, nouveau_statut: int):
"""Change le statut d'un devis"""
@ -750,210 +601,58 @@ def contact_read(req: CodeRequest):
@app.post("/sage/commandes/list", dependencies=[Depends(verify_token)])
def commandes_list(limit: int = 100, statut: Optional[int] = None):
"""
📋 Liste toutes les commandes
✅ AJOUT: Retourne maintenant DO_Ref pour tracer les transformations
"""
try:
if not sage or not sage.cial:
raise HTTPException(503, "Service Sage indisponible")
with sage._com_context(), sage._lock_com:
factory = sage.cial.FactoryDocumentVente
commandes = []
index = 1
max_iterations = limit * 10
erreurs_consecutives = 0
max_erreurs = 100
logger.info(f"🔍 Recherche commandes (limit={limit}, statut={statut})")
while (
len(commandes) < limit
and index < max_iterations
and erreurs_consecutives < max_erreurs
def commandes_list(
limit: int = Query(100, description="Nombre max de commandes"),
statut: Optional[int] = Query(None, description="Filtrer par statut"),
filtre: str = Query("", description="Filtre texte")
):
"""
📋 Liste rapide des commandes depuis le CACHE (sans lignes)
⚡ ULTRA-RAPIDE: Utilise le cache mémoire
"""
try:
persist = factory.List(index)
if persist is None:
break
commandes = sage.lister_toutes_commandes_cache(filtre)
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
doc.Read()
if statut is not None:
commandes = [c for c in commandes if c.get("statut") == statut]
doc_type = getattr(doc, "DO_Type", -1)
commandes = commandes[:limit]
# Filtrer sur type 10 (BON_COMMANDE)
if doc_type != settings.SAGE_TYPE_BON_COMMANDE:
index += 1
continue
doc_statut = getattr(doc, "DO_Statut", 0)
# Filtre statut
if statut is not None and doc_statut != statut:
index += 1
continue
# Charger client
client_code = ""
client_intitule = ""
try:
client_obj = getattr(doc, "Client", None)
if client_obj:
client_obj.Read()
client_code = getattr(client_obj, "CT_Num", "").strip()
client_intitule = getattr(
client_obj, "CT_Intitule", ""
).strip()
except Exception as e:
logger.debug(f"Erreur chargement client: {e}")
commande = {
"numero": getattr(doc, "DO_Piece", ""),
"reference": getattr(doc, "DO_Ref", ""), # ✅ AJOUT
"date": str(getattr(doc, "DO_Date", "")),
"client_code": client_code,
"client_intitule": client_intitule,
"total_ht": float(getattr(doc, "DO_TotalHT", 0.0)),
"total_ttc": float(getattr(doc, "DO_TotalTTC", 0.0)),
"statut": doc_statut,
}
commandes.append(commande)
erreurs_consecutives = 0
index += 1
except Exception as e:
erreurs_consecutives += 1
index += 1
if erreurs_consecutives >= max_erreurs:
break
logger.info(f"✅ {len(commandes)} commandes retournées")
logger.info(f"✅ {len(commandes)} commandes retournées depuis le cache")
return {"success": True, "data": commandes}
except HTTPException:
raise
except Exception as e:
logger.error(f"❌ Erreur liste commandes: {e}", exc_info=True)
raise HTTPException(500, str(e))
@app.post("/sage/factures/list", dependencies=[Depends(verify_token)])
def factures_list(limit: int = 100, statut: Optional[int] = None):
def factures_list(
limit: int = Query(100, description="Nombre max de factures"),
statut: Optional[int] = Query(None, description="Filtrer par statut"),
filtre: str = Query("", description="Filtre texte")
):
"""
📋 Liste toutes les factures avec leurs lignes
✅ AJOUT: Retourne maintenant DO_Ref pour tracer les transformations
📋 Liste rapide des factures depuis le CACHE (sans lignes)
⚡ ULTRA-RAPIDE: Utilise le cache mémoire
💡 Pour les détails avec lignes, utiliser /sage/documents/get
"""
try:
with sage._com_context(), sage._lock_com:
factory = sage.cial.FactoryDocumentVente
factures = []
index = 1
max_iterations = limit * 3
factures = sage.lister_toutes_factures_cache(filtre)
while len(factures) < limit and index < max_iterations:
try:
persist = factory.List(index)
if persist is None:
break
if statut is not None:
factures = [f for f in factures if f.get("statut") == statut]
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
doc.Read()
factures = factures[:limit]
# Filtrer factures (type 60)
if getattr(doc, "DO_Type", -1) != settings.SAGE_TYPE_FACTURE:
index += 1
continue
logger.info(f"✅ {len(factures)} factures retournées depuis le cache")
doc_statut = getattr(doc, "DO_Statut", 0)
if statut is None or doc_statut == statut:
# Charger client
client_code = ""
client_intitule = ""
try:
client_obj = getattr(doc, "Client", None)
if client_obj:
client_obj.Read()
client_code = getattr(client_obj, "CT_Num", "").strip()
client_intitule = getattr(client_obj, "CT_Intitule", "").strip()
except:
pass
# Charger les lignes
lignes = []
try:
factory_lignes = getattr(doc, "FactoryDocumentLigne", None)
if not factory_lignes:
factory_lignes = getattr(doc, "FactoryDocumentVenteLigne", None)
if factory_lignes:
ligne_index = 1
while ligne_index <= 100:
try:
ligne_persist = factory_lignes.List(ligne_index)
if ligne_persist is None:
break
ligne = win32com.client.CastTo(ligne_persist, "IBODocumentLigne3")
ligne.Read()
# Charger référence article
article_ref = ""
try:
article_ref = getattr(ligne, "AR_Ref", "").strip()
if not article_ref:
article_obj = getattr(ligne, "Article", None)
if article_obj:
article_obj.Read()
article_ref = getattr(article_obj, "AR_Ref", "").strip()
except:
pass
lignes.append({
"article": article_ref,
"designation": getattr(ligne, "DL_Design", ""),
"quantite": float(getattr(ligne, "DL_Qte", 0.0)),
"prix_unitaire": float(getattr(ligne, "DL_PrixUnitaire", 0.0)),
"montant_ht": float(getattr(ligne, "DL_MontantHT", 0.0))
})
ligne_index += 1
except Exception as e:
logger.debug(f"Erreur ligne {ligne_index}: {e}")
break
except Exception as e:
logger.debug(f"Erreur chargement lignes: {e}")
factures.append({
"numero": getattr(doc, "DO_Piece", ""),
"reference": getattr(doc, "DO_Ref", ""), # ✅ AJOUT
"date": str(getattr(doc, "DO_Date", "")),
"client_code": client_code,
"client_intitule": client_intitule,
"total_ht": float(getattr(doc, "DO_TotalHT", 0.0)),
"total_ttc": float(getattr(doc, "DO_TotalTTC", 0.0)),
"statut": doc_statut,
"lignes": lignes
})
index += 1
except:
index += 1
continue
logger.info(f"✅ {len(factures)} factures retournées")
return {"success": True, "data": factures}
except Exception as e:
logger.error(f"Erreur liste factures: {e}")
logger.error(f"❌ Erreur liste factures: {e}", exc_info=True)
raise HTTPException(500, str(e))
@app.post("/sage/client/remise-max", dependencies=[Depends(verify_token)])
@ -2670,15 +2369,16 @@ def prospect_get(req: CodeRequest):
@app.post("/sage/fournisseurs/list", dependencies=[Depends(verify_token)])
def fournisseurs_list(req: FiltreRequest):
"""
⚡ ENDPOINT DIRECT : Liste fournisseurs via FactoryFournisseur
⚡ Liste rapide des fournisseurs depuis le CACHE
✅ Utilise la méthode corrigée qui ne dépend pas du cache
✅ Utilise le cache mémoire pour une réponse instantanée
🔄 Cache actualisé automatiquement toutes les 15 minutes
"""
try:
# ✅ Appel direct à la méthode corrigée
fournisseurs = sage.lister_tous_fournisseurs(req.filtre)
# ✅ Utiliser le cache au lieu de la lecture directe
fournisseurs = sage.lister_tous_fournisseurs_cache(req.filtre)
logger.info(f"✅ {len(fournisseurs)} fournisseurs retournés via endpoint")
logger.info(f"✅ {len(fournisseurs)} fournisseurs retournés depuis le cache")
return {"success": True, "data": fournisseurs}
@ -3022,6 +2722,68 @@ def modifier_facture_endpoint(req: FactureUpdateGatewayRequest):
raise HTTPException(500, str(e))
@app.post("/sage/documents/generate-pdf", dependencies=[Depends(verify_token)])
def generer_pdf_document(req: PDFGenerationRequest):
"""
📄 Génération PDF d'un document (endpoint généralisé)
**Supporte tous les types de documents Sage:**
- Devis (0)
- Bons de commande (10)
- Bons de livraison (30)
- Factures (60)
- Avoirs (50)
**Process:**
1. Charge le document depuis Sage
2. Génère le PDF via l'état Sage correspondant
3. Retourne le PDF en base64
Args:
req: Requête contenant doc_id et type_doc
Returns:
{
"success": true,
"data": {
"pdf_base64": "JVBERi0xLjQK...",
"taille_octets": 12345,
"type_doc": 0,
"numero": "DE00001"
}
}
"""
try:
logger.info(f"📄 Génération PDF: {req.doc_id} (type={req.type_doc})")
# Appel au connecteur Sage
pdf_bytes = sage.generer_pdf_document(req.doc_id, req.type_doc)
if not pdf_bytes:
raise HTTPException(500, "PDF vide généré")
# Encoder en base64 pour le transport JSON
import base64
pdf_base64 = base64.b64encode(pdf_bytes).decode('utf-8')
logger.info(f"✅ PDF généré: {len(pdf_bytes)} octets")
return {
"success": True,
"data": {
"pdf_base64": pdf_base64,
"taille_octets": len(pdf_bytes),
"type_doc": req.type_doc,
"numero": req.doc_id
}
}
except HTTPException:
raise
except Exception as e:
logger.error(f"❌ Erreur génération PDF: {e}", exc_info=True)
raise HTTPException(500, str(e))
# =====================================================
# LANCEMENT
# =====================================================