""" Validation de factures Sage 100c Module: utils/documents/validation.py (Windows Server) COM pour écriture, SQL pour lecture """ from typing import Dict import win32com.client import logging logger = logging.getLogger(__name__) # ============================================================ # 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 "", } # ============================================================ # INTROSPECTION - Pour découvrir les méthodes de validation # ============================================================ def introspecter_validation(connector, numero_facture: str = None) -> Dict: """ 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. """ 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("_")] # 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) # 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 # 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() ] # Toutes les méthodes callable result["doc_methods"] = [ a for a in doc_attrs if callable(getattr(doc, a, None)) and not a.startswith("_") ] # 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) 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) # 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) # 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) except Exception as e: result["global_error"] = str(e) return result # ============================================================ # FONCTIONS COM (ÉCRITURE) - À adapter selon l'introspection # ============================================================ 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") 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 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) - 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") 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 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" ) def _valider_document_com(connector, numero_facture: str, valider: bool = True) -> bool: """ 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. """ erreurs = [] valeur_cible = 1 if valider else 0 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: CreateProcess_ValiderBon try: process = connector.cial.CreateProcess_ValiderBon() logger.info(f" Tentative via CreateProcess_ValiderBon...") # Ajouter le document au process process.Document = persist # Définir l'action (valider ou dévalider) try: process.Valider = valider except Exception: pass # 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 except AttributeError: erreurs.append("CreateProcess_ValiderBon n'existe pas") except Exception as e: erreurs.append(f"CreateProcess_ValiderBon: {e}") # APPROCHE 2: Méthode Valider() ou Devalider() sur le document try: doc = win32com.client.CastTo(persist, "IBODocumentVente3") doc.Read() 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}") except Exception as e: erreurs.append(f"Cast IBODocumentVente3: {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])}") # ============================================================ # FONCTIONS UTILITAIRES # ============================================================ def _build_response_sql( connector, numero_facture: str, deja_valide: bool, action: str ) -> Dict: """Construit la réponse via SQL (pas COM)""" 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", ]