""" Validation de factures Sage 100c Module: utils/documents/validation.py (Windows Server) 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) # ============================================================ def get_statut_validation(connector, numero_facture: str) -> Dict: """ Retourne le statut de validation d'une facture (SQL) """ with connector._get_sql_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT DO_Valide, DO_Statut, DO_TotalHT, DO_TotalTTC, ISNULL(DO_MontantRegle, 0), CT_NumPayeur, 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") valide = int(row[0]) if row[0] is not None else 0 statut = int(row[1]) if row[1] is not None else 0 total_ht = float(row[2]) if row[2] else 0.0 total_ttc = float(row[3]) if row[3] else 0.0 montant_regle = float(row[4]) if row[4] else 0.0 client_code = row[5].strip() if row[5] else "" date_facture = row[6] reference = row[7].strip() if row[7] else "" solde = max(0.0, total_ttc - montant_regle) return { "numero_facture": numero_facture, "est_validee": valide == 1, "statut": statut, "statut_libelle": _get_statut_libelle(statut), "peut_etre_modifiee": valide == 0 and statut not in (5, 6), "peut_etre_devalidee": valide == 1 and montant_regle < 0.01 and statut not in (5, 6), "total_ht": total_ht, "total_ttc": total_ttc, "montant_regle": montant_regle, "solde_restant": solde, "client_code": client_code, "date_facture": date_facture.strftime("%Y-%m-%d") if date_facture else None, "reference": reference, } def _get_facture_info_sql(connector, numero_facture: str) -> Dict: """ Récupère les infos d'une facture via SQL (pour vérifications préalables) """ with connector._get_sql_connection() as conn: cursor = conn.cursor() cursor.execute( """ SELECT DO_Valide, DO_Statut, DO_TotalHT, DO_TotalTTC, ISNULL(DO_MontantRegle, 0) as MontantRegle, CT_NumPayeur FROM F_DOCENTETE WHERE DO_Piece = ? AND DO_Type = 6 """, (numero_facture,), ) row = cursor.fetchone() if not row: return None return { "valide": int(row[0]) if row[0] is not None else 0, "statut": int(row[1]) if row[1] is not None else 0, "total_ht": float(row[2]) if row[2] else 0.0, "total_ttc": float(row[3]) if row[3] else 0.0, "montant_regle": float(row[4]) if row[4] else 0.0, "client_code": row[5].strip() if row[5] else "", } # ============================================================ # LATE BINDING - Contourner le cache gen_py # ============================================================ def _get_dynamic_dispatch(com_object): """ 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. """ try: # 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 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] # 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 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] 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 # ============================================================ # FONCTIONS COM (ÉCRITURE) avec late binding # ============================================================ def _valider_document_com(connector, numero_facture: str, valider: bool = True) -> bool: """ Valide ou dévalide un document via COM en late binding. 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" 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: erreurs.append(f"Late binding dynamique: {e}") logger.debug(f" Late binding dynamique échoué: {e}") # APPROCHE 2: Invoke direct avec PROPERTYPUT try: logger.info(" Tentative Invoke PROPERTYPUT...") # Cast puis invoke doc = win32com.client.CastTo(persist, "IBODocumentVente3") doc.Read() # 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) """ if not connector.cial: raise RuntimeError("Connexion Sage non établie") logger.info(f"🔒 Validation facture {numero_facture}") # 1. Vérifications préalables via SQL info = _get_facture_info_sql(connector, numero_facture) if not info: raise ValueError(f"Facture {numero_facture} introuvable") if info["statut"] == 6: raise ValueError(f"Facture {numero_facture} annulée, validation impossible") if info["valide"] == 1: logger.info(f"Facture {numero_facture} déjà validée") return _build_response_sql( connector, numero_facture, deja_valide=True, action="validation" ) # 2. Modification via COM (late binding) try: with connector._com_context(), connector._lock_com: success = _valider_document_com(connector, numero_facture, valider=True) if not success: raise RuntimeError("La validation COM a échoué") logger.info(f"✅ Facture {numero_facture} validée") except ValueError: raise except Exception as e: logger.error(f"❌ Erreur COM validation {numero_facture}: {e}", exc_info=True) raise RuntimeError(f"Échec validation: {str(e)}") # 3. Vérification et réponse via SQL info_apres = _get_facture_info_sql(connector, numero_facture) if info_apres and info_apres["valide"] != 1: raise RuntimeError( "Échec validation: DO_Valide non modifié après l'opération COM" ) return _build_response_sql( connector, numero_facture, deja_valide=False, action="validation" ) def devalider_facture(connector, numero_facture: str) -> Dict: """ Dévalide une facture (retire le cadenas) """ if not connector.cial: raise RuntimeError("Connexion Sage non établie") logger.info(f"🔓 Dévalidation facture {numero_facture}") # 1. Vérifications préalables via SQL info = _get_facture_info_sql(connector, numero_facture) if not info: raise ValueError(f"Facture {numero_facture} introuvable") if info["statut"] == 6: raise ValueError(f"Facture {numero_facture} annulée") if info["statut"] == 5: raise ValueError( f"Facture {numero_facture} transformée, dévalidation impossible" ) if info["montant_regle"] > 0.01: raise ValueError( f"Facture {numero_facture} partiellement réglée ({info['montant_regle']:.2f}€), " "dévalidation impossible" ) if info["valide"] == 0: logger.info(f"Facture {numero_facture} déjà non validée") return _build_response_sql( connector, numero_facture, deja_valide=True, action="devalidation" ) # 2. Modification via COM (late binding) try: with connector._com_context(), connector._lock_com: success = _valider_document_com(connector, numero_facture, valider=False) if not success: raise RuntimeError("La dévalidation COM a échoué") logger.info(f"✅ Facture {numero_facture} dévalidée") except ValueError: raise except Exception as e: logger.error(f"❌ Erreur COM dévalidation {numero_facture}: {e}", exc_info=True) raise RuntimeError(f"Échec dévalidation: {str(e)}") # 3. Vérification et réponse via SQL info_apres = _get_facture_info_sql(connector, numero_facture) if info_apres and info_apres["valide"] != 0: raise RuntimeError( "Échec dévalidation: DO_Valide non modifié après l'opération COM" ) return _build_response_sql( connector, numero_facture, deja_valide=False, action="devalidation" ) # ============================================================ # INTROSPECTION # ============================================================ def introspecter_validation(connector, numero_facture: str = None) -> Dict: """ Introspection pour découvrir les méthodes de validation disponibles. """ if not connector.cial: raise RuntimeError("Connexion Sage non établie") result = {} try: 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 ] if numero_facture: factory = connector.cial.FactoryDocumentVente if factory.ExistPiece(60, numero_facture): persist = factory.ReadPiece(60, numero_facture) # Test late binding try: doc_dyn = _get_dynamic_dispatch(persist) doc_dyn.Read() # Lister les attributs via late binding result["late_binding_test"] = { "DO_Valide_readable": True, "DO_Valide_value": doc_dyn.DO_Valide, } # 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) except Exception as e: result["late_binding_error"] = str(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: result["global_error"] = str(e) return result # ============================================================ # FONCTIONS UTILITAIRES # ============================================================ def _build_response_sql( connector, numero_facture: str, deja_valide: bool, action: str ) -> Dict: """Construit la réponse via SQL""" info = _get_facture_info_sql(connector, numero_facture) if action == "validation": message = ( "Facture déjà validée" if deja_valide else "Facture validée avec succès" ) else: message = ( "Facture déjà non validée" if deja_valide else "Facture dévalidée avec succès" ) return { "numero_facture": numero_facture, "est_validee": info["valide"] == 1 if info else False, "statut": info["statut"] if info else 0, "statut_libelle": _get_statut_libelle(info["statut"]) if info else "Inconnu", "total_ht": info["total_ht"] if info else 0.0, "total_ttc": info["total_ttc"] if info else 0.0, "client_code": info["client_code"] if info else "", "message": message, "action_effectuee": not deja_valide, } def _get_statut_libelle(statut: int) -> str: """Retourne le libellé d'un statut de document""" statuts = { 0: "Brouillon", 1: "Confirmé", 2: "En cours", 3: "Imprimé", 4: "Suspendu", 5: "Transformé", 6: "Annulé", } return statuts.get(statut, f"Statut {statut}") __all__ = [ "valider_facture", "devalider_facture", "get_statut_validation", "introspecter_validation", ]