from typing import Dict, List, Optional import win32com.client import pywintypes from datetime import datetime from schemas.documents.reglements import ModeReglement import time import logging logger = logging.getLogger(__name__) def _get_journal_auto(self, mode_reglement: int) -> str: with self._get_sql_connection() as conn: cursor = conn.cursor() if mode_reglement == ModeReglement.ESPECES: cursor.execute( "SELECT TOP 1 JO_Num FROM F_JOURNAUX WHERE JO_Type = 2 AND JO_Reglement = 1 AND CG_Num LIKE '53%' ORDER BY JO_Num" ) else: cursor.execute( "SELECT TOP 1 JO_Num FROM F_JOURNAUX WHERE JO_Type = 2 AND JO_Reglement = 1 AND CG_Num LIKE '51%' ORDER BY JO_Num" ) row = cursor.fetchone() if row: return row[0].strip() cursor.execute( "SELECT TOP 1 JO_Num FROM F_JOURNAUX WHERE JO_Type = 2 AND JO_Reglement = 1 ORDER BY JO_Num" ) row = cursor.fetchone() if row: return row[0].strip() raise ValueError("Aucun journal de trésorerie configuré") def lire_journaux_banque(self) -> List[Dict]: if not self.cial: raise RuntimeError("Connexion Sage non établie") with self._get_sql_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT JO_Num, JO_Intitule, CG_Num FROM F_JOURNAUX WHERE JO_Type = 2 AND JO_Reglement = 1 ORDER BY JO_Num" ) return [ { "code": row[0].strip(), "intitule": row[1].strip() if row[1] else "", "compte_general": row[2].strip() if row[2] else "", "type": "Caisse" if (row[2] or "").startswith("53") else "Banque", } for row in cursor.fetchall() ] def lire_tous_journaux(self) -> List[Dict]: if not self.cial: raise RuntimeError("Connexion Sage non établie") types_libelles = { 0: "Achats", 1: "Ventes", 2: "Trésorerie", 3: "Général", 4: "Situation", } with self._get_sql_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT JO_Num, JO_Intitule, JO_Type, JO_Reglement, CG_Num FROM F_JOURNAUX ORDER BY JO_Type, JO_Num" ) return [ { "code": row[0].strip(), "intitule": row[1].strip() if row[1] else "", "type_code": row[2], "type_libelle": types_libelles.get(row[2], f"Type {row[2]}"), "reglement_actif": row[3] == 1, "compte_general": row[4].strip() if row[4] else "", } for row in cursor.fetchall() ] def regler_facture( self, numero_facture: str, montant: float, mode_reglement: int = ModeReglement.CHEQUE, date_reglement: datetime = None, reference: str = "", libelle: str = "", ) -> Dict: if not self.cial: raise RuntimeError("Connexion Sage non établie") if montant <= 0: raise ValueError("Le montant du règlement doit être positif") date_reglement = date_reglement or datetime.now() code_journal = _get_journal_auto(self, mode_reglement) logger.info( f"Règlement facture {numero_facture}: {montant}€ (mode: {mode_reglement}, journal: {code_journal})" ) try: with self._com_context(), self._lock_com: factory = self.cial.FactoryDocumentVente if not factory.ExistPiece(60, numero_facture): raise ValueError(f"Facture {numero_facture} introuvable") persist = factory.ReadPiece(60, numero_facture) doc = win32com.client.CastTo(persist, "IBODocumentVente3") doc.Read() total_ttc = float(getattr(doc, "DO_TotalTTC", 0.0)) montant_deja_regle = float(getattr(doc, "DO_MontantRegle", 0.0)) statut = getattr(doc, "DO_Statut", 0) if statut == 6: raise ValueError(f"Facture {numero_facture} annulée") solde_actuel = total_ttc - montant_deja_regle if montant > solde_actuel + 0.01: raise ValueError( f"Montant ({montant}€) supérieur au solde ({solde_actuel:.2f}€)" ) client_code = "" try: client_obj = getattr(doc, "Client", None) if client_obj: client_obj.Read() client_code = getattr(client_obj, "CT_Num", "").strip() except Exception: pass echeance = _get_premiere_echeance(doc) if not echeance: raise ValueError(f"Facture {numero_facture} sans échéance") numero_reglement = _executer_reglement_com( self, doc, echeance, montant, mode_reglement, date_reglement, reference, libelle, code_journal, client_code, numero_facture, ) time.sleep(0.3) doc.Read() nouveau_montant_regle = float(getattr(doc, "DO_MontantRegle", 0.0)) if abs(nouveau_montant_regle - montant_deja_regle) < 0.01: raise RuntimeError( "Le règlement n'a pas été appliqué (DO_MontantRegle inchangé)" ) nouveau_solde = total_ttc - nouveau_montant_regle logger.info(f"Règlement effectué - Solde restant: {nouveau_solde:.2f}€") return { "numero_facture": numero_facture, "numero_reglement": numero_reglement, "montant_regle": montant, "date_reglement": date_reglement.strftime("%Y-%m-%d"), "mode_reglement": mode_reglement, "mode_reglement_libelle": ModeReglement.get_libelle(mode_reglement), "reference": reference, "libelle": libelle, "code_journal": code_journal, "total_facture": total_ttc, "solde_restant": nouveau_solde, "facture_soldee": nouveau_solde < 0.01, "client_code": client_code, } except ValueError: raise except Exception as e: logger.error(f"Erreur règlement: {e}", exc_info=True) raise RuntimeError(f"Échec règlement facture: {str(e)}") def _get_premiere_echeance(doc): try: factory_ech = getattr(doc, "FactoryDocumentEcheance", None) if factory_ech: ech_list = factory_ech.List if ech_list: echeance = ech_list.Item(1) if echeance: for iface in ["IBODocumentEcheance3", "IBODocumentEcheance"]: try: echeance = win32com.client.CastTo(echeance, iface) logger.info(f" Échéance castée vers {iface}") break except Exception: continue echeance.Read() return echeance except Exception as e: logger.warning(f" Pas d'échéance: {e}") return None def _executer_reglement_com( self, doc, echeance, montant, mode_reglement, date_reglement, reference, libelle, code_journal, client_code, numero_facture, ): erreurs = [] # APPROCHE PRINCIPALE: Créer règlement complet, l'écrire, puis l'assigner au process try: logger.info("Création du règlement via FactoryDocumentReglement...") # 1. Créer le règlement factory_reg = self.cial.FactoryDocumentReglement reg = factory_reg.Create() reg = win32com.client.CastTo(reg, "IBODocumentReglement") logger.info(" Règlement créé et casté vers IBODocumentReglement") # 2. Configurer le Journal (objet) try: journal_factory = self.cial.CptaApplication.FactoryJournal journal_persist = journal_factory.ReadNumero(code_journal) if journal_persist: reg.Journal = journal_persist logger.info(f" Journal défini: {code_journal}") except Exception as e: logger.warning(f" Journal: {e}") # 3. Configurer le TiersPayeur (objet client) try: factory_client = self.cial.CptaApplication.FactoryClient if client_code: client_persist = factory_client.ReadNumero(client_code) if client_persist: reg.TiersPayeur = client_persist logger.info(f" TiersPayeur défini: {client_code}") except Exception as e: logger.warning(f" TiersPayeur: {e}") # 4. Configurer les champs simples try: reg.RG_Date = pywintypes.Time(date_reglement) logger.info(f" RG_Date: {date_reglement}") except Exception as e: logger.warning(f" RG_Date: {e}") try: reg.RG_Montant = montant logger.info(f" RG_Montant: {montant}") except Exception as e: logger.warning(f" RG_Montant: {e}") # 5. Mode de règlement via l'objet Reglement try: # Lire le mode de règlement depuis la base mode_factory = getattr( self.cial.CptaApplication, "FactoryModeReglement", None ) if mode_factory: mode_obj = mode_factory.ReadNumero(mode_reglement) if mode_obj: reg.Reglement = mode_obj logger.info(f" Mode règlement défini: {mode_reglement}") except Exception as e: logger.debug(f" Mode règlement via factory: {e}") if reference: try: reg.RG_Reference = reference except Exception: pass if libelle: try: reg.RG_Libelle = libelle except Exception: pass try: reg.RG_Impute = 1 # Imputé except Exception: pass try: reg.RG_Compta = 0 # Non comptabilisé except Exception: pass # 6. ÉCRIRE le règlement reg.Write() numero = getattr(reg, "RG_Piece", None) logger.info(f" Règlement écrit avec numéro: {numero}") # 7. Créer le lien règlement-échéance via la factory DU RÈGLEMENT try: logger.info(" Création du lien règlement-échéance...") factory_reg_ech = getattr(reg, "FactoryDocumentReglementEcheance", None) if factory_reg_ech: reg_ech = factory_reg_ech.Create() # Cast vers IBODocumentReglementEcheance for iface in [ "IBODocumentReglementEcheance3", "IBODocumentReglementEcheance", ]: try: reg_ech = win32com.client.CastTo(reg_ech, iface) logger.info(f" Cast vers {iface} réussi") break except Exception: continue # Définir l'échéance - le Reglement est déjà lié via la factory try: reg_ech.Echeance = echeance logger.info(" Echeance définie") except Exception as e: logger.warning(f" Echeance: {e}") # Définir le montant try: reg_ech.RC_Montant = montant logger.info(f" RC_Montant: {montant}") except Exception as e: logger.warning(f" RC_Montant: {e}") # Écrire le lien try: reg_ech.SetDefault() except Exception: pass reg_ech.Write() logger.info(" Lien règlement-échéance écrit!") return str(numero) if numero else None except Exception as e: erreurs.append(f"Lien échéance: {e}") logger.warning(f" Erreur création lien: {e}") # Si le lien a échoué, essayer via le process logger.info(" Tentative via CreateProcess_ReglerEcheances...") try: process = self.cial.CreateProcess_ReglerEcheances() # Assigner le règlement déjà écrit process.Reglement = reg logger.info(" Règlement assigné au process") # Ajouter l'échéance try: process.AddDocumentEcheanceMontant(echeance, montant) logger.info(" Échéance ajoutée avec montant") except Exception: process.AddDocumentEcheance(echeance) logger.info(" Échéance ajoutée") can_process = getattr(process, "CanProcess", True) logger.info(f" CanProcess: {can_process}") if can_process: process.Process() logger.info(" Process() réussi!") return str(numero) if numero else None else: # Vérifier les erreurs try: errors = process.Errors if errors: for i in range(1, errors.Count + 1): logger.warning(f" Erreur [{i}]: {errors.Item(i)}") except Exception: pass except Exception as e: erreurs.append(f"Process: {e}") logger.warning(f" Process échoué: {e}") except Exception as e: erreurs.append(f"FactoryDocumentReglement: {e}") logger.error(f"FactoryDocumentReglement échoué: {e}") # APPROCHE ALTERNATIVE: Via le mode règlement de l'échéance try: logger.info("Tentative via modification directe de l'échéance...") # L'échéance a un attribut Reglement qui est le mode de règlement mode_obj = getattr(echeance, "Reglement", None) if mode_obj: attrs = [a for a in dir(mode_obj) if not a.startswith("_")] logger.info(f" Attributs Reglement échéance: {attrs[:15]}...") # Vérifier si l'échéance a FactoryDocumentReglementEcheance factory_reg_ech = getattr(echeance, "FactoryDocumentReglementEcheance", None) if factory_reg_ech: logger.info(" FactoryDocumentReglementEcheance trouvée sur échéance") # Créer le lien depuis l'échéance reg_ech = factory_reg_ech.Create() for iface in [ "IBODocumentReglementEcheance3", "IBODocumentReglementEcheance", ]: try: reg_ech = win32com.client.CastTo(reg_ech, iface) logger.info(f" Cast vers {iface}") break except Exception: continue # Ici, l'échéance devrait déjà être liée # Il faut définir le règlement try: # Créer un nouveau règlement pour ce lien factory_reg = self.cial.FactoryDocumentReglement new_reg = factory_reg.Create() new_reg = win32com.client.CastTo(new_reg, "IBODocumentReglement") # Configurer minimalement journal_factory = self.cial.CptaApplication.FactoryJournal journal_persist = journal_factory.ReadNumero(code_journal) if journal_persist: new_reg.Journal = journal_persist factory_client = self.cial.CptaApplication.FactoryClient if client_code: client_persist = factory_client.ReadNumero(client_code) if client_persist: new_reg.TiersPayeur = client_persist new_reg.RG_Date = pywintypes.Time(date_reglement) new_reg.RG_Montant = montant new_reg.RG_Impute = 1 # Écrire le règlement new_reg.Write() logger.info( f" Nouveau règlement créé: {getattr(new_reg, 'RG_Piece', None)}" ) # Assigner au lien - ici on doit peut-être utiliser un autre attribut # Puisque reg_ech.Reglement n'est pas settable, essayons via SetDefault try: reg_ech.SetDefault() except Exception: pass reg_ech.RC_Montant = montant reg_ech.Write() logger.info(" Lien créé depuis échéance!") return str(getattr(new_reg, "RG_Piece", None)) except Exception as e: logger.warning(f" Erreur: {e}") except Exception as e: erreurs.append(f"Via échéance: {e}") logger.warning(f"Via échéance échoué: {e}") raise RuntimeError(f"Aucune méthode n'a fonctionné. Erreurs: {'; '.join(erreurs)}") def introspecter_reglement(self): if not self.cial: raise RuntimeError("Connexion Sage non établie") result = {} try: with self._com_context(), self._lock_com: # IBODocumentReglement et sa factory de liens try: factory = self.cial.FactoryDocumentReglement reg = factory.Create() reg = win32com.client.CastTo(reg, "IBODocumentReglement") result["IBODocumentReglement"] = [ a for a in dir(reg) if not a.startswith("_") ] # FactoryDocumentReglementEcheance depuis le règlement factory_lien = getattr(reg, "FactoryDocumentReglementEcheance", None) if factory_lien: lien = factory_lien.Create() result["ReglementEcheance_base"] = [ a for a in dir(lien) if not a.startswith("_") ] for iface in [ "IBODocumentReglementEcheance3", "IBODocumentReglementEcheance", ]: try: lien_cast = win32com.client.CastTo(lien, iface) result[f"ReglementEcheance_{iface}"] = [ a for a in dir(lien_cast) if not a.startswith("_") ] except Exception as e: result[f"cast_{iface}_error"] = str(e) except Exception as e: result["error_reglement"] = str(e) # Process try: process = self.cial.CreateProcess_ReglerEcheances() result["Process"] = [a for a in dir(process) if not a.startswith("_")] except Exception as e: result["error_process"] = str(e) # Échéance et ses attributs try: factory_doc = self.cial.FactoryDocumentVente doc_list = factory_doc.List for i in range(1, 20): try: doc = doc_list.Item(i) if doc: doc.Read() if getattr(doc, "DO_Type", 0) == 6: factory_ech = getattr( doc, "FactoryDocumentEcheance", None ) if factory_ech: ech_list = factory_ech.List if ech_list: ech = ech_list.Item(1) if ech: ech = win32com.client.CastTo( ech, "IBODocumentEcheance3" ) ech.Read() result["IBODocumentEcheance3"] = [ a for a in dir(ech) if not a.startswith("_") ] # FactoryDocumentReglementEcheance depuis l'échéance factory_lien_ech = getattr( ech, "FactoryDocumentReglementEcheance", None, ) if factory_lien_ech: lien_ech = factory_lien_ech.Create() result[ "EcheanceReglementEcheance_base" ] = [ a for a in dir(lien_ech) if not a.startswith("_") ] for iface in [ "IBODocumentReglementEcheance3", "IBODocumentReglementEcheance", ]: try: lien_ech_cast = ( win32com.client.CastTo( lien_ech, iface ) ) result[ f"EcheanceReglementEcheance_{iface}" ] = [ a for a in dir(lien_ech_cast) if not a.startswith("_") ] except Exception: pass # Reglement de l'échéance (mode) mode = getattr(ech, "Reglement", None) if mode: result["Echeance_Reglement_mode"] = [ a for a in dir(mode) if not a.startswith("_") ] break except Exception: continue except Exception as e: result["error_echeance"] = str(e) except Exception as e: result["global_error"] = str(e) return result def regler_factures_client( self, client_code, montant_total, mode_reglement=ModeReglement.CHEQUE, date_reglement=None, reference="", libelle="", numeros_factures=None, ): if not self.cial: raise RuntimeError("Connexion Sage non établie") if montant_total <= 0: raise ValueError("Le montant total doit être positif") date_reglement = date_reglement or datetime.now() factures = _get_factures_non_soldees_client_sql(self, client_code, numeros_factures) if not factures: raise ValueError(f"Aucune facture à régler pour {client_code}") solde_total = sum(f["solde"] for f in factures) if montant_total > solde_total + 0.01: raise ValueError( f"Montant ({montant_total}€) supérieur au solde ({solde_total:.2f}€)" ) reglements = [] restant = montant_total for fac in factures: if restant < 0.01: break a_regler = min(restant, fac["solde"]) try: res = regler_facture( self, fac["numero"], a_regler, mode_reglement, date_reglement, reference, libelle, ) reglements.append(res) restant -= a_regler except Exception as e: logger.error(f"Erreur {fac['numero']}: {e}") break if not reglements: raise RuntimeError("Aucun règlement effectué") return { "client_code": client_code, "montant_demande": montant_total, "montant_effectif": sum(r["montant_regle"] for r in reglements), "nb_factures_reglees": len(reglements), "nb_factures_soldees": sum(1 for r in reglements if r["facture_soldee"]), "date_reglement": date_reglement.strftime("%Y-%m-%d"), "mode_reglement": mode_reglement, "mode_reglement_libelle": ModeReglement.get_libelle(mode_reglement), "reference": reference, "reglements": reglements, } def _get_factures_non_soldees_client_sql(self, client_code, numeros=None): with self._get_sql_connection() as conn: cursor = conn.cursor() if numeros: placeholders = ",".join("?" * len(numeros)) query = f"SELECT DO_Piece, DO_Date, DO_TotalTTC, DO_MontantRegle FROM F_DOCENTETE WHERE DO_Type = 6 AND DO_Tiers = ? AND DO_Piece IN ({placeholders}) AND DO_Statut <> 6 AND (DO_TotalTTC - ISNULL(DO_MontantRegle, 0)) > 0.01 ORDER BY DO_Date ASC" params = [client_code] + numeros else: query = "SELECT DO_Piece, DO_Date, DO_TotalTTC, DO_MontantRegle FROM F_DOCENTETE WHERE DO_Type = 6 AND DO_Tiers = ? AND DO_Statut <> 6 AND (DO_TotalTTC - ISNULL(DO_MontantRegle, 0)) > 0.01 ORDER BY DO_Date ASC" params = [client_code] cursor.execute(query, params) return [ { "numero": r[0].strip(), "date": r[1].strftime("%Y-%m-%d") if r[1] else None, "total_ttc": float(r[2] or 0), "montant_regle": float(r[3] or 0), "solde": float(r[2] or 0) - float(r[3] or 0), } for r in cursor.fetchall() ] def lire_reglements_facture(self, numero_facture): if not self.cial: raise RuntimeError("Connexion Sage non établie") with self._get_sql_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT DO_TotalTTC, DO_MontantRegle, DO_Tiers, DO_Date, DO_Ref FROM F_DOCENTETE WHERE DO_Piece = ? AND DO_Type = 6", (numero_facture,), ) row = cursor.fetchone() if not row: raise ValueError(f"Facture {numero_facture} introuvable") total = float(row[0] or 0) regle = float(row[1] or 0) solde = max(0, total - regle) return { "numero_facture": numero_facture, "client_code": (row[2] or "").strip(), "date_facture": row[3].strftime("%Y-%m-%d") if row[3] else None, "reference": (row[4] or "").strip(), "total_ttc": total, "total_regle": regle, "solde_restant": solde, "est_soldee": solde < 0.01, } def lire_reglements_client( self, client_code, date_debut=None, date_fin=None, inclure_soldees=True ): if not self.cial: raise RuntimeError("Connexion Sage non établie") with self._get_sql_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT CT_Intitule FROM F_COMPTET WHERE CT_Num = ?", (client_code,) ) row = cursor.fetchone() if not row: raise ValueError(f"Client {client_code} introuvable") intitule = (row[0] or "").strip() query = "SELECT DO_Piece, DO_Date, DO_TotalTTC, DO_MontantRegle, DO_Ref FROM F_DOCENTETE WHERE DO_Type = 6 AND DO_Tiers = ?" params = [client_code] if date_debut: query += " AND DO_Date >= ?" params.append(date_debut) if date_fin: query += " AND DO_Date <= ?" params.append(date_fin) query += " ORDER BY DO_Date ASC" cursor.execute(query, params) factures = [] for r in cursor.fetchall(): total = float(r[2] or 0) regle = float(r[3] or 0) solde = max(0, total - regle) soldee = solde < 0.01 if inclure_soldees or not soldee: factures.append( { "numero_facture": r[0].strip(), "date_facture": r[1].strftime("%Y-%m-%d") if r[1] else None, "total_ttc": total, "reference": (r[4] or "").strip(), "total_regle": regle, "solde_restant": solde, "est_soldee": soldee, } ) return { "client_code": client_code, "client_intitule": intitule, "nb_factures": len(factures), "nb_factures_soldees": sum(1 for f in factures if f["est_soldee"]), "nb_factures_en_cours": sum(1 for f in factures if not f["est_soldee"]), "total_factures": sum(f["total_ttc"] for f in factures), "total_regle": sum(f["total_regle"] for f in factures), "solde_global": sum(f["solde_restant"] for f in factures), "factures": factures, } __all__ = [ "ModeReglement", "lire_journaux_banque", "lire_tous_journaux", "introspecter_reglement", "regler_facture", "regler_factures_client", "lire_reglements_facture", "lire_reglements_client", ]