723 lines
27 KiB
Python
723 lines
27 KiB
Python
from typing import Dict, List, Optional
|
|
import win32com.client
|
|
import pywintypes
|
|
from datetime import datetime
|
|
from schemas.documents.reglements import ModeReglement
|
|
import time
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _get_journal_auto(self, mode_reglement: int) -> str:
|
|
with self._get_sql_connection() as conn:
|
|
cursor = conn.cursor()
|
|
if mode_reglement == ModeReglement.ESPECES:
|
|
cursor.execute(
|
|
"SELECT TOP 1 JO_Num FROM F_JOURNAUX WHERE JO_Type = 2 AND JO_Reglement = 1 AND CG_Num LIKE '53%' ORDER BY JO_Num"
|
|
)
|
|
else:
|
|
cursor.execute(
|
|
"SELECT TOP 1 JO_Num FROM F_JOURNAUX WHERE JO_Type = 2 AND JO_Reglement = 1 AND CG_Num LIKE '51%' ORDER BY JO_Num"
|
|
)
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return row[0].strip()
|
|
cursor.execute(
|
|
"SELECT TOP 1 JO_Num FROM F_JOURNAUX WHERE JO_Type = 2 AND JO_Reglement = 1 ORDER BY JO_Num"
|
|
)
|
|
row = cursor.fetchone()
|
|
if row:
|
|
return row[0].strip()
|
|
raise ValueError("Aucun journal de trésorerie configuré")
|
|
|
|
|
|
def lire_journaux_banque(self) -> List[Dict]:
|
|
if not self.cial:
|
|
raise RuntimeError("Connexion Sage non établie")
|
|
with self._get_sql_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT JO_Num, JO_Intitule, CG_Num FROM F_JOURNAUX WHERE JO_Type = 2 AND JO_Reglement = 1 ORDER BY JO_Num"
|
|
)
|
|
return [
|
|
{
|
|
"code": row[0].strip(),
|
|
"intitule": row[1].strip() if row[1] else "",
|
|
"compte_general": row[2].strip() if row[2] else "",
|
|
"type": "Caisse" if (row[2] or "").startswith("53") else "Banque",
|
|
}
|
|
for row in cursor.fetchall()
|
|
]
|
|
|
|
|
|
def lire_tous_journaux(self) -> List[Dict]:
|
|
if not self.cial:
|
|
raise RuntimeError("Connexion Sage non établie")
|
|
types_libelles = {
|
|
0: "Achats",
|
|
1: "Ventes",
|
|
2: "Trésorerie",
|
|
3: "Général",
|
|
4: "Situation",
|
|
}
|
|
with self._get_sql_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT JO_Num, JO_Intitule, JO_Type, JO_Reglement, CG_Num FROM F_JOURNAUX ORDER BY JO_Type, JO_Num"
|
|
)
|
|
return [
|
|
{
|
|
"code": row[0].strip(),
|
|
"intitule": row[1].strip() if row[1] else "",
|
|
"type_code": row[2],
|
|
"type_libelle": types_libelles.get(row[2], f"Type {row[2]}"),
|
|
"reglement_actif": row[3] == 1,
|
|
"compte_general": row[4].strip() if row[4] else "",
|
|
}
|
|
for row in cursor.fetchall()
|
|
]
|
|
|
|
|
|
def regler_facture(
|
|
self,
|
|
numero_facture: str,
|
|
montant: float,
|
|
mode_reglement: int = ModeReglement.CHEQUE,
|
|
date_reglement: datetime = None,
|
|
reference: str = "",
|
|
libelle: str = "",
|
|
) -> Dict:
|
|
if not self.cial:
|
|
raise RuntimeError("Connexion Sage non établie")
|
|
if montant <= 0:
|
|
raise ValueError("Le montant du règlement doit être positif")
|
|
|
|
date_reglement = date_reglement or datetime.now()
|
|
code_journal = _get_journal_auto(self, mode_reglement)
|
|
logger.info(
|
|
f"Règlement facture {numero_facture}: {montant}€ (mode: {mode_reglement}, journal: {code_journal})"
|
|
)
|
|
|
|
try:
|
|
with self._com_context(), self._lock_com:
|
|
factory = self.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()
|
|
|
|
total_ttc = float(getattr(doc, "DO_TotalTTC", 0.0))
|
|
montant_deja_regle = float(getattr(doc, "DO_MontantRegle", 0.0))
|
|
statut = getattr(doc, "DO_Statut", 0)
|
|
|
|
if statut == 6:
|
|
raise ValueError(f"Facture {numero_facture} annulée")
|
|
|
|
solde_actuel = total_ttc - montant_deja_regle
|
|
if montant > solde_actuel + 0.01:
|
|
raise ValueError(
|
|
f"Montant ({montant}€) supérieur au solde ({solde_actuel:.2f}€)"
|
|
)
|
|
|
|
client_code = ""
|
|
try:
|
|
client_obj = getattr(doc, "Client", None)
|
|
if client_obj:
|
|
client_obj.Read()
|
|
client_code = getattr(client_obj, "CT_Num", "").strip()
|
|
except Exception:
|
|
pass
|
|
|
|
echeance = _get_premiere_echeance(doc)
|
|
if not echeance:
|
|
raise ValueError(f"Facture {numero_facture} sans échéance")
|
|
|
|
transaction_active = False
|
|
try:
|
|
self.cial.CptaApplication.BeginTrans()
|
|
transaction_active = True
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
numero_reglement = _executer_reglement_com(
|
|
self,
|
|
doc,
|
|
echeance,
|
|
montant,
|
|
mode_reglement,
|
|
date_reglement,
|
|
reference,
|
|
libelle,
|
|
code_journal,
|
|
client_code,
|
|
numero_facture,
|
|
)
|
|
|
|
if transaction_active:
|
|
try:
|
|
self.cial.CptaApplication.CommitTrans()
|
|
except Exception:
|
|
pass
|
|
|
|
time.sleep(0.5)
|
|
doc.Read()
|
|
nouveau_montant_regle = float(getattr(doc, "DO_MontantRegle", 0.0))
|
|
|
|
if abs(nouveau_montant_regle - montant_deja_regle) < 0.01:
|
|
raise RuntimeError(
|
|
"Le règlement n'a pas été appliqué (DO_MontantRegle inchangé)"
|
|
)
|
|
|
|
nouveau_solde = total_ttc - nouveau_montant_regle
|
|
logger.info(f"Règlement effectué - Solde restant: {nouveau_solde:.2f}€")
|
|
|
|
return {
|
|
"numero_facture": numero_facture,
|
|
"numero_reglement": numero_reglement,
|
|
"montant_regle": montant,
|
|
"date_reglement": date_reglement.strftime("%Y-%m-%d"),
|
|
"mode_reglement": mode_reglement,
|
|
"mode_reglement_libelle": ModeReglement.get_libelle(mode_reglement),
|
|
"reference": reference,
|
|
"libelle": libelle,
|
|
"code_journal": code_journal,
|
|
"total_facture": total_ttc,
|
|
"solde_restant": nouveau_solde,
|
|
"facture_soldee": nouveau_solde < 0.01,
|
|
"client_code": client_code,
|
|
}
|
|
except Exception:
|
|
if transaction_active:
|
|
try:
|
|
self.cial.CptaApplication.RollbackTrans()
|
|
except Exception:
|
|
pass
|
|
raise
|
|
except ValueError:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Erreur règlement: {e}", exc_info=True)
|
|
raise RuntimeError(f"Échec règlement facture: {str(e)}")
|
|
|
|
|
|
def _get_premiere_echeance(doc):
|
|
try:
|
|
factory_ech = getattr(doc, "FactoryDocumentEcheance", None)
|
|
if factory_ech:
|
|
ech_list = factory_ech.List
|
|
if ech_list:
|
|
echeance = ech_list.Item(1)
|
|
if echeance:
|
|
for iface in ["IBODocumentEcheance3", "IBODocumentEcheance"]:
|
|
try:
|
|
echeance = win32com.client.CastTo(echeance, iface)
|
|
logger.info(f" Échéance castée vers {iface}")
|
|
break
|
|
except Exception:
|
|
continue
|
|
echeance.Read()
|
|
return echeance
|
|
except Exception as e:
|
|
logger.warning(f" Pas d'échéance: {e}")
|
|
return None
|
|
|
|
|
|
def _executer_reglement_com(
|
|
self,
|
|
doc,
|
|
echeance,
|
|
montant,
|
|
mode_reglement,
|
|
date_reglement,
|
|
reference,
|
|
libelle,
|
|
code_journal,
|
|
client_code,
|
|
numero_facture,
|
|
):
|
|
erreurs = []
|
|
|
|
# Approche 1: CreateProcess_ReglerEcheances - Créer règlement puis l'assigner
|
|
try:
|
|
logger.info(
|
|
"Tentative via CreateProcess_ReglerEcheances avec règlement créé..."
|
|
)
|
|
process = self.cial.CreateProcess_ReglerEcheances()
|
|
|
|
if process:
|
|
# D'abord créer un règlement via FactoryDocumentReglement
|
|
factory_reg = self.cial.FactoryDocumentReglement
|
|
reg = factory_reg.Create()
|
|
reg = win32com.client.CastTo(reg, "IBODocumentReglement")
|
|
|
|
# Configurer le règlement
|
|
try:
|
|
journal_factory = self.cial.CptaApplication.FactoryJournal
|
|
journal_persist = journal_factory.ReadNumero(code_journal)
|
|
if journal_persist:
|
|
reg.Journal = journal_persist
|
|
logger.info(f" Journal: {code_journal}")
|
|
except Exception as e:
|
|
logger.warning(f" Journal: {e}")
|
|
|
|
try:
|
|
factory_client = self.cial.CptaApplication.FactoryClient
|
|
if client_code:
|
|
client_persist = factory_client.ReadNumero(client_code)
|
|
if client_persist:
|
|
reg.TiersPayeur = client_persist
|
|
logger.info(f" TiersPayeur: {client_code}")
|
|
except Exception as e:
|
|
logger.warning(f" TiersPayeur: {e}")
|
|
|
|
try:
|
|
reg.RG_Date = pywintypes.Time(date_reglement)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
reg.RG_Montant = montant
|
|
except Exception:
|
|
pass
|
|
if reference:
|
|
try:
|
|
reg.RG_Reference = reference
|
|
except Exception:
|
|
pass
|
|
if libelle:
|
|
try:
|
|
reg.RG_Libelle = libelle
|
|
except Exception:
|
|
pass
|
|
try:
|
|
reg.RG_Impute = 1
|
|
except Exception:
|
|
pass
|
|
|
|
# Assigner le règlement au process
|
|
try:
|
|
process.Reglement = reg
|
|
logger.info(" Règlement assigné au process")
|
|
except Exception as e:
|
|
logger.warning(f" Assignation règlement: {e}")
|
|
|
|
# Ajouter l'échéance avec montant
|
|
try:
|
|
process.AddDocumentEcheanceMontant(echeance, montant)
|
|
logger.info(" Échéance ajoutée avec montant")
|
|
except Exception as e1:
|
|
logger.debug(f" AddDocumentEcheanceMontant: {e1}")
|
|
try:
|
|
process.AddDocumentEcheance(echeance)
|
|
logger.info(" Échéance ajoutée")
|
|
except Exception as e2:
|
|
raise RuntimeError(f"AddEcheance: {e2}")
|
|
|
|
can_process = getattr(process, "CanProcess", True)
|
|
logger.info(f" CanProcess: {can_process}")
|
|
|
|
if can_process:
|
|
process.Process()
|
|
logger.info(" Process() réussi")
|
|
|
|
numero = None
|
|
try:
|
|
result = getattr(process, "ReglementResult", None)
|
|
if result:
|
|
result.Read()
|
|
numero = getattr(result, "RG_Piece", "")
|
|
except Exception:
|
|
pass
|
|
|
|
return str(numero) if numero else None
|
|
|
|
except Exception as e:
|
|
erreurs.append(f"CreateProcess avec règlement: {e}")
|
|
logger.warning(f"CreateProcess avec règlement échoué: {e}")
|
|
|
|
# Approche 2: Configurer le Reglement du process directement (toutes propriétés)
|
|
try:
|
|
logger.info("Tentative via configuration complète Process.Reglement...")
|
|
process = self.cial.CreateProcess_ReglerEcheances()
|
|
|
|
if process:
|
|
reglement = getattr(process, "Reglement", None)
|
|
if reglement:
|
|
# Lister TOUS les attributs
|
|
reg_attrs = [a for a in dir(reglement) if not a.startswith("_")]
|
|
logger.info(f" Attributs Reglement: {reg_attrs}")
|
|
|
|
# Configurer TOUT
|
|
_set_safe(
|
|
reglement, ["RG_Date", "Date"], pywintypes.Time(date_reglement)
|
|
)
|
|
_set_safe(reglement, ["RG_Montant", "Montant"], montant)
|
|
_set_safe(reglement, ["JO_Num", "Journal", "CodeJournal"], code_journal)
|
|
_set_safe(
|
|
reglement,
|
|
["CT_NumPayeur", "CT_Num", "Tiers", "Client"],
|
|
client_code,
|
|
)
|
|
_set_safe(
|
|
reglement,
|
|
["N_Reglement", "ModeReglement", "RG_ModeReglement"],
|
|
mode_reglement,
|
|
)
|
|
_set_safe(reglement, ["RG_Type", "Type"], 0) # 0 = Client
|
|
_set_safe(reglement, ["RG_Impute", "Impute"], 1)
|
|
_set_safe(reglement, ["RG_Compta", "Compta"], 0)
|
|
|
|
if reference:
|
|
_set_safe(reglement, ["RG_Reference", "Reference"], reference)
|
|
if libelle:
|
|
_set_safe(reglement, ["RG_Libelle", "Libelle"], libelle)
|
|
|
|
# Essayer SetDefault
|
|
try:
|
|
reglement.SetDefault()
|
|
logger.info(" SetDefault() appelé")
|
|
except Exception:
|
|
pass
|
|
|
|
logger.info(" Reglement configuré")
|
|
|
|
# Ajouter l'échéance
|
|
try:
|
|
process.AddDocumentEcheanceMontant(echeance, montant)
|
|
logger.info(" Échéance ajoutée")
|
|
except Exception as e1:
|
|
try:
|
|
process.AddDocumentEcheance(echeance)
|
|
logger.info(" Échéance ajoutée (sans montant)")
|
|
except Exception as e2:
|
|
raise RuntimeError(f"AddEcheance: {e2}")
|
|
|
|
can_process = getattr(process, "CanProcess", True)
|
|
logger.info(f" CanProcess: {can_process}")
|
|
|
|
# Vérifier les erreurs du process
|
|
try:
|
|
errors = getattr(process, "Errors", None)
|
|
if errors:
|
|
err_count = getattr(errors, "Count", 0)
|
|
for i in range(1, err_count + 1):
|
|
err = errors.Item(i)
|
|
logger.warning(f" Erreur process [{i}]: {err}")
|
|
except Exception:
|
|
pass
|
|
|
|
if can_process:
|
|
process.Process()
|
|
logger.info(" Process() réussi")
|
|
|
|
numero = None
|
|
try:
|
|
result = getattr(process, "ReglementResult", None)
|
|
if result:
|
|
result.Read()
|
|
numero = getattr(result, "RG_Piece", "")
|
|
except Exception:
|
|
pass
|
|
|
|
return str(numero) if numero else None
|
|
else:
|
|
logger.warning(" CanProcess = False!")
|
|
|
|
except Exception as e:
|
|
erreurs.append(f"Config complète: {e}")
|
|
logger.warning(f"Config complète échouée: {e}")
|
|
|
|
# Approche 3: Utiliser SetDefaultReglement sur le document
|
|
try:
|
|
logger.info("Tentative via doc.SetDefaultReglement...")
|
|
|
|
if hasattr(doc, "SetDefaultReglement"):
|
|
doc.SetDefaultReglement()
|
|
logger.info(" SetDefaultReglement() appelé")
|
|
|
|
# Configurer via le règlement par défaut
|
|
reg = getattr(doc, "Reglement", None)
|
|
if reg:
|
|
attrs = [a for a in dir(reg) if not a.startswith("_")]
|
|
logger.info(f" Attributs doc.Reglement: {attrs[:15]}...")
|
|
|
|
except Exception as e:
|
|
erreurs.append(f"SetDefaultReglement: {e}")
|
|
logger.warning(f"SetDefaultReglement: {e}")
|
|
|
|
raise RuntimeError(f"Aucune méthode n'a fonctionné. Erreurs: {'; '.join(erreurs)}")
|
|
|
|
|
|
def _set_safe(obj, attrs, value):
|
|
for attr in attrs:
|
|
try:
|
|
setattr(obj, attr, value)
|
|
logger.debug(f" {attr} = {value}")
|
|
return True
|
|
except Exception:
|
|
continue
|
|
return False
|
|
|
|
|
|
def introspecter_reglement(self):
|
|
if not self.cial:
|
|
raise RuntimeError("Connexion Sage non établie")
|
|
result = {}
|
|
try:
|
|
with self._com_context(), self._lock_com:
|
|
# Process et son Reglement
|
|
try:
|
|
process = self.cial.CreateProcess_ReglerEcheances()
|
|
result["Process"] = [a for a in dir(process) if not a.startswith("_")]
|
|
|
|
reglement = getattr(process, "Reglement", None)
|
|
if reglement:
|
|
result["Process_Reglement"] = [
|
|
a for a in dir(reglement) if not a.startswith("_")
|
|
]
|
|
|
|
# Essayer de lire les valeurs par défaut
|
|
for attr in ["RG_Type", "RG_Impute", "JO_Num", "CT_NumPayeur"]:
|
|
try:
|
|
val = getattr(reglement, attr, "N/A")
|
|
result[f"Reglement_{attr}"] = str(val)
|
|
except Exception:
|
|
pass
|
|
except Exception as e:
|
|
result["error_process"] = str(e)
|
|
|
|
# IBODocumentReglement
|
|
try:
|
|
factory = self.cial.FactoryDocumentReglement
|
|
reg = factory.Create()
|
|
reg = win32com.client.CastTo(reg, "IBODocumentReglement")
|
|
result["IBODocumentReglement"] = [
|
|
a for a in dir(reg) if not a.startswith("_")
|
|
]
|
|
except Exception as e:
|
|
result["error_reglement"] = str(e)
|
|
|
|
# Échéance
|
|
try:
|
|
factory_doc = self.cial.FactoryDocumentVente
|
|
doc_list = factory_doc.List
|
|
for i in range(1, 20):
|
|
try:
|
|
doc = doc_list.Item(i)
|
|
if doc:
|
|
doc.Read()
|
|
if getattr(doc, "DO_Type", 0) == 6:
|
|
factory_ech = getattr(
|
|
doc, "FactoryDocumentEcheance", None
|
|
)
|
|
if factory_ech:
|
|
ech_list = factory_ech.List
|
|
if ech_list:
|
|
ech = ech_list.Item(1)
|
|
if ech:
|
|
ech = win32com.client.CastTo(
|
|
ech, "IBODocumentEcheance3"
|
|
)
|
|
ech.Read()
|
|
result["IBODocumentEcheance3"] = [
|
|
a
|
|
for a in dir(ech)
|
|
if not a.startswith("_")
|
|
]
|
|
|
|
# Mode règlement de l'échéance
|
|
mode = getattr(ech, "Reglement", None)
|
|
if mode:
|
|
result["Echeance_Reglement"] = [
|
|
a
|
|
for a in dir(mode)
|
|
if not a.startswith("_")
|
|
]
|
|
break
|
|
except Exception:
|
|
continue
|
|
except Exception as e:
|
|
result["error_echeance"] = str(e)
|
|
except Exception as e:
|
|
result["global_error"] = str(e)
|
|
return result
|
|
|
|
|
|
def regler_factures_client(
|
|
self,
|
|
client_code,
|
|
montant_total,
|
|
mode_reglement=ModeReglement.CHEQUE,
|
|
date_reglement=None,
|
|
reference="",
|
|
libelle="",
|
|
numeros_factures=None,
|
|
):
|
|
if not self.cial:
|
|
raise RuntimeError("Connexion Sage non établie")
|
|
if montant_total <= 0:
|
|
raise ValueError("Le montant total doit être positif")
|
|
date_reglement = date_reglement or datetime.now()
|
|
factures = _get_factures_non_soldees_client_sql(self, client_code, numeros_factures)
|
|
if not factures:
|
|
raise ValueError(f"Aucune facture à régler pour {client_code}")
|
|
solde_total = sum(f["solde"] for f in factures)
|
|
if montant_total > solde_total + 0.01:
|
|
raise ValueError(
|
|
f"Montant ({montant_total}€) supérieur au solde ({solde_total:.2f}€)"
|
|
)
|
|
reglements = []
|
|
restant = montant_total
|
|
for fac in factures:
|
|
if restant < 0.01:
|
|
break
|
|
a_regler = min(restant, fac["solde"])
|
|
try:
|
|
res = regler_facture(
|
|
self,
|
|
fac["numero"],
|
|
a_regler,
|
|
mode_reglement,
|
|
date_reglement,
|
|
reference,
|
|
libelle,
|
|
)
|
|
reglements.append(res)
|
|
restant -= a_regler
|
|
except Exception as e:
|
|
logger.error(f"Erreur {fac['numero']}: {e}")
|
|
break
|
|
if not reglements:
|
|
raise RuntimeError("Aucun règlement effectué")
|
|
return {
|
|
"client_code": client_code,
|
|
"montant_demande": montant_total,
|
|
"montant_effectif": sum(r["montant_regle"] for r in reglements),
|
|
"nb_factures_reglees": len(reglements),
|
|
"nb_factures_soldees": sum(1 for r in reglements if r["facture_soldee"]),
|
|
"date_reglement": date_reglement.strftime("%Y-%m-%d"),
|
|
"mode_reglement": mode_reglement,
|
|
"mode_reglement_libelle": ModeReglement.get_libelle(mode_reglement),
|
|
"reference": reference,
|
|
"reglements": reglements,
|
|
}
|
|
|
|
|
|
def _get_factures_non_soldees_client_sql(self, client_code, numeros=None):
|
|
with self._get_sql_connection() as conn:
|
|
cursor = conn.cursor()
|
|
if numeros:
|
|
placeholders = ",".join("?" * len(numeros))
|
|
query = f"SELECT DO_Piece, DO_Date, DO_TotalTTC, DO_MontantRegle FROM F_DOCENTETE WHERE DO_Type = 6 AND DO_Tiers = ? AND DO_Piece IN ({placeholders}) AND DO_Statut <> 6 AND (DO_TotalTTC - ISNULL(DO_MontantRegle, 0)) > 0.01 ORDER BY DO_Date ASC"
|
|
params = [client_code] + numeros
|
|
else:
|
|
query = "SELECT DO_Piece, DO_Date, DO_TotalTTC, DO_MontantRegle FROM F_DOCENTETE WHERE DO_Type = 6 AND DO_Tiers = ? AND DO_Statut <> 6 AND (DO_TotalTTC - ISNULL(DO_MontantRegle, 0)) > 0.01 ORDER BY DO_Date ASC"
|
|
params = [client_code]
|
|
cursor.execute(query, params)
|
|
return [
|
|
{
|
|
"numero": r[0].strip(),
|
|
"date": r[1].strftime("%Y-%m-%d") if r[1] else None,
|
|
"total_ttc": float(r[2] or 0),
|
|
"montant_regle": float(r[3] or 0),
|
|
"solde": float(r[2] or 0) - float(r[3] or 0),
|
|
}
|
|
for r in cursor.fetchall()
|
|
]
|
|
|
|
|
|
def lire_reglements_facture(self, numero_facture):
|
|
if not self.cial:
|
|
raise RuntimeError("Connexion Sage non établie")
|
|
with self._get_sql_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT DO_TotalTTC, DO_MontantRegle, DO_Tiers, 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")
|
|
total = float(row[0] or 0)
|
|
regle = float(row[1] or 0)
|
|
solde = max(0, total - regle)
|
|
return {
|
|
"numero_facture": numero_facture,
|
|
"client_code": (row[2] or "").strip(),
|
|
"date_facture": row[3].strftime("%Y-%m-%d") if row[3] else None,
|
|
"reference": (row[4] or "").strip(),
|
|
"total_ttc": total,
|
|
"total_regle": regle,
|
|
"solde_restant": solde,
|
|
"est_soldee": solde < 0.01,
|
|
}
|
|
|
|
|
|
def lire_reglements_client(
|
|
self, client_code, date_debut=None, date_fin=None, inclure_soldees=True
|
|
):
|
|
if not self.cial:
|
|
raise RuntimeError("Connexion Sage non établie")
|
|
with self._get_sql_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT CT_Intitule FROM F_COMPTET WHERE CT_Num = ?", (client_code,)
|
|
)
|
|
row = cursor.fetchone()
|
|
if not row:
|
|
raise ValueError(f"Client {client_code} introuvable")
|
|
intitule = (row[0] or "").strip()
|
|
query = "SELECT DO_Piece, DO_Date, DO_TotalTTC, DO_MontantRegle, DO_Ref FROM F_DOCENTETE WHERE DO_Type = 6 AND DO_Tiers = ?"
|
|
params = [client_code]
|
|
if date_debut:
|
|
query += " AND DO_Date >= ?"
|
|
params.append(date_debut)
|
|
if date_fin:
|
|
query += " AND DO_Date <= ?"
|
|
params.append(date_fin)
|
|
query += " ORDER BY DO_Date ASC"
|
|
cursor.execute(query, params)
|
|
factures = []
|
|
for r in cursor.fetchall():
|
|
total = float(r[2] or 0)
|
|
regle = float(r[3] or 0)
|
|
solde = max(0, total - regle)
|
|
soldee = solde < 0.01
|
|
if inclure_soldees or not soldee:
|
|
factures.append(
|
|
{
|
|
"numero_facture": r[0].strip(),
|
|
"date_facture": r[1].strftime("%Y-%m-%d") if r[1] else None,
|
|
"total_ttc": total,
|
|
"reference": (r[4] or "").strip(),
|
|
"total_regle": regle,
|
|
"solde_restant": solde,
|
|
"est_soldee": soldee,
|
|
}
|
|
)
|
|
return {
|
|
"client_code": client_code,
|
|
"client_intitule": intitule,
|
|
"nb_factures": len(factures),
|
|
"nb_factures_soldees": sum(1 for f in factures if f["est_soldee"]),
|
|
"nb_factures_en_cours": sum(1 for f in factures if not f["est_soldee"]),
|
|
"total_factures": sum(f["total_ttc"] for f in factures),
|
|
"total_regle": sum(f["total_regle"] for f in factures),
|
|
"solde_global": sum(f["solde_restant"] for f in factures),
|
|
"factures": factures,
|
|
}
|
|
|
|
|
|
__all__ = [
|
|
"ModeReglement",
|
|
"lire_journaux_banque",
|
|
"lire_tous_journaux",
|
|
"introspecter_reglement",
|
|
"regler_facture",
|
|
"regler_factures_client",
|
|
"lire_reglements_facture",
|
|
"lire_reglements_client",
|
|
]
|