652 lines
23 KiB
Python
652 lines
23 KiB
Python
from typing import Dict
|
||
import win32com.client
|
||
import pywintypes
|
||
import time
|
||
from schemas.documents.doc_config import TypeDocumentVente, ConfigDocument
|
||
import logging
|
||
from utils.functions.functions import normaliser_date
|
||
from utils.tiers.clients.clients_data import _cast_client
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def creer_document_vente(
|
||
self, doc_data: dict, type_document: TypeDocumentVente
|
||
) -> Dict:
|
||
if not self.cial:
|
||
raise RuntimeError("Connexion Sage non établie")
|
||
|
||
config = ConfigDocument(type_document)
|
||
logger.info(
|
||
f" Début création {config.nom_document} pour client {doc_data['client']['code']}"
|
||
)
|
||
|
||
try:
|
||
with self._com_context(), self._lock_com:
|
||
transaction_active = False
|
||
|
||
try:
|
||
try:
|
||
self.cial.CptaApplication.BeginTrans()
|
||
transaction_active = True
|
||
logger.debug(" Transaction Sage démarrée")
|
||
except Exception as e:
|
||
logger.warning(f"BeginTrans échoué (non critique): {e}")
|
||
|
||
process = self.cial.CreateProcess_Document(config.type_sage)
|
||
doc = process.Document
|
||
|
||
try:
|
||
doc = win32com.client.CastTo(doc, "IBODocumentVente3")
|
||
except Exception:
|
||
pass
|
||
|
||
logger.info(f" Document {config.nom_document} créé")
|
||
|
||
date_principale = normaliser_date(
|
||
doc_data.get(config.champ_date_principale)
|
||
)
|
||
doc.DO_Date = pywintypes.Time(date_principale)
|
||
|
||
try:
|
||
doc.DO_Heure = pywintypes.Time(date_principale)
|
||
logger.debug(
|
||
f"DO_Heure défini: {date_principale.strftime('%H:%M:%S')}"
|
||
)
|
||
except Exception as e:
|
||
logger.debug(f"DO_Heure non défini: {e}")
|
||
|
||
if config.champ_date_secondaire and doc_data.get(
|
||
config.champ_date_secondaire
|
||
):
|
||
doc.DO_DateLivr = pywintypes.Time(
|
||
normaliser_date(doc_data[config.champ_date_secondaire])
|
||
)
|
||
logger.info(
|
||
f" {config.champ_date_secondaire}: {doc_data[config.champ_date_secondaire]}"
|
||
)
|
||
|
||
factory_client = self.cial.CptaApplication.FactoryClient
|
||
persist_client = factory_client.ReadNumero(doc_data["client"]["code"])
|
||
|
||
if not persist_client:
|
||
raise ValueError(f"Client {doc_data['client']['code']} introuvable")
|
||
|
||
client_obj = _cast_client(persist_client)
|
||
if not client_obj:
|
||
raise ValueError("Impossible de charger le client")
|
||
|
||
doc.SetDefaultClient(client_obj)
|
||
doc.Write()
|
||
logger.info(f" Client {doc_data['client']['code']} associé")
|
||
|
||
if doc_data.get("reference"):
|
||
try:
|
||
doc.DO_Ref = doc_data["reference"]
|
||
logger.info(f" Référence: {doc_data['reference']}")
|
||
except Exception as e:
|
||
logger.warning(f"Référence non définie: {e}")
|
||
|
||
if type_document == TypeDocumentVente.FACTURE:
|
||
_configurer_facture(self, doc)
|
||
|
||
try:
|
||
factory_lignes = doc.FactoryDocumentLigne
|
||
except Exception:
|
||
factory_lignes = doc.FactoryDocumentVenteLigne
|
||
|
||
factory_article = self.cial.FactoryArticle
|
||
|
||
logger.info(f" Ajout de {len(doc_data['lignes'])} lignes...")
|
||
|
||
for idx, ligne_data in enumerate(doc_data["lignes"], 1):
|
||
_ajouter_ligne_document(
|
||
cial=self.cial,
|
||
factory_lignes=factory_lignes,
|
||
factory_article=factory_article,
|
||
ligne_data=ligne_data,
|
||
idx=idx,
|
||
doc=doc,
|
||
)
|
||
|
||
logger.info(" Validation du document...")
|
||
|
||
if type_document == TypeDocumentVente.FACTURE:
|
||
try:
|
||
doc.SetClient(client_obj)
|
||
logger.debug(" ↳ Client réassocié avant validation")
|
||
except Exception:
|
||
try:
|
||
doc.SetDefaultClient(client_obj)
|
||
except Exception:
|
||
pass
|
||
|
||
doc.Write()
|
||
|
||
if type_document != TypeDocumentVente.DEVIS:
|
||
process.Process()
|
||
logger.info(" Process() appelé")
|
||
else:
|
||
try:
|
||
process.Process()
|
||
logger.info(" Process() appelé (devis)")
|
||
except Exception:
|
||
logger.debug(" ↳ Process() ignoré pour devis brouillon")
|
||
|
||
if transaction_active:
|
||
try:
|
||
self.cial.CptaApplication.CommitTrans()
|
||
logger.info(" Transaction committée")
|
||
except Exception:
|
||
pass
|
||
|
||
time.sleep(2)
|
||
|
||
numero_document = _recuperer_numero_document(process, doc)
|
||
|
||
if not numero_document:
|
||
raise RuntimeError(
|
||
f"Numéro {config.nom_document} vide après création"
|
||
)
|
||
|
||
logger.info(f" Numéro: {numero_document}")
|
||
|
||
doc_final_data = _relire_document_final(
|
||
self,
|
||
config=config,
|
||
numero_document=numero_document,
|
||
doc_data=doc_data,
|
||
)
|
||
|
||
logger.info(
|
||
f" {config.nom_document.upper()} CRÉÉ: "
|
||
f"{numero_document} - {doc_final_data['total_ttc']}€ TTC"
|
||
)
|
||
|
||
return doc_final_data
|
||
|
||
except Exception:
|
||
if transaction_active:
|
||
try:
|
||
self.cial.CptaApplication.RollbackTrans()
|
||
logger.error(" Transaction annulée (rollback)")
|
||
except Exception:
|
||
pass
|
||
raise
|
||
|
||
except Exception as e:
|
||
logger.error(
|
||
f" ERREUR CRÉATION {config.nom_document.upper()}: {e}", exc_info=True
|
||
)
|
||
raise RuntimeError(f"Échec création {config.nom_document}: {str(e)}")
|
||
|
||
|
||
def _appliquer_remise_ligne(ligne_obj, remise_pourcent: float) -> bool:
|
||
"""Applique la remise via FromString - SOLUTION FINALE"""
|
||
try:
|
||
import pythoncom
|
||
|
||
dispatch = ligne_obj._oleobj_
|
||
|
||
dispid = dispatch.GetIDsOfNames(0, "Remise")
|
||
remise_obj = dispatch.Invoke(dispid, 0, pythoncom.DISPATCH_PROPERTYGET, 1)
|
||
remise_wrapper = win32com.client.Dispatch(remise_obj)
|
||
|
||
remise_wrapper.FromString(f"{remise_pourcent}%")
|
||
|
||
try:
|
||
remise_wrapper.Calcul()
|
||
except Exception:
|
||
pass
|
||
|
||
ligne_obj.Write()
|
||
|
||
logger.info(f" Remise {remise_pourcent}% appliquée")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f" Erreur remise: {e}")
|
||
return False
|
||
|
||
|
||
def _ajouter_ligne_document(
|
||
cial, factory_lignes, factory_article, ligne_data: dict, idx: int, doc
|
||
) -> None:
|
||
"""VERSION FINALE AVEC REMISES FONCTIONNELLES"""
|
||
logger.info(f" ├─ Ligne {idx}: {ligne_data['article_code']}")
|
||
|
||
persist_article = factory_article.ReadReference(ligne_data["article_code"])
|
||
if not persist_article:
|
||
raise ValueError(f"Article {ligne_data['article_code']} introuvable")
|
||
|
||
article_obj = win32com.client.CastTo(persist_article, "IBOArticle3")
|
||
article_obj.Read()
|
||
|
||
prix_sage = float(getattr(article_obj, "AR_PrixVen", 0.0))
|
||
designation_sage = getattr(article_obj, "AR_Design", "")
|
||
|
||
ligne_persist = factory_lignes.Create()
|
||
try:
|
||
ligne_obj = win32com.client.CastTo(ligne_persist, "IBODocumentLigne3")
|
||
except Exception:
|
||
ligne_obj = win32com.client.CastTo(ligne_persist, "IBODocumentVenteLigne3")
|
||
|
||
quantite = float(ligne_data["quantite"])
|
||
|
||
try:
|
||
ligne_obj.SetDefaultArticleReference(ligne_data["article_code"], quantite)
|
||
except Exception:
|
||
try:
|
||
ligne_obj.SetDefaultArticle(article_obj, quantite)
|
||
except Exception:
|
||
ligne_obj.DL_Design = designation_sage
|
||
ligne_obj.DL_Qte = quantite
|
||
|
||
prix_auto = float(getattr(ligne_obj, "DL_PrixUnitaire", 0.0))
|
||
prix_perso = ligne_data.get("prix_unitaire_ht")
|
||
|
||
if prix_perso and prix_perso > 0:
|
||
ligne_obj.DL_PrixUnitaire = float(prix_perso)
|
||
elif prix_auto == 0 and prix_sage > 0:
|
||
ligne_obj.DL_PrixUnitaire = float(prix_sage)
|
||
|
||
prix_final = float(getattr(ligne_obj, "DL_PrixUnitaire", 0))
|
||
logger.info(f" Prix: {prix_final}€")
|
||
|
||
ligne_obj.Write()
|
||
|
||
remise = ligne_data.get("remise_pourcentage", 0)
|
||
if remise and remise > 0:
|
||
logger.info(f" Application remise {remise}%...")
|
||
_appliquer_remise_ligne(ligne_obj, remise)
|
||
|
||
logger.info(f" Ligne {idx} terminée")
|
||
|
||
|
||
def _configurer_facture(self, doc) -> None:
|
||
"""Configuration spécifique pour les factures"""
|
||
logger.info(" Configuration spécifique facture...")
|
||
|
||
try:
|
||
if hasattr(doc, "DO_CodeJournal"):
|
||
try:
|
||
param_societe = self.cial.CptaApplication.ParametreSociete
|
||
journal_defaut = getattr(param_societe, "P_CodeJournalVte", "VTE")
|
||
doc.DO_CodeJournal = journal_defaut
|
||
logger.debug(f" Code journal: {journal_defaut}")
|
||
except Exception:
|
||
doc.DO_CodeJournal = "VTE"
|
||
logger.debug(" Code journal: VTE (défaut)")
|
||
except Exception as e:
|
||
logger.debug(f" Code journal: {e}")
|
||
|
||
try:
|
||
if hasattr(doc, "DO_Souche"):
|
||
doc.DO_Souche = 0
|
||
logger.debug(" Souche: 0")
|
||
except Exception:
|
||
pass
|
||
|
||
try:
|
||
if hasattr(doc, "DO_Regime"):
|
||
doc.DO_Regime = 0
|
||
logger.debug(" Régime: 0")
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def _recuperer_numero_document(process, doc) -> str:
|
||
"""Récupère le numéro du document créé"""
|
||
numero = None
|
||
|
||
try:
|
||
doc_result = process.DocumentResult
|
||
if doc_result:
|
||
doc_result = win32com.client.CastTo(doc_result, "IBODocumentVente3")
|
||
doc_result.Read()
|
||
numero = getattr(doc_result, "DO_Piece", "")
|
||
except Exception:
|
||
pass
|
||
|
||
if not numero:
|
||
numero = getattr(doc, "DO_Piece", "")
|
||
|
||
return numero
|
||
|
||
|
||
def _relire_document_final(
|
||
self,
|
||
config: ConfigDocument,
|
||
numero_document: str,
|
||
doc_data: dict,
|
||
client_code_fallback: str = None,
|
||
) -> Dict:
|
||
"""
|
||
Relit le document pour obtenir les totaux calculés par Sage
|
||
"""
|
||
factory_doc = self.cial.FactoryDocumentVente
|
||
persist_reread = factory_doc.ReadPiece(config.type_sage, numero_document)
|
||
|
||
client_code = None
|
||
date_secondaire_value = None
|
||
|
||
if persist_reread:
|
||
doc_final = win32com.client.CastTo(persist_reread, "IBODocumentVente3")
|
||
doc_final.Read()
|
||
|
||
total_ht = float(getattr(doc_final, "DO_TotalHT", 0.0))
|
||
total_ttc = float(getattr(doc_final, "DO_TotalTTC", 0.0))
|
||
reference_finale = getattr(doc_final, "DO_Ref", "")
|
||
|
||
try:
|
||
client_obj = getattr(doc_final, "Client", None)
|
||
if client_obj:
|
||
client_obj.Read()
|
||
client_code = getattr(client_obj, "CT_Num", "").strip()
|
||
except Exception:
|
||
pass
|
||
|
||
if config.champ_date_secondaire:
|
||
try:
|
||
date_livr = getattr(doc_final, "DO_DateLivr", None)
|
||
if date_livr:
|
||
date_secondaire_value = date_livr.strftime("%Y-%m-%d")
|
||
except Exception:
|
||
pass
|
||
else:
|
||
total_ht = 0.0
|
||
total_ttc = 0.0
|
||
reference_finale = doc_data.get("reference", "")
|
||
date_secondaire_value = doc_data.get(config.champ_date_secondaire)
|
||
|
||
if not client_code:
|
||
client_code = client_code_fallback or doc_data.get("client", {}).get("code", "")
|
||
|
||
resultat = {
|
||
config.champ_numero: numero_document,
|
||
"total_ht": total_ht,
|
||
"total_ttc": total_ttc,
|
||
"nb_lignes": len(doc_data.get("lignes", [])),
|
||
"client_code": client_code,
|
||
config.champ_date_principale: str(
|
||
normaliser_date(doc_data.get(config.champ_date_principale))
|
||
)
|
||
if doc_data.get(config.champ_date_principale)
|
||
else None,
|
||
"reference": reference_finale,
|
||
}
|
||
|
||
if config.champ_date_secondaire:
|
||
resultat[config.champ_date_secondaire] = date_secondaire_value
|
||
|
||
return resultat
|
||
|
||
|
||
def modifier_document_vente(
|
||
self, numero: str, doc_data: dict, type_document: TypeDocumentVente
|
||
) -> Dict:
|
||
if not self.cial:
|
||
raise RuntimeError("Connexion Sage non établie")
|
||
|
||
config = ConfigDocument(type_document)
|
||
logger.info(f" === MODIFICATION {config.nom_document.upper()} {numero} ===")
|
||
|
||
try:
|
||
with self._com_context(), self._lock_com:
|
||
logger.info(" Chargement document...")
|
||
factory = self.cial.FactoryDocumentVente
|
||
persist = None
|
||
|
||
for type_test in [config.type_sage, 50]:
|
||
try:
|
||
persist_test = factory.ReadPiece(type_test, numero)
|
||
if persist_test:
|
||
persist = persist_test
|
||
logger.info(f" Document trouvé (type={type_test})")
|
||
break
|
||
except Exception:
|
||
continue
|
||
|
||
if not persist:
|
||
raise ValueError(
|
||
f"{config.nom_document.capitalize()} {numero} introuvable"
|
||
)
|
||
|
||
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
|
||
doc.Read()
|
||
|
||
statut_actuel = getattr(doc, "DO_Statut", 0)
|
||
|
||
if statut_actuel == 5:
|
||
raise ValueError(f"Le {config.nom_document} a déjà été transformé")
|
||
if statut_actuel == 6:
|
||
raise ValueError(f"Le {config.nom_document} est annulé")
|
||
|
||
client_code_initial = ""
|
||
try:
|
||
client_obj = getattr(doc, "Client", None)
|
||
if client_obj:
|
||
client_obj.Read()
|
||
client_code_initial = getattr(client_obj, "CT_Num", "").strip()
|
||
logger.info(f" Client initial: {client_code_initial}")
|
||
except Exception as e:
|
||
logger.error(f" Erreur lecture client: {e}")
|
||
|
||
if not client_code_initial:
|
||
raise ValueError("Client introuvable dans le document")
|
||
|
||
nb_lignes_initial = 0
|
||
try:
|
||
factory_lignes = getattr(doc, "FactoryDocumentLigne", None) or getattr(
|
||
doc, "FactoryDocumentVenteLigne", None
|
||
)
|
||
index = 1
|
||
while index <= 100:
|
||
try:
|
||
ligne_p = factory_lignes.List(index)
|
||
if ligne_p is None:
|
||
break
|
||
nb_lignes_initial += 1
|
||
index += 1
|
||
except Exception:
|
||
break
|
||
logger.info(f" Lignes existantes: {nb_lignes_initial}")
|
||
except Exception as e:
|
||
logger.warning(f" Erreur comptage lignes: {e}")
|
||
|
||
champs_modifies = []
|
||
|
||
modif_date = config.champ_date_principale in doc_data
|
||
modif_date_sec = (
|
||
config.champ_date_secondaire
|
||
and config.champ_date_secondaire in doc_data
|
||
)
|
||
modif_statut = "statut" in doc_data
|
||
modif_ref = "reference" in doc_data
|
||
modif_lignes = "lignes" in doc_data and doc_data["lignes"] is not None
|
||
|
||
logger.info(" Modifications demandées:")
|
||
logger.info(f" {config.champ_date_principale}: {modif_date}")
|
||
if config.champ_date_secondaire:
|
||
logger.info(f" {config.champ_date_secondaire}: {modif_date_sec}")
|
||
logger.info(f" Statut: {modif_statut}")
|
||
logger.info(f" Référence: {modif_ref}")
|
||
logger.info(f" Lignes: {modif_lignes}")
|
||
|
||
doc_data_temp = doc_data.copy()
|
||
reference_a_modifier = None
|
||
statut_a_modifier = None
|
||
|
||
if modif_lignes:
|
||
if modif_ref:
|
||
reference_a_modifier = doc_data_temp.pop("reference")
|
||
modif_ref = False
|
||
if modif_statut:
|
||
statut_a_modifier = doc_data_temp.pop("statut")
|
||
modif_statut = False
|
||
|
||
logger.info(" Test Write() basique...")
|
||
try:
|
||
doc.Write()
|
||
doc.Read()
|
||
logger.info(" Write() basique OK")
|
||
except Exception as e:
|
||
logger.error(f" Document verrouillé: {e}")
|
||
raise ValueError(f"Document verrouillé: {e}")
|
||
|
||
if not modif_lignes and (
|
||
modif_date or modif_date_sec or modif_statut or modif_ref
|
||
):
|
||
logger.info(" Modifications simples...")
|
||
|
||
if modif_date:
|
||
date_principale = normaliser_date(
|
||
doc_data_temp.get(config.champ_date_principale)
|
||
)
|
||
doc.DO_Date = pywintypes.Time(date_principale)
|
||
try:
|
||
doc.DO_Heure = pywintypes.Time(date_principale)
|
||
except Exception:
|
||
pass
|
||
champs_modifies.append(config.champ_date_principale)
|
||
|
||
if modif_date_sec:
|
||
doc.DO_DateLivr = pywintypes.Time(
|
||
normaliser_date(doc_data_temp[config.champ_date_secondaire])
|
||
)
|
||
champs_modifies.append(config.champ_date_secondaire)
|
||
|
||
if modif_statut:
|
||
doc.DO_Statut = doc_data_temp["statut"]
|
||
champs_modifies.append("statut")
|
||
|
||
if modif_ref:
|
||
doc.DO_Ref = doc_data_temp["reference"]
|
||
champs_modifies.append("reference")
|
||
|
||
if type_document == TypeDocumentVente.FACTURE:
|
||
_configurer_facture(self, doc)
|
||
|
||
doc.Write()
|
||
logger.info(" Modifications appliquées")
|
||
|
||
elif modif_lignes:
|
||
logger.info(" REMPLACEMENT COMPLET DES LIGNES...")
|
||
|
||
if modif_date:
|
||
doc.DO_Date = pywintypes.Time(
|
||
normaliser_date(doc_data_temp.get(config.champ_date_principale))
|
||
)
|
||
champs_modifies.append(config.champ_date_principale)
|
||
|
||
if modif_date_sec:
|
||
doc.DO_DateLivr = pywintypes.Time(
|
||
normaliser_date(doc_data_temp[config.champ_date_secondaire])
|
||
)
|
||
champs_modifies.append(config.champ_date_secondaire)
|
||
|
||
if type_document == TypeDocumentVente.FACTURE:
|
||
_configurer_facture(self, doc)
|
||
|
||
nouvelles_lignes = doc_data["lignes"]
|
||
nb_nouvelles = len(nouvelles_lignes)
|
||
|
||
logger.info(f" {nb_lignes_initial} → {nb_nouvelles} lignes")
|
||
|
||
try:
|
||
factory_lignes = doc.FactoryDocumentLigne
|
||
except Exception:
|
||
factory_lignes = doc.FactoryDocumentVenteLigne
|
||
|
||
factory_article = self.cial.FactoryArticle
|
||
|
||
if nb_lignes_initial > 0:
|
||
logger.info(f" ️ Suppression {nb_lignes_initial} lignes...")
|
||
for idx in range(nb_lignes_initial, 0, -1):
|
||
try:
|
||
ligne_p = factory_lignes.List(idx)
|
||
if ligne_p:
|
||
ligne = win32com.client.CastTo(
|
||
ligne_p, "IBODocumentLigne3"
|
||
)
|
||
ligne.Read()
|
||
ligne.Remove()
|
||
except Exception as e:
|
||
logger.warning(f" Ligne {idx}: {e}")
|
||
logger.info(" Lignes supprimées")
|
||
|
||
logger.info(f" Ajout {nb_nouvelles} lignes...")
|
||
for idx, ligne_data in enumerate(nouvelles_lignes, 1):
|
||
_ajouter_ligne_document(
|
||
cial=self.cial,
|
||
factory_lignes=factory_lignes,
|
||
factory_article=factory_article,
|
||
ligne_data=ligne_data,
|
||
idx=idx,
|
||
doc=doc,
|
||
)
|
||
|
||
logger.info(" Nouvelles lignes ajoutées avec remises")
|
||
|
||
doc.Write()
|
||
time.sleep(0.5)
|
||
doc.Read()
|
||
|
||
champs_modifies.append("lignes")
|
||
|
||
if reference_a_modifier is not None:
|
||
try:
|
||
logger.info(
|
||
f" Modification référence: '{reference_a_modifier}'"
|
||
)
|
||
doc.DO_Ref = (
|
||
str(reference_a_modifier) if reference_a_modifier else ""
|
||
)
|
||
doc.Write()
|
||
time.sleep(0.5)
|
||
doc.Read()
|
||
champs_modifies.append("reference")
|
||
except Exception as e:
|
||
logger.warning(f" Référence: {e}")
|
||
|
||
if statut_a_modifier is not None:
|
||
try:
|
||
logger.info(f" Modification statut: {statut_a_modifier}")
|
||
doc.DO_Statut = int(statut_a_modifier)
|
||
doc.Write()
|
||
time.sleep(0.5)
|
||
doc.Read()
|
||
champs_modifies.append("statut")
|
||
except Exception as e:
|
||
logger.warning(f" Statut: {e}")
|
||
|
||
resultat = _relire_document_final(
|
||
self, config, numero, doc_data, client_code_fallback=client_code_initial
|
||
)
|
||
resultat["champs_modifies"] = champs_modifies
|
||
|
||
logger.info(f" {config.nom_document.upper()} {numero} MODIFIÉ")
|
||
logger.info(
|
||
f" Totaux: {resultat['total_ht']}€ HT / {resultat['total_ttc']}€ TTC"
|
||
)
|
||
logger.info(f" Champs modifiés: {champs_modifies}")
|
||
|
||
return resultat
|
||
|
||
except ValueError as e:
|
||
logger.error(f" ERREUR MÉTIER: {e}")
|
||
raise
|
||
except Exception as e:
|
||
logger.error(f" ERREUR TECHNIQUE: {e}", exc_info=True)
|
||
raise RuntimeError(f"Erreur Sage: {str(e)}")
|
||
|
||
|
||
__all__ = [
|
||
"creer_document_vente",
|
||
"_ajouter_ligne_document",
|
||
"_configurer_facture",
|
||
"_recuperer_numero_document",
|
||
"_relire_document_final",
|
||
"modifier_document_vente",
|
||
]
|