fanilo/Sage100-ws
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: fanilo:3dd863accf178c84930f740997c160aa548d6502
Branches Tags
fanilo:main
fanilo:develop
..
compare: fanilo:6b4b603aee07855eaa2056c3547361352985ca9a
Branches Tags
fanilo:develop
fanilo:main

No commits in common. "3dd863accf178c84930f740997c160aa548d6502" and "6b4b603aee07855eaa2056c3547361352985ca9a" have entirely different histories.
3dd863accf ... 6b4b603aee

13 changed files with 798 additions and 1574 deletions
Show stats Expand all files Collapse all files

7
.env.example
View file

@ -1,9 +1,14 @@
# ============================================================================
# SAGE 100 CLOUD - CONNEXION BOI/COM
# ============================================================================
CHEMIN_BASE=<CHEMIN_VERS_LE_FICHIER_GCM>
UTILISATEUR=<UTILISATEUR_SAGE100>
MOT_DE_PASSE=<MOT_DE_PASSE_SAGE100>
SAGE_GATEWAY_TOKEN=<TOKEN_SAGE_GATEWAY>
# ============================================================================
# API - CONFIGURATION SERVEUR
# ============================================================================
API_HOST=0.0.0.0
API_PORT=8000

6
.gitignore vendored
View file

@ -1,3 +1,7 @@
# ================================
# Python / FastAPI
# ================================
# Environnements virtuels
venv/
.env
@ -34,4 +38,4 @@ htmlcov/
dist/
*clean*.py
cleaner.py

186
main.py
View file

@ -2,7 +2,7 @@ from fastapi import FastAPI, HTTPException, Header, Depends, Query
from fastapi.middleware.cors import CORSMiddleware
from typing import Optional
from datetime import datetime, date
from datetime import datetime
import uvicorn
import logging
import win32com.client
@ -86,7 +86,7 @@ sage: Optional[SageConnector] = None
def startup():
global sage
logger.info(" Démarrage Sage Gateway Windows...")
logger.info("🚀 Démarrage Sage Gateway Windows...")
try:
validate_settings()
@ -113,7 +113,7 @@ def startup():
def shutdown():
if sage:
sage.deconnecter()
logger.info(" Sage Gateway arrêté")
logger.info("👋 Sage Gateway arrêté")
@app.get("/health")
@ -323,7 +323,7 @@ def transformer_document(
):
try:
logger.info(
f" Transformation demandée: {numero_source} "
f"🔄 Transformation demandée: {numero_source} "
f"(type {type_source}) → type {type_cible}"
)
@ -796,7 +796,7 @@ def creer_avoir_endpoint(req: AvoirCreate):
@app.post("/sage/avoirs/update", dependencies=[Depends(verify_token)])
def modifier_avoir_endpoint(req: AvoirUpdate):
"""
️ Modification d'un avoir dans Sage
✏️ Modification d'un avoir dans Sage
"""
try:
resultat = sage.modifier_avoir(req.numero, req.avoir_data)
@ -1182,7 +1182,7 @@ def lister_depots():
def creer_entree_stock(req: EntreeStock):
try:
logger.info(
f" [ENTREE STOCK] Création bon d'entrée : {len(req.lignes)} ligne(s)"
f"📦 [ENTREE STOCK] Création bon d'entrée : {len(req.lignes)} ligne(s)"
)
entree_data = {
@ -1212,7 +1212,7 @@ def creer_entree_stock(req: EntreeStock):
def creer_sortie_stock(req: SortieStock):
try:
logger.info(
f" [SORTIE STOCK] Création bon de sortie : {len(req.lignes)} ligne(s)"
f"📤 [SORTIE STOCK] Création bon de sortie : {len(req.lignes)} ligne(s)"
)
sortie_data = {
@ -1457,11 +1457,6 @@ def regler_facture_endpoint(
date_reglement=date_reg,
reference=req.reference or "",
libelle=req.libelle or "",
code_journal=req.code_journal, # None = auto
devise_code=req.devise_code,
cours_devise=req.cours_devise,
tva_encaissement=req.tva_encaissement,
compte_general=req.compte_general,
)
return {
"success": True,
@ -1479,23 +1474,15 @@ def regler_facture_endpoint(
@app.post("/sage/reglements/multiple", dependencies=[Depends(verify_token)])
def regler_factures_client_endpoint(req: ReglementMultipleRequest):
try:
date_reg = (
datetime.combine(req.date_reglement, datetime.min.time())
if req.date_reglement
else None
)
resultat = sage.regler_factures_client(
client_code=req.client_code,
montant_total=req.montant_total,
mode_reglement=req.mode_reglement,
date_reglement=date_reg,
date_reglement=req.date_reglement,
reference=req.reference or "",
libelle=req.libelle or "",
code_journal=req.code_journal,
numeros_factures=req.numeros_factures,
devise_code=req.devise_code,
cours_devise=req.cours_devise,
tva_encaissement=req.tva_encaissement,
)
return {"success": True, "data": resultat}
@ -1604,75 +1591,21 @@ def get_statut_validation_endpoint(numero_facture: str):
@app.get("/sage/reglements/modes", dependencies=[Depends(verify_token)])
def get_modes_reglement_endpoint():
"""Récupère les modes de règlement depuis Sage"""
try:
modes = sage.lire_modes_reglement()
return {"success": True, "data": {"modes": modes}}
except Exception as e:
logger.error(f"Erreur lecture modes règlement: {e}")
raise HTTPException(500, str(e))
@app.get("/sage/devises", dependencies=[Depends(verify_token)])
def get_devises_endpoint():
"""Récupère les devises disponibles depuis Sage"""
try:
devises = sage.lire_devises()
return {"success": True, "data": {"devises": devises}}
except Exception as e:
logger.error(f"Erreur lecture devises: {e}")
raise HTTPException(500, str(e))
@app.get("/sage/journaux/tresorerie", dependencies=[Depends(verify_token)])
def get_journaux_tresorerie_endpoint():
"""Récupère les journaux de trésorerie (banque + caisse)"""
try:
journaux = sage.lire_journaux_tresorerie()
return {"success": True, "data": {"journaux": journaux}}
except Exception as e:
logger.error(f"Erreur lecture journaux trésorerie: {e}")
raise HTTPException(500, str(e))
@app.get("/sage/comptes-generaux", dependencies=[Depends(verify_token)])
def get_comptes_generaux_endpoint(
prefixe: Optional[str] = Query(None, description="Filtre par préfixe (ex: 41, 51)"),
type_compte: Optional[str] = Query(
None,
description="Type: client, fournisseur, banque, caisse, tva, produit, charge",
),
):
"""Récupère les comptes généraux"""
try:
comptes = sage.lire_comptes_generaux(prefixe=prefixe, type_compte=type_compte)
return {"success": True, "data": {"comptes": comptes, "total": len(comptes)}}
except Exception as e:
logger.error(f"Erreur lecture comptes généraux: {e}")
raise HTTPException(500, str(e))
@app.get("/sage/tva/taux", dependencies=[Depends(verify_token)])
def get_tva_taux_endpoint():
"""Récupère les taux de TVA"""
try:
taux = sage.lire_tva_taux()
return {"success": True, "data": {"taux": taux}}
except Exception as e:
logger.error(f"Erreur lecture taux TVA: {e}")
raise HTTPException(500, str(e))
@app.get("/sage/parametres/encaissement", dependencies=[Depends(verify_token)])
def get_parametres_encaissement_endpoint():
"""Récupère les paramètres TVA sur encaissement"""
try:
params = sage.lire_parametres_encaissement()
return {"success": True, "data": params}
except Exception as e:
logger.error(f"Erreur lecture paramètres encaissement: {e}")
raise HTTPException(500, str(e))
def get_modes_reglement():
return {
"success": True,
"data": {
"modes": [
{"code": 1, "libelle": "Virement"},
{"code": 2, "libelle": "Chèque"},
{"code": 3, "libelle": "Traite"},
{"code": 4, "libelle": "Carte bancaire"},
{"code": 5, "libelle": "LCR"},
{"code": 6, "libelle": "Prélèvement"},
{"code": 7, "libelle": "Espèces"},
]
},
}
@app.get("/sage/journaux")
@ -1680,6 +1613,7 @@ def get_tous_journaux():
try:
journaux = sage.lire_tous_journaux()
# Grouper par type
par_type = {}
for j in journaux:
t = j["type_libelle"]
@ -1744,6 +1678,7 @@ def introspection_com():
}
with sage._com_context(), sage._lock_com:
# Attributs de cial
try:
for attr in dir(sage.cial):
if not attr.startswith("_"):
@ -1751,6 +1686,7 @@ def introspection_com():
except Exception as e:
resultats["cial_error"] = str(e)
# Attributs de BaseCpta
try:
base_cpta = sage.cial.BaseCpta
for attr in dir(base_cpta):
@ -1759,12 +1695,14 @@ def introspection_com():
except Exception as e:
resultats["base_cpta_error"] = str(e)
# Attributs de ParametreDossier
try:
param = sage.cial.BaseCpta.ParametreDossier
for attr in dir(param):
if not attr.startswith("_"):
resultats["param_dossier_attributes"].append(attr)
# Tester spécifiquement les attributs logo possibles
resultats["logo_tests"] = {}
for logo_attr in [
"Logo",
@ -1789,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}
@ -1799,6 +1745,12 @@ def introspection_validation(numero_facture: str = None):
@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}
@ -1807,60 +1759,6 @@ def introspection_document_complet(numero_facture: str):
raise HTTPException(status_code=500, detail=str(e))
@app.get("/sage/reglements", dependencies=[Depends(verify_token)])
def get_tous_reglements(
date_debut: Optional[date] = Query(None, description="Date de début (YYYY-MM-DD)"),
date_fin: Optional[date] = Query(None, description="Date de fin (YYYY-MM-DD)"),
client_code: Optional[str] = Query(None, description="Filtrer par code client"),
type_reglement: Optional[str] = Query(
None, description="Type: 'client' ou 'fournisseur'"
),
limit: int = Query(500, ge=1, le=2000, description="Nombre max de résultats"),
):
try:
date_debut_dt = (
datetime.combine(date_debut, datetime.min.time()) if date_debut else None
)
date_fin_dt = (
datetime.combine(date_fin, datetime.max.time()) if date_fin else None
)
result = sage.lire_tous_reglements(
date_debut=date_debut_dt,
date_fin=date_fin_dt,
client_code=client_code,
type_reglement=type_reglement,
)
return {"success": True, "data": result}
except Exception as e:
logger.error(f"Erreur lecture règlements: {e}")
raise HTTPException(500, str(e))
@app.get("/sage/reglements/facture/{facture_no}", dependencies=[Depends(verify_token)])
def get_reglement_facture_detail(facture_no):
try:
result = sage.lire_facture_reglement_detail(facture_no)
return {"success": True, "data": result}
except ValueError as e:
raise HTTPException(404, str(e))
except Exception as e:
logger.error(f"Erreur détail règlement {facture_no}: {e}")
raise HTTPException(500, str(e))
@app.get("/sage/reglements/{rg_no}", dependencies=[Depends(verify_token)])
def get_reglement_detail(rg_no):
try:
result = sage.lire_reglement_detail(rg_no)
return {"success": True, "data": result}
except ValueError as e:
raise HTTPException(404, str(e))
except Exception as e:
logger.error(f"Erreur détail règlement {rg_no}: {e}")
raise HTTPException(500, str(e))
if __name__ == "__main__":
uvicorn.run(
"main:app",

434
sage_connector.py
View file

File diff suppressed because it is too large Load diff

7
schemas/articles/famille_d_articles.py
View file

@ -1,7 +1,10 @@
from pydantic import BaseModel, Field
from typing import Optional
from pydantic import BaseModel, Field, validator, EmailStr, field_validator
from typing import Optional, List, Dict
from enum import Enum, IntEnum
from datetime import datetime, date
class FamilleCreate(BaseModel):
"""Modèle pour créer une famille d'articles"""

12
schemas/documents/reglements.py
View file

@ -1,3 +1,8 @@
"""
Schémas Pydantic pour les règlements de factures
Module: schemas/reglements.py
"""
from pydantic import BaseModel, Field, field_validator
from typing import List, Optional
from datetime import datetime
@ -64,10 +69,6 @@ class ReglementFactureRequest(BaseModel):
code_journal: str = Field(
default="BEU", max_length=6, description="Code journal comptable"
)
devise_code: Optional[int] = Field(0)
cours_devise: Optional[float] = Field(1.0)
tva_encaissement: Optional[bool] = Field(False)
compte_general: Optional[str] = Field(None)
@field_validator("montant")
def validate_montant(cls, v):
@ -102,9 +103,6 @@ class ReglementMultipleRequest(BaseModel):
default=None,
description="Liste des factures à régler (sinon: plus anciennes d'abord)",
)
devise_code: Optional[int] = Field(0)
cours_devise: Optional[float] = Field(1.0)
tva_encaissement: Optional[bool] = Field(False)
@field_validator("client_code", mode="before")
def strip_client_code(cls, v):

40
utils/articles/articles_data_sql.py
View file

@ -929,6 +929,10 @@ def enrichir_conditionnements(articles: List[Dict], cursor) -> List[Dict]:
def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
"""
Mappe les données brutes de la DB vers le format ArticleResponse
avec normalisation des types et génération des libellés
"""
article = {}
def get_val(sql_col, default=None, convert_type=None):
@ -948,6 +952,7 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
return val
# === CHAMPS DE BASE ===
article["reference"] = get_val("AR_Ref", convert_type=str)
article["designation"] = get_val("AR_Design", convert_type=str)
article["code_ean"] = get_val("AR_CodeBarre", convert_type=str)
@ -955,6 +960,7 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
article["edi_code"] = get_val("AR_EdiCode", convert_type=str)
article["raccourci"] = get_val("AR_Raccourci", convert_type=str)
# === PRIX ===
article["prix_vente"] = get_val("AR_PrixVen", 0.0, float)
article["prix_achat"] = get_val("AR_PrixAch", 0.0, float)
article["coef"] = get_val("AR_Coef", 0.0, float)
@ -968,35 +974,44 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
article["cout_standard"] = get_val("AR_CoutStd", 0.0, float)
# === UNITÉS ET POIDS (avec normalisation string) ===
article["unite_vente"] = normalize_string_field(get_val("AR_UniteVen"))
article["unite_poids"] = normalize_string_field(get_val("AR_UnitePoids"))
article["poids_net"] = get_val("AR_PoidsNet", 0.0, float)
article["poids_brut"] = get_val("AR_PoidsBrut", 0.0, float)
# === GAMMES (avec normalisation string) ===
article["gamme_1"] = normalize_string_field(get_val("AR_Gamme1"))
article["gamme_2"] = normalize_string_field(get_val("AR_Gamme2"))
# === TYPE ARTICLE (avec libellé) ===
type_val = get_val("AR_Type", 0, int)
article["type_article"] = type_val
article["type_article_libelle"] = TypeArticle.get_label(type_val)
# === FAMILLE ===
article["famille_code"] = get_val("FA_CodeFamille", convert_type=str)
# === NATURE ET GARANTIE ===
article["nature"] = get_val("AR_Nature", 0, int)
article["garantie"] = get_val("AR_Garantie", 0, int)
article["code_fiscal"] = normalize_string_field(get_val("AR_CodeFiscal"))
article["pays"] = normalize_string_field(get_val("AR_Pays"))
# === FOURNISSEUR ===
article["fournisseur_principal"] = get_val("CO_No", 0, int)
# === CONDITIONNEMENT (avec normalisation string) ===
article["conditionnement"] = normalize_string_field(get_val("AR_Condition"))
article["nb_colis"] = get_val("AR_NbColis", 0, int)
article["prevision"] = get_val("AR_Prevision", False, bool)
# === SUIVI STOCK (avec libellé) ===
suivi_stock_val = normalize_enum_to_int(get_val("AR_SuiviStock"))
article["suivi_stock"] = suivi_stock_val
article["suivi_stock_libelle"] = SuiviStockType.get_label(suivi_stock_val)
# === NOMENCLATURE (avec libellé) ===
nomenclature_val = normalize_enum_to_int(get_val("AR_Nomencl"))
article["nomenclature"] = nomenclature_val
article["nomenclature_libelle"] = NomenclatureType.get_label(nomenclature_val)
@ -1004,6 +1019,7 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
article["qte_composant"] = get_val("AR_QteComp", 0.0, float)
article["qte_operatoire"] = get_val("AR_QteOperatoire", 0.0, float)
# === STATUT ARTICLE ===
sommeil = get_val("AR_Sommeil", 0, int)
article["est_actif"] = sommeil == 0
article["en_sommeil"] = sommeil == 1
@ -1011,6 +1027,7 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
article["soumis_escompte"] = get_val("AR_Escompte", False, bool)
article["delai"] = get_val("AR_Delai", 0, int)
# === STATISTIQUES ===
article["stat_01"] = get_val("AR_Stat01", convert_type=str)
article["stat_02"] = get_val("AR_Stat02", convert_type=str)
article["stat_03"] = get_val("AR_Stat03", convert_type=str)
@ -1018,14 +1035,17 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
article["stat_05"] = get_val("AR_Stat05", convert_type=str)
article["hors_statistique"] = get_val("AR_HorsStat", False, bool)
# === CATÉGORIES COMPTABLES ===
article["categorie_1"] = get_val("CL_No1", 0, int)
article["categorie_2"] = get_val("CL_No2", 0, int)
article["categorie_3"] = get_val("CL_No3", 0, int)
article["categorie_4"] = get_val("CL_No4", 0, int)
# === DATE MODIFICATION ===
date_modif = get_val("AR_DateModif")
article["date_modification"] = str(date_modif) if date_modif else None
# === PARAMÈTRES DE VENTE ===
article["vente_debit"] = get_val("AR_VteDebit", False, bool)
article["non_imprimable"] = get_val("AR_NotImp", False, bool)
article["transfere"] = get_val("AR_Transfere", False, bool)
@ -1038,6 +1058,7 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
article["sous_traitance"] = get_val("AR_SousTraitance", False, bool)
article["criticite"] = get_val("AR_Criticite", 0, int)
# === PARAMÈTRES DE PRODUCTION ===
article["reprise_code_defaut"] = normalize_string_field(get_val("RP_CodeDefaut"))
article["delai_fabrication"] = get_val("AR_DelaiFabrication", 0, int)
article["delai_peremption"] = get_val("AR_DelaiPeremption", 0, int)
@ -1045,10 +1066,12 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
article["type_lancement"] = get_val("AR_TypeLancement", 0, int)
article["cycle"] = get_val("AR_Cycle", 1, int)
# === MÉDIA ET LANGUES ===
article["photo"] = get_val("AR_Photo", convert_type=str)
article["langue_1"] = get_val("AR_Langue1", convert_type=str)
article["langue_2"] = get_val("AR_Langue2", convert_type=str)
# === FRAIS ===
article["frais_01_denomination"] = get_val(
"AR_Frais01FR_Denomination", convert_type=str
)
@ -1059,6 +1082,7 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
"AR_Frais03FR_Denomination", convert_type=str
)
# === CHAMPS PERSONNALISÉS ===
article["marque_commerciale"] = get_val("Marque commerciale", convert_type=str)
objectif_val = get_val("Objectif / Qtés vendues")
@ -1083,6 +1107,7 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
article["interdire_commande"] = get_val("AR_InterdireCommande", False, bool)
article["exclure"] = get_val("AR_Exclure", False, bool)
# === INITIALISATION DES CHAMPS DE STOCK (remplis par enrichissement) ===
article["stock_reel"] = 0.0
article["stock_mini"] = 0.0
article["stock_maxi"] = 0.0
@ -1090,6 +1115,7 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
article["stock_commande"] = 0.0
article["stock_disponible"] = 0.0
# === INITIALISATION DES CHAMPS DE FAMILLE (remplis par enrichissement) ===
article["famille_libelle"] = None
article["famille_type"] = None
article["famille_unite_vente"] = None
@ -1106,6 +1132,7 @@ def _mapper_article_depuis_row(row_data: Dict, colonnes_config: Dict) -> Dict:
article["famille_hors_stat"] = None
article["famille_pays"] = None
# === INITIALISATION DES CHAMPS FOURNISSEUR/TVA (remplis par enrichissement) ===
article["fournisseur_nom"] = None
article["tva_code"] = None
article["tva_taux"] = None
@ -1185,7 +1212,7 @@ def enrichir_fournisseurs_articles(articles: List[Dict], cursor) -> List[Dict]:
)
if not nums_fournisseurs:
logger.warning(" Aucun numéro de fournisseur trouvé dans les articles")
logger.warning(" ⚠ Aucun numéro de fournisseur trouvé dans les articles")
for article in articles:
article["fournisseur_nom"] = None
return articles
@ -1211,7 +1238,7 @@ def enrichir_fournisseurs_articles(articles: List[Dict], cursor) -> List[Dict]:
if len(fournisseur_rows) == 0:
logger.warning(
f" Aucun fournisseur trouvé pour CT_Type=1 et CT_Num IN {nums_fournisseurs[:5]}"
f" ⚠ Aucun fournisseur trouvé pour CT_Type=1 et CT_Num IN {nums_fournisseurs[:5]}"
)
cursor.execute(
f"SELECT CT_Num, CT_Type FROM F_COMPTET WHERE CT_Num IN ({placeholders})",
@ -1234,6 +1261,7 @@ def enrichir_fournisseurs_articles(articles: List[Dict], cursor) -> List[Dict]:
nb_enrichis = 0
for article in articles:
num_fourn = article.get("fournisseur_principal")
# Convertir en string pour correspondre au fournisseur_map
num_fourn_str = (
str(num_fourn).strip() if num_fourn not in (None, "", " ") else None
)
@ -1273,7 +1301,7 @@ def enrichir_familles_articles(articles: List[Dict], cursor) -> List[Dict]:
)
if not codes_familles:
logger.warning(" Aucun code famille trouvé dans les articles")
logger.warning(" ⚠ Aucun code famille trouvé dans les articles")
for article in articles:
_init_champs_famille_vides(article)
return articles
@ -1464,6 +1492,12 @@ def enrichir_tva_articles(articles: List[Dict], cursor) -> List[Dict]:
return articles
def _get_type_article_libelle(type_val: int) -> str:
"""Retourne le libellé du type d'article"""
types = {0: "Article", 1: "Prestation", 2: "Divers / Frais", 3: "Nomenclature"}
return types.get(type_val, f"Type {type_val}")
def _cast_article(persist_obj):
try:
obj = win32com.client.CastTo(persist_obj, "IBOArticle3")

1169
utils/documents/settle.py
View file

File diff suppressed because it is too large Load diff

345
utils/documents/validations.py
View file

@ -1,10 +1,22 @@
from typing import Dict
"""
Validation de factures Sage 100c
Module: utils/documents/validation.py (Windows Server)
Version diagnostic - Introspection complète pour trouver la méthode de validation
"""
from typing import Dict, List
import win32com.client
import logging
logger = logging.getLogger(__name__)
# ============================================================
# FONCTIONS SQL (LECTURE)
# ============================================================
def get_statut_validation(connector, numero_facture: str) -> Dict:
"""Retourne le statut de validation d'une facture (SQL)"""
with connector._get_sql_connection() as conn:
@ -83,7 +95,16 @@ def _get_facture_info_sql(connector, numero_facture: str) -> Dict:
}
# ============================================================
# INTROSPECTION COMPLÈTE
# ============================================================
def introspecter_document_complet(connector, numero_facture: str) -> Dict:
"""
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")
@ -104,6 +125,7 @@ def introspecter_document_complet(connector, numero_facture: str) -> Dict:
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"] = []
@ -126,6 +148,7 @@ def introspecter_document_complet(connector, numero_facture: str) -> Dict:
{"name": attr, "error": str(e)[:50]}
)
# Chercher spécifiquement les attributs liés à validation/valide
result["persist"]["validation_related"] = [
a
for a in persist_attrs
@ -135,6 +158,7 @@ def introspecter_document_complet(connector, numero_facture: str) -> Dict:
)
]
# 2. IBODocumentVente3
try:
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
doc.Read()
@ -144,14 +168,16 @@ def introspecter_document_complet(connector, numero_facture: str) -> Dict:
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 Exception:
except:
pass
# Chercher DO_* properties
result["IBODocumentVente3"]["DO_properties"] = []
for attr in doc_attrs:
if attr.startswith("DO_"):
@ -165,6 +191,7 @@ def introspecter_document_complet(connector, numero_facture: str) -> Dict:
{"name": attr, "error": str(e)[:50]}
)
# Chercher les attributs liés à validation
result["IBODocumentVente3"]["validation_related"] = [
a
for a in doc_attrs
@ -177,6 +204,7 @@ def introspecter_document_complet(connector, numero_facture: str) -> Dict:
except Exception as e:
result["IBODocumentVente3"]["error"] = str(e)
# 3. IBODocument3
try:
doc3 = win32com.client.CastTo(persist, "IBODocument3")
doc3.Read()
@ -195,6 +223,7 @@ def introspecter_document_complet(connector, numero_facture: str) -> Dict:
except Exception as e:
result["IBODocument3"]["error"] = str(e)
# 4. IPMDocument
try:
pmdoc = win32com.client.CastTo(persist, "IPMDocument")
@ -207,6 +236,7 @@ def introspecter_document_complet(connector, numero_facture: str) -> Dict:
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:
@ -219,6 +249,9 @@ def introspecter_document_complet(connector, numero_facture: str) -> Dict:
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")
@ -226,11 +259,13 @@ def introspecter_validation(connector, numero_facture: str = None) -> Dict:
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)()
@ -244,6 +279,7 @@ def introspecter_validation(connector, numero_facture: str = None) -> Dict:
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
@ -256,71 +292,125 @@ def introspecter_validation(connector, numero_facture: str = None) -> Dict:
def valider_facture(connector, numero_facture: str) -> Dict:
logger.info(f" Validation facture {numero_facture} (SQL direct)")
with connector._get_sql_connection() as conn:
cursor = conn.cursor()
cursor.execute(
"""
SELECT DO_Valide, DO_Statut, DO_TotalTTC, DO_MontantRegle
FROM F_DOCENTETE
WHERE DO_Piece = ? AND DO_Type = 6
""",
(numero_facture,),
)
Valide via late binding forcé + invocation COM correcte
"""
import win32com.client
import win32com.client.dynamic
import pythoncom
row = cursor.fetchone()
if not row:
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")
valide_avant, statut, total_ttc, montant_regle = row
persist = factory.ReadPiece(60, numero_facture)
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
doc.Read()
if valide_avant == 1:
# 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}
if statut == 6: # Annulé
raise ValueError("Facture annulée, validation impossible")
oleobj = doc._oleobj_
cursor.execute(
"""
UPDATE F_DOCENTETE
SET DO_Valide = 1, DO_Imprim = 0
WHERE DO_Piece = ? AND DO_Type = 6
""",
(numero_facture,),
# 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,
}
conn.commit()
logger.info(f" DISPIDs trouvés: {dispids}")
cursor.execute(
"""
SELECT DO_Valide, DO_Imprim
FROM F_DOCENTETE
WHERE DO_Piece = ? AND DO_Type = 6
""",
(numero_facture,),
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}")
valide_apres, imprim_apres = cursor.fetchone()
logger.info(f" SQL: DO_Valide={valide_apres}, DO_Imprim={imprim_apres}")
# 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,
"methode": "SQL_DIRECT",
"DO_Valide": valide_apres == 1,
"DO_Imprim": imprim_apres == 1,
"success": valide_apres == 1,
"warning": "Validation SQL directe - règles métier Sage contournées",
"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:
"""
Dévalide une facture (retire le cadenas)
"""
if not connector.cial:
raise RuntimeError("Connexion Sage non établie")
logger.info(f" Dévalidation facture {numero_facture}")
logger.info(f"🔓 Dévalidation facture {numero_facture}")
info = _get_facture_info_sql(connector, numero_facture)
if not info:
@ -352,12 +442,12 @@ def devalider_facture(connector, numero_facture: str) -> Dict:
if not success:
raise RuntimeError("La dévalidation COM a échoué")
logger.info(f" Facture {numero_facture} dévalidée")
logger.info(f"✅ Facture {numero_facture} dévalidée")
except ValueError:
raise
except Exception as e:
logger.error(f" Erreur COM dévalidation {numero_facture}: {e}", exc_info=True)
logger.error(f"❌ Erreur COM dévalidation {numero_facture}: {e}", exc_info=True)
raise RuntimeError(f"Échec dévalidation: {str(e)}")
info_apres = _get_facture_info_sql(connector, numero_facture)
@ -372,6 +462,9 @@ def devalider_facture(connector, numero_facture: str) -> Dict:
def _valider_document_com(connector, numero_facture: str, valider: bool = True) -> bool:
"""
Tente de valider/dévalider un document via COM.
"""
erreurs = []
action = "validation" if valider else "dévalidation"
valeur_cible = 1 if valider else 0
@ -384,6 +477,7 @@ def _valider_document_com(connector, numero_facture: str, valider: bool = True)
if not persist:
raise ValueError(f"Impossible de lire la facture {numero_facture}")
# APPROCHE 1: Accès direct à DO_Valide sur IBODocumentVente3
try:
logger.info(
" APPROCHE 1: Modification directe DO_Valide sur IBODocumentVente3..."
@ -391,18 +485,21 @@ def _valider_document_com(connector, numero_facture: str, valider: bool = True)
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(" DO_Valide modifié avec succès!")
logger.info(f" ✅ DO_Valide modifié avec succès!")
return True
else:
erreurs.append(
@ -413,6 +510,7 @@ def _valider_document_com(connector, numero_facture: str, valider: bool = True)
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")
@ -429,6 +527,7 @@ def _valider_document_com(connector, numero_facture: str, valider: bool = True)
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]
@ -443,6 +542,7 @@ def _valider_document_com(connector, numero_facture: str, valider: bool = True)
for proc_name in validation_processes:
try:
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}")
@ -460,6 +560,7 @@ def _valider_document_com(connector, numero_facture: str, valider: bool = True)
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")
@ -478,71 +579,127 @@ def _valider_document_com(connector, numero_facture: str, valider: bool = True)
raise RuntimeError(f"Échec {action}: {'; '.join(erreurs[:5])}")
def explorer_toutes_interfaces_validation(connector, numero_facture: str) -> Dict:
"""Explorer TOUTES les interfaces possibles pour trouver un setter DO_Valide"""
result = {"numero_facture": numero_facture, "interfaces": {}}
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()
interfaces = [
"IBODocumentVente3",
"IBODocument3",
"IBIPersistObject",
"IBIDocument",
"IPMDocument",
"IDispatch",
]
for iface_name in interfaces:
# 1. CreateProcess_Document SANS paramètre
try:
obj = win32com.client.CastTo(persist, iface_name)
if hasattr(obj, "Read"):
obj.Read()
oleobj = obj._oleobj_
type_info = oleobj.GetTypeInfo()
type_attr = type_info.GetTypeAttr()
props = {}
for i in range(type_attr.cFuncs):
func_desc = type_info.GetFuncDesc(i)
names = type_info.GetNames(func_desc.memid)
if names and names[0] in (
"DO_Valide",
"DO_Imprim",
"Valider",
"Validate",
"Lock",
):
props[names[0]] = {
"memid": func_desc.memid,
"invkind": func_desc.invkind,
"has_setter": (func_desc.invkind & 4) == 4,
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"]
)
],
}
result["interfaces"][iface_name] = {
"success": True,
"properties": props,
}
# Essayer d'assigner le document
if "Document" in attrs:
try:
process.Document = doc
result["CreateProcess_Document_no_param"]["Document_assigned"] = (
True
)
except Exception as e:
result["CreateProcess_Document_no_param"]["Document_error"] = str(e)
if "SetDocument" in attrs:
try:
process.SetDocument(doc)
result["CreateProcess_Document_no_param"]["SetDocument_ok"] = True
except Exception as e:
result["CreateProcess_Document_no_param"]["SetDocument_error"] = (
str(e)
)
except Exception as e:
result["interfaces"][iface_name] = {"error": str(e)[:100]}
result["CreateProcess_Document_no_param_error"] = str(e)
# 2. CreateProcess_Document avec type (60 = facture)
try:
factory_attrs = [a for a in dir(factory) if not a.startswith("_")]
result["factory_methods"] = [
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 factory_attrs
if any(x in a.lower() for x in ["valid", "lock", "confirm", "imprim"])
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"])
]
except Exception:
pass
# 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
# ============================================================
# FONCTIONS UTILITAIRES
# ============================================================
def _build_response_sql(
connector, numero_facture: str, deja_valide: bool, action: str
) -> Dict:

136
utils/functions/data/create_doc.py
View file

@ -19,7 +19,7 @@ def creer_document_vente(
config = ConfigDocument(type_document)
logger.info(
f" Début création {config.nom_document} pour client {doc_data['client']['code']}"
f"📝 Début création {config.nom_document} pour client {doc_data['client']['code']}"
)
try:
@ -27,13 +27,15 @@ def creer_document_vente(
transaction_active = False
try:
# Démarrage transaction
try:
self.cial.CptaApplication.BeginTrans()
transaction_active = True
logger.debug(" Transaction Sage démarrée")
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
@ -42,13 +44,15 @@ def creer_document_vente(
except Exception:
pass
logger.info(f" Document {config.nom_document} créé")
logger.info(f"✓ Document {config.nom_document} créé")
# ===== DATES =====
date_principale = normaliser_date(
doc_data.get(config.champ_date_principale)
)
doc.DO_Date = pywintypes.Time(date_principale)
# Heure - même datetime, Sage extrait la composante horaire
try:
doc.DO_Heure = pywintypes.Time(date_principale)
logger.debug(
@ -57,6 +61,7 @@ def creer_document_vente(
except Exception as e:
logger.debug(f"DO_Heure non défini: {e}")
# Date secondaire (livraison, etc.)
if config.champ_date_secondaire and doc_data.get(
config.champ_date_secondaire
):
@ -64,9 +69,10 @@ def creer_document_vente(
normaliser_date(doc_data[config.champ_date_secondaire])
)
logger.info(
f" {config.champ_date_secondaire}: {doc_data[config.champ_date_secondaire]}"
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"])
@ -79,18 +85,21 @@ def creer_document_vente(
doc.SetDefaultClient(client_obj)
doc.Write()
logger.info(f" Client {doc_data['client']['code']} associé")
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']}")
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:
@ -98,8 +107,9 @@ def creer_document_vente(
factory_article = self.cial.FactoryArticle
logger.info(f" Ajout de {len(doc_data['lignes'])} lignes...")
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,
@ -110,8 +120,10 @@ def creer_document_vente(
doc=doc,
)
logger.info(" Validation du document...")
# ===== 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)
@ -124,25 +136,28 @@ def creer_document_vente(
doc.Write()
# Process() sauf pour devis en brouillon
if type_document != TypeDocumentVente.DEVIS:
process.Process()
logger.info(" Process() appelé")
logger.info("✓ Process() appelé")
else:
try:
process.Process()
logger.info(" Process() appelé (devis)")
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")
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:
@ -150,8 +165,9 @@ def creer_document_vente(
f"Numéro {config.nom_document} vide après création"
)
logger.info(f" Numéro: {numero_document}")
logger.info(f"📄 Numéro: {numero_document}")
# ===== RELECTURE POUR TOTAUX =====
doc_final_data = _relire_document_final(
self,
config=config,
@ -189,17 +205,21 @@ def _appliquer_remise_ligne(ligne_obj, remise_pourcent: float) -> bool:
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")
@ -216,6 +236,7 @@ def _ajouter_ligne_document(
"""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")
@ -234,6 +255,7 @@ def _ajouter_ligne_document(
quantite = float(ligne_data["quantite"])
# ===== ASSOCIATION ARTICLE =====
try:
ligne_obj.SetDefaultArticleReference(ligne_data["article_code"], quantite)
except Exception:
@ -243,6 +265,7 @@ def _ajouter_ligne_document(
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")
@ -252,21 +275,23 @@ def _ajouter_ligne_document(
ligne_obj.DL_PrixUnitaire = float(prix_sage)
prix_final = float(getattr(ligne_obj, "DL_PrixUnitaire", 0))
logger.info(f" Prix: {prix_final}€")
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}%...")
logger.info(f" 🎯 Application remise {remise}%...")
_appliquer_remise_ligne(ligne_obj, remise)
logger.info(f" Ligne {idx} terminée")
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...")
logger.info(" 🔧 Configuration spécifique facture...")
try:
if hasattr(doc, "DO_CodeJournal"):
@ -274,24 +299,24 @@ def _configurer_facture(self, doc) -> None:
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}")
logger.debug(f" ✓ Code journal: {journal_defaut}")
except Exception:
doc.DO_CodeJournal = "VTE"
logger.debug(" Code journal: VTE (défaut)")
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")
logger.debug(" ✓ Souche: 0")
except Exception:
pass
try:
if hasattr(doc, "DO_Regime"):
doc.DO_Regime = 0
logger.debug(" Régime: 0")
logger.debug(" ✓ Régime: 0")
except Exception:
pass
@ -339,6 +364,7 @@ def _relire_document_final(
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:
@ -347,6 +373,7 @@ def _relire_document_final(
except Exception:
pass
# Date secondaire
if config.champ_date_secondaire:
try:
date_livr = getattr(doc_final, "DO_DateLivr", None)
@ -355,14 +382,17 @@ def _relire_document_final(
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,
@ -377,6 +407,7 @@ def _relire_document_final(
"reference": reference_finale,
}
# Ajout date secondaire si applicable
if config.champ_date_secondaire:
resultat[config.champ_date_secondaire] = date_secondaire_value
@ -390,11 +421,14 @@ def modifier_document_vente(
raise RuntimeError("Connexion Sage non établie")
config = ConfigDocument(type_document)
logger.info(f" === MODIFICATION {config.nom_document.upper()} {numero} ===")
logger.info(f"📝 === MODIFICATION {config.nom_document.upper()} {numero} ===")
try:
with self._com_context(), self._lock_com:
logger.info(" Chargement document...")
# ==========================================
# 1. CHARGEMENT DOCUMENT
# ==========================================
logger.info("📂 Chargement document...")
factory = self.cial.FactoryDocumentVente
persist = None
@ -403,7 +437,7 @@ def modifier_document_vente(
persist_test = factory.ReadPiece(type_test, numero)
if persist_test:
persist = persist_test
logger.info(f" Document trouvé (type={type_test})")
logger.info(f" ✓ Document trouvé (type={type_test})")
break
except Exception:
continue
@ -416,6 +450,7 @@ def modifier_document_vente(
doc = win32com.client.CastTo(persist, "IBODocumentVente3")
doc.Read()
# Vérifications
statut_actuel = getattr(doc, "DO_Statut", 0)
if statut_actuel == 5:
@ -423,6 +458,9 @@ def modifier_document_vente(
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)
@ -436,6 +474,9 @@ def modifier_document_vente(
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(
@ -455,6 +496,9 @@ def modifier_document_vente(
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
@ -466,7 +510,7 @@ def modifier_document_vente(
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("📋 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}")
@ -474,6 +518,7 @@ def modifier_document_vente(
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
@ -486,19 +531,25 @@ def modifier_document_vente(
statut_a_modifier = doc_data_temp.pop("statut")
modif_statut = False
logger.info(" Test Write() basique...")
# ==========================================
# 5. TEST WRITE BASIQUE
# ==========================================
logger.info("🔍 Test Write() basique...")
try:
doc.Write()
doc.Read()
logger.info(" Write() basique OK")
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...")
logger.info("📝 Modifications simples...")
if modif_date:
date_principale = normaliser_date(
@ -525,15 +576,20 @@ def modifier_document_vente(
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")
logger.info(" ✓ Modifications appliquées")
# ==========================================
# 7. REMPLACEMENT LIGNES
# ==========================================
elif modif_lignes:
logger.info(" REMPLACEMENT COMPLET DES 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))
@ -546,6 +602,7 @@ def modifier_document_vente(
)
champs_modifies.append(config.champ_date_secondaire)
# 🔥 CONFIGURATION SPÉCIFIQUE FACTURE (avant lignes)
if type_document == TypeDocumentVente.FACTURE:
_configurer_facture(self, doc)
@ -561,8 +618,9 @@ def modifier_document_vente(
factory_article = self.cial.FactoryArticle
# Suppression lignes existantes
if nb_lignes_initial > 0:
logger.info(f" ️ Suppression {nb_lignes_initial} lignes...")
logger.info(f" 🗑️ Suppression {nb_lignes_initial} lignes...")
for idx in range(nb_lignes_initial, 0, -1):
try:
ligne_p = factory_lignes.List(idx)
@ -574,10 +632,12 @@ def modifier_document_vente(
ligne.Remove()
except Exception as e:
logger.warning(f" Ligne {idx}: {e}")
logger.info(" Lignes supprimées")
logger.info(" ✓ Lignes supprimées")
logger.info(f" Ajout {nb_nouvelles} lignes...")
# 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,
@ -587,7 +647,7 @@ def modifier_document_vente(
doc=doc,
)
logger.info(" Nouvelles lignes ajoutées avec remises")
logger.info(" ✓ Nouvelles lignes ajoutées avec remises")
doc.Write()
time.sleep(0.5)
@ -595,10 +655,13 @@ def modifier_document_vente(
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}'"
f" 📝 Modification référence: '{reference_a_modifier}'"
)
doc.DO_Ref = (
str(reference_a_modifier) if reference_a_modifier else ""
@ -612,7 +675,7 @@ def modifier_document_vente(
if statut_a_modifier is not None:
try:
logger.info(f" Modification statut: {statut_a_modifier}")
logger.info(f" 📝 Modification statut: {statut_a_modifier}")
doc.DO_Statut = int(statut_a_modifier)
doc.Write()
time.sleep(0.5)
@ -621,6 +684,9 @@ def modifier_document_vente(
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
)

2
utils/functions/functions.py
View file

@ -167,8 +167,10 @@ def _parser_heure_sage(do_heure) -> str:
return "00:00:00"
try:
# Convertir en entier pour éliminer les zéros de padding SQL
heure_int = int(str(do_heure).strip())
# Formatter en string 6 caractères (HHMMSS)
heure_str = str(heure_int).zfill(6)
hh = int(heure_str[0:2])

13
utils/functions/items_to_dict.py
View file

@ -80,6 +80,7 @@ def contact_to_dict(row) -> Dict:
def _collaborators_to_dict(row) -> Optional[dict]:
"""Convertit une ligne SQL en dictionnaire collaborateur"""
# Vérifier si le collaborateur existe
if not hasattr(row, "Collab_CO_No") or row.Collab_CO_No is None:
return None
@ -156,6 +157,7 @@ def collaborators_to_dict(row):
def tiers_to_dict(row) -> dict:
"""Convertit une ligne SQL en dictionnaire tiers"""
tiers = {
# IDENTIFICATION
"numero": _safe_strip(row.CT_Num),
"intitule": _safe_strip(row.CT_Intitule),
"type_tiers": row.CT_Type,
@ -165,6 +167,7 @@ def tiers_to_dict(row) -> dict:
"siret": _safe_strip(row.CT_Siret),
"tva_intra": _safe_strip(row.CT_Identifiant),
"code_naf": _safe_strip(row.CT_Ape),
# ADRESSE
"contact": _safe_strip(row.CT_Contact),
"adresse": _safe_strip(row.CT_Adresse),
"complement": _safe_strip(row.CT_Complement),
@ -172,16 +175,19 @@ def tiers_to_dict(row) -> dict:
"ville": _safe_strip(row.CT_Ville),
"region": _safe_strip(row.CT_CodeRegion),
"pays": _safe_strip(row.CT_Pays),
# TELECOM
"telephone": _safe_strip(row.CT_Telephone),
"telecopie": _safe_strip(row.CT_Telecopie),
"email": _safe_strip(row.CT_EMail),
"site_web": _safe_strip(row.CT_Site),
"facebook": _safe_strip(row.CT_Facebook),
"linkedin": _safe_strip(row.CT_LinkedIn),
# TAUX
"taux01": row.CT_Taux01,
"taux02": row.CT_Taux02,
"taux03": row.CT_Taux03,
"taux04": row.CT_Taux04,
# STATISTIQUES
"statistique01": _safe_strip(row.CT_Statistique01),
"statistique02": _safe_strip(row.CT_Statistique02),
"statistique03": _safe_strip(row.CT_Statistique03),
@ -192,11 +198,13 @@ def tiers_to_dict(row) -> dict:
"statistique08": _safe_strip(row.CT_Statistique08),
"statistique09": _safe_strip(row.CT_Statistique09),
"statistique10": _safe_strip(row.CT_Statistique10),
# COMMERCIAL
"encours_autorise": row.CT_Encours,
"assurance_credit": row.CT_Assurance,
"langue": row.CT_Langue,
"commercial_code": row.CO_No,
"commercial": _collaborators_to_dict(row),
# FACTURATION
"lettrage_auto": (row.CT_Lettrage == 1),
"est_actif": (row.CT_Sommeil == 0),
"type_facture": row.CT_Facture,
@ -208,12 +216,16 @@ def tiers_to_dict(row) -> dict:
"exclure_relance": (row.CT_NotRappel == 1),
"exclure_penalites": (row.CT_NotPenal == 1),
"bon_a_payer": row.CT_BonAPayer,
# LOGISTIQUE
"priorite_livraison": row.CT_PrioriteLivr,
"livraison_partielle": row.CT_LivrPartielle,
"delai_transport": row.CT_DelaiTransport,
"delai_appro": row.CT_DelaiAppro,
# COMMENTAIRE
"commentaire": _safe_strip(row.CT_Commentaire),
# ANALYTIQUE
"section_analytique": _safe_strip(row.CA_Num),
# ORGANISATION / SURVEILLANCE
"mode_reglement_code": row.MR_No,
"surveillance_active": (row.CT_Surveillance == 1),
"coface": _safe_strip(row.CT_Coface),
@ -224,6 +236,7 @@ def tiers_to_dict(row) -> dict:
"sv_objet_maj": _safe_strip(row.CT_SvObjetMaj),
"sv_chiffre_affaires": row.CT_SvCA,
"sv_resultat": row.CT_SvResultat,
# COMPTE GENERAL ET CATEGORIES
"compte_general": _safe_strip(row.CG_NumPrinc),
"categorie_tarif": row.N_CatTarif,
"categorie_compta": row.N_CatCompta,

1
utils/tiers/tiers_data_sql.py
View file

@ -1,4 +1,5 @@
def _build_tiers_select_query():
"""Construit la partie SELECT de la requête avec tous les champs tiers + collaborateur"""
return """
SELECT
-- IDENTIFICATION TIERS (9)