From 7d58a607f54d1f145f2e755d6b394aa9e404911f Mon Sep 17 00:00:00 2001 From: fanilo Date: Thu, 15 Jan 2026 09:42:11 +0100 Subject: [PATCH] Trying again --- utils/documents/validations.py | 501 +++++++++++++++------------------ 1 file changed, 234 insertions(+), 267 deletions(-) diff --git a/utils/documents/validations.py b/utils/documents/validations.py index 8062704..90caa2b 100644 --- a/utils/documents/validations.py +++ b/utils/documents/validations.py @@ -2,15 +2,22 @@ Validation de factures Sage 100c Module: utils/documents/validation.py (Windows Server) -COM pour écriture, SQL pour lecture +COM pour écriture (late binding), SQL pour lecture +Solution: Contourner le cache gen_py avec late binding pur """ from typing import Dict import win32com.client +import pythoncom import logging logger = logging.getLogger(__name__) +# Constantes COM pour Invoke +DISPATCH_PROPERTYPUT = 4 +DISPATCH_PROPERTYGET = 2 +DISPATCH_METHOD = 1 + # ============================================================ # FONCTIONS SQL (LECTURE) @@ -101,156 +108,202 @@ def _get_facture_info_sql(connector, numero_facture: str) -> Dict: # ============================================================ -# INTROSPECTION - Pour découvrir les méthodes de validation +# LATE BINDING - Contourner le cache gen_py # ============================================================ -def introspecter_validation(connector, numero_facture: str = None) -> Dict: +def _get_dynamic_dispatch(com_object): """ - Introspection pour découvrir les méthodes de validation disponibles dans Sage. - - Appeler cette fonction pour voir quelles méthodes/process sont disponibles - pour valider un document. + Convertit un objet COM early-binding en late-binding dynamique. + Cela contourne le cache gen_py qui marque certaines propriétés en read-only. """ - if not connector.cial: - raise RuntimeError("Connexion Sage non établie") - - result = {} - try: - with connector._com_context(), connector._lock_com: - # 1. Méthodes sur cial (BSCIALApplication100c) - cial_attrs = [a for a in dir(connector.cial) if not a.startswith("_")] + # Récupérer l'objet COM brut et créer un dispatch dynamique + oleobj = com_object._oleobj_ + return win32com.client.Dispatch(oleobj.QueryInterface(pythoncom.IID_IDispatch)) + except Exception: + return com_object - # Filtrer les CreateProcess et méthodes liées à la validation - result["cial_createprocess"] = [ - a - for a in cial_attrs - if "Process" in a or "Valider" in a or "Valid" in a - ] - # 2. Si un numéro de facture est fourni, inspecter le document - if numero_facture: - factory = connector.cial.FactoryDocumentVente - if factory.ExistPiece(60, numero_facture): - persist = factory.ReadPiece(60, numero_facture) +def _set_property_via_invoke(com_object, prop_name: str, value) -> bool: + """ + Définit une propriété COM via Invoke direct (DISPATCH_PROPERTYPUT). + Contourne les restrictions du cache gen_py. + """ + try: + oleobj = com_object._oleobj_ + # Récupérer le DISPID de la propriété + dispid = oleobj.GetIDsOfNames(0, prop_name) + if isinstance(dispid, tuple): + dispid = dispid[0] - # Attributs du persist brut - persist_attrs = [a for a in dir(persist) if not a.startswith("_")] - result["persist_attrs_validation"] = [ - a - for a in persist_attrs - if "valide" in a.lower() - or "valid" in a.lower() - or "lock" in a.lower() - or "statut" in a.lower() - ] - result["persist_attrs_all"] = persist_attrs + # Invoke avec PROPERTYPUT + oleobj.Invoke(dispid, 0, DISPATCH_PROPERTYPUT, True, value) + return True + except Exception as e: + logger.debug(f" Invoke PROPERTYPUT échoué pour {prop_name}: {e}") + return False - # Cast vers IBODocumentVente3 - try: - doc = win32com.client.CastTo(persist, "IBODocumentVente3") - doc.Read() - doc_attrs = [a for a in dir(doc) if not a.startswith("_")] - # Attributs liés à la validation - result["doc_attrs_validation"] = [ - a - for a in doc_attrs - if "valide" in a.lower() - or "valid" in a.lower() - or "lock" in a.lower() - or "statut" in a.lower() - ] +def _call_method_via_invoke(com_object, method_name: str) -> bool: + """ + Appelle une méthode COM via Invoke direct. + """ + try: + oleobj = com_object._oleobj_ + dispid = oleobj.GetIDsOfNames(0, method_name) + if isinstance(dispid, tuple): + dispid = dispid[0] - # Toutes les méthodes callable - result["doc_methods"] = [ - a - for a in doc_attrs - if callable(getattr(doc, a, None)) and not a.startswith("_") - ] + oleobj.Invoke(dispid, 0, DISPATCH_METHOD, True) + return True + except Exception as e: + logger.debug(f" Invoke METHOD échoué pour {method_name}: {e}") + return False - # Valeur actuelle de DO_Valide et DO_Statut - try: - result["DO_Valide_current"] = getattr( - doc, "DO_Valide", "NOT_FOUND" - ) - except Exception as e: - result["DO_Valide_error"] = str(e) - try: - result["DO_Statut_current"] = getattr( - doc, "DO_Statut", "NOT_FOUND" - ) - except Exception as e: - result["DO_Statut_error"] = str(e) +# ============================================================ +# FONCTIONS COM (ÉCRITURE) avec late binding +# ============================================================ - except Exception as e: - result["cast_error"] = str(e) - # Essayer IBPersistDocument - try: - persist_doc = win32com.client.CastTo( - persist, "IBPersistDocument" - ) - persist_doc_attrs = [ - a for a in dir(persist_doc) if not a.startswith("_") - ] - result["IBPersistDocument_attrs"] = persist_doc_attrs - except Exception as e: - result["IBPersistDocument_error"] = str(e) +def _valider_document_com(connector, numero_facture: str, valider: bool = True) -> bool: + """ + Valide ou dévalide un document via COM en late binding. - # 3. Explorer chaque CreateProcess pertinent - for process_name in result.get("cial_createprocess", []): - if process_name.startswith("CreateProcess"): - try: - process = getattr(connector.cial, process_name)() - process_attrs = [ - a for a in dir(process) if not a.startswith("_") - ] - result[f"{process_name}_attrs"] = process_attrs - except Exception as e: - result[f"{process_name}_error"] = str(e) + Utilise Invoke direct pour contourner les restrictions gen_py. + """ + erreurs = [] + valeur_cible = 1 if valider else 0 + valeur_bool = True if valider else False + action = "validation" if valider else "dévalidation" - # 4. Chercher spécifiquement CreateProcess_ValiderBon ou similaire - validation_processes = [ - "CreateProcess_ValiderBon", - "CreateProcess_ValiderDocument", - "CreateProcess_ValiderPiece", - "CreateProcess_Validation", - ] - for vp in validation_processes: - if vp not in result: - try: - process = getattr(connector.cial, vp)() - process_attrs = [ - a for a in dir(process) if not a.startswith("_") - ] - result[f"{vp}_attrs"] = process_attrs - result[f"{vp}_exists"] = True - except AttributeError: - result[f"{vp}_exists"] = False - except Exception as e: - result[f"{vp}_error"] = str(e) + factory = connector.cial.FactoryDocumentVente + if not factory.ExistPiece(60, numero_facture): + raise ValueError(f"Facture {numero_facture} introuvable") + + persist = factory.ReadPiece(60, numero_facture) + if not persist: + raise ValueError(f"Impossible de lire la facture {numero_facture}") + + # APPROCHE 1: Late binding via Dispatch dynamique + try: + logger.info(" Tentative late binding dynamique...") + + # Convertir en dispatch dynamique (late binding) + doc_dyn = _get_dynamic_dispatch(persist) + doc_dyn.Read() + + # Lire la valeur actuelle + current = doc_dyn.DO_Valide + logger.info(" DO_Valide actuel (late): {current}") + + # Modifier + doc_dyn.DO_Valide = valeur_bool + doc_dyn.Write() + + logger.info(" Late binding dynamique réussi!") + return True except Exception as e: - result["global_error"] = str(e) + erreurs.append(f"Late binding dynamique: {e}") + logger.debug(f" Late binding dynamique échoué: {e}") - return result + # APPROCHE 2: Invoke direct avec PROPERTYPUT + try: + logger.info(" Tentative Invoke PROPERTYPUT...") + # Cast puis invoke + doc = win32com.client.CastTo(persist, "IBODocumentVente3") + doc.Read() -# ============================================================ -# FONCTIONS COM (ÉCRITURE) - À adapter selon l'introspection -# ============================================================ + # Essayer avec valeur bool + success = _set_property_via_invoke(doc, "DO_Valide", valeur_bool) + if success: + _call_method_via_invoke(doc, "Write") + logger.info(" Invoke PROPERTYPUT (bool) réussi!") + return True + + # Essayer avec valeur int + success = _set_property_via_invoke(doc, "DO_Valide", valeur_cible) + if success: + _call_method_via_invoke(doc, "Write") + logger.info(" Invoke PROPERTYPUT (int) réussi!") + return True + + erreurs.append("Invoke PROPERTYPUT échoué pour bool et int") + + except Exception as e: + erreurs.append(f"Invoke PROPERTYPUT: {e}") + logger.debug(f" Invoke PROPERTYPUT échoué: {e}") + + # APPROCHE 3: Dispatch frais sans cache + try: + logger.info(" Tentative Dispatch frais (sans cache)...") + + # Créer un nouveau Dispatch sans utiliser le cache gen_py + # En passant par l'IDispatch de l'objet + oleobj = persist._oleobj_ + idisp = oleobj.QueryInterface(pythoncom.IID_IDispatch) + + # Créer un wrapper dynamique + doc_fresh = win32com.client.dynamic.Dispatch(idisp) + doc_fresh.Read() + + logger.info(" DO_Valide actuel: {doc_fresh.DO_Valide}") + doc_fresh.DO_Valide = valeur_bool + doc_fresh.Write() + + logger.info(" Dispatch frais réussi!") + return True + + except Exception as e: + erreurs.append(f"Dispatch frais: {e}") + logger.debug(f" Dispatch frais échoué: {e}") + + # APPROCHE 4: Via IBPersistDocument + try: + logger.info(" Tentative via IBPersistDocument...") + + doc = win32com.client.CastTo(persist, "IBPersistDocument") + doc.Read() + + doc_dyn = _get_dynamic_dispatch(doc) + doc_dyn.DO_Valide = valeur_bool + doc_dyn.Write() + + logger.info(" IBPersistDocument réussi!") + return True + + except Exception as e: + erreurs.append(f"IBPersistDocument: {e}") + logger.debug(f" IBPersistDocument échoué: {e}") + + # APPROCHE 5: Accès direct sans Cast + try: + logger.info(" Tentative accès direct sans Cast...") + + persist.Read() + persist_dyn = _get_dynamic_dispatch(persist) + + logger.info(" DO_Valide actuel: {persist_dyn.DO_Valide}") + persist_dyn.DO_Valide = valeur_bool + persist_dyn.Write() + + logger.info(" Accès direct sans Cast réussi!") + return True + + except Exception as e: + erreurs.append(f"Direct sans Cast: {e}") + logger.debug(f" Direct sans Cast échoué: {e}") + + logger.error(f" Toutes les approches de {action} ont échoué: {erreurs}") + raise RuntimeError(f"Échec {action}: {'; '.join(erreurs[:3])}") def valider_facture(connector, numero_facture: str) -> Dict: """ Valide une facture (pose le cadenas) - - - Vérifications: SQL - - Modification: COM (via Process ou méthode appropriée) - - Réponse: SQL """ if not connector.cial: raise RuntimeError("Connexion Sage non établie") @@ -271,7 +324,7 @@ def valider_facture(connector, numero_facture: str) -> Dict: connector, numero_facture, deja_valide=True, action="validation" ) - # 2. Modification via COM + # 2. Modification via COM (late binding) try: with connector._com_context(), connector._lock_com: success = _valider_document_com(connector, numero_facture, valider=True) @@ -301,10 +354,6 @@ def valider_facture(connector, numero_facture: str) -> Dict: def devalider_facture(connector, numero_facture: str) -> Dict: """ Dévalide une facture (retire le cadenas) - - - Vérifications: SQL - - Modification: COM (via Process ou méthode appropriée) - - Réponse: SQL """ if not connector.cial: raise RuntimeError("Connexion Sage non établie") @@ -336,7 +385,7 @@ def devalider_facture(connector, numero_facture: str) -> Dict: connector, numero_facture, deja_valide=True, action="devalidation" ) - # 2. Modification via COM + # 2. Modification via COM (late binding) try: with connector._com_context(), connector._lock_com: success = _valider_document_com(connector, numero_facture, valider=False) @@ -363,157 +412,75 @@ def devalider_facture(connector, numero_facture: str) -> Dict: ) -def _valider_document_com(connector, numero_facture: str, valider: bool = True) -> bool: +# ============================================================ +# INTROSPECTION +# ============================================================ + + +def introspecter_validation(connector, numero_facture: str = None) -> Dict: """ - Valide ou dévalide un document via COM. - - Essaie plusieurs approches dans l'ordre: - 1. CreateProcess_ValiderBon (si disponible) - 2. Méthode Valider() sur le document - 3. SetFieldValue sur DO_Valide - 4. Accès direct à DO_Valide - - Retourne True si réussi. + Introspection pour découvrir les méthodes de validation disponibles. """ - erreurs = [] - valeur_cible = 1 if valider else 0 - action = "validation" if valider else "dévalidation" + if not connector.cial: + raise RuntimeError("Connexion Sage non établie") - factory = connector.cial.FactoryDocumentVente - if not factory.ExistPiece(60, numero_facture): - raise ValueError(f"Facture {numero_facture} introuvable") + result = {} - persist = factory.ReadPiece(60, numero_facture) - if not persist: - raise ValueError(f"Impossible de lire la facture {numero_facture}") - - # APPROCHE 1: CreateProcess_ValiderBon try: - process = connector.cial.CreateProcess_ValiderBon() - logger.info(f" Tentative via CreateProcess_ValiderBon...") + with connector._com_context(), connector._lock_com: + cial_attrs = [a for a in dir(connector.cial) if not a.startswith("_")] + result["cial_createprocess"] = [ + a + for a in cial_attrs + if "Process" in a or "Valider" in a or "Valid" in a + ] - # Ajouter le document au process - process.Document = persist + if numero_facture: + factory = connector.cial.FactoryDocumentVente + if factory.ExistPiece(60, numero_facture): + persist = factory.ReadPiece(60, numero_facture) - # Définir l'action (valider ou dévalider) - try: - process.Valider = valider - except Exception: - pass + # Test late binding + try: + doc_dyn = _get_dynamic_dispatch(persist) + doc_dyn.Read() - # Vérifier si on peut traiter - can_process = getattr(process, "CanProcess", True) - if can_process: - process.Process() - logger.info(f" CreateProcess_ValiderBon.Process() réussi") - return True - else: - # Lire les erreurs - try: - errors = process.Errors - if errors and errors.Count > 0: - for i in range(1, errors.Count + 1): - erreurs.append(f"ValiderBon error: {errors.Item(i)}") - except Exception: - pass + # Lister les attributs via late binding + result["late_binding_test"] = { + "DO_Valide_readable": True, + "DO_Valide_value": doc_dyn.DO_Valide, + } - except AttributeError: - erreurs.append("CreateProcess_ValiderBon n'existe pas") - except Exception as e: - erreurs.append(f"CreateProcess_ValiderBon: {e}") + # Tester si writable + try: + original = doc_dyn.DO_Valide + doc_dyn.DO_Valide = ( + original # Essayer de réécrire la même valeur + ) + result["late_binding_test"]["DO_Valide_writable"] = True + except Exception as e: + result["late_binding_test"]["DO_Valide_writable"] = False + result["late_binding_test"]["write_error"] = str(e) - # APPROCHE 2: Méthode Valider() ou Devalider() sur le document - try: - doc = win32com.client.CastTo(persist, "IBODocumentVente3") - doc.Read() + except Exception as e: + result["late_binding_error"] = str(e) - if valider: - method_names = ["Valider", "Validate", "Lock", "SetValide"] - else: - method_names = ["Devalider", "Devalidate", "Unlock", "SetDevalide"] - - for method_name in method_names: - try: - method = getattr(doc, method_name, None) - if method and callable(method): - logger.info(f" Tentative via doc.{method_name}()...") - method() - doc.Write() - logger.info(f" doc.{method_name}() réussi") - return True - except Exception as e: - erreurs.append(f"doc.{method_name}: {e}") + # Test des process + for process_name in result.get("cial_createprocess", []): + if process_name.startswith("CreateProcess"): + try: + process = getattr(connector.cial, process_name)() + process_attrs = [ + a for a in dir(process) if not a.startswith("_") + ] + result[f"{process_name}_attrs"] = process_attrs + except Exception as e: + result[f"{process_name}_error"] = str(e) except Exception as e: - erreurs.append(f"Cast IBODocumentVente3: {e}") + result["global_error"] = str(e) - # APPROCHE 3: SetFieldValue (méthode générique Sage) - try: - doc = win32com.client.CastTo(persist, "IBODocumentVente3") - doc.Read() - - try: - logger.info( - f" Tentative via SetFieldValue('DO_Valide', {valeur_cible})..." - ) - doc.SetFieldValue("DO_Valide", valeur_cible) - doc.Write() - logger.info(f" SetFieldValue réussi") - return True - except Exception as e: - erreurs.append(f"SetFieldValue: {e}") - - except Exception as e: - erreurs.append(f"SetFieldValue cast: {e}") - - # APPROCHE 4: Accès direct via différentes interfaces - interfaces = [ - "IBODocumentVente3", - "IBODocumentVente2", - "IBODocumentVente", - "IBODocument3", - ] - - for iface in interfaces: - try: - doc = win32com.client.CastTo(persist, iface) - doc.Read() - - logger.info(f" Tentative accès direct DO_Valide via {iface}...") - - # Essayer d'accéder et modifier - try: - current_val = doc.DO_Valide - logger.info(f" Valeur actuelle: {current_val}") - doc.DO_Valide = valeur_cible - doc.Write() - logger.info(f" Accès direct via {iface} réussi") - return True - except AttributeError as e: - erreurs.append(f"{iface} DO_Valide AttributeError: {e}") - except Exception as e: - erreurs.append(f"{iface} DO_Valide: {e}") - - except Exception as e: - erreurs.append(f"Cast {iface}: {e}") - - # APPROCHE 5: Via IPMDocument si disponible - try: - doc = win32com.client.CastTo(persist, "IPMDocument") - logger.info(f" Tentative via IPMDocument...") - - if valider: - doc.Valider() - else: - doc.Devalider() - doc.Write() - logger.info(f" IPMDocument.{'Valider' if valider else 'Devalider'}() réussi") - return True - except Exception as e: - erreurs.append(f"IPMDocument: {e}") - - logger.error(f" Toutes les approches de {action} ont échoué: {erreurs}") - raise RuntimeError(f"Échec {action}: {'; '.join(erreurs[:5])}") + return result # ============================================================ @@ -524,7 +491,7 @@ def _valider_document_com(connector, numero_facture: str, valider: bool = True) def _build_response_sql( connector, numero_facture: str, deja_valide: bool, action: str ) -> Dict: - """Construit la réponse via SQL (pas COM)""" + """Construit la réponse via SQL""" info = _get_facture_info_sql(connector, numero_facture) if action == "validation":