782 lines
30 KiB
Python
782 lines
30 KiB
Python
from typing import Dict, List
|
|
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")
|
|
|
|
numero_reglement = _executer_reglement_com(
|
|
self,
|
|
doc,
|
|
echeance,
|
|
montant,
|
|
mode_reglement,
|
|
date_reglement,
|
|
reference,
|
|
libelle,
|
|
code_journal,
|
|
client_code,
|
|
numero_facture,
|
|
)
|
|
|
|
time.sleep(0.3)
|
|
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 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 PRINCIPALE: Créer règlement complet, l'écrire, puis l'assigner au process
|
|
try:
|
|
logger.info("Création du règlement via FactoryDocumentReglement...")
|
|
|
|
# 1. Créer le règlement
|
|
factory_reg = self.cial.FactoryDocumentReglement
|
|
reg = factory_reg.Create()
|
|
reg = win32com.client.CastTo(reg, "IBODocumentReglement")
|
|
logger.info(" Règlement créé et casté vers IBODocumentReglement")
|
|
|
|
# 2. Configurer le Journal (objet)
|
|
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 défini: {code_journal}")
|
|
except Exception as e:
|
|
logger.warning(f" Journal: {e}")
|
|
|
|
# 3. Configurer le TiersPayeur (objet client)
|
|
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 défini: {client_code}")
|
|
except Exception as e:
|
|
logger.warning(f" TiersPayeur: {e}")
|
|
|
|
# 4. Configurer les champs simples
|
|
try:
|
|
reg.RG_Date = pywintypes.Time(date_reglement)
|
|
logger.info(f" RG_Date: {date_reglement}")
|
|
except Exception as e:
|
|
logger.warning(f" RG_Date: {e}")
|
|
|
|
try:
|
|
reg.RG_Montant = montant
|
|
logger.info(f" RG_Montant: {montant}")
|
|
except Exception as e:
|
|
logger.warning(f" RG_Montant: {e}")
|
|
|
|
# 5. Mode de règlement via l'objet Reglement
|
|
try:
|
|
# Lire le mode de règlement depuis la base
|
|
mode_factory = getattr(
|
|
self.cial.CptaApplication, "FactoryModeReglement", None
|
|
)
|
|
if mode_factory:
|
|
mode_obj = mode_factory.ReadNumero(mode_reglement)
|
|
if mode_obj:
|
|
reg.Reglement = mode_obj
|
|
logger.info(f" Mode règlement défini: {mode_reglement}")
|
|
except Exception as e:
|
|
logger.debug(f" Mode règlement via factory: {e}")
|
|
|
|
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 # Imputé
|
|
except Exception:
|
|
pass
|
|
|
|
try:
|
|
reg.RG_Compta = 0 # Non comptabilisé
|
|
except Exception:
|
|
pass
|
|
|
|
# 6. ÉCRIRE le règlement
|
|
reg.Write()
|
|
numero = getattr(reg, "RG_Piece", None)
|
|
logger.info(f" Règlement écrit avec numéro: {numero}")
|
|
|
|
# 7. Créer le lien règlement-échéance via la factory DU RÈGLEMENT
|
|
try:
|
|
logger.info(" Création du lien règlement-échéance...")
|
|
factory_reg_ech = getattr(reg, "FactoryDocumentReglementEcheance", None)
|
|
|
|
if factory_reg_ech:
|
|
reg_ech = factory_reg_ech.Create()
|
|
|
|
# Cast vers IBODocumentReglementEcheance
|
|
for iface in [
|
|
"IBODocumentReglementEcheance3",
|
|
"IBODocumentReglementEcheance",
|
|
]:
|
|
try:
|
|
reg_ech = win32com.client.CastTo(reg_ech, iface)
|
|
logger.info(f" Cast vers {iface} réussi")
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
# Définir l'échéance - le Reglement est déjà lié via la factory
|
|
try:
|
|
reg_ech.Echeance = echeance
|
|
logger.info(" Echeance définie")
|
|
except Exception as e:
|
|
logger.warning(f" Echeance: {e}")
|
|
|
|
# Définir le montant
|
|
try:
|
|
reg_ech.RC_Montant = montant
|
|
logger.info(f" RC_Montant: {montant}")
|
|
except Exception as e:
|
|
logger.warning(f" RC_Montant: {e}")
|
|
|
|
# Écrire le lien
|
|
try:
|
|
reg_ech.SetDefault()
|
|
except Exception:
|
|
pass
|
|
|
|
reg_ech.Write()
|
|
logger.info(" Lien règlement-échéance écrit!")
|
|
|
|
return str(numero) if numero else None
|
|
|
|
except Exception as e:
|
|
erreurs.append(f"Lien échéance: {e}")
|
|
logger.warning(f" Erreur création lien: {e}")
|
|
|
|
# Si le lien a échoué, essayer via le process
|
|
logger.info(" Tentative via CreateProcess_ReglerEcheances...")
|
|
try:
|
|
process = self.cial.CreateProcess_ReglerEcheances()
|
|
|
|
# Assigner le règlement déjà écrit
|
|
process.Reglement = reg
|
|
logger.info(" Règlement assigné au process")
|
|
|
|
# Ajouter l'échéance
|
|
try:
|
|
process.AddDocumentEcheanceMontant(echeance, montant)
|
|
logger.info(" Échéance ajoutée avec montant")
|
|
except Exception:
|
|
process.AddDocumentEcheance(echeance)
|
|
logger.info(" Échéance ajoutée")
|
|
|
|
can_process = getattr(process, "CanProcess", True)
|
|
logger.info(f" CanProcess: {can_process}")
|
|
|
|
if can_process:
|
|
process.Process()
|
|
logger.info(" Process() réussi!")
|
|
return str(numero) if numero else None
|
|
else:
|
|
# Vérifier les erreurs
|
|
try:
|
|
errors = process.Errors
|
|
if errors:
|
|
for i in range(1, errors.Count + 1):
|
|
logger.warning(f" Erreur [{i}]: {errors.Item(i)}")
|
|
except Exception:
|
|
pass
|
|
|
|
except Exception as e:
|
|
erreurs.append(f"Process: {e}")
|
|
logger.warning(f" Process échoué: {e}")
|
|
|
|
except Exception as e:
|
|
erreurs.append(f"FactoryDocumentReglement: {e}")
|
|
logger.error(f"FactoryDocumentReglement échoué: {e}")
|
|
|
|
# APPROCHE ALTERNATIVE: Via le mode règlement de l'échéance
|
|
try:
|
|
logger.info("Tentative via modification directe de l'échéance...")
|
|
|
|
# L'échéance a un attribut Reglement qui est le mode de règlement
|
|
mode_obj = getattr(echeance, "Reglement", None)
|
|
if mode_obj:
|
|
attrs = [a for a in dir(mode_obj) if not a.startswith("_")]
|
|
logger.info(f" Attributs Reglement échéance: {attrs[:15]}...")
|
|
|
|
# Vérifier si l'échéance a FactoryDocumentReglementEcheance
|
|
factory_reg_ech = getattr(echeance, "FactoryDocumentReglementEcheance", None)
|
|
if factory_reg_ech:
|
|
logger.info(" FactoryDocumentReglementEcheance trouvée sur échéance")
|
|
|
|
# Créer le lien depuis l'échéance
|
|
reg_ech = factory_reg_ech.Create()
|
|
|
|
for iface in [
|
|
"IBODocumentReglementEcheance3",
|
|
"IBODocumentReglementEcheance",
|
|
]:
|
|
try:
|
|
reg_ech = win32com.client.CastTo(reg_ech, iface)
|
|
logger.info(f" Cast vers {iface}")
|
|
break
|
|
except Exception:
|
|
continue
|
|
|
|
# Ici, l'échéance devrait déjà être liée
|
|
# Il faut définir le règlement
|
|
try:
|
|
# Créer un nouveau règlement pour ce lien
|
|
factory_reg = self.cial.FactoryDocumentReglement
|
|
new_reg = factory_reg.Create()
|
|
new_reg = win32com.client.CastTo(new_reg, "IBODocumentReglement")
|
|
|
|
# Configurer minimalement
|
|
journal_factory = self.cial.CptaApplication.FactoryJournal
|
|
journal_persist = journal_factory.ReadNumero(code_journal)
|
|
if journal_persist:
|
|
new_reg.Journal = journal_persist
|
|
|
|
factory_client = self.cial.CptaApplication.FactoryClient
|
|
if client_code:
|
|
client_persist = factory_client.ReadNumero(client_code)
|
|
if client_persist:
|
|
new_reg.TiersPayeur = client_persist
|
|
|
|
new_reg.RG_Date = pywintypes.Time(date_reglement)
|
|
new_reg.RG_Montant = montant
|
|
new_reg.RG_Impute = 1
|
|
|
|
# Écrire le règlement
|
|
new_reg.Write()
|
|
logger.info(
|
|
f" Nouveau règlement créé: {getattr(new_reg, 'RG_Piece', None)}"
|
|
)
|
|
|
|
# Assigner au lien - ici on doit peut-être utiliser un autre attribut
|
|
# Puisque reg_ech.Reglement n'est pas settable, essayons via SetDefault
|
|
try:
|
|
reg_ech.SetDefault()
|
|
except Exception:
|
|
pass
|
|
|
|
reg_ech.RC_Montant = montant
|
|
reg_ech.Write()
|
|
|
|
logger.info(" Lien créé depuis échéance!")
|
|
return str(getattr(new_reg, "RG_Piece", None))
|
|
|
|
except Exception as e:
|
|
logger.warning(f" Erreur: {e}")
|
|
|
|
except Exception as e:
|
|
erreurs.append(f"Via échéance: {e}")
|
|
logger.warning(f"Via échéance échoué: {e}")
|
|
|
|
raise RuntimeError(f"Aucune méthode n'a fonctionné. Erreurs: {'; '.join(erreurs)}")
|
|
|
|
|
|
def introspecter_reglement(self):
|
|
if not self.cial:
|
|
raise RuntimeError("Connexion Sage non établie")
|
|
result = {}
|
|
try:
|
|
with self._com_context(), self._lock_com:
|
|
# IBODocumentReglement et sa factory de liens
|
|
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("_")
|
|
]
|
|
|
|
# FactoryDocumentReglementEcheance depuis le règlement
|
|
factory_lien = getattr(reg, "FactoryDocumentReglementEcheance", None)
|
|
if factory_lien:
|
|
lien = factory_lien.Create()
|
|
result["ReglementEcheance_base"] = [
|
|
a for a in dir(lien) if not a.startswith("_")
|
|
]
|
|
|
|
for iface in [
|
|
"IBODocumentReglementEcheance3",
|
|
"IBODocumentReglementEcheance",
|
|
]:
|
|
try:
|
|
lien_cast = win32com.client.CastTo(lien, iface)
|
|
result[f"ReglementEcheance_{iface}"] = [
|
|
a for a in dir(lien_cast) if not a.startswith("_")
|
|
]
|
|
except Exception as e:
|
|
result[f"cast_{iface}_error"] = str(e)
|
|
|
|
except Exception as e:
|
|
result["error_reglement"] = str(e)
|
|
|
|
# Process
|
|
try:
|
|
process = self.cial.CreateProcess_ReglerEcheances()
|
|
result["Process"] = [a for a in dir(process) if not a.startswith("_")]
|
|
except Exception as e:
|
|
result["error_process"] = str(e)
|
|
|
|
# Échéance et ses attributs
|
|
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("_")
|
|
]
|
|
|
|
# FactoryDocumentReglementEcheance depuis l'échéance
|
|
factory_lien_ech = getattr(
|
|
ech,
|
|
"FactoryDocumentReglementEcheance",
|
|
None,
|
|
)
|
|
if factory_lien_ech:
|
|
lien_ech = factory_lien_ech.Create()
|
|
result[
|
|
"EcheanceReglementEcheance_base"
|
|
] = [
|
|
a
|
|
for a in dir(lien_ech)
|
|
if not a.startswith("_")
|
|
]
|
|
|
|
for iface in [
|
|
"IBODocumentReglementEcheance3",
|
|
"IBODocumentReglementEcheance",
|
|
]:
|
|
try:
|
|
lien_ech_cast = (
|
|
win32com.client.CastTo(
|
|
lien_ech, iface
|
|
)
|
|
)
|
|
result[
|
|
f"EcheanceReglementEcheance_{iface}"
|
|
] = [
|
|
a
|
|
for a in dir(lien_ech_cast)
|
|
if not a.startswith("_")
|
|
]
|
|
except Exception:
|
|
pass
|
|
|
|
# Reglement de l'échéance (mode)
|
|
mode = getattr(ech, "Reglement", None)
|
|
if mode:
|
|
result["Echeance_Reglement_mode"] = [
|
|
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",
|
|
]
|