""" Validation de factures Sage 100c Module: utils/documents/validation.py (Windows Server) COM pour écriture, SQL pour lecture Solution robuste avec late binding pour éviter les erreurs intermittentes """ from typing import Dict import win32com.client import win32com.client.dynamic 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 "", } # ============================================================ # FONCTIONS COM (ÉCRITURE) # ============================================================ def _set_do_valide_com(doc, valeur: int) -> bool: """ Définit DO_Valide via COM avec plusieurs méthodes de fallback Retourne True si réussi """ erreurs = [] # Méthode 1: Accès direct (early binding) try: doc.DO_Valide = valeur doc.Write() logger.debug(" DO_Valide défini via accès direct") return True except AttributeError as e: erreurs.append(f"Direct: {e}") except Exception as e: erreurs.append(f"Direct: {e}") # Méthode 2: Via setattr try: setattr(doc, "DO_Valide", valeur) doc.Write() logger.debug(" DO_Valide défini via setattr") return True except Exception as e: erreurs.append(f"setattr: {e}") # Méthode 3: Late binding - récupérer l'objet COM sous-jacent try: # Accès via _oleobj_ (objet COM brut) disp = win32com.client.dynamic.Dispatch(doc._oleobj_) disp.DO_Valide = valeur disp.Write() logger.debug(" DO_Valide défini via late binding _oleobj_") return True except Exception as e: erreurs.append(f"Late _oleobj_: {e}") # Méthode 4: Invoke direct sur la propriété try: import pythoncom from win32com.client import VARIANT # DISPATCH_PROPERTYPUT = 4 doc._oleobj_.Invoke( doc._oleobj_.GetIDsOfNames(0, "DO_Valide")[0], 0, 4, 0, valeur ) doc.Write() logger.debug(" DO_Valide défini via Invoke") return True except Exception as e: erreurs.append(f"Invoke: {e}") logger.error(f" Toutes les méthodes ont échoué: {erreurs}") return False def _get_document_com(connector, numero_facture: str): """ Récupère le document COM avec plusieurs tentatives d'interface Retourne (doc, interface_utilisee) ou lève une exception """ factory = connector.cial.FactoryDocumentVente if not factory.ExistPiece(60, numero_facture): raise ValueError(f"Facture {numero_facture} introuvable dans Sage") persist = factory.ReadPiece(60, numero_facture) if not persist: raise ValueError(f"Impossible de lire la facture {numero_facture}") # Essayer plusieurs interfaces par ordre de préférence interfaces = [ "IBODocumentVente3", "IBODocumentVente2", "IBODocumentVente", "IBODocument3", "IBPersistDocument", ] for iface in interfaces: try: doc = win32com.client.CastTo(persist, iface) doc.Read() logger.debug(f" Document casté vers {iface}") return doc, iface except Exception as e: logger.debug(f" Cast {iface} échoué: {e}") continue # Fallback: utiliser persist directement try: persist.Read() logger.debug(" Utilisation de persist directement (sans cast)") return persist, "persist" except Exception as e: raise RuntimeError(f"Impossible d'accéder au document: {e}") def valider_facture(connector, numero_facture: str) -> Dict: """ Valide une facture (pose le cadenas) - Vérifications: SQL - Modification DO_Valide: COM - 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: doc, iface = _get_document_com(connector, numero_facture) logger.debug(f" Interface utilisée: {iface}") success = _set_do_valide_com(doc, 1) if not success: raise RuntimeError("Impossible de définir DO_Valide") logger.info(f"✅ Facture {numero_facture} validée (COM via {iface})") 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 Write()") 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 DO_Valide: COM - 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: doc, iface = _get_document_com(connector, numero_facture) logger.debug(f" Interface utilisée: {iface}") success = _set_do_valide_com(doc, 0) if not success: raise RuntimeError("Impossible de définir DO_Valide") logger.info(f"✅ Facture {numero_facture} dévalidée (COM via {iface})") 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 Write()") return _build_response_sql( connector, numero_facture, deja_valide=False, action="devalidation" ) # ============================================================ # 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", ] # ============================================================ # 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", ]