valider_facture quasi fonctionelle
This commit is contained in:
parent
7d58a607f5
commit
6b4b603aee
3 changed files with 542 additions and 301 deletions
24
main.py
24
main.py
|
|
@ -1727,6 +1727,14 @@ def introspection_com():
|
|||
|
||||
@app.get("/sage/debug/introspection-validation")
|
||||
def introspection_validation(numero_facture: str = None):
|
||||
"""
|
||||
Introspection des méthodes de validation disponibles dans Sage.
|
||||
|
||||
Utiliser cette route pour découvrir comment valider un document.
|
||||
|
||||
Args:
|
||||
numero_facture: Optionnel - numéro de facture pour inspecter le document
|
||||
"""
|
||||
try:
|
||||
resultat = sage.introspecter_validation(numero_facture)
|
||||
return {"success": True, "data": resultat}
|
||||
|
|
@ -1735,6 +1743,22 @@ def introspection_validation(numero_facture: str = None):
|
|||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
@app.get("/sage/debug/introspection-document/{numero_facture}")
|
||||
def introspection_document_complet(numero_facture: str):
|
||||
"""
|
||||
Introspection COMPLÈTE d'un document spécifique.
|
||||
|
||||
Retourne tous les attributs, méthodes et propriétés disponibles
|
||||
sur le document pour découvrir la méthode de validation.
|
||||
"""
|
||||
try:
|
||||
resultat = sage.introspecter_document_complet(numero_facture)
|
||||
return {"success": True, "data": resultat}
|
||||
except Exception as e:
|
||||
logger.error(f"Erreur introspection document: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=str(e))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
uvicorn.run(
|
||||
"main:app",
|
||||
|
|
|
|||
|
|
@ -108,6 +108,7 @@ from utils.documents.validations import (
|
|||
devalider_facture as _devalider,
|
||||
get_statut_validation as _get_statut,
|
||||
introspecter_validation as _introspect,
|
||||
explorer_impression_validation as _introspect_doc,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
@ -8098,3 +8099,6 @@ class SageConnector:
|
|||
|
||||
def introspecter_validation(self, numero_facture: str = None) -> dict:
|
||||
return _introspect(self, numero_facture)
|
||||
|
||||
def introspecter_document_complet(self, numero_facture: str) -> dict:
|
||||
return _introspect_doc(self, numero_facture)
|
||||
|
|
|
|||
|
|
@ -2,22 +2,15 @@
|
|||
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
|
||||
Version diagnostic - Introspection complète pour trouver la méthode de validation
|
||||
"""
|
||||
|
||||
from typing import Dict
|
||||
from typing import Dict, List
|
||||
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)
|
||||
|
|
@ -25,9 +18,7 @@ DISPATCH_METHOD = 1
|
|||
|
||||
|
||||
def get_statut_validation(connector, numero_facture: str) -> Dict:
|
||||
"""
|
||||
Retourne le statut de validation d'une facture (SQL)
|
||||
"""
|
||||
"""Retourne le statut de validation d'une facture (SQL)"""
|
||||
with connector._get_sql_connection() as conn:
|
||||
cursor = conn.cursor()
|
||||
cursor.execute(
|
||||
|
|
@ -76,17 +67,14 @@ def get_statut_validation(connector, numero_facture: str) -> Dict:
|
|||
|
||||
|
||||
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)
|
||||
"""
|
||||
"""Récupère les infos d'une facture via 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) as MontantRegle,
|
||||
CT_NumPayeur
|
||||
ISNULL(DO_MontantRegle, 0), CT_NumPayeur
|
||||
FROM F_DOCENTETE
|
||||
WHERE DO_Piece = ? AND DO_Type = 6
|
||||
""",
|
||||
|
|
@ -108,247 +96,311 @@ def _get_facture_info_sql(connector, numero_facture: str) -> Dict:
|
|||
|
||||
|
||||
# ============================================================
|
||||
# LATE BINDING - Contourner le cache gen_py
|
||||
# INTROSPECTION COMPLÈTE
|
||||
# ============================================================
|
||||
|
||||
|
||||
def _get_dynamic_dispatch(com_object):
|
||||
def introspecter_document_complet(connector, numero_facture: str) -> Dict:
|
||||
"""
|
||||
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)
|
||||
Introspection COMPLÈTE d'un document pour découvrir toutes les méthodes
|
||||
et propriétés disponibles.
|
||||
"""
|
||||
if not connector.cial:
|
||||
raise RuntimeError("Connexion Sage non établie")
|
||||
|
||||
logger.info(f"🔒 Validation facture {numero_facture}")
|
||||
result = {
|
||||
"numero_facture": numero_facture,
|
||||
"persist": {},
|
||||
"IBODocumentVente3": {},
|
||||
"IBODocument3": {},
|
||||
"IPMDocument": {},
|
||||
}
|
||||
|
||||
# 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é")
|
||||
factory = connector.cial.FactoryDocumentVente
|
||||
|
||||
logger.info(f"✅ Facture {numero_facture} validée")
|
||||
if not factory.ExistPiece(60, numero_facture):
|
||||
raise ValueError(f"Facture {numero_facture} introuvable")
|
||||
|
||||
except ValueError:
|
||||
raise
|
||||
persist = factory.ReadPiece(60, numero_facture)
|
||||
|
||||
# 1. Attributs du persist brut
|
||||
persist_attrs = [a for a in dir(persist) if not a.startswith("_")]
|
||||
result["persist"]["all_attrs"] = persist_attrs
|
||||
result["persist"]["methods"] = []
|
||||
result["persist"]["properties"] = []
|
||||
|
||||
for attr in persist_attrs:
|
||||
try:
|
||||
val = getattr(persist, attr, None)
|
||||
if callable(val):
|
||||
result["persist"]["methods"].append(attr)
|
||||
else:
|
||||
result["persist"]["properties"].append(
|
||||
{
|
||||
"name": attr,
|
||||
"value": str(val)[:100] if val is not None else None,
|
||||
}
|
||||
)
|
||||
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"
|
||||
result["persist"]["properties"].append(
|
||||
{"name": attr, "error": str(e)[:50]}
|
||||
)
|
||||
|
||||
return _build_response_sql(
|
||||
connector, numero_facture, deja_valide=False, action="validation"
|
||||
# Chercher spécifiquement les attributs liés à validation/valide
|
||||
result["persist"]["validation_related"] = [
|
||||
a
|
||||
for a in persist_attrs
|
||||
if any(
|
||||
x in a.lower()
|
||||
for x in ["valid", "lock", "confirm", "statut", "etat"]
|
||||
)
|
||||
]
|
||||
|
||||
# 2. IBODocumentVente3
|
||||
try:
|
||||
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
|
||||
doc.Read()
|
||||
|
||||
doc_attrs = [a for a in dir(doc) if not a.startswith("_")]
|
||||
result["IBODocumentVente3"]["all_attrs"] = doc_attrs
|
||||
result["IBODocumentVente3"]["methods"] = []
|
||||
result["IBODocumentVente3"]["properties_with_values"] = []
|
||||
|
||||
# Lister les méthodes
|
||||
for attr in doc_attrs:
|
||||
try:
|
||||
val = getattr(doc, attr, None)
|
||||
if callable(val):
|
||||
result["IBODocumentVente3"]["methods"].append(attr)
|
||||
except:
|
||||
pass
|
||||
|
||||
# Chercher DO_* properties
|
||||
result["IBODocumentVente3"]["DO_properties"] = []
|
||||
for attr in doc_attrs:
|
||||
if attr.startswith("DO_"):
|
||||
try:
|
||||
val = getattr(doc, attr, "ERROR")
|
||||
result["IBODocumentVente3"]["DO_properties"].append(
|
||||
{"name": attr, "value": str(val)[:50]}
|
||||
)
|
||||
except Exception as e:
|
||||
result["IBODocumentVente3"]["DO_properties"].append(
|
||||
{"name": attr, "error": str(e)[:50]}
|
||||
)
|
||||
|
||||
# Chercher les attributs liés à validation
|
||||
result["IBODocumentVente3"]["validation_related"] = [
|
||||
a
|
||||
for a in doc_attrs
|
||||
if any(
|
||||
x in a.lower()
|
||||
for x in ["valid", "lock", "confirm", "statut", "etat"]
|
||||
)
|
||||
]
|
||||
|
||||
except Exception as e:
|
||||
result["IBODocumentVente3"]["error"] = str(e)
|
||||
|
||||
# 3. IBODocument3
|
||||
try:
|
||||
doc3 = win32com.client.CastTo(persist, "IBODocument3")
|
||||
doc3.Read()
|
||||
|
||||
doc3_attrs = [a for a in dir(doc3) if not a.startswith("_")]
|
||||
result["IBODocument3"]["all_attrs"] = doc3_attrs
|
||||
result["IBODocument3"]["validation_related"] = [
|
||||
a
|
||||
for a in doc3_attrs
|
||||
if any(
|
||||
x in a.lower()
|
||||
for x in ["valid", "lock", "confirm", "statut", "etat"]
|
||||
)
|
||||
]
|
||||
|
||||
except Exception as e:
|
||||
result["IBODocument3"]["error"] = str(e)
|
||||
|
||||
# 4. IPMDocument
|
||||
try:
|
||||
pmdoc = win32com.client.CastTo(persist, "IPMDocument")
|
||||
|
||||
pmdoc_attrs = [a for a in dir(pmdoc) if not a.startswith("_")]
|
||||
result["IPMDocument"]["all_attrs"] = pmdoc_attrs
|
||||
result["IPMDocument"]["methods"] = [
|
||||
a for a in pmdoc_attrs if callable(getattr(pmdoc, a, None))
|
||||
]
|
||||
|
||||
except Exception as e:
|
||||
result["IPMDocument"]["error"] = str(e)
|
||||
|
||||
# 5. Chercher FactoryDocument* sur le document
|
||||
result["factories_on_doc"] = []
|
||||
for attr in persist_attrs:
|
||||
if "Factory" in attr:
|
||||
result["factories_on_doc"].append(attr)
|
||||
|
||||
except Exception as e:
|
||||
result["global_error"] = str(e)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def introspecter_validation(connector, numero_facture: str = None) -> Dict:
|
||||
"""
|
||||
Introspection pour découvrir les méthodes de validation.
|
||||
"""
|
||||
if not connector.cial:
|
||||
raise RuntimeError("Connexion Sage non établie")
|
||||
|
||||
result = {}
|
||||
|
||||
try:
|
||||
with connector._com_context(), connector._lock_com:
|
||||
# Tous les CreateProcess
|
||||
cial_attrs = [a for a in dir(connector.cial) if not a.startswith("_")]
|
||||
result["all_createprocess"] = [
|
||||
a for a in cial_attrs if "CreateProcess" in a
|
||||
]
|
||||
|
||||
# Explorer chaque process
|
||||
for process_name in result["all_createprocess"]:
|
||||
try:
|
||||
process = getattr(connector.cial, process_name)()
|
||||
process_attrs = [a for a in dir(process) if not a.startswith("_")]
|
||||
result[process_name] = {
|
||||
"attrs": process_attrs,
|
||||
"has_valider": "Valider" in process_attrs
|
||||
or "Valid" in str(process_attrs),
|
||||
"has_document": "Document" in process_attrs,
|
||||
}
|
||||
except Exception as e:
|
||||
result[process_name] = {"error": str(e)}
|
||||
|
||||
# Introspection document si fourni
|
||||
if numero_facture:
|
||||
result["document"] = introspecter_document_complet(
|
||||
connector, numero_facture
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
result["global_error"] = str(e)
|
||||
|
||||
return result
|
||||
|
||||
|
||||
def valider_facture(connector, numero_facture: str) -> Dict:
|
||||
"""
|
||||
Valide via late binding forcé + invocation COM correcte
|
||||
"""
|
||||
import win32com.client
|
||||
import win32com.client.dynamic
|
||||
import pythoncom
|
||||
|
||||
if not connector.cial:
|
||||
raise RuntimeError("Connexion Sage non établie")
|
||||
|
||||
logger.info(f"🔒 Validation facture {numero_facture} (v5)")
|
||||
|
||||
with connector._com_context(), connector._lock_com:
|
||||
factory = connector.cial.FactoryDocumentVente
|
||||
|
||||
if not factory.ExistPiece(60, numero_facture):
|
||||
raise ValueError(f"Facture {numero_facture} introuvable")
|
||||
|
||||
persist = factory.ReadPiece(60, numero_facture)
|
||||
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
|
||||
doc.Read()
|
||||
|
||||
# Lecture via getattr
|
||||
valide_avant = getattr(doc, "DO_Valide", None)
|
||||
imprim_avant = getattr(doc, "DO_Imprim", None)
|
||||
logger.info(f" Avant: DO_Imprim={imprim_avant}, DO_Valide={valide_avant}")
|
||||
|
||||
if valide_avant == True:
|
||||
return {"numero_facture": numero_facture, "deja_valide": True}
|
||||
|
||||
oleobj = doc._oleobj_
|
||||
|
||||
# Explorer les DISPID
|
||||
type_info = oleobj.GetTypeInfo()
|
||||
type_attr = type_info.GetTypeAttr()
|
||||
|
||||
dispids = {}
|
||||
for i in range(type_attr.cFuncs):
|
||||
func_desc = type_info.GetFuncDesc(i)
|
||||
names = type_info.GetNames(func_desc.memid)
|
||||
if names:
|
||||
name = names[0]
|
||||
if name in ("DO_Valide", "DO_Imprim"):
|
||||
invkind = func_desc.invkind
|
||||
logger.info(
|
||||
f" {name}: memid={func_desc.memid}, invkind={invkind}, cParams={func_desc.cParamsOpt}"
|
||||
)
|
||||
dispids[name] = {
|
||||
"memid": func_desc.memid,
|
||||
"invkind": invkind,
|
||||
"cParams": func_desc.cParamsOpt,
|
||||
}
|
||||
|
||||
logger.info(f" DISPIDs trouvés: {dispids}")
|
||||
|
||||
errors = []
|
||||
|
||||
# Méthode A: Invoke PROPERTYPUT
|
||||
if "DO_Valide" in dispids:
|
||||
memid = dispids["DO_Valide"]["memid"]
|
||||
try:
|
||||
oleobj.Invoke(memid, 0, pythoncom.DISPATCH_PROPERTYPUT, False, True)
|
||||
doc.Write()
|
||||
logger.info(f" ✅ Méthode A: Invoke PROPERTYPUT OK")
|
||||
except Exception as e:
|
||||
errors.append(f"Invoke PROPERTYPUT: {e}")
|
||||
logger.warning(f" Méthode A échouée: {e}")
|
||||
|
||||
# Méthode B: DumbDispatch
|
||||
try:
|
||||
raw_dispatch = win32com.client.dynamic.DumbDispatch(oleobj)
|
||||
raw_dispatch.DO_Valide = True
|
||||
doc.Write()
|
||||
logger.info(f" ✅ Méthode B: DumbDispatch OK")
|
||||
except Exception as e:
|
||||
errors.append(f"DumbDispatch: {e}")
|
||||
logger.warning(f" Méthode B échouée: {e}")
|
||||
|
||||
# Méthode C: Dispatch dynamique combiné
|
||||
try:
|
||||
if "DO_Valide" in dispids:
|
||||
memid = dispids["DO_Valide"]["memid"]
|
||||
oleobj.Invoke(
|
||||
memid,
|
||||
0,
|
||||
pythoncom.DISPATCH_PROPERTYPUT | pythoncom.DISPATCH_METHOD,
|
||||
False,
|
||||
True,
|
||||
)
|
||||
doc.Write()
|
||||
logger.info(f" ✅ Méthode C: Combined flags OK")
|
||||
except Exception as e:
|
||||
errors.append(f"Combined flags: {e}")
|
||||
logger.warning(f" Méthode C échouée: {e}")
|
||||
|
||||
# Vérification
|
||||
doc.Read()
|
||||
valide_apres = getattr(doc, "DO_Valide", None)
|
||||
imprim_apres = getattr(doc, "DO_Imprim", None)
|
||||
logger.info(f" Après: DO_Imprim={imprim_apres}, DO_Valide={valide_apres}")
|
||||
|
||||
return {
|
||||
"numero_facture": numero_facture,
|
||||
"avant": {"DO_Imprim": imprim_avant, "DO_Valide": valide_avant},
|
||||
"apres": {"DO_Imprim": imprim_apres, "DO_Valide": valide_apres},
|
||||
"dispids": dispids,
|
||||
"errors": errors,
|
||||
"success": valide_apres == True,
|
||||
}
|
||||
|
||||
|
||||
def devalider_facture(connector, numero_facture: str) -> Dict:
|
||||
|
|
@ -360,7 +412,6 @@ def devalider_facture(connector, numero_facture: str) -> Dict:
|
|||
|
||||
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")
|
||||
|
|
@ -385,7 +436,6 @@ def devalider_facture(connector, numero_facture: str) -> Dict:
|
|||
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)
|
||||
|
|
@ -400,7 +450,6 @@ def devalider_facture(connector, numero_facture: str) -> Dict:
|
|||
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(
|
||||
|
|
@ -412,73 +461,236 @@ def devalider_facture(connector, numero_facture: str) -> Dict:
|
|||
)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# INTROSPECTION
|
||||
# ============================================================
|
||||
|
||||
|
||||
def introspecter_validation(connector, numero_facture: str = None) -> Dict:
|
||||
def _valider_document_com(connector, numero_facture: str, valider: bool = True) -> bool:
|
||||
"""
|
||||
Introspection pour découvrir les méthodes de validation disponibles.
|
||||
Tente de valider/dévalider un document via COM.
|
||||
"""
|
||||
if not connector.cial:
|
||||
raise RuntimeError("Connexion Sage non établie")
|
||||
erreurs = []
|
||||
action = "validation" if valider else "dévalidation"
|
||||
valeur_cible = 1 if valider else 0
|
||||
|
||||
result = {}
|
||||
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: Accès direct à DO_Valide sur IBODocumentVente3
|
||||
try:
|
||||
with connector._com_context(), connector._lock_com:
|
||||
cial_attrs = [a for a in dir(connector.cial) if not a.startswith("_")]
|
||||
result["cial_createprocess"] = [
|
||||
logger.info(
|
||||
" APPROCHE 1: Modification directe DO_Valide sur IBODocumentVente3..."
|
||||
)
|
||||
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
|
||||
doc.Read()
|
||||
|
||||
# Vérifier la valeur actuelle
|
||||
valeur_avant = getattr(doc, "DO_Valide", None)
|
||||
logger.info(f" DO_Valide avant: {valeur_avant}")
|
||||
|
||||
# Tenter la modification
|
||||
doc.DO_Valide = valeur_cible
|
||||
doc.Write()
|
||||
|
||||
# Relire pour vérifier
|
||||
doc.Read()
|
||||
valeur_apres = getattr(doc, "DO_Valide", None)
|
||||
logger.info(f" DO_Valide après: {valeur_apres}")
|
||||
|
||||
if valeur_apres == valeur_cible:
|
||||
logger.info(f" ✅ DO_Valide modifié avec succès!")
|
||||
return True
|
||||
else:
|
||||
erreurs.append(
|
||||
f"DO_Valide non modifié (avant={valeur_avant}, après={valeur_apres})"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
erreurs.append(f"IBODocumentVente3.DO_Valide: {e}")
|
||||
logger.warning(f" Erreur: {e}")
|
||||
|
||||
# APPROCHE 2: Via IBODocument3 (interface parent)
|
||||
try:
|
||||
logger.info(" APPROCHE 2: Via IBODocument3...")
|
||||
doc3 = win32com.client.CastTo(persist, "IBODocument3")
|
||||
doc3.Read()
|
||||
|
||||
if hasattr(doc3, "DO_Valide"):
|
||||
doc3.DO_Valide = valeur_cible
|
||||
doc3.Write()
|
||||
logger.info(f" IBODocument3.DO_Valide = {valeur_cible} OK")
|
||||
return True
|
||||
else:
|
||||
erreurs.append("IBODocument3 n'a pas DO_Valide")
|
||||
|
||||
except Exception as e:
|
||||
erreurs.append(f"IBODocument3: {e}")
|
||||
|
||||
# APPROCHE 3: Chercher un CreateProcess de validation
|
||||
try:
|
||||
logger.info(" APPROCHE 3: Recherche CreateProcess de validation...")
|
||||
cial_attrs = [a for a in dir(connector.cial) if "CreateProcess" in a]
|
||||
validation_processes = [
|
||||
a
|
||||
for a in cial_attrs
|
||||
if "Process" in a or "Valider" in a or "Valid" in a
|
||||
if any(x in a.lower() for x in ["valid", "confirm", "lock"])
|
||||
]
|
||||
logger.info(f" CreateProcess trouvés: {cial_attrs}")
|
||||
logger.info(f" Liés à validation: {validation_processes}")
|
||||
|
||||
if numero_facture:
|
||||
factory = connector.cial.FactoryDocumentVente
|
||||
if factory.ExistPiece(60, numero_facture):
|
||||
persist = factory.ReadPiece(60, numero_facture)
|
||||
|
||||
# Test late binding
|
||||
for proc_name in validation_processes:
|
||||
try:
|
||||
doc_dyn = _get_dynamic_dispatch(persist)
|
||||
doc_dyn.Read()
|
||||
process = getattr(connector.cial, proc_name)()
|
||||
# Lister les attributs du process
|
||||
proc_attrs = [a for a in dir(process) if not a.startswith("_")]
|
||||
logger.info(f" {proc_name} attrs: {proc_attrs}")
|
||||
|
||||
# Lister les attributs via late binding
|
||||
result["late_binding_test"] = {
|
||||
"DO_Valide_readable": True,
|
||||
"DO_Valide_value": doc_dyn.DO_Valide,
|
||||
if hasattr(process, "Document"):
|
||||
process.Document = persist
|
||||
if hasattr(process, "Valider"):
|
||||
process.Valider = valider
|
||||
if hasattr(process, "Process"):
|
||||
process.Process()
|
||||
logger.info(f" {proc_name}.Process() exécuté!")
|
||||
return True
|
||||
except Exception as e:
|
||||
erreurs.append(f"{proc_name}: {e}")
|
||||
|
||||
except Exception as e:
|
||||
erreurs.append(f"CreateProcess: {e}")
|
||||
|
||||
# APPROCHE 4: WriteDefault avec paramètres
|
||||
try:
|
||||
logger.info(" APPROCHE 4: WriteDefault...")
|
||||
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
|
||||
doc.Read()
|
||||
doc.DO_Valide = valeur_cible
|
||||
|
||||
if hasattr(doc, "WriteDefault"):
|
||||
doc.WriteDefault()
|
||||
logger.info(" WriteDefault() exécuté")
|
||||
return True
|
||||
|
||||
except Exception as e:
|
||||
erreurs.append(f"WriteDefault: {e}")
|
||||
|
||||
logger.error(f" Toutes les approches de {action} ont échoué: {erreurs}")
|
||||
raise RuntimeError(f"Échec {action}: {'; '.join(erreurs[:5])}")
|
||||
|
||||
|
||||
def explorer_impression_validation(connector, numero_facture: str) -> Dict:
|
||||
"""Explorer les méthodes d'impression/validation pour les factures"""
|
||||
result = {"numero_facture": numero_facture}
|
||||
|
||||
with connector._com_context(), connector._lock_com:
|
||||
factory = connector.cial.FactoryDocumentVente
|
||||
persist = factory.ReadPiece(60, numero_facture)
|
||||
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
|
||||
doc.Read()
|
||||
|
||||
# 1. CreateProcess_Document SANS paramètre
|
||||
try:
|
||||
process = connector.cial.CreateProcess_Document()
|
||||
attrs = [a for a in dir(process) if not a.startswith("_")]
|
||||
result["CreateProcess_Document_no_param"] = {
|
||||
"attrs": attrs,
|
||||
"print_related": [
|
||||
a
|
||||
for a in attrs
|
||||
if any(
|
||||
x in a.lower()
|
||||
for x in ["print", "imprim", "edit", "model", "bgc", "valid"]
|
||||
)
|
||||
],
|
||||
}
|
||||
|
||||
# Tester si writable
|
||||
# Essayer d'assigner le document
|
||||
if "Document" in attrs:
|
||||
try:
|
||||
original = doc_dyn.DO_Valide
|
||||
doc_dyn.DO_Valide = (
|
||||
original # Essayer de réécrire la même valeur
|
||||
process.Document = doc
|
||||
result["CreateProcess_Document_no_param"]["Document_assigned"] = (
|
||||
True
|
||||
)
|
||||
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)
|
||||
result["CreateProcess_Document_no_param"]["Document_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"):
|
||||
if "SetDocument" in attrs:
|
||||
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
|
||||
process.SetDocument(doc)
|
||||
result["CreateProcess_Document_no_param"]["SetDocument_ok"] = True
|
||||
except Exception as e:
|
||||
result[f"{process_name}_error"] = str(e)
|
||||
result["CreateProcess_Document_no_param"]["SetDocument_error"] = (
|
||||
str(e)
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
result["global_error"] = str(e)
|
||||
result["CreateProcess_Document_no_param_error"] = str(e)
|
||||
|
||||
# 2. CreateProcess_Document avec type (60 = facture)
|
||||
try:
|
||||
process = connector.cial.CreateProcess_Document(60)
|
||||
attrs = [a for a in dir(process) if not a.startswith("_")]
|
||||
result["CreateProcess_Document_type60"] = {"attrs": attrs}
|
||||
except Exception as e:
|
||||
result["CreateProcess_Document_type60_error"] = str(e)
|
||||
|
||||
# 3. Explorer TOUS les CreateProcess
|
||||
cial_attrs = [a for a in dir(connector.cial) if "CreateProcess" in a]
|
||||
result["all_createprocess"] = cial_attrs
|
||||
|
||||
# 4. Chercher spécifiquement impression/validation
|
||||
for proc_name in cial_attrs:
|
||||
if any(
|
||||
x in proc_name.lower()
|
||||
for x in ["imprim", "print", "edit", "valid", "confirm"]
|
||||
):
|
||||
try:
|
||||
proc = getattr(connector.cial, proc_name)()
|
||||
proc_attrs = [a for a in dir(proc) if not a.startswith("_")]
|
||||
result[proc_name] = {
|
||||
"attrs": proc_attrs,
|
||||
"has_modele": [
|
||||
a
|
||||
for a in proc_attrs
|
||||
if "model" in a.lower() or "bgc" in a.lower()
|
||||
],
|
||||
"has_document": "Document" in proc_attrs
|
||||
or "SetDocument" in proc_attrs,
|
||||
}
|
||||
except Exception as e:
|
||||
result[proc_name] = {"error": str(e)}
|
||||
|
||||
# 5. Explorer le document pour méthodes Print/Imprimer
|
||||
doc_attrs = [a for a in dir(doc) if not a.startswith("_")]
|
||||
result["doc_print_methods"] = [
|
||||
a
|
||||
for a in doc_attrs
|
||||
if any(x in a.lower() for x in ["print", "imprim", "edit", "valid"])
|
||||
]
|
||||
|
||||
# 6. Chercher IPMDocument (Process Manager)
|
||||
try:
|
||||
pm = win32com.client.CastTo(persist, "IPMDocument")
|
||||
pm_attrs = [a for a in dir(pm) if not a.startswith("_")]
|
||||
result["IPMDocument"] = {
|
||||
"attrs": pm_attrs,
|
||||
"print_related": [
|
||||
a
|
||||
for a in pm_attrs
|
||||
if any(
|
||||
x in a.lower()
|
||||
for x in ["print", "imprim", "edit", "valid", "process"]
|
||||
)
|
||||
],
|
||||
}
|
||||
except Exception as e:
|
||||
result["IPMDocument_error"] = str(e)
|
||||
|
||||
# 7. Chemin du modèle BGC
|
||||
result["modele_bgc_path"] = (
|
||||
r"C:\Users\Public\Documents\Sage\Entreprise 100c\fr-FR\Documents standards\Gestion commerciale\Ventes\Facture client.bgc"
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
|
|
@ -537,4 +749,5 @@ __all__ = [
|
|||
"devalider_facture",
|
||||
"get_statut_validation",
|
||||
"introspecter_validation",
|
||||
"introspecter_document_complet",
|
||||
]
|
||||
|
|
|
|||
Loading…
Reference in a new issue