from fastapi import FastAPI, HTTPException, Header, Depends, Query from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel, Field from typing import Optional, List, Dict from datetime import datetime, date from enum import Enum import uvicorn import logging import win32com.client from config import settings, validate_settings from sage_connector import SageConnector # ===================================================== # LOGGING # ===================================================== logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", handlers=[logging.FileHandler("sage_gateway.log"), logging.StreamHandler()], ) logger = logging.getLogger(__name__) # ===================================================== # ENUMS # ===================================================== class TypeDocument(int, Enum): DEVIS = 0 BON_LIVRAISON = 1 BON_RETOUR = 2 COMMANDE = 3 PREPARATION = 4 FACTURE = 5 # ===================================================== # MODÈLES # ===================================================== class FiltreRequest(BaseModel): filtre: Optional[str] = "" class CodeRequest(BaseModel): code: str class ChampLibreRequest(BaseModel): doc_id: str type_doc: int nom_champ: str valeur: str class DevisRequest(BaseModel): client_id: str date_devis: Optional[date] = None lignes: List[ Dict ] # Format: {article_code, quantite, prix_unitaire_ht, remise_pourcentage} class TransformationRequest(BaseModel): numero_source: str type_source: int type_cible: int class StatutRequest(BaseModel): nouveau_statut: int # ===================================================== # SÉCURITÉ # ===================================================== def verify_token(x_sage_token: str = Header(...)): """Vérification du token d'authentification""" if x_sage_token != settings.sage_gateway_token: logger.warning(f"❌ Token invalide reçu: {x_sage_token[:20]}...") raise HTTPException(401, "Token invalide") return True # ===================================================== # APPLICATION # ===================================================== app = FastAPI( title="Sage Gateway - Windows Server", version="1.0.0", description="Passerelle d'accès à Sage 100c pour VPS Linux", ) app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, allow_methods=["*"], allow_headers=["*"], allow_credentials=True, ) sage: Optional[SageConnector] = None # ===================================================== # LIFECYCLE # ===================================================== @app.on_event("startup") def startup(): global sage logger.info("🚀 Démarrage Sage Gateway Windows...") # Validation config try: validate_settings() logger.info("✅ Configuration validée") except ValueError as e: logger.error(f"❌ Configuration invalide: {e}") raise # Connexion Sage sage = SageConnector( settings.chemin_base, settings.utilisateur, settings.mot_de_passe ) if not sage.connecter(): raise RuntimeError("❌ Impossible de se connecter à Sage 100c") logger.info("✅ Sage Gateway démarré et connecté") @app.on_event("shutdown") def shutdown(): if sage: sage.deconnecter() logger.info("👋 Sage Gateway arrêté") # ===================================================== # ENDPOINTS - SYSTÈME # ===================================================== @app.get("/health") def health(): """Health check""" return { "status": "ok", "sage_connected": sage is not None and sage.cial is not None, "cache_info": sage.get_cache_info() if sage else None, "timestamp": datetime.now().isoformat(), } # ===================================================== # ENDPOINTS - CLIENTS # ===================================================== @app.post("/sage/clients/list", dependencies=[Depends(verify_token)]) def clients_list(req: FiltreRequest): """Liste des clients avec filtre optionnel""" try: clients = sage.lister_tous_clients(req.filtre) return {"success": True, "data": clients} except Exception as e: logger.error(f"Erreur liste clients: {e}") raise HTTPException(500, str(e)) @app.post("/sage/clients/get", dependencies=[Depends(verify_token)]) def client_get(req: CodeRequest): """Lecture d'un client par code""" try: client = sage.lire_client(req.code) if not client: raise HTTPException(404, f"Client {req.code} non trouvé") return {"success": True, "data": client} except HTTPException: raise except Exception as e: logger.error(f"Erreur lecture client: {e}") raise HTTPException(500, str(e)) # ===================================================== # ENDPOINTS - ARTICLES # ===================================================== @app.post("/sage/articles/list", dependencies=[Depends(verify_token)]) def articles_list(req: FiltreRequest): """Liste des articles avec filtre optionnel""" try: articles = sage.lister_tous_articles(req.filtre) return {"success": True, "data": articles} except Exception as e: logger.error(f"Erreur liste articles: {e}") raise HTTPException(500, str(e)) @app.post("/sage/articles/get", dependencies=[Depends(verify_token)]) def article_get(req: CodeRequest): """Lecture d'un article par référence""" try: article = sage.lire_article(req.code) if not article: raise HTTPException(404, f"Article {req.code} non trouvé") return {"success": True, "data": article} except HTTPException: raise except Exception as e: logger.error(f"Erreur lecture article: {e}") raise HTTPException(500, str(e)) # ===================================================== # ENDPOINTS - DEVIS # ===================================================== @app.post("/sage/devis/create", dependencies=[Depends(verify_token)]) def creer_devis(req: DevisRequest): """Création d'un devis""" try: # Transformer en format attendu par sage_connector devis_data = { "client": {"code": req.client_id, "intitule": ""}, "date_devis": req.date_devis or date.today(), "lignes": req.lignes, } resultat = sage.creer_devis_enrichi(devis_data) return {"success": True, "data": resultat} except Exception as e: logger.error(f"Erreur création devis: {e}") raise HTTPException(500, str(e)) @app.post("/sage/devis/get", dependencies=[Depends(verify_token)]) def lire_devis(req: CodeRequest): """Lecture d'un devis""" try: devis = sage.lire_devis(req.code) if not devis: raise HTTPException(404, f"Devis {req.code} non trouvé") return {"success": True, "data": devis} except HTTPException: raise except Exception as e: logger.error(f"Erreur lecture devis: {e}") raise HTTPException(500, str(e)) @app.post("/sage/devis/list", dependencies=[Depends(verify_token)]) def devis_list( limit: int = 100, statut: Optional[int] = None, inclure_lignes: bool = Query(True) ): """ 📋 Liste tous les devis avec filtres optionnels 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 """ 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 devis_list = [] index = 1 max_iterations = limit * 3 erreurs_consecutives = 0 max_erreurs = 50 logger.info( f"🔍 Recherche devis (limit={limit}, statut={statut}, inclure_lignes={inclure_lignes})" ) 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)" ) 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""" try: with sage._com_context(), sage._lock_com: factory = sage.cial.FactoryDocumentVente persist = factory.ReadPiece(0, numero) if not persist: raise HTTPException(404, f"Devis {numero} introuvable") doc = win32com.client.CastTo(persist, "IBODocumentVente3") doc.Read() statut_actuel = getattr(doc, "DO_Statut", 0) doc.DO_Statut = nouveau_statut doc.Write() logger.info(f"✅ Statut devis {numero}: {statut_actuel} → {nouveau_statut}") return { "success": True, "data": { "numero": numero, "statut_ancien": statut_actuel, "statut_nouveau": nouveau_statut, }, } except HTTPException: raise except Exception as e: logger.error(f"Erreur changement statut: {e}") raise HTTPException(500, str(e)) # ===================================================== # ENDPOINTS - DOCUMENTS # ===================================================== @app.post("/sage/documents/get", dependencies=[Depends(verify_token)]) def lire_document(numero: str, type_doc: int): """Lecture d'un document (commande, facture, etc.)""" try: doc = sage.lire_document(numero, type_doc) if not doc: raise HTTPException(404, f"Document {numero} non trouvé") return {"success": True, "data": doc} except HTTPException: raise except Exception as e: logger.error(f"Erreur lecture document: {e}") raise HTTPException(500, str(e)) @app.post("/sage/documents/transform", dependencies=[Depends(verify_token)]) def transformer_document( numero_source: str = Query(..., description="Numéro du document source"), type_source: int = Query(..., description="Type document source"), type_cible: int = Query(..., description="Type document cible"), ): """ 🔧 Transformation de document ✅ CORRECTION : Utilise les VRAIS types Sage Dataven Types valides : - 0: Devis - 10: Bon de commande - 20: Préparation - 30: Bon de livraison - 40: Bon de retour - 50: Bon d'avoir - 60: Facture Transformations autorisées : - Devis (0) → Commande (10) - Commande (10) → Bon livraison (30) - Commande (10) → Facture (60) - Bon livraison (30) → Facture (60) """ try: logger.info( f"🔄 Transformation demandée: {numero_source} " f"(type {type_source}) → type {type_cible}" ) # ✅ Matrice des transformations valides pour VOTRE Sage transformations_valides = { (0, 10), # Devis → Commande (10, 30), # Commande → Bon de livraison (10, 60), # Commande → Facture (30, 60), # Bon de livraison → Facture (0, 60), # Devis → Facture (si autorisé) } if (type_source, type_cible) not in transformations_valides: logger.error( f"❌ Transformation non autorisée: {type_source} → {type_cible}" ) raise HTTPException( 400, f"Transformation non autorisée: type {type_source} → type {type_cible}. " f"Transformations valides: {transformations_valides}", ) # Appel au connecteur Sage resultat = sage.transformer_document(numero_source, type_source, type_cible) logger.info( f"✅ Transformation réussie: {numero_source} → " f"{resultat.get('document_cible', '?')} " f"({resultat.get('nb_lignes', 0)} lignes)" ) return {"success": True, "data": resultat} except HTTPException: raise except ValueError as e: logger.error(f"❌ Erreur métier transformation: {e}") raise HTTPException(400, str(e)) except Exception as e: logger.error(f"❌ Erreur technique transformation: {e}", exc_info=True) raise HTTPException(500, f"Erreur transformation: {str(e)}") @app.post("/sage/documents/champ-libre", dependencies=[Depends(verify_token)]) def maj_champ_libre(req: ChampLibreRequest): """Mise à jour d'un champ libre""" try: success = sage.mettre_a_jour_champ_libre( req.doc_id, req.type_doc, req.nom_champ, req.valeur ) return {"success": success} except Exception as e: logger.error(f"Erreur MAJ champ libre: {e}") raise HTTPException(500, str(e)) @app.post("/sage/documents/derniere-relance", dependencies=[Depends(verify_token)]) def maj_derniere_relance(doc_id: str, type_doc: int): """📅 Met à jour le champ 'Dernière relance' d'un document""" try: success = sage.mettre_a_jour_derniere_relance(doc_id, type_doc) return {"success": success} except Exception as e: logger.error(f"Erreur MAJ dernière relance: {e}") raise HTTPException(500, str(e)) # ===================================================== # ENDPOINTS - CONTACTS # ===================================================== @app.post("/sage/contact/read", dependencies=[Depends(verify_token)]) def contact_read(req: CodeRequest): """Lecture du contact principal d'un client""" try: contact = sage.lire_contact_principal_client(req.code) if not contact: raise HTTPException(404, f"Contact non trouvé pour client {req.code}") return {"success": True, "data": contact} except HTTPException: raise except Exception as e: logger.error(f"Erreur lecture contact: {e}") raise HTTPException(500, str(e)) @app.post("/sage/commandes/list", dependencies=[Depends(verify_token)]) def commandes_list(limit: int = 100, statut: Optional[int] = None): """ 📋 Liste toutes les commandes ✅ CORRECTION: Filtre sur type 10 (BON_COMMANDE) """ 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 ): try: persist = factory.List(index) if persist is None: break doc = win32com.client.CastTo(persist, "IBODocumentVente3") doc.Read() doc_type = getattr(doc, "DO_Type", -1) # ✅ CRITIQUE : 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", ""), "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") 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): """ 📋 Liste toutes les factures ✅ CORRECTION: Filtre sur type 60 (FACTURE) """ try: with sage._com_context(), sage._lock_com: factory = sage.cial.FactoryDocumentVente factures = [] index = 1 max_iterations = limit * 3 while len(factures) < limit and index < max_iterations: try: persist = factory.List(index) if persist is None: break doc = win32com.client.CastTo(persist, "IBODocumentVente3") doc.Read() # ✅ CRITIQUE: Filtrer factures (type 60) if getattr(doc, "DO_Type", -1) != settings.SAGE_TYPE_FACTURE: index += 1 continue 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 factures.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, } ) 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}") raise HTTPException(500, str(e)) @app.post("/sage/client/remise-max", dependencies=[Depends(verify_token)]) def lire_remise_max_client(code: str): """Récupère la remise max autorisée pour un client""" try: client_obj = sage._lire_client_obj(code) if not client_obj: raise HTTPException(404, f"Client {code} introuvable") remise_max = 10.0 # Défaut try: remise_max = float(getattr(client_obj, "CT_RemiseMax", 10.0)) except: pass logger.info(f"✅ Remise max client {code}: {remise_max}%") return { "success": True, "data": {"client_code": code, "remise_max": remise_max}, } except HTTPException: raise except Exception as e: logger.error(f"Erreur lecture remise: {e}") raise HTTPException(500, str(e)) # ===================================================== # ENDPOINTS - ADMIN # ===================================================== @app.post("/sage/cache/refresh", dependencies=[Depends(verify_token)]) def refresh_cache(): """Force le rafraîchissement du cache""" try: sage.forcer_actualisation_cache() return { "success": True, "message": "Cache actualisé", "info": sage.get_cache_info(), } except Exception as e: logger.error(f"Erreur refresh cache: {e}") raise HTTPException(500, str(e)) @app.get("/sage/cache/info", dependencies=[Depends(verify_token)]) def cache_info_get(): """Informations sur le cache (endpoint GET)""" try: return {"success": True, "data": sage.get_cache_info()} except Exception as e: logger.error(f"Erreur info cache: {e}") raise HTTPException(500, str(e)) # Script à ajouter temporairement dans main.py pour diagnostiquer @app.get("/sage/devis/{numero}/diagnostic", dependencies=[Depends(verify_token)]) def diagnostiquer_devis(numero: str): """ ENDPOINT DE DIAGNOSTIC: Affiche TOUTES les infos d'un devis Permet de comprendre pourquoi un devis ne peut pas être transformé """ 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 # Essayer ReadPiece persist = factory.ReadPiece(0, numero) # Si échec, chercher dans List() if not persist: logger.info(f"[DIAG] ReadPiece echoue, recherche dans List()...") index = 1 while index < 10000: try: persist_test = factory.List(index) if persist_test is None: break doc_test = win32com.client.CastTo( persist_test, "IBODocumentVente3" ) doc_test.Read() if ( getattr(doc_test, "DO_Type", -1) == 0 and getattr(doc_test, "DO_Piece", "") == numero ): persist = persist_test logger.info(f"[DIAG] Trouve a l'index {index}") break index += 1 except: index += 1 if not persist: raise HTTPException(404, f"Devis {numero} introuvable") doc = win32com.client.CastTo(persist, "IBODocumentVente3") doc.Read() # EXTRACTION COMPLÈTE diagnostic = { "numero": getattr(doc, "DO_Piece", ""), "type": getattr(doc, "DO_Type", -1), "statut": getattr(doc, "DO_Statut", -1), "statut_libelle": { 0: "Brouillon", 1: "Soumis", 2: "Accepte", 3: "Realise partiellement", 4: "Realise totalement", 5: "Transforme", 6: "Annule", }.get(getattr(doc, "DO_Statut", -1), "Inconnu"), "date": str(getattr(doc, "DO_Date", "")), "total_ht": float(getattr(doc, "DO_TotalHT", 0.0)), "total_ttc": float(getattr(doc, "DO_TotalTTC", 0.0)), "est_transformable": False, "raison_blocage": None, } # Client try: client_obj = getattr(doc, "Client", None) if client_obj: client_obj.Read() diagnostic["client_code"] = getattr( client_obj, "CT_Num", "" ).strip() diagnostic["client_intitule"] = getattr( client_obj, "CT_Intitule", "" ).strip() except Exception as e: diagnostic["erreur_client"] = str(e) # Lignes try: factory_lignes = doc.FactoryDocumentLigne except: factory_lignes = doc.FactoryDocumentVenteLigne lignes = [] index = 1 while index <= 100: try: ligne_p = factory_lignes.List(index) if ligne_p is None: break ligne = win32com.client.CastTo(ligne_p, "IBODocumentLigne3") ligne.Read() 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( { "index": index, "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)), } ) index += 1 except: break diagnostic["nb_lignes"] = len(lignes) diagnostic["lignes"] = lignes # ANALYSE TRANSFORMABILITÉ statut = diagnostic["statut"] if statut == 5: diagnostic["raison_blocage"] = "Document deja transforme (statut=5)" elif statut == 6: diagnostic["raison_blocage"] = "Document annule (statut=6)" elif statut in [3, 4]: diagnostic["raison_blocage"] = ( f"Document deja realise partiellement ou totalement (statut={statut}). " f"Une commande/BL/facture existe probablement deja." ) diagnostic["suggestion"] = ( "Cherchez les documents lies a ce devis dans Sage. " "Il a peut-etre deja ete transforme manuellement." ) elif statut == 0: diagnostic["est_transformable"] = True diagnostic["action_requise"] = ( "Statut 'Brouillon'. Le systeme le passera automatiquement a 'Accepte' " "avant transformation." ) elif statut == 2: diagnostic["est_transformable"] = True diagnostic["action_requise"] = ( "Statut 'Accepte'. Transformation possible." ) else: diagnostic["raison_blocage"] = f"Statut inconnu ou non gere: {statut}" # Champs libres (pour Universign, etc.) champs_libres = {} try: for champ in ["UniversignID", "DerniereRelance", "DO_Ref"]: try: valeur = getattr(doc, f"DO_{champ}", None) if valeur: champs_libres[champ] = str(valeur) except: pass except: pass if champs_libres: diagnostic["champs_libres"] = champs_libres logger.info( f"[DIAG] Devis {numero}: statut={statut}, transformable={diagnostic['est_transformable']}" ) return {"success": True, "diagnostic": diagnostic} except HTTPException: raise except Exception as e: logger.error(f"[DIAG] Erreur diagnostic: {e}", exc_info=True) raise HTTPException(500, str(e)) @app.get("/sage/diagnostic/configuration", dependencies=[Depends(verify_token)]) def diagnostic_configuration(): """ DIAGNOSTIC COMPLET de la configuration Sage Teste: - Quelles méthodes COM sont disponibles - Quels types de documents sont autorisés - Quelles permissions l'utilisateur a - Version de Sage """ try: if not sage or not sage.cial: raise HTTPException(503, "Service Sage indisponible") with sage._com_context(), sage._lock_com: diagnostic = { "connexion": "OK", "chemin_base": sage.chemin_base, "utilisateur": sage.utilisateur, } # Version Sage try: version = getattr(sage.cial, "Version", "Inconnue") diagnostic["version_sage"] = str(version) except: diagnostic["version_sage"] = "Non disponible" # Test des méthodes disponibles sur IBSCIALApplication3 methodes_disponibles = [] methodes_a_tester = [ "CreateProcess_Document", "FactoryDocumentVente", "FactoryArticle", "CptaApplication", "BeginTrans", "CommitTrans", "RollbackTrans", ] for methode in methodes_a_tester: try: if hasattr(sage.cial, methode): methodes_disponibles.append(methode) except: pass diagnostic["methodes_cial_disponibles"] = methodes_disponibles # Test des types de documents autorisés types_autorises = [] types_bloques = [] for type_doc in range(6): # 0-5 try: # Essayer de créer un process (sans le valider) process = sage.cial.CreateProcess_Document(type_doc) if process: types_autorises.append( { "type": type_doc, "libelle": { 0: "Devis", 1: "Bon de livraison", 2: "Bon de retour", 3: "Commande", 4: "Preparation", 5: "Facture", }[type_doc], } ) # Ne pas valider, juste tester del process except Exception as e: types_bloques.append( { "type": type_doc, "libelle": { 0: "Devis", 1: "Bon de livraison", 2: "Bon de retour", 3: "Commande", 4: "Preparation", 5: "Facture", }[type_doc], "erreur": str(e)[:200], } ) diagnostic["types_documents_autorises"] = types_autorises diagnostic["types_documents_bloques"] = types_bloques # Test TransformInto() sur un devis test try: factory = sage.cial.FactoryDocumentVente # Chercher n'importe quel devis index = 1 devis_test = None while index < 100: try: persist = factory.List(index) if persist is None: break doc = win32com.client.CastTo(persist, "IBODocumentVente3") doc.Read() if getattr(doc, "DO_Type", -1) == 0: # Devis devis_test = doc break index += 1 except: index += 1 if devis_test: # Tester si TransformInto existe if hasattr(devis_test, "TransformInto"): diagnostic["transforminto_disponible"] = True diagnostic["transforminto_test"] = "Methode existe (non testee)" else: diagnostic["transforminto_disponible"] = False diagnostic["transforminto_test"] = ( "Methode TransformInto() inexistante" ) else: diagnostic["transforminto_disponible"] = ( "Impossible de tester (aucun devis trouve)" ) except Exception as e: diagnostic["transforminto_disponible"] = False diagnostic["transforminto_erreur"] = str(e) # Modules Sage actifs try: # Tester l'accès aux différentes factories modules = {} try: sage.cial.FactoryDocumentVente modules["Ventes"] = "OK" except: modules["Ventes"] = "INACCESSIBLE" try: sage.cial.CptaApplication.FactoryClient modules["Clients"] = "OK" except: modules["Clients"] = "INACCESSIBLE" try: sage.cial.FactoryArticle modules["Articles"] = "OK" except: modules["Articles"] = "INACCESSIBLE" diagnostic["modules_actifs"] = modules except Exception as e: diagnostic["modules_actifs_erreur"] = str(e) # Compter documents existants try: counts = {} factory = sage.cial.FactoryDocumentVente for type_doc in range(6): count = 0 index = 1 while index < 1000: try: persist = factory.List(index) if persist is None: break doc = win32com.client.CastTo(persist, "IBODocumentVente3") doc.Read() if getattr(doc, "DO_Type", -1) == type_doc: count += 1 index += 1 except: index += 1 break counts[ { 0: "Devis", 1: "Bons_livraison", 2: "Bons_retour", 3: "Commandes", 4: "Preparations", 5: "Factures", }[type_doc] ] = count diagnostic["documents_existants"] = counts except Exception as e: diagnostic["documents_existants_erreur"] = str(e) logger.info("[DIAG] Configuration Sage analysee") return {"success": True, "diagnostic": diagnostic} except HTTPException: raise except Exception as e: logger.error(f"[DIAG] Erreur diagnostic config: {e}", exc_info=True) raise HTTPException(500, str(e)) @app.get("/sage/diagnostic/types-reels", dependencies=[Depends(verify_token)]) def decouvrir_types_reels(): """ DIAGNOSTIC CRITIQUE: Découvre les VRAIS types de documents Sage Au lieu de deviner les types (0-5), on va: 1. Créer manuellement un document de chaque type dans Sage 2. Les lister ici pour voir leurs vrais numéros de type """ 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 # Parcourir TOUS les documents documents_par_type = {} index = 1 max_docs = 500 # Limiter pour ne pas bloquer logger.info("[DIAG] Scan de tous les documents...") while index < max_docs: try: persist = factory.List(index) if persist is None: break doc = win32com.client.CastTo(persist, "IBODocumentVente3") doc.Read() # Récupérer le type ET le sous-type type_doc = getattr(doc, "DO_Type", -1) piece = getattr(doc, "DO_Piece", "") statut = getattr(doc, "DO_Statut", -1) # Essayer de récupérer le domaine (vente/achat) domaine = "Inconnu" try: domaine_val = getattr(doc, "DO_Domaine", -1) domaine = {0: "Vente", 1: "Achat"}.get( domaine_val, f"Code {domaine_val}" ) except: pass # Récupérer la catégorie categorie = "Inconnue" try: cat_val = getattr(doc, "DO_Categorie", -1) categorie = str(cat_val) except: pass # Grouper par type if type_doc not in documents_par_type: documents_par_type[type_doc] = { "count": 0, "exemples": [], "domaine": domaine, "categorie": categorie, } documents_par_type[type_doc]["count"] += 1 # Garder quelques exemples if len(documents_par_type[type_doc]["exemples"]) < 3: documents_par_type[type_doc]["exemples"].append( { "numero": piece, "statut": statut, "domaine": domaine, "categorie": categorie, } ) index += 1 except Exception as e: logger.debug(f"Erreur index {index}: {e}") index += 1 # Formater le résultat types_trouves = [] for type_num, infos in sorted(documents_par_type.items()): types_trouves.append( { "type_code": type_num, "nombre_documents": infos["count"], "domaine": infos["domaine"], "categorie": infos["categorie"], "exemples": infos["exemples"], "suggestion_libelle": _deviner_libelle_type( type_num, infos["exemples"] ), } ) logger.info( f"[DIAG] {len(types_trouves)} types de documents distincts trouves" ) return { "success": True, "types_documents_reels": types_trouves, "instructions": ( "Pour identifier les types corrects:\n" "1. Creez manuellement dans Sage: 1 Bon de commande, 1 BL, 1 Facture\n" "2. Appelez de nouveau cet endpoint\n" "3. Les nouveaux types apparaitront avec leurs numeros corrects" ), "total_documents_scannes": index - 1, } except Exception as e: logger.error(f"[DIAG] Erreur decouverte types: {e}", exc_info=True) raise HTTPException(500, str(e)) def _deviner_libelle_type(type_num, exemples): """Devine le libellé d'un type basé sur les numéros de pièce""" if not exemples: return "Type inconnu" # Analyser les préfixes des numéros prefixes = [ex["numero"][:2] for ex in exemples if ex["numero"]] prefix_commun = max(set(prefixes), key=prefixes.count) if prefixes else "" # Deviner selon le type_num et les préfixes suggestions = { 0: "Devis (DE)", 1: "Bon de livraison (BL)", 2: "Bon de retour (BR)", 3: "Bon de commande (BC)", 4: "Preparation de livraison (PL)", 5: "Facture (FA)", 6: "Facture d'avoir (AV)", 7: "Bon d'avoir financier (BA)", } libelle_base = suggestions.get(type_num, f"Type {type_num}") if prefix_commun: libelle_base += f" - Detecte: prefix '{prefix_commun}'" return libelle_base @app.post("/sage/test-creation-par-type", dependencies=[Depends(verify_token)]) def tester_creation_par_type(type_doc: int = Query(..., ge=0, le=20)): """ TEST: Essaie de créer un document d'un type spécifique Permet de tester tous les types possibles (0-20) pour trouver lesquels fonctionnent sur votre installation """ try: if not sage or not sage.cial: raise HTTPException(503, "Service Sage indisponible") with sage._com_context(), sage._lock_com: logger.info(f"[TEST] Tentative creation type {type_doc}...") try: # Essayer de créer un process process = sage.cial.CreateProcess_Document(type_doc) if not process: return { "success": False, "type": type_doc, "resultat": "Process NULL retourne", } # Si on arrive ici, le type est valide ! doc = process.Document try: doc = win32com.client.CastTo(doc, "IBODocumentVente3") except: pass # Récupérer les infos du document créé type_reel = getattr(doc, "DO_Type", -1) domaine = getattr(doc, "DO_Domaine", -1) # NE PAS VALIDER le document (pas de Write/Process) # On veut juste savoir si la création est possible del process del doc return { "success": True, "type_demande": type_doc, "type_reel_doc": type_reel, "domaine": {0: "Vente", 1: "Achat"}.get(domaine, domaine), "resultat": "CREATION POSSIBLE", "note": "Document non valide (test uniquement)", } except Exception as e: return { "success": False, "type": type_doc, "erreur": str(e), "resultat": "CREATION IMPOSSIBLE", } except Exception as e: logger.error(f"[TEST] Erreur test type {type_doc}: {e}") raise HTTPException(500, str(e)) # ===================================================== # LANCEMENT # ===================================================== if __name__ == "__main__": uvicorn.run( "main:app", host=settings.api_host, port=settings.api_port, reload=False, # Pas de reload en production log_level="info", )