from typing import Dict import win32com.client import pywintypes import time from schemas.documents.doc_config import TypeDocumentVente, ConfigDocument import logging from utils.functions.functions import normaliser_date from utils.tiers.clients.clients_data import _cast_client logger = logging.getLogger(__name__) def creer_document_vente( self, doc_data: dict, type_document: TypeDocumentVente ) -> Dict: if not self.cial: raise RuntimeError("Connexion Sage non établie") config = ConfigDocument(type_document) logger.info( f"📝 Début création {config.nom_document} pour client {doc_data['client']['code']}" ) try: with self._com_context(), self._lock_com: transaction_active = False try: # Démarrage transaction try: self.cial.CptaApplication.BeginTrans() transaction_active = True logger.debug("✓ Transaction Sage démarrée") except Exception as e: logger.warning(f"BeginTrans échoué (non critique): {e}") # Création du document process = self.cial.CreateProcess_Document(config.type_sage) doc = process.Document try: doc = win32com.client.CastTo(doc, "IBODocumentVente3") except Exception: pass logger.info(f"✓ Document {config.nom_document} créé") # ===== DATES ===== doc.DO_Date = pywintypes.Time( normaliser_date(doc_data.get(config.champ_date_principale)) ) # Date secondaire (livraison, etc.) if config.champ_date_secondaire and doc_data.get( config.champ_date_secondaire ): doc.DO_DateLivr = pywintypes.Time( normaliser_date(doc_data[config.champ_date_secondaire]) ) logger.info( f"✓ {config.champ_date_secondaire}: {doc_data[config.champ_date_secondaire]}" ) # ===== CLIENT ===== factory_client = self.cial.CptaApplication.FactoryClient persist_client = factory_client.ReadNumero(doc_data["client"]["code"]) if not persist_client: raise ValueError(f"Client {doc_data['client']['code']} introuvable") client_obj = _cast_client(persist_client) if not client_obj: raise ValueError("Impossible de charger le client") doc.SetDefaultClient(client_obj) doc.Write() logger.info(f"✓ Client {doc_data['client']['code']} associé") # ===== RÉFÉRENCE ===== if doc_data.get("reference"): try: doc.DO_Ref = doc_data["reference"] logger.info(f"✓ Référence: {doc_data['reference']}") except Exception as e: logger.warning(f"Référence non définie: {e}") # ===== CONFIGURATION SPÉCIFIQUE FACTURE ===== if type_document == TypeDocumentVente.FACTURE: _configurer_facture(self, doc) # ===== FACTORY LIGNES ===== try: factory_lignes = doc.FactoryDocumentLigne except Exception: factory_lignes = doc.FactoryDocumentVenteLigne factory_article = self.cial.FactoryArticle logger.info(f"📦 Ajout de {len(doc_data['lignes'])} lignes...") # ===== TRAITEMENT DES LIGNES ===== for idx, ligne_data in enumerate(doc_data["lignes"], 1): _ajouter_ligne_document( cial=self.cial, factory_lignes=factory_lignes, factory_article=factory_article, ligne_data=ligne_data, idx=idx, doc=doc, ) # ===== VALIDATION ===== logger.info("💾 Validation du document...") # Pour les factures, réassocier le client avant validation if type_document == TypeDocumentVente.FACTURE: try: doc.SetClient(client_obj) logger.debug(" ↳ Client réassocié avant validation") except Exception: try: doc.SetDefaultClient(client_obj) except Exception: pass doc.Write() # Process() sauf pour devis en brouillon if type_document != TypeDocumentVente.DEVIS: process.Process() logger.info("✓ Process() appelé") else: try: process.Process() logger.info("✓ Process() appelé (devis)") except Exception: logger.debug(" ↳ Process() ignoré pour devis brouillon") # Commit transaction if transaction_active: try: self.cial.CptaApplication.CommitTrans() logger.info("✓ Transaction committée") except Exception: pass time.sleep(2) # ===== RÉCUPÉRATION DU NUMÉRO ===== numero_document = _recuperer_numero_document(process, doc) if not numero_document: raise RuntimeError( f"Numéro {config.nom_document} vide après création" ) logger.info(f"📄 Numéro: {numero_document}") # ===== RELECTURE POUR TOTAUX ===== doc_final_data = _relire_document_final( self, config=config, numero_document=numero_document, doc_data=doc_data, ) logger.info( f"✅ {config.nom_document.upper()} CRÉÉ: " f"{numero_document} - {doc_final_data['total_ttc']}€ TTC" ) return doc_final_data except Exception: if transaction_active: try: self.cial.CptaApplication.RollbackTrans() logger.error(" Transaction annulée (rollback)") except Exception: pass raise except Exception as e: logger.error( f" ERREUR CRÉATION {config.nom_document.upper()}: {e}", exc_info=True ) raise RuntimeError(f"Échec création {config.nom_document}: {str(e)}") def _appliquer_remise_ligne(ligne_obj, remise_pourcent: float) -> bool: """Applique la remise via FromString - SOLUTION FINALE""" try: import pythoncom dispatch = ligne_obj._oleobj_ # 1. Récupérer l'objet Remise dispid = dispatch.GetIDsOfNames(0, "Remise") remise_obj = dispatch.Invoke(dispid, 0, pythoncom.DISPATCH_PROPERTYGET, 1) remise_wrapper = win32com.client.Dispatch(remise_obj) # 2. Définir la remise via FromString remise_wrapper.FromString(f"{remise_pourcent}%") # 3. Calcul (optionnel mais recommandé) try: remise_wrapper.Calcul() except Exception: pass # 4. Write la ligne ligne_obj.Write() logger.info(f" ✅ Remise {remise_pourcent}% appliquée") return True except Exception as e: logger.error(f" ❌ Erreur remise: {e}") return False def _ajouter_ligne_document( cial, factory_lignes, factory_article, ligne_data: dict, idx: int, doc ) -> None: """VERSION FINALE AVEC REMISES FONCTIONNELLES""" logger.info(f" ├─ Ligne {idx}: {ligne_data['article_code']}") # ===== CRÉATION LIGNE ===== persist_article = factory_article.ReadReference(ligne_data["article_code"]) if not persist_article: raise ValueError(f"Article {ligne_data['article_code']} introuvable") article_obj = win32com.client.CastTo(persist_article, "IBOArticle3") article_obj.Read() prix_sage = float(getattr(article_obj, "AR_PrixVen", 0.0)) designation_sage = getattr(article_obj, "AR_Design", "") ligne_persist = factory_lignes.Create() try: ligne_obj = win32com.client.CastTo(ligne_persist, "IBODocumentLigne3") except Exception: ligne_obj = win32com.client.CastTo(ligne_persist, "IBODocumentVenteLigne3") quantite = float(ligne_data["quantite"]) # ===== ASSOCIATION ARTICLE ===== try: ligne_obj.SetDefaultArticleReference(ligne_data["article_code"], quantite) except Exception: try: ligne_obj.SetDefaultArticle(article_obj, quantite) except Exception: ligne_obj.DL_Design = designation_sage ligne_obj.DL_Qte = quantite # ===== PRIX ===== prix_auto = float(getattr(ligne_obj, "DL_PrixUnitaire", 0.0)) prix_perso = ligne_data.get("prix_unitaire_ht") if prix_perso and prix_perso > 0: ligne_obj.DL_PrixUnitaire = float(prix_perso) elif prix_auto == 0 and prix_sage > 0: ligne_obj.DL_PrixUnitaire = float(prix_sage) prix_final = float(getattr(ligne_obj, "DL_PrixUnitaire", 0)) logger.info(f" 💰 Prix: {prix_final}€") # ===== WRITE INITIAL ===== ligne_obj.Write() # ===== APPLICATION REMISE (TOUTES LES LIGNES!) ===== remise = ligne_data.get("remise_pourcentage", 0) if remise and remise > 0: logger.info(f" 🎯 Application remise {remise}%...") _appliquer_remise_ligne(ligne_obj, remise) logger.info(f" ✓ Ligne {idx} terminée") def _configurer_facture(self, doc) -> None: """Configuration spécifique pour les factures""" logger.info(" 🔧 Configuration spécifique facture...") try: if hasattr(doc, "DO_CodeJournal"): try: param_societe = self.cial.CptaApplication.ParametreSociete journal_defaut = getattr(param_societe, "P_CodeJournalVte", "VTE") doc.DO_CodeJournal = journal_defaut logger.debug(f" ✓ Code journal: {journal_defaut}") except Exception: doc.DO_CodeJournal = "VTE" logger.debug(" ✓ Code journal: VTE (défaut)") except Exception as e: logger.debug(f" Code journal: {e}") try: if hasattr(doc, "DO_Souche"): doc.DO_Souche = 0 logger.debug(" ✓ Souche: 0") except Exception: pass try: if hasattr(doc, "DO_Regime"): doc.DO_Regime = 0 logger.debug(" ✓ Régime: 0") except Exception: pass def _recuperer_numero_document(process, doc) -> str: """Récupère le numéro du document créé""" numero = None try: doc_result = process.DocumentResult if doc_result: doc_result = win32com.client.CastTo(doc_result, "IBODocumentVente3") doc_result.Read() numero = getattr(doc_result, "DO_Piece", "") except Exception: pass if not numero: numero = getattr(doc, "DO_Piece", "") return numero def _relire_document_final( self, config: ConfigDocument, numero_document: str, doc_data: dict ) -> Dict: """ Relit le document pour obtenir les totaux calculés par Sage """ factory_doc = self.cial.FactoryDocumentVente persist_reread = factory_doc.ReadPiece(config.type_sage, numero_document) if persist_reread: doc_final = win32com.client.CastTo(persist_reread, "IBODocumentVente3") doc_final.Read() total_ht = float(getattr(doc_final, "DO_TotalHT", 0.0)) total_ttc = float(getattr(doc_final, "DO_TotalTTC", 0.0)) reference_finale = getattr(doc_final, "DO_Ref", "") # Date secondaire date_secondaire_value = None if config.champ_date_secondaire: try: date_livr = getattr(doc_final, "DO_DateLivr", None) if date_livr: date_secondaire_value = date_livr.strftime("%Y-%m-%d") except Exception: pass else: # Valeurs par défaut si relecture échoue total_ht = 0.0 total_ttc = 0.0 reference_finale = doc_data.get("reference", "") date_secondaire_value = doc_data.get(config.champ_date_secondaire) # Construction du résultat resultat = { config.champ_numero: numero_document, "total_ht": total_ht, "total_ttc": total_ttc, "nb_lignes": len(doc_data["lignes"]), "client_code": doc_data["client"]["code"], config.champ_date_principale: str( normaliser_date(doc_data.get(config.champ_date_principale)) ), "reference": reference_finale, } # Ajout date secondaire si applicable if config.champ_date_secondaire: resultat[config.champ_date_secondaire] = date_secondaire_value return resultat __all__ = [ "creer_document_vente", "_ajouter_ligne_document", "_configurer_facture", "_recuperer_numero_document", "_relire_document_final", ]