Sage100-ws/utils/functions/data/create_doc.py

707 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

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:
# Démarrage transaction
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}")
# Création du document
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éé")
# ===== DATES =====
doc.DO_Date = pywintypes.Time(
normaliser_date(doc_data.get(config.champ_date_principale))
)
# Date secondaire (livraison, etc.)
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]}"
)
# ===== CLIENT =====
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é")
# ===== RÉFÉRENCE =====
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}")
# ===== CONFIGURATION SPÉCIFIQUE FACTURE =====
if type_document == TypeDocumentVente.FACTURE:
_configurer_facture(self, doc)
# ===== FACTORY LIGNES =====
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...")
# ===== TRAITEMENT DES 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,
)
# ===== VALIDATION =====
logger.info("💾 Validation du document...")
# Pour les factures, réassocier le client avant validation
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()
# Process() sauf pour devis en brouillon
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")
# Commit transaction
if transaction_active:
try:
self.cial.CptaApplication.CommitTrans()
logger.info("✓ Transaction committée")
except Exception:
pass
time.sleep(2)
# ===== RÉCUPÉRATION DU NUMÉRO =====
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}")
# ===== RELECTURE POUR TOTAUX =====
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_
# 1. Récupérer l'objet Remise
dispid = dispatch.GetIDsOfNames(0, "Remise")
remise_obj = dispatch.Invoke(dispid, 0, pythoncom.DISPATCH_PROPERTYGET, 1)
remise_wrapper = win32com.client.Dispatch(remise_obj)
# 2. Définir la remise via FromString
remise_wrapper.FromString(f"{remise_pourcent}%")
# 3. Calcul (optionnel mais recommandé)
try:
remise_wrapper.Calcul()
except Exception:
pass
# 4. Write la ligne
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']}")
# ===== CRÉATION LIGNE =====
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"])
# ===== ASSOCIATION ARTICLE =====
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 =====
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}")
# ===== WRITE INITIAL =====
ligne_obj.Write()
# ===== APPLICATION REMISE (TOUTES LES LIGNES!) =====
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", "")
# Récupérer le client depuis le document Sage
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
# Date secondaire
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:
# Valeurs par défaut si relecture échoue
total_ht = 0.0
total_ttc = 0.0
reference_finale = doc_data.get("reference", "")
date_secondaire_value = doc_data.get(config.champ_date_secondaire)
# Fallback pour le code client (priorité: Sage > fallback > doc_data)
if not client_code:
client_code = client_code_fallback or doc_data.get("client", {}).get("code", "")
# Construction du résultat
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,
}
# Ajout date secondaire si applicable
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:
"""
Méthode unifiée de modification de documents de vente
RÉUTILISE les mêmes sous-méthodes que la création
"""
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:
# ==========================================
# 1. CHARGEMENT DOCUMENT
# ==========================================
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()
# Vérifications
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é")
# ==========================================
# 2. SAUVEGARDE CLIENT INITIAL
# ==========================================
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")
# ==========================================
# 3. COMPTAGE LIGNES EXISTANTES
# ==========================================
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}")
# ==========================================
# 4. ANALYSE MODIFICATIONS
# ==========================================
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}")
# Reporter référence et statut après les 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
# ==========================================
# 5. TEST WRITE BASIQUE
# ==========================================
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}")
# ==========================================
# 6. MODIFICATIONS SIMPLES (sans lignes)
# ==========================================
if not modif_lignes and (
modif_date or modif_date_sec or modif_statut or modif_ref
):
logger.info("📝 Modifications simples...")
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 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")
# 🔥 CONFIGURATION SPÉCIFIQUE FACTURE
if type_document == TypeDocumentVente.FACTURE:
_configurer_facture(self, doc)
doc.Write()
logger.info(" ✓ Modifications appliquées")
# ==========================================
# 7. REMPLACEMENT LIGNES
# ==========================================
elif modif_lignes:
logger.info("🔄 REMPLACEMENT COMPLET DES LIGNES...")
# Dates
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)
# 🔥 CONFIGURATION SPÉCIFIQUE FACTURE (avant lignes)
if type_document == TypeDocumentVente.FACTURE:
_configurer_facture(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
# Suppression lignes existantes
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")
# Ajout nouvelles lignes avec REMISES
logger.info(f" Ajout {nb_nouvelles} lignes...")
for idx, ligne_data in enumerate(nouvelles_lignes, 1):
# 🔥 UTILISE _ajouter_ligne_document qui applique les remises
_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")
# ==========================================
# 8. MODIFICATIONS REPORTÉES
# ==========================================
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}")
# ==========================================
# 9. RELECTURE FINALE
# ==========================================
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",
]