Reviewed and refactored files' structures and indentation

This commit is contained in:
fanilo 2025-12-31 09:25:05 +01:00
parent 3bb1aee4b4
commit 0788f46bde
29 changed files with 1417 additions and 6237 deletions

View file

@ -15,26 +15,21 @@ class Settings(BaseSettings):
SAGE_TYPE_BON_AVOIR: int = 50 SAGE_TYPE_BON_AVOIR: int = 50
SAGE_TYPE_FACTURE: int = 60 SAGE_TYPE_FACTURE: int = 60
# === SAGE 100c (Windows uniquement) ===
chemin_base: str chemin_base: str
utilisateur: str = "Administrateur" utilisateur: str = "Administrateur"
mot_de_passe: str mot_de_passe: str
# === Sécurité Gateway === sage_gateway_token: str
sage_gateway_token: str # Token partagé avec le VPS Linux
# === SMTP (optionnel sur Windows) ===
smtp_host: Optional[str] = None smtp_host: Optional[str] = None
smtp_port: int = 587 smtp_port: int = 587
smtp_user: Optional[str] = None smtp_user: Optional[str] = None
smtp_password: Optional[str] = None smtp_password: Optional[str] = None
smtp_from: Optional[str] = None smtp_from: Optional[str] = None
# === API Windows ===
api_host: str = "0.0.0.0" api_host: str = "0.0.0.0"
api_port: int = 8000 api_port: int = 8000
# === CORS ===
cors_origins: List[str] = ["*"] cors_origins: List[str] = ["*"]

View file

@ -1,12 +0,0 @@
======================================================================
DIAGNOSTIC CRYSTAL REPORTS
======================================================================
Installation détectée: True
DLL trouvées:
ProgID valides:
Architecture OK: False
PROBLÈMES:
SOLUTIONS:

49
main.py
View file

@ -48,8 +48,8 @@ from schemas import (
SortieStockRequest, SortieStockRequest,
FamilleCreate, FamilleCreate,
PDFGenerationRequest, PDFGenerationRequest,
DevisUpdateGatewayRequest DevisUpdateGatewayRequest,
) )
logging.basicConfig( logging.basicConfig(
level=logging.INFO, level=logging.INFO,
@ -58,6 +58,7 @@ logging.basicConfig(
) )
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def verify_token(x_sage_token: str = Header(...)): def verify_token(x_sage_token: str = Header(...)):
"""Vérification du token d'authentification""" """Vérification du token d'authentification"""
if x_sage_token != settings.sage_gateway_token: if x_sage_token != settings.sage_gateway_token:
@ -278,7 +279,9 @@ def changer_statut_document(numero: str, type_doc: int, nouveau_statut: int):
doc.DO_Statut = nouveau_statut doc.DO_Statut = nouveau_statut
doc.Write() doc.Write()
logger.info(f" Statut document {numero}: {statut_actuel}{nouveau_statut}") logger.info(
f" Statut document {numero}: {statut_actuel}{nouveau_statut}"
)
return { return {
"success": True, "success": True,
@ -323,11 +326,11 @@ def transformer_document(
) )
transformations_valides = { transformations_valides = {
(0, 10), # Devis → Commande (0, 10),
(10, 30), # Commande → Bon de livraison (10, 30),
(10, 60), # Commande → Facture (10, 60),
(30, 60), # Bon de livraison → Facture (30, 60),
(0, 60), # Devis → Facture (si autorisé) (0, 60),
} }
if (type_source, type_cible) not in transformations_valides: if (type_source, type_cible) not in transformations_valides:
@ -443,11 +446,11 @@ def lire_remise_max_client(code: str):
if not client_obj: if not client_obj:
raise HTTPException(404, f"Client {code} introuvable") raise HTTPException(404, f"Client {code} introuvable")
remise_max = 10.0 # Défaut remise_max = 10.0
try: try:
remise_max = float(getattr(client_obj, "CT_RemiseMax", 10.0)) remise_max = float(getattr(client_obj, "CT_RemiseMax", 10.0))
except: except Exception:
pass pass
logger.info(f" Remise max client {code}: {remise_max}%") logger.info(f" Remise max client {code}: {remise_max}%")
@ -633,7 +636,7 @@ def livraisons_list(
livraisons = sage.lister_toutes_livraisons_cache(filtre) livraisons = sage.lister_toutes_livraisons_cache(filtre)
if statut is not None: if statut is not None:
livraisons = [l for l in livraisons if l.get("statut") == statut] livraisons = [ligne for ligne in livraisons if ligne.get("statut") == statut]
livraisons = livraisons[:limit] livraisons = livraisons[:limit]
@ -1026,7 +1029,7 @@ def lister_depots():
factory_depot = sage.cial.FactoryDepot factory_depot = sage.cial.FactoryDepot
index = 1 index = 1
while index <= 100: # Max 100 dépôts while index <= 100:
try: try:
persist = factory_depot.List(index) persist = factory_depot.List(index)
@ -1045,30 +1048,30 @@ def lister_depots():
try: try:
code = getattr(depot, "DE_Code", "").strip() code = getattr(depot, "DE_Code", "").strip()
except: except Exception:
pass pass
try: try:
numero = int(getattr(depot, "Compteur", 0)) numero = int(getattr(depot, "Compteur", 0))
except: except Exception:
try: try:
numero = int(code) numero = int(code)
except: except Exception:
numero = 0 numero = 0
try: try:
intitule = getattr(depot, "DE_Intitule", "") intitule = getattr(depot, "DE_Intitule", "")
except: except Exception:
pass pass
try: try:
contact = getattr(depot, "DE_Contact", "") contact = getattr(depot, "DE_Contact", "")
except: except Exception:
pass pass
try: try:
exclu = getattr(depot, "DE_Exclure", False) exclu = getattr(depot, "DE_Exclure", False)
except: except Exception:
pass pass
if not code: if not code:
@ -1085,9 +1088,9 @@ def lister_depots():
cp = getattr(adresse_obj, "CodePostal", "") cp = getattr(adresse_obj, "CodePostal", "")
ville = getattr(adresse_obj, "Ville", "") ville = getattr(adresse_obj, "Ville", "")
adresse_complete = f"{adresse} {cp} {ville}".strip() adresse_complete = f"{adresse} {cp} {ville}".strip()
except: except Exception:
pass pass
except: except Exception:
pass pass
principal = False principal = False
@ -1095,8 +1098,8 @@ def lister_depots():
principal = True principal = True
depot_info = { depot_info = {
"code": code, # ⭐ "01", "02" "code": code,
"numero": numero, # ⭐ 1, 2 (depuis Compteur) "numero": numero,
"intitule": intitule, "intitule": intitule,
"adresse": adresse_complete, "adresse": adresse_complete,
"contact": contact, "contact": contact,
@ -1341,6 +1344,6 @@ if __name__ == "__main__":
"main:app", "main:app",
host=settings.api_host, host=settings.api_host,
port=settings.api_port, port=settings.api_port,
reload=False, # Pas de reload en production reload=False,
log_level="info", log_level="info",
) )

File diff suppressed because it is too large Load diff

File diff suppressed because it is too large Load diff

View file

@ -1,9 +1,7 @@
from pydantic import BaseModel
from pydantic import BaseModel, Field, validator, EmailStr, field_validator
from typing import Optional, List, Dict from typing import Optional, List, Dict
from enum import Enum, IntEnum from datetime import date
from datetime import datetime, date
class AvoirCreateGatewayRequest(BaseModel): class AvoirCreateGatewayRequest(BaseModel):
"""Création d'un avoir côté gateway""" """Création d'un avoir côté gateway"""

View file

@ -1,8 +1,7 @@
from pydantic import BaseModel
from pydantic import BaseModel, Field, validator, EmailStr, field_validator
from typing import Optional, List, Dict from typing import Optional, List, Dict
from enum import Enum, IntEnum from datetime import date
from datetime import datetime, date
class CommandeCreateRequest(BaseModel): class CommandeCreateRequest(BaseModel):
"""Création d'une commande""" """Création d'une commande"""

View file

@ -1,9 +1,7 @@
from pydantic import BaseModel
from pydantic import BaseModel, Field, validator, EmailStr, field_validator
from typing import Optional, List, Dict from typing import Optional, List, Dict
from enum import Enum, IntEnum from datetime import date
from datetime import datetime, date
class DevisRequest(BaseModel): class DevisRequest(BaseModel):
client_id: str client_id: str
@ -12,8 +10,9 @@ class DevisRequest(BaseModel):
reference: Optional[str] = None reference: Optional[str] = None
lignes: List[Dict] lignes: List[Dict]
class DevisUpdateGatewayRequest(BaseModel): class DevisUpdateGatewayRequest(BaseModel):
"""Modèle pour modification devis côté gateway""" """Modèle pour modification devis côté gateway"""
numero: str numero: str
devis_data: Dict devis_data: Dict

View file

@ -1,8 +1,6 @@
from pydantic import BaseModel, Field
from enum import Enum
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 TypeDocument(int, Enum): class TypeDocument(int, Enum):
DEVIS = 0 DEVIS = 0
@ -17,16 +15,15 @@ class DocumentGetRequest(BaseModel):
numero: str numero: str
type_doc: int type_doc: int
class TransformationRequest(BaseModel): class TransformationRequest(BaseModel):
numero_source: str numero_source: str
type_source: int type_source: int
type_cible: int type_cible: int
class PDFGenerationRequest(BaseModel): class PDFGenerationRequest(BaseModel):
"""Modèle pour génération PDF""" """Modèle pour génération PDF"""
doc_id: str = Field(..., description="Numéro du document") doc_id: str = Field(..., description="Numéro du document")
type_doc: int = Field(..., ge=0, le=60, description="Type de document Sage") type_doc: int = Field(..., ge=0, le=60, description="Type de document Sage")

View file

@ -1,9 +1,7 @@
from pydantic import BaseModel
from pydantic import BaseModel, Field, validator, EmailStr, field_validator
from typing import Optional, List, Dict from typing import Optional, List, Dict
from enum import Enum, IntEnum from datetime import date
from datetime import datetime, date
class FactureCreateGatewayRequest(BaseModel): class FactureCreateGatewayRequest(BaseModel):
"""Création d'une facture côté gateway""" """Création d'une facture côté gateway"""

View file

@ -1,8 +1,7 @@
from pydantic import BaseModel
from pydantic import BaseModel, Field, validator, EmailStr, field_validator
from typing import Optional, List, Dict from typing import Optional, List, Dict
from enum import Enum, IntEnum from datetime import date
from datetime import datetime, date
class LivraisonCreateGatewayRequest(BaseModel): class LivraisonCreateGatewayRequest(BaseModel):
"""Création d'une livraison côté gateway""" """Création d'une livraison côté gateway"""

View file

@ -1,6 +1,6 @@
from pydantic import BaseModel
from typing import Optional
from pydantic import BaseModel, Field, validator, EmailStr, field_validator
from typing import Optional, List, Dict
class FiltreRequest(BaseModel): class FiltreRequest(BaseModel):
filtre: Optional[str] = "" filtre: Optional[str] = ""

View file

@ -1,5 +1,6 @@
from pydantic import BaseModel, Field, validator, EmailStr, field_validator from pydantic import BaseModel, Field, field_validator
from typing import Optional, List, Dict from typing import Optional, Dict
class ClientCreateRequest(BaseModel): class ClientCreateRequest(BaseModel):
intitule: str = Field( intitule: str = Field(
@ -348,7 +349,7 @@ class ClientCreateRequest(BaseModel):
"statistique08": self.statistique08, "statistique08": self.statistique08,
"statistique09": self.statistique09, "statistique09": self.statistique09,
"statistique10": self.statistique10, "statistique10": self.statistique10,
"secteur": self.secteur, # Gardé pour compatibilité "secteur": self.secteur,
"encours_autorise": self.encours_autorise, "encours_autorise": self.encours_autorise,
"assurance_credit": self.assurance_credit, "assurance_credit": self.assurance_credit,
"langue": self.langue, "langue": self.langue,

View file

@ -1,9 +1,10 @@
from pydantic import BaseModel
from typing import Optional, Dict
from pydantic import BaseModel, Field, validator, EmailStr, field_validator
from typing import Optional, List, Dict
class ContactCreateRequest(BaseModel): class ContactCreateRequest(BaseModel):
"""Requête de création de contact""" """Requête de création de contact"""
numero: str numero: str
civilite: Optional[str] = None civilite: Optional[str] = None
nom: str nom: str
@ -21,17 +22,20 @@ class ContactCreateRequest(BaseModel):
class ContactListRequest(BaseModel): class ContactListRequest(BaseModel):
"""Requête de liste des contacts""" """Requête de liste des contacts"""
numero: str numero: str
class ContactGetRequest(BaseModel): class ContactGetRequest(BaseModel):
"""Requête de récupération d'un contact""" """Requête de récupération d'un contact"""
numero: str numero: str
contact_numero: int contact_numero: int
class ContactUpdateRequest(BaseModel): class ContactUpdateRequest(BaseModel):
"""Requête de modification d'un contact""" """Requête de modification d'un contact"""
numero: str numero: str
contact_numero: int contact_numero: int
updates: Dict updates: Dict
@ -39,5 +43,6 @@ class ContactUpdateRequest(BaseModel):
class ContactDeleteRequest(BaseModel): class ContactDeleteRequest(BaseModel):
"""Requête de suppression d'un contact""" """Requête de suppression d'un contact"""
numero: str numero: str
contact_numero: int contact_numero: int

View file

@ -1,21 +1,5 @@
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, validator, EmailStr, field_validator from typing import Optional, Dict
from typing import Optional, List, Dict
from enum import Enum, IntEnum
from datetime import datetime, date
class FournisseurCreateRequest(BaseModel):
intitule: str = Field(..., description="Raison sociale du fournisseur")
compte_collectif: str = Field("401000", description="Compte général rattaché")
num: Optional[str] = Field(None, description="Code fournisseur (auto si vide)")
adresse: Optional[str] = None
code_postal: Optional[str] = None
ville: Optional[str] = None
pays: Optional[str] = None
email: Optional[str] = None
telephone: Optional[str] = None
siret: Optional[str] = None
tva_intra: Optional[str] = None
class FournisseurCreateRequest(BaseModel): class FournisseurCreateRequest(BaseModel):
@ -37,5 +21,3 @@ class FournisseurUpdateGatewayRequest(BaseModel):
code: str code: str
fournisseur_data: Dict fournisseur_data: Dict

View file

@ -1,17 +1,16 @@
from pydantic import BaseModel, Field, validator, EmailStr, field_validator from pydantic import BaseModel, Field
from typing import Optional, List, Dict from typing import Optional
from enum import Enum, IntEnum from enum import IntEnum
class TiersListRequest(BaseModel): class TiersListRequest(BaseModel):
"""Requête de listage des tiers""" """Requête de listage des tiers"""
type_tiers: Optional[str] = Field( type_tiers: Optional[str] = Field(
None, None, description="Type: client, fournisseur, prospect, all"
description="Type: client, fournisseur, prospect, all"
)
filtre: str = Field(
"",
description="Filtre sur code ou intitulé"
) )
filtre: str = Field("", description="Filtre sur code ou intitulé")
class TypeTiers(IntEnum): class TypeTiers(IntEnum):
"""CT_Type - Type de tiers""" """CT_Type - Type de tiers"""
@ -19,4 +18,4 @@ class TypeTiers(IntEnum):
CLIENT = 0 CLIENT = 0
FOURNISSEUR = 1 FOURNISSEUR = 1
SALARIE = 2 SALARIE = 2
AUTRE = 3 AUTRE = 3

323
test.py
View file

@ -1,323 +0,0 @@
"""
🔬 DIAGNOSTIC APPROFONDI CRYSTAL REPORTS
Découvre pourquoi Crystal ne fonctionne pas après redémarrage
"""
import os
import winreg
import subprocess
import sys
def diagnostic_complet_crystal():
"""Diagnostic exhaustif de l'installation Crystal"""
print("="*70)
print("🔬 DIAGNOSTIC APPROFONDI CRYSTAL REPORTS")
print("="*70)
problemes = []
solutions = []
# ==========================================
# 1. VÉRIFIER SI CRYSTAL EST INSTALLÉ
# ==========================================
print("\n📁 1. Vérification présence des fichiers...")
chemins_installation = [
r"C:\Program Files\SAP BusinessObjects",
r"C:\Program Files (x86)\SAP BusinessObjects",
r"C:\Program Files\SAP\Crystal Reports",
r"C:\Program Files (x86)\SAP\Crystal Reports",
]
crystal_trouve = False
chemin_crystal = None
for chemin in chemins_installation:
if os.path.exists(chemin):
print(f" Dossier trouvé : {chemin}")
crystal_trouve = True
chemin_crystal = chemin
# Afficher taille
try:
total_size = 0
for dirpath, dirnames, filenames in os.walk(chemin):
for f in filenames:
fp = os.path.join(dirpath, f)
try:
total_size += os.path.getsize(fp)
except:
pass
size_mb = total_size / (1024 * 1024)
print(f" Taille : {size_mb:.1f} MB")
if size_mb < 100:
print(f" Taille suspecte (attendu: 300-800 MB)")
problemes.append("Installation incomplète (taille trop petite)")
except Exception as e:
print(f" Impossible de calculer taille : {e}")
else:
print(f" Absent : {chemin}")
if not crystal_trouve:
print("\n PROBLÈME MAJEUR : Crystal Reports n'est pas installé")
problemes.append("Crystal Reports non installé")
solutions.append("Télécharger et installer SAP Crystal Reports Runtime")
return {"problemes": problemes, "solutions": solutions, "installe": False}
# ==========================================
# 2. CHERCHER LES DLL CRITIQUES
# ==========================================
print("\n📦 2. Recherche DLL critiques...")
dll_critiques = {
'crpe32.dll': 'Crystal Reports Print Engine (CRITIQUE)',
'crxf_pdf.dll': 'Export PDF (CRITIQUE)',
'crdb_adoplus.dll': 'Connexion base de données',
'CrystalDecisions.CrystalReports.Engine.dll': 'Moteur Crystal .NET',
}
dll_trouvees = {}
for dll_nom, description in dll_critiques.items():
trouve = False
for root, dirs, files in os.walk(chemin_crystal):
if dll_nom.lower() in [f.lower() for f in files]:
dll_path = os.path.join(root, dll_nom)
dll_trouvees[dll_nom] = dll_path
print(f" {dll_nom}")
print(f" {dll_path}")
trouve = True
break
if not trouve:
print(f" {dll_nom} - {description}")
if "CRITIQUE" in description:
problemes.append(f"{dll_nom} manquante")
if len(dll_trouvees) < 2:
print("\n Trop peu de DLL trouvées - Installation corrompue")
problemes.append("DLL manquantes - Installation corrompue")
solutions.append("Réinstaller Crystal Reports Runtime")
# ==========================================
# 3. VÉRIFIER LE REGISTRE
# ==========================================
print("\n📋 3. Vérification registre Windows...")
prog_ids = [
"CrystalRuntime.Application.140",
"CrystalRuntime.Application.13",
"CrystalRuntime.Application.12",
"CrystalRuntime.Application",
"CrystalDesignRunTime.Application",
]
prog_ids_trouves = []
for prog_id in prog_ids:
try:
# Vérifier existence
key = winreg.OpenKey(winreg.HKEY_CLASSES_ROOT, prog_id)
print(f" {prog_id}")
# Lire le CLSID
try:
clsid_key = winreg.OpenKey(key, "CLSID")
clsid, _ = winreg.QueryValueEx(clsid_key, "")
print(f" CLSID: {clsid}")
# Vérifier que le CLSID existe aussi
try:
clsid_path = f"CLSID\\{clsid}"
clsid_key_check = winreg.OpenKey(winreg.HKEY_CLASSES_ROOT, clsid_path)
# Lire InprocServer32 (chemin DLL)
try:
server_key = winreg.OpenKey(clsid_key_check, "InprocServer32")
dll_path, _ = winreg.QueryValueEx(server_key, "")
print(f" DLL: {dll_path}")
# Vérifier que la DLL existe
if not os.path.exists(dll_path):
print(f" DLL INTROUVABLE: {dll_path}")
problemes.append(f"{prog_id}: DLL manquante ({dll_path})")
else:
prog_ids_trouves.append(prog_id)
except:
print(f" InprocServer32 non trouvé")
except:
print(f" CLSID {clsid} non trouvé dans registre")
problemes.append(f"{prog_id}: CLSID cassé")
except:
print(f" Pas de CLSID")
winreg.CloseKey(key)
except:
print(f" {prog_id}")
if not prog_ids_trouves:
print("\n Aucun ProgID valide - Enregistrement COM échoué")
problemes.append("ProgID non enregistrés correctement")
solutions.append("Réenregistrer les DLL Crystal avec regsvr32")
# ==========================================
# 4. VÉRIFIER ARCHITECTURE (32 vs 64 bit)
# ==========================================
print("\n🔧 4. Vérification architecture...")
# Architecture Python
python_arch = "64-bit" if sys.maxsize > 2**32 else "32-bit"
print(f" Python : {python_arch}")
# Architecture système
import platform
sys_arch = platform.machine()
print(f" Système : {sys_arch}")
# Détecter architecture Crystal installée
crystal_arch = None
if dll_trouvees:
first_dll = list(dll_trouvees.values())[0]
if "win64_x64" in first_dll or "x64" in first_dll:
crystal_arch = "64-bit"
elif "win32_x86" in first_dll or "x86" in first_dll:
crystal_arch = "32-bit"
if crystal_arch:
print(f" Crystal : {crystal_arch}")
if python_arch != crystal_arch:
print(f"\n INCOMPATIBILITÉ ARCHITECTURE")
print(f" Python {python_arch} ne peut pas utiliser Crystal {crystal_arch}")
problemes.append(f"Incompatibilité: Python {python_arch} vs Crystal {crystal_arch}")
solutions.append(f"Réinstaller Crystal en version {python_arch}")
# ==========================================
# 5. VÉRIFIER SERVICES WINDOWS
# ==========================================
print("\n🔄 5. Vérification services Windows...")
try:
result = subprocess.run(
['sc', 'query', 'type=', 'service'],
capture_output=True,
text=True,
timeout=10
)
services_crystal_attendus = [
"SAP Crystal Reports",
"Crystal Reports",
"CrystalReports",
]
services_trouves = []
for service in services_crystal_attendus:
if service.lower() in result.stdout.lower():
services_trouves.append(service)
print(f" Service trouvé: {service}")
if not services_trouves:
print(f" Aucun service Crystal trouvé")
print(f" (Normal pour Runtime léger)")
except Exception as e:
print(f" Impossible de vérifier services: {e}")
# ==========================================
# 6. TEST INSTANCIATION COM DÉTAILLÉ
# ==========================================
print("\n🧪 6. Test instanciation COM détaillé...")
import win32com.client
for prog_id in prog_ids_trouves:
print(f"\n Test: {prog_id}")
try:
obj = win32com.client.Dispatch(prog_id)
print(f" Instanciation RÉUSSIE")
# Lister méthodes disponibles
print(f" Méthodes disponibles:")
for attr in dir(obj):
if not attr.startswith('_') and callable(getattr(obj, attr, None)):
print(f" - {attr}()")
return {
"problemes": problemes,
"solutions": solutions,
"installe": True,
"prog_id_fonctionnel": prog_id
}
except Exception as e:
print(f" Échec: {e}")
print(f" Type erreur: {type(e).__name__}")
print(f" Code: {e.args if hasattr(e, 'args') else 'N/A'}")
# ==========================================
# 7. RÉSUMÉ ET RECOMMANDATIONS
# ==========================================
print("\n" + "="*70)
print("📊 RÉSUMÉ DU DIAGNOSTIC")
print("="*70)
print(f"\n📁 Installation détectée: {'OUI' if crystal_trouve else 'NON'}")
print(f"📦 DLL trouvées: {len(dll_trouvees)}/{len(dll_critiques)}")
print(f"📋 ProgID valides: {len(prog_ids_trouves)}")
print(f"🔧 Architecture: Python {python_arch}, Crystal {crystal_arch or 'INCONNUE'}")
if problemes:
print(f"\n PROBLÈMES DÉTECTÉS ({len(problemes)}):")
for i, pb in enumerate(problemes, 1):
print(f" {i}. {pb}")
if solutions:
print(f"\n💡 SOLUTIONS RECOMMANDÉES:")
for i, sol in enumerate(solutions, 1):
print(f" {i}. {sol}")
print("\n" + "="*70)
return {
"problemes": problemes,
"solutions": solutions,
"installe": crystal_trouve,
"dll_trouvees": list(dll_trouvees.keys()),
"prog_ids_valides": prog_ids_trouves,
"architecture_ok": python_arch == crystal_arch if crystal_arch else False
}
if __name__ == "__main__":
resultats = diagnostic_complet_crystal()
print("\n💾 Résultats sauvegardés dans diagnostic_crystal.txt")
# Sauvegarder dans fichier
with open("diagnostic_crystal.txt", "w", encoding="utf-8") as f:
f.write("="*70 + "\n")
f.write("DIAGNOSTIC CRYSTAL REPORTS\n")
f.write("="*70 + "\n\n")
f.write(f"Installation détectée: {resultats['installe']}\n")
f.write(f"DLL trouvées: {', '.join(resultats.get('dll_trouvees', []))}\n")
f.write(f"ProgID valides: {', '.join(resultats.get('prog_ids_valides', []))}\n")
f.write(f"Architecture OK: {resultats.get('architecture_ok', False)}\n\n")
f.write("PROBLÈMES:\n")
for pb in resultats['problemes']:
f.write(f" - {pb}\n")
f.write("\nSOLUTIONS:\n")
for sol in resultats['solutions']:
f.write(f" - {sol}\n")

View file

@ -2,6 +2,7 @@ import logging
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _extraire_article(article_obj): def _extraire_article(article_obj):
try: try:
data = { data = {
@ -23,68 +24,64 @@ def _extraire_article(article_obj):
if code_barre1: if code_barre1:
data["code_ean"] = code_barre1 data["code_ean"] = code_barre1
data["code_barre"] = code_barre1 data["code_barre"] = code_barre1
except: except Exception:
pass pass
try: try:
data["prix_vente"] = float(getattr(article_obj, "AR_PrixVen", 0.0)) data["prix_vente"] = float(getattr(article_obj, "AR_PrixVen", 0.0))
except: except Exception:
data["prix_vente"] = 0.0 data["prix_vente"] = 0.0
try: try:
data["prix_achat"] = float(getattr(article_obj, "AR_PrixAch", 0.0)) data["prix_achat"] = float(getattr(article_obj, "AR_PrixAch", 0.0))
except: except Exception:
data["prix_achat"] = 0.0 data["prix_achat"] = 0.0
try: try:
data["prix_revient"] = float( data["prix_revient"] = float(getattr(article_obj, "AR_PrixRevient", 0.0))
getattr(article_obj, "AR_PrixRevient", 0.0) except Exception:
)
except:
data["prix_revient"] = 0.0 data["prix_revient"] = 0.0
try: try:
data["stock_reel"] = float(getattr(article_obj, "AR_Stock", 0.0)) data["stock_reel"] = float(getattr(article_obj, "AR_Stock", 0.0))
except: except Exception:
data["stock_reel"] = 0.0 data["stock_reel"] = 0.0
try: try:
data["stock_mini"] = float(getattr(article_obj, "AR_StockMini", 0.0)) data["stock_mini"] = float(getattr(article_obj, "AR_StockMini", 0.0))
except: except Exception:
data["stock_mini"] = 0.0 data["stock_mini"] = 0.0
try: try:
data["stock_maxi"] = float(getattr(article_obj, "AR_StockMaxi", 0.0)) data["stock_maxi"] = float(getattr(article_obj, "AR_StockMaxi", 0.0))
except: except Exception:
data["stock_maxi"] = 0.0 data["stock_maxi"] = 0.0
try: try:
data["stock_reserve"] = float(getattr(article_obj, "AR_QteCom", 0.0)) data["stock_reserve"] = float(getattr(article_obj, "AR_QteCom", 0.0))
except: except Exception:
data["stock_reserve"] = 0.0 data["stock_reserve"] = 0.0
try: try:
data["stock_commande"] = float( data["stock_commande"] = float(getattr(article_obj, "AR_QteComFou", 0.0))
getattr(article_obj, "AR_QteComFou", 0.0) except Exception:
)
except:
data["stock_commande"] = 0.0 data["stock_commande"] = 0.0
try: try:
data["stock_disponible"] = data["stock_reel"] - data["stock_reserve"] data["stock_disponible"] = data["stock_reel"] - data["stock_reserve"]
except: except Exception:
data["stock_disponible"] = data["stock_reel"] data["stock_disponible"] = data["stock_reel"]
try: try:
commentaire = getattr(article_obj, "AR_Commentaire", "").strip() commentaire = getattr(article_obj, "AR_Commentaire", "").strip()
data["description"] = commentaire data["description"] = commentaire
except: except Exception:
data["description"] = "" data["description"] = ""
try: try:
design2 = getattr(article_obj, "AR_Design2", "").strip() design2 = getattr(article_obj, "AR_Design2", "").strip()
data["designation_complementaire"] = design2 data["designation_complementaire"] = design2
except: except Exception:
data["designation_complementaire"] = "" data["designation_complementaire"] = ""
try: try:
@ -95,7 +92,7 @@ def _extraire_article(article_obj):
1: "Prestation", 1: "Prestation",
2: "Divers", 2: "Divers",
}.get(type_art, "Inconnu") }.get(type_art, "Inconnu")
except: except Exception:
data["type_article"] = 0 data["type_article"] = 0
data["type_article_libelle"] = "Article" data["type_article_libelle"] = "Article"
@ -113,11 +110,11 @@ def _extraire_article(article_obj):
).strip() ).strip()
else: else:
data["famille_libelle"] = "" data["famille_libelle"] = ""
except: except Exception:
data["famille_libelle"] = "" data["famille_libelle"] = ""
else: else:
data["famille_libelle"] = "" data["famille_libelle"] = ""
except: except Exception:
data["famille_code"] = "" data["famille_code"] = ""
data["famille_libelle"] = "" data["famille_libelle"] = ""
@ -135,39 +132,39 @@ def _extraire_article(article_obj):
).strip() ).strip()
else: else:
data["fournisseur_nom"] = "" data["fournisseur_nom"] = ""
except: except Exception:
data["fournisseur_nom"] = "" data["fournisseur_nom"] = ""
else: else:
data["fournisseur_nom"] = "" data["fournisseur_nom"] = ""
except: except Exception:
data["fournisseur_principal"] = "" data["fournisseur_principal"] = ""
data["fournisseur_nom"] = "" data["fournisseur_nom"] = ""
try: try:
data["unite_vente"] = getattr(article_obj, "AR_UniteVen", "").strip() data["unite_vente"] = getattr(article_obj, "AR_UniteVen", "").strip()
except: except Exception:
data["unite_vente"] = "" data["unite_vente"] = ""
try: try:
data["unite_achat"] = getattr(article_obj, "AR_UniteAch", "").strip() data["unite_achat"] = getattr(article_obj, "AR_UniteAch", "").strip()
except: except Exception:
data["unite_achat"] = "" data["unite_achat"] = ""
try: try:
data["poids"] = float(getattr(article_obj, "AR_Poids", 0.0)) data["poids"] = float(getattr(article_obj, "AR_Poids", 0.0))
except: except Exception:
data["poids"] = 0.0 data["poids"] = 0.0
try: try:
data["volume"] = float(getattr(article_obj, "AR_Volume", 0.0)) data["volume"] = float(getattr(article_obj, "AR_Volume", 0.0))
except: except Exception:
data["volume"] = 0.0 data["volume"] = 0.0
try: try:
sommeil = getattr(article_obj, "AR_Sommeil", 0) sommeil = getattr(article_obj, "AR_Sommeil", 0)
data["est_actif"] = sommeil == 0 data["est_actif"] = sommeil == 0
data["en_sommeil"] = sommeil == 1 data["en_sommeil"] = sommeil == 1
except: except Exception:
data["est_actif"] = True data["est_actif"] = True
data["en_sommeil"] = False data["en_sommeil"] = False
@ -182,22 +179,22 @@ def _extraire_article(article_obj):
data["tva_taux"] = float(getattr(tva_obj, "TA_Taux", 20.0)) data["tva_taux"] = float(getattr(tva_obj, "TA_Taux", 20.0))
else: else:
data["tva_taux"] = 20.0 data["tva_taux"] = 20.0
except: except Exception:
data["tva_taux"] = 20.0 data["tva_taux"] = 20.0
except: except Exception:
data["tva_code"] = "" data["tva_code"] = ""
data["tva_taux"] = 20.0 data["tva_taux"] = 20.0
try: try:
date_creation = getattr(article_obj, "AR_DateCreate", None) date_creation = getattr(article_obj, "AR_DateCreate", None)
data["date_creation"] = str(date_creation) if date_creation else "" data["date_creation"] = str(date_creation) if date_creation else ""
except: except Exception:
data["date_creation"] = "" data["date_creation"] = ""
try: try:
date_modif = getattr(article_obj, "AR_DateModif", None) date_modif = getattr(article_obj, "AR_DateModif", None)
data["date_modification"] = str(date_modif) if date_modif else "" data["date_modification"] = str(date_modif) if date_modif else ""
except: except Exception:
data["date_modification"] = "" data["date_modification"] = ""
return data return data
@ -238,7 +235,7 @@ def _extraire_article(article_obj):
"date_modification": "", "date_modification": "",
} }
__all__ = [ __all__ = [
"_extraire_article", "_extraire_article",
] ]

File diff suppressed because it is too large Load diff

View file

@ -2,39 +2,39 @@ import logging
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def verifier_stock_suffisant(article_ref, quantite, cursor, depot=None): def verifier_stock_suffisant(article_ref, quantite, cursor, depot=None):
"""Version thread-safe avec lock SQL""" """Version thread-safe avec lock SQL"""
try: try:
cursor.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
cursor.execute("BEGIN TRANSACTION")
cursor.execute("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE") try:
cursor.execute("BEGIN TRANSACTION") cursor.execute(
"""
try:
cursor.execute(
"""
SELECT SUM(AS_QteSto) SELECT SUM(AS_QteSto)
FROM F_ARTSTOCK WITH (UPDLOCK, ROWLOCK) FROM F_ARTSTOCK WITH (UPDLOCK, ROWLOCK)
WHERE AR_Ref = ? WHERE AR_Ref = ?
""", """,
(article_ref.upper(),), (article_ref.upper(),),
) )
row = cursor.fetchone() row = cursor.fetchone()
stock_dispo = float(row[0]) if row and row[0] else 0.0 stock_dispo = float(row[0]) if row and row[0] else 0.0
suffisant = stock_dispo >= quantite suffisant = stock_dispo >= quantite
cursor.execute("COMMIT") cursor.execute("COMMIT")
return { return {
"suffisant": suffisant, "suffisant": suffisant,
"stock_disponible": stock_dispo, "stock_disponible": stock_dispo,
"quantite_demandee": quantite, "quantite_demandee": quantite,
} }
except: except Exception:
cursor.execute("ROLLBACK") cursor.execute("ROLLBACK")
raise raise
except Exception as e: except Exception as e:
logger.error(f"Erreur vérification stock: {e}") logger.error(f"Erreur vérification stock: {e}")

View file

@ -3,6 +3,7 @@ import logging
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _rechercher_devis_dans_liste(numero_devis, factory_doc): def _rechercher_devis_dans_liste(numero_devis, factory_doc):
"""Recherche un devis dans les 100 premiers éléments de la liste.""" """Recherche un devis dans les 100 premiers éléments de la liste."""
index = 1 index = 1
@ -23,11 +24,12 @@ def _rechercher_devis_dans_liste(numero_devis, factory_doc):
return persist_test return persist_test
index += 1 index += 1
except: except Exception:
index += 1 index += 1
return None return None
def _recuperer_numero_devis(process, doc): def _recuperer_numero_devis(process, doc):
"""Récupère le numéro du devis créé via plusieurs méthodes.""" """Récupère le numéro du devis créé via plusieurs méthodes."""
numero_devis = None numero_devis = None
@ -38,7 +40,7 @@ def _recuperer_numero_devis(process, doc):
doc_result = win32com.client.CastTo(doc_result, "IBODocumentVente3") doc_result = win32com.client.CastTo(doc_result, "IBODocumentVente3")
doc_result.Read() doc_result.Read()
numero_devis = getattr(doc_result, "DO_Piece", "") numero_devis = getattr(doc_result, "DO_Piece", "")
except: except Exception:
pass pass
if not numero_devis: if not numero_devis:
@ -50,12 +52,10 @@ def _recuperer_numero_devis(process, doc):
doc.Write() doc.Write()
doc.Read() doc.Read()
numero_devis = getattr(doc, "DO_Piece", "") numero_devis = getattr(doc, "DO_Piece", "")
except: except Exception:
pass pass
return numero_devis return numero_devis
__all__ = [
"_recuperer_numero_devis", __all__ = ["_recuperer_numero_devis", "_rechercher_devis_dans_liste"]
"_rechercher_devis_dans_liste"
]

View file

@ -1,5 +1,6 @@
from typing import Dict, List, Optional, Any from typing import Dict, List, Optional, Any
def _extraire_infos_devis(doc, numero: str, champs_modifies: list) -> Dict: def _extraire_infos_devis(doc, numero: str, champs_modifies: list) -> Dict:
"""Extrait les informations complètes du devis.""" """Extrait les informations complètes du devis."""
total_ht = float(getattr(doc, "DO_TotalHT", 0.0)) total_ht = float(getattr(doc, "DO_TotalHT", 0.0))
@ -12,7 +13,7 @@ def _extraire_infos_devis(doc, numero: str, champs_modifies: list) -> Dict:
date_doc = getattr(doc, "DO_Date", None) date_doc = getattr(doc, "DO_Date", None)
if date_doc: if date_doc:
date_devis = date_doc.strftime("%Y-%m-%d") date_devis = date_doc.strftime("%Y-%m-%d")
except: except Exception:
pass pass
date_livraison = None date_livraison = None
@ -20,7 +21,7 @@ def _extraire_infos_devis(doc, numero: str, champs_modifies: list) -> Dict:
date_livr = getattr(doc, "DO_DateLivr", None) date_livr = getattr(doc, "DO_DateLivr", None)
if date_livr: if date_livr:
date_livraison = date_livr.strftime("%Y-%m-%d") date_livraison = date_livr.strftime("%Y-%m-%d")
except: except Exception:
pass pass
client_code = "" client_code = ""
@ -29,7 +30,7 @@ def _extraire_infos_devis(doc, numero: str, champs_modifies: list) -> Dict:
if client_obj: if client_obj:
client_obj.Read() client_obj.Read()
client_code = getattr(client_obj, "CT_Num", "") client_code = getattr(client_obj, "CT_Num", "")
except: except Exception:
pass pass
return { return {
@ -44,6 +45,7 @@ def _extraire_infos_devis(doc, numero: str, champs_modifies: list) -> Dict:
"client_code": client_code, "client_code": client_code,
} }
__all__ = [ __all__ = [
"_extraire_infos_devis", "_extraire_infos_devis",
] ]

View file

@ -2,14 +2,16 @@ import win32com.client
from typing import Optional from typing import Optional
import logging import logging
logger = logging.getLogger(__name__)
from utils.functions.functions import ( from utils.functions.functions import (
_convertir_type_depuis_sql, _convertir_type_depuis_sql,
_convertir_type_pour_sql, _convertir_type_pour_sql,
_safe_strip _safe_strip,
) )
logger = logging.getLogger(__name__)
def _afficher_etat_document(doc, titre: str): def _afficher_etat_document(doc, titre: str):
"""Affiche l'état complet d'un document.""" """Affiche l'état complet d'un document."""
logger.info("-" * 80) logger.info("-" * 80)
@ -19,15 +21,15 @@ def _afficher_etat_document(doc, titre: str):
logger.info(f" DO_Piece: {getattr(doc, 'DO_Piece', 'N/A')}") logger.info(f" DO_Piece: {getattr(doc, 'DO_Piece', 'N/A')}")
logger.info(f" DO_Ref: '{getattr(doc, 'DO_Ref', 'N/A')}'") logger.info(f" DO_Ref: '{getattr(doc, 'DO_Ref', 'N/A')}'")
logger.info(f" DO_Statut: {getattr(doc, 'DO_Statut', 'N/A')}") logger.info(f" DO_Statut: {getattr(doc, 'DO_Statut', 'N/A')}")
date_doc = getattr(doc, 'DO_Date', None) date_doc = getattr(doc, "DO_Date", None)
date_str = date_doc.strftime('%Y-%m-%d') if date_doc else 'None' date_str = date_doc.strftime("%Y-%m-%d") if date_doc else "None"
logger.info(f" DO_Date: {date_str}") logger.info(f" DO_Date: {date_str}")
date_livr = getattr(doc, 'DO_DateLivr', None) date_livr = getattr(doc, "DO_DateLivr", None)
date_livr_str = date_livr.strftime('%Y-%m-%d') if date_livr else 'None' date_livr_str = date_livr.strftime("%Y-%m-%d") if date_livr else "None"
logger.info(f" DO_DateLivr: {date_livr_str}") logger.info(f" DO_DateLivr: {date_livr_str}")
logger.info(f" DO_TotalHT: {getattr(doc, 'DO_TotalHT', 0)}") logger.info(f" DO_TotalHT: {getattr(doc, 'DO_TotalHT', 0)}")
logger.info(f" DO_TotalTTC: {getattr(doc, 'DO_TotalTTC', 0)}") logger.info(f" DO_TotalTTC: {getattr(doc, 'DO_TotalTTC', 0)}")
except Exception as e: except Exception as e:
@ -40,7 +42,7 @@ def _compter_lignes_document(doc) -> int:
try: try:
try: try:
factory_lignes = doc.FactoryDocumentLigne factory_lignes = doc.FactoryDocumentLigne
except: except Exception:
factory_lignes = doc.FactoryDocumentVenteLigne factory_lignes = doc.FactoryDocumentVenteLigne
count = 0 count = 0
@ -52,7 +54,7 @@ def _compter_lignes_document(doc) -> int:
break break
count += 1 count += 1
index += 1 index += 1
except: except Exception:
break break
return count return count
except Exception as e: except Exception as e:
@ -63,7 +65,7 @@ def _compter_lignes_document(doc) -> int:
def _rechercher_devis_par_numero(numero: str, factory): def _rechercher_devis_par_numero(numero: str, factory):
"""Recherche un devis par numéro dans la liste.""" """Recherche un devis par numéro dans la liste."""
logger.info(f" Recherche de {numero} dans la liste...") logger.info(f" Recherche de {numero} dans la liste...")
index = 1 index = 1
while index < 10000: while index < 10000:
try: try:
@ -82,7 +84,7 @@ def _rechercher_devis_par_numero(numero: str, factory):
return persist_test return persist_test
index += 1 index += 1
except: except Exception:
index += 1 index += 1
logger.error(f" Devis {numero} non trouvé dans la liste") logger.error(f" Devis {numero} non trouvé dans la liste")
@ -91,7 +93,6 @@ def _rechercher_devis_par_numero(numero: str, factory):
def _lire_document_sql(cursor, numero: str, type_doc: int): def _lire_document_sql(cursor, numero: str, type_doc: int):
try: try:
query = """ query = """
SELECT SELECT
d.DO_Piece, d.DO_Date, d.DO_Ref, d.DO_TotalHT, d.DO_TotalTTC, d.DO_Piece, d.DO_Date, d.DO_Ref, d.DO_TotalHT, d.DO_TotalTTC,
@ -126,71 +127,56 @@ def _lire_document_sql(cursor, numero: str, type_doc: int):
numero_piece = _safe_strip(row[0]) numero_piece = _safe_strip(row[0])
logger.info(f"[SQL READ] Document trouvé: {numero_piece}") logger.info(f"[SQL READ] Document trouvé: {numero_piece}")
doc = { doc = {
"numero": numero_piece, "numero": numero_piece,
"reference": _safe_strip(row[2]), # DO_Ref "reference": _safe_strip(row[2]),
"date": str(row[1]) if row[1] else "", # DO_Date "date": str(row[1]) if row[1] else "",
"date_livraison": (str(row[7]) if row[7] else ""), # DO_DateLivr "date_livraison": (str(row[7]) if row[7] else ""),
"date_expedition": ( "date_expedition": (str(row[8]) if row[8] else ""),
str(row[8]) if row[8] else "" "client_code": _safe_strip(row[6]),
), # DO_DateExpedition "client_intitule": _safe_strip(row[39]),
"client_code": _safe_strip(row[6]), # DO_Tiers "client_adresse": _safe_strip(row[40]),
"client_intitule": _safe_strip(row[39]), # CT_Intitule "client_code_postal": _safe_strip(row[41]),
"client_adresse": _safe_strip(row[40]), # CT_Adresse "client_ville": _safe_strip(row[42]),
"client_code_postal": _safe_strip(row[41]), # CT_CodePostal "client_telephone": _safe_strip(row[43]),
"client_ville": _safe_strip(row[42]), # CT_Ville "client_email": _safe_strip(row[44]),
"client_telephone": _safe_strip(row[43]), # CT_Telephone "contact": _safe_strip(row[9]),
"client_email": _safe_strip(row[44]), # CT_EMail "total_ht": float(row[3]) if row[3] else 0.0,
"contact": _safe_strip(row[9]), # DO_Contact "total_ht_net": float(row[10]) if row[10] else 0.0,
"total_ht": float(row[3]) if row[3] else 0.0, # DO_TotalHT "total_ttc": float(row[4]) if row[4] else 0.0,
"total_ht_net": float(row[10]) if row[10] else 0.0, # DO_TotalHTNet "net_a_payer": float(row[11]) if row[11] else 0.0,
"total_ttc": float(row[4]) if row[4] else 0.0, # DO_TotalTTC "montant_regle": (float(row[12]) if row[12] else 0.0),
"net_a_payer": float(row[11]) if row[11] else 0.0, # DO_NetAPayer "reliquat": float(row[13]) if row[13] else 0.0,
"montant_regle": ( "taux_escompte": (float(row[14]) if row[14] else 0.0),
float(row[12]) if row[12] else 0.0 "escompte": float(row[15]) if row[15] else 0.0,
), # DO_MontantRegle "taxe1": float(row[16]) if row[16] else 0.0,
"reliquat": float(row[13]) if row[13] else 0.0, # DO_Reliquat "taxe2": float(row[17]) if row[17] else 0.0,
"taux_escompte": ( "taxe3": float(row[18]) if row[18] else 0.0,
float(row[14]) if row[14] else 0.0 "code_taxe1": _safe_strip(row[19]),
), # DO_TxEscompte "code_taxe2": _safe_strip(row[20]),
"escompte": float(row[15]) if row[15] else 0.0, # DO_Escompte "code_taxe3": _safe_strip(row[21]),
"taxe1": float(row[16]) if row[16] else 0.0, # DO_Taxe1 "statut": int(row[5]) if row[5] is not None else 0,
"taxe2": float(row[17]) if row[17] else 0.0, # DO_Taxe2
"taxe3": float(row[18]) if row[18] else 0.0, # DO_Taxe3
"code_taxe1": _safe_strip(row[19]), # DO_CodeTaxe1
"code_taxe2": _safe_strip(row[20]), # DO_CodeTaxe2
"code_taxe3": _safe_strip(row[21]), # DO_CodeTaxe3
"statut": int(row[5]) if row[5] is not None else 0, # DO_Statut
"statut_estatut": ( "statut_estatut": (
int(row[22]) if row[22] is not None else 0 int(row[22]) if row[22] is not None else 0
), # DO_EStatut ),
"imprime": int(row[23]) if row[23] is not None else 0, # DO_Imprim "imprime": int(row[23]) if row[23] is not None else 0,
"valide": int(row[24]) if row[24] is not None else 0, # DO_Valide "valide": int(row[24]) if row[24] is not None else 0,
"cloture": int(row[25]) if row[25] is not None else 0, # DO_Cloture "cloture": int(row[25]) if row[25] is not None else 0,
"transfere": ( "transfere": (int(row[26]) if row[26] is not None else 0),
int(row[26]) if row[26] is not None else 0 "souche": int(row[27]) if row[27] is not None else 0,
), # DO_Transfere "piece_origine": _safe_strip(row[28]),
"souche": int(row[27]) if row[27] is not None else 0, # DO_Souche "guid": _safe_strip(row[29]),
"piece_origine": _safe_strip(row[28]), # DO_PieceOrig "ca_num": _safe_strip(row[30]),
"guid": _safe_strip(row[29]), # DO_GUID "cg_num": _safe_strip(row[31]),
"ca_num": _safe_strip(row[30]), # CA_Num "expedition": (int(row[32]) if row[32] is not None else 1),
"cg_num": _safe_strip(row[31]), # CG_Num "condition": (int(row[33]) if row[33] is not None else 1),
"expedition": ( "tarif": int(row[34]) if row[34] is not None else 1,
int(row[32]) if row[32] is not None else 1 "type_frais": (int(row[35]) if row[35] is not None else 0),
), # DO_Expedit "valeur_frais": float(row[36]) if row[36] else 0.0,
"condition": (
int(row[33]) if row[33] is not None else 1
), # DO_Condition
"tarif": int(row[34]) if row[34] is not None else 1, # DO_Tarif
"type_frais": (
int(row[35]) if row[35] is not None else 0
), # DO_TypeFrais
"valeur_frais": float(row[36]) if row[36] else 0.0, # DO_ValFrais
"type_franco": ( "type_franco": (
int(row[37]) if row[37] is not None else 0 int(row[37]) if row[37] is not None else 0
), # DO_TypeFranco ),
"valeur_franco": float(row[38]) if row[38] else 0.0, # DO_ValFranco "valeur_franco": float(row[38]) if row[38] else 0.0,
} }
cursor.execute( cursor.execute(
@ -216,8 +202,7 @@ def _lire_document_sql(cursor, numero: str, type_doc: int):
) )
montant_net = ( montant_net = (
float(ligne_row.DL_MontantNet) float(ligne_row.DL_MontantNet)
if hasattr(ligne_row, "DL_MontantNet") if hasattr(ligne_row, "DL_MontantNet") and ligne_row.DL_MontantNet
and ligne_row.DL_MontantNet
else montant_ht else montant_ht
) )
@ -245,15 +230,11 @@ def _lire_document_sql(cursor, numero: str, type_doc: int):
montant_taxe3 = montant_net * (taux_taxe3 / 100) montant_taxe3 = montant_net * (taux_taxe3 / 100)
ligne = { ligne = {
"numero_ligne": ( "numero_ligne": (int(ligne_row.DL_Ligne) if ligne_row.DL_Ligne else 0),
int(ligne_row.DL_Ligne) if ligne_row.DL_Ligne else 0
),
"article_code": _safe_strip(ligne_row.AR_Ref), "article_code": _safe_strip(ligne_row.AR_Ref),
"designation": _safe_strip(ligne_row.DL_Design), "designation": _safe_strip(ligne_row.DL_Design),
"designation_article": _safe_strip(ligne_row.AR_Design), "designation_article": _safe_strip(ligne_row.AR_Design),
"quantite": ( "quantite": (float(ligne_row.DL_Qte) if ligne_row.DL_Qte else 0.0),
float(ligne_row.DL_Qte) if ligne_row.DL_Qte else 0.0
),
"quantite_livree": ( "quantite_livree": (
float(ligne_row.DL_QteLiv) float(ligne_row.DL_QteLiv)
if hasattr(ligne_row, "DL_QteLiv") and ligne_row.DL_QteLiv if hasattr(ligne_row, "DL_QteLiv") and ligne_row.DL_QteLiv
@ -311,9 +292,7 @@ def _lire_document_sql(cursor, numero: str, type_doc: int):
else 0 else 0
), ),
"remise_article": ( "remise_article": (
float(ligne_row.AR_Escompte) float(ligne_row.AR_Escompte) if ligne_row.AR_Escompte else 0.0
if ligne_row.AR_Escompte
else 0.0
), ),
"taux_taxe1": taux_taxe1, "taux_taxe1": taux_taxe1,
"montant_taxe1": montant_taxe1, "montant_taxe1": montant_taxe1,
@ -333,20 +312,15 @@ def _lire_document_sql(cursor, numero: str, type_doc: int):
float(ligne_row.AR_CoutStd) if ligne_row.AR_CoutStd else 0.0 float(ligne_row.AR_CoutStd) if ligne_row.AR_CoutStd else 0.0
), ),
"poids_net": ( "poids_net": (
float(ligne_row.AR_PoidsNet) float(ligne_row.AR_PoidsNet) if ligne_row.AR_PoidsNet else 0.0
if ligne_row.AR_PoidsNet
else 0.0
), ),
"poids_brut": ( "poids_brut": (
float(ligne_row.AR_PoidsBrut) float(ligne_row.AR_PoidsBrut) if ligne_row.AR_PoidsBrut else 0.0
if ligne_row.AR_PoidsBrut
else 0.0
), ),
"unite_vente": _safe_strip(ligne_row.AR_UniteVen), "unite_vente": _safe_strip(ligne_row.AR_UniteVen),
"date_livraison_ligne": ( "date_livraison_ligne": (
str(ligne_row.DL_DateLivr) str(ligne_row.DL_DateLivr)
if hasattr(ligne_row, "DL_DateLivr") if hasattr(ligne_row, "DL_DateLivr") and ligne_row.DL_DateLivr
and ligne_row.DL_DateLivr
else "" else ""
), ),
"statut_ligne": ( "statut_ligne": (
@ -356,9 +330,7 @@ def _lire_document_sql(cursor, numero: str, type_doc: int):
else 0 else 0
), ),
"depot": ( "depot": (
_safe_strip(ligne_row.DE_No) _safe_strip(ligne_row.DE_No) if hasattr(ligne_row, "DE_No") else ""
if hasattr(ligne_row, "DE_No")
else ""
), ),
"numero_commande": ( "numero_commande": (
_safe_strip(ligne_row.DL_NoColis) _safe_strip(ligne_row.DL_NoColis)
@ -376,9 +348,9 @@ def _lire_document_sql(cursor, numero: str, type_doc: int):
doc["lignes"] = lignes doc["lignes"] = lignes
doc["nb_lignes"] = len(lignes) doc["nb_lignes"] = len(lignes)
total_ht_calcule = sum(l.get("montant_ligne_ht", 0) for l in lignes) total_ht_calcule = sum(ligne.get("montant_ligne_ht", 0) for ligne in lignes)
total_ttc_calcule = sum(l.get("montant_ligne_ttc", 0) for l in lignes) total_ttc_calcule = sum(ligne.get("montant_ligne_ttc", 0) for ligne in lignes)
total_taxes_calcule = sum(l.get("total_taxes", 0) for l in lignes) total_taxes_calcule = sum(ligne.get("total_taxes", 0) for ligne in lignes)
doc["total_ht_calcule"] = total_ht_calcule doc["total_ht_calcule"] = total_ht_calcule
doc["total_ttc_calcule"] = total_ttc_calcule doc["total_ttc_calcule"] = total_ttc_calcule
@ -390,6 +362,7 @@ def _lire_document_sql(cursor, numero: str, type_doc: int):
logger.error(f" Erreur SQL lecture document {numero}: {e}", exc_info=True) logger.error(f" Erreur SQL lecture document {numero}: {e}", exc_info=True)
return None return None
def _lister_documents_avec_lignes_sql( def _lister_documents_avec_lignes_sql(
cursor, cursor,
type_doc: int, type_doc: int,
@ -423,7 +396,9 @@ def _lister_documents_avec_lignes_sql(
params = [type_doc_sql] params = [type_doc_sql]
if filtre: if filtre:
query += " AND (d.DO_Piece LIKE ? OR c.CT_Intitule LIKE ? OR d.DO_Ref LIKE ?)" query += (
" AND (d.DO_Piece LIKE ? OR c.CT_Intitule LIKE ? OR d.DO_Ref LIKE ?)"
)
params.extend([f"%{filtre}%", f"%{filtre}%", f"%{filtre}%"]) params.extend([f"%{filtre}%", f"%{filtre}%", f"%{filtre}%"])
query += " ORDER BY d.DO_Date DESC" query += " ORDER BY d.DO_Date DESC"
@ -450,7 +425,7 @@ def _lister_documents_avec_lignes_sql(
for idx, entete in enumerate(entetes): for idx, entete in enumerate(entetes):
numero = _safe_strip(entete.DO_Piece) numero = _safe_strip(entete.DO_Piece)
logger.info( logger.info(
f"[SQL LIST] [{idx+1}/{len(entetes)}] Traitement {numero}..." f"[SQL LIST] [{idx + 1}/{len(entetes)}] Traitement {numero}..."
) )
try: try:
@ -469,9 +444,7 @@ def _lister_documents_avec_lignes_sql(
numero.upper().startswith(p) for p in prefixes_acceptes numero.upper().startswith(p) for p in prefixes_acceptes
) )
if not est_vente: if not est_vente:
logger.info( logger.info(f"[SQL LIST] {numero} : exclu (préfixe achat)")
f"[SQL LIST] {numero} : exclu (préfixe achat)"
)
stats["exclus_prefixe"] += 1 stats["exclus_prefixe"] += 1
continue continue
@ -488,9 +461,7 @@ def _lister_documents_avec_lignes_sql(
"reference": _safe_strip(entete.DO_Ref), "reference": _safe_strip(entete.DO_Ref),
"date": str(entete.DO_Date) if entete.DO_Date else "", "date": str(entete.DO_Date) if entete.DO_Date else "",
"date_livraison": ( "date_livraison": (
str(entete.DO_DateLivr) str(entete.DO_DateLivr) if entete.DO_DateLivr else ""
if entete.DO_DateLivr
else ""
), ),
"date_expedition": ( "date_expedition": (
str(entete.DO_DateExpedition) str(entete.DO_DateExpedition)
@ -500,34 +471,22 @@ def _lister_documents_avec_lignes_sql(
"client_code": _safe_strip(entete.DO_Tiers), "client_code": _safe_strip(entete.DO_Tiers),
"client_intitule": _safe_strip(entete.CT_Intitule), "client_intitule": _safe_strip(entete.CT_Intitule),
"client_adresse": _safe_strip(entete.CT_Adresse), "client_adresse": _safe_strip(entete.CT_Adresse),
"client_code_postal": _safe_strip( "client_code_postal": _safe_strip(entete.CT_CodePostal),
entete.CT_CodePostal
),
"client_ville": _safe_strip(entete.CT_Ville), "client_ville": _safe_strip(entete.CT_Ville),
"client_telephone": _safe_strip( "client_telephone": _safe_strip(entete.CT_Telephone),
entete.CT_Telephone
),
"client_email": _safe_strip(entete.CT_EMail), "client_email": _safe_strip(entete.CT_EMail),
"contact": _safe_strip(entete.DO_Contact), "contact": _safe_strip(entete.DO_Contact),
"total_ht": ( "total_ht": (
float(entete.DO_TotalHT) float(entete.DO_TotalHT) if entete.DO_TotalHT else 0.0
if entete.DO_TotalHT
else 0.0
), ),
"total_ht_net": ( "total_ht_net": (
float(entete.DO_TotalHTNet) float(entete.DO_TotalHTNet) if entete.DO_TotalHTNet else 0.0
if entete.DO_TotalHTNet
else 0.0
), ),
"total_ttc": ( "total_ttc": (
float(entete.DO_TotalTTC) float(entete.DO_TotalTTC) if entete.DO_TotalTTC else 0.0
if entete.DO_TotalTTC
else 0.0
), ),
"net_a_payer": ( "net_a_payer": (
float(entete.DO_NetAPayer) float(entete.DO_NetAPayer) if entete.DO_NetAPayer else 0.0
if entete.DO_NetAPayer
else 0.0
), ),
"montant_regle": ( "montant_regle": (
float(entete.DO_MontantRegle) float(entete.DO_MontantRegle)
@ -535,36 +494,22 @@ def _lister_documents_avec_lignes_sql(
else 0.0 else 0.0
), ),
"reliquat": ( "reliquat": (
float(entete.DO_Reliquat) float(entete.DO_Reliquat) if entete.DO_Reliquat else 0.0
if entete.DO_Reliquat
else 0.0
), ),
"taux_escompte": ( "taux_escompte": (
float(entete.DO_TxEscompte) float(entete.DO_TxEscompte) if entete.DO_TxEscompte else 0.0
if entete.DO_TxEscompte
else 0.0
), ),
"escompte": ( "escompte": (
float(entete.DO_Escompte) float(entete.DO_Escompte) if entete.DO_Escompte else 0.0
if entete.DO_Escompte
else 0.0
),
"taxe1": (
float(entete.DO_Taxe1) if entete.DO_Taxe1 else 0.0
),
"taxe2": (
float(entete.DO_Taxe2) if entete.DO_Taxe2 else 0.0
),
"taxe3": (
float(entete.DO_Taxe3) if entete.DO_Taxe3 else 0.0
), ),
"taxe1": (float(entete.DO_Taxe1) if entete.DO_Taxe1 else 0.0),
"taxe2": (float(entete.DO_Taxe2) if entete.DO_Taxe2 else 0.0),
"taxe3": (float(entete.DO_Taxe3) if entete.DO_Taxe3 else 0.0),
"code_taxe1": _safe_strip(entete.DO_CodeTaxe1), "code_taxe1": _safe_strip(entete.DO_CodeTaxe1),
"code_taxe2": _safe_strip(entete.DO_CodeTaxe2), "code_taxe2": _safe_strip(entete.DO_CodeTaxe2),
"code_taxe3": _safe_strip(entete.DO_CodeTaxe3), "code_taxe3": _safe_strip(entete.DO_CodeTaxe3),
"statut": ( "statut": (
int(entete.DO_Statut) int(entete.DO_Statut) if entete.DO_Statut is not None else 0
if entete.DO_Statut is not None
else 0
), ),
"statut_estatut": ( "statut_estatut": (
int(entete.DO_EStatut) int(entete.DO_EStatut)
@ -572,14 +517,10 @@ def _lister_documents_avec_lignes_sql(
else 0 else 0
), ),
"imprime": ( "imprime": (
int(entete.DO_Imprim) int(entete.DO_Imprim) if entete.DO_Imprim is not None else 0
if entete.DO_Imprim is not None
else 0
), ),
"valide": ( "valide": (
int(entete.DO_Valide) int(entete.DO_Valide) if entete.DO_Valide is not None else 0
if entete.DO_Valide is not None
else 0
), ),
"cloture": ( "cloture": (
int(entete.DO_Cloture) int(entete.DO_Cloture)
@ -605,9 +546,7 @@ def _lister_documents_avec_lignes_sql(
else 0 else 0
), ),
"valeur_frais": ( "valeur_frais": (
float(entete.DO_ValFrais) float(entete.DO_ValFrais) if entete.DO_ValFrais else 0.0
if entete.DO_ValFrais
else 0.0
), ),
"type_franco": ( "type_franco": (
int(entete.DO_TypeFranco) int(entete.DO_TypeFranco)
@ -615,16 +554,12 @@ def _lister_documents_avec_lignes_sql(
else 0 else 0
), ),
"valeur_franco": ( "valeur_franco": (
float(entete.DO_ValFranco) float(entete.DO_ValFranco) if entete.DO_ValFranco else 0.0
if entete.DO_ValFranco
else 0.0
), ),
"lignes": [], "lignes": [],
} }
logger.debug( logger.debug(f"[SQL LIST] {numero} : document de base créé")
f"[SQL LIST] {numero} : document de base créé"
)
except Exception as e: except Exception as e:
logger.error( logger.error(
@ -665,20 +600,17 @@ def _lister_documents_avec_lignes_sql(
taux_taxe1 = ( taux_taxe1 = (
float(ligne_row.DL_Taxe1) float(ligne_row.DL_Taxe1)
if hasattr(ligne_row, "DL_Taxe1") if hasattr(ligne_row, "DL_Taxe1") and ligne_row.DL_Taxe1
and ligne_row.DL_Taxe1
else 0.0 else 0.0
) )
taux_taxe2 = ( taux_taxe2 = (
float(ligne_row.DL_Taxe2) float(ligne_row.DL_Taxe2)
if hasattr(ligne_row, "DL_Taxe2") if hasattr(ligne_row, "DL_Taxe2") and ligne_row.DL_Taxe2
and ligne_row.DL_Taxe2
else 0.0 else 0.0
) )
taux_taxe3 = ( taux_taxe3 = (
float(ligne_row.DL_Taxe3) float(ligne_row.DL_Taxe3)
if hasattr(ligne_row, "DL_Taxe3") if hasattr(ligne_row, "DL_Taxe3") and ligne_row.DL_Taxe3
and ligne_row.DL_Taxe3
else 0.0 else 0.0
) )
@ -691,21 +623,13 @@ def _lister_documents_avec_lignes_sql(
ligne = { ligne = {
"numero_ligne": ( "numero_ligne": (
int(ligne_row.DL_Ligne) int(ligne_row.DL_Ligne) if ligne_row.DL_Ligne else 0
if ligne_row.DL_Ligne
else 0
), ),
"article_code": _safe_strip(ligne_row.AR_Ref), "article_code": _safe_strip(ligne_row.AR_Ref),
"designation": _safe_strip( "designation": _safe_strip(ligne_row.DL_Design),
ligne_row.DL_Design "designation_article": _safe_strip(ligne_row.AR_Design),
),
"designation_article": _safe_strip(
ligne_row.AR_Design
),
"quantite": ( "quantite": (
float(ligne_row.DL_Qte) float(ligne_row.DL_Qte) if ligne_row.DL_Qte else 0.0
if ligne_row.DL_Qte
else 0.0
), ),
"quantite_livree": ( "quantite_livree": (
float(ligne_row.DL_QteLiv) float(ligne_row.DL_QteLiv)
@ -785,18 +709,12 @@ def _lister_documents_avec_lignes_sql(
"total_taxes": montant_taxe1 "total_taxes": montant_taxe1
+ montant_taxe2 + montant_taxe2
+ montant_taxe3, + montant_taxe3,
"famille_article": _safe_strip( "famille_article": _safe_strip(ligne_row.FA_CodeFamille),
ligne_row.FA_CodeFamille
),
"gamme1": _safe_strip(ligne_row.AR_Gamme1), "gamme1": _safe_strip(ligne_row.AR_Gamme1),
"gamme2": _safe_strip(ligne_row.AR_Gamme2), "gamme2": _safe_strip(ligne_row.AR_Gamme2),
"code_barre": _safe_strip( "code_barre": _safe_strip(ligne_row.AR_CodeBarre),
ligne_row.AR_CodeBarre
),
"type_article": _safe_strip(ligne_row.AR_Type), "type_article": _safe_strip(ligne_row.AR_Type),
"nature_article": _safe_strip( "nature_article": _safe_strip(ligne_row.AR_Nature),
ligne_row.AR_Nature
),
"garantie": _safe_strip(ligne_row.AR_Garantie), "garantie": _safe_strip(ligne_row.AR_Garantie),
"cout_standard": ( "cout_standard": (
float(ligne_row.AR_CoutStd) float(ligne_row.AR_CoutStd)
@ -813,9 +731,7 @@ def _lister_documents_avec_lignes_sql(
if ligne_row.AR_PoidsBrut if ligne_row.AR_PoidsBrut
else 0.0 else 0.0
), ),
"unite_vente": _safe_strip( "unite_vente": _safe_strip(ligne_row.AR_UniteVen),
ligne_row.AR_UniteVen
),
"date_livraison_ligne": ( "date_livraison_ligne": (
str(ligne_row.DL_DateLivr) str(ligne_row.DL_DateLivr)
if hasattr(ligne_row, "DL_DateLivr") if hasattr(ligne_row, "DL_DateLivr")
@ -848,13 +764,13 @@ def _lister_documents_avec_lignes_sql(
doc["nb_lignes"] = len(doc["lignes"]) doc["nb_lignes"] = len(doc["lignes"])
doc["total_ht_calcule"] = sum( doc["total_ht_calcule"] = sum(
l.get("montant_ligne_ht", 0) for l in doc["lignes"] ligne.get("montant_ligne_ht", 0) for ligne in doc["lignes"]
) )
doc["total_ttc_calcule"] = sum( doc["total_ttc_calcule"] = sum(
l.get("montant_ligne_ttc", 0) for l in doc["lignes"] ligne.get("montant_ligne_ttc", 0) for ligne in doc["lignes"]
) )
doc["total_taxes_calcule"] = sum( doc["total_taxes_calcule"] = sum(
l.get("total_taxes", 0) for l in doc["lignes"] ligne.get("total_taxes", 0) for ligne in doc["lignes"]
) )
logger.debug( logger.debug(
@ -881,13 +797,11 @@ def _lister_documents_avec_lignes_sql(
) )
continue continue
logger.info(f"[SQL LIST] ═══════════════════════════") logger.info("[SQL LIST] ═══════════════════════════")
logger.info(f"[SQL LIST] STATISTIQUES FINALES:") logger.info("[SQL LIST] STATISTIQUES FINALES:")
logger.info(f"[SQL LIST] Total SQL: {stats['total']}") logger.info(f"[SQL LIST] Total SQL: {stats['total']}")
logger.info(f"[SQL LIST] Exclus préfixe: {stats['exclus_prefixe']}") logger.info(f"[SQL LIST] Exclus préfixe: {stats['exclus_prefixe']}")
logger.info( logger.info(f"[SQL LIST] Erreur construction: {stats['erreur_construction']}")
f"[SQL LIST] Erreur construction: {stats['erreur_construction']}"
)
logger.info(f"[SQL LIST] Erreur lignes: {stats['erreur_lignes']}") logger.info(f"[SQL LIST] Erreur lignes: {stats['erreur_lignes']}")
logger.info( logger.info(
f"[SQL LIST] Erreur transformations: {stats['erreur_transformations']}" f"[SQL LIST] Erreur transformations: {stats['erreur_transformations']}"
@ -895,7 +809,7 @@ def _lister_documents_avec_lignes_sql(
logger.info(f"[SQL LIST] Erreur liaisons: {stats['erreur_liaisons']}") logger.info(f"[SQL LIST] Erreur liaisons: {stats['erreur_liaisons']}")
logger.info(f"[SQL LIST] SUCCÈS: {stats['succes']}") logger.info(f"[SQL LIST] SUCCÈS: {stats['succes']}")
logger.info(f"[SQL LIST] Documents retournés: {len(documents)}") logger.info(f"[SQL LIST] Documents retournés: {len(documents)}")
logger.info(f"[SQL LIST] ═══════════════════════════") logger.info("[SQL LIST] ═══════════════════════════")
return documents return documents
@ -910,4 +824,4 @@ __all__ = [
"_rechercher_devis_par_numero", "_rechercher_devis_par_numero",
"_lire_document_sql", "_lire_document_sql",
"_lister_documents_avec_lignes_sql", "_lister_documents_avec_lignes_sql",
] ]

View file

@ -4,9 +4,10 @@ import logging
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _clean_str(value, max_len: int) -> str: def _clean_str(value, max_len: int) -> str:
"""Nettoie et tronque une chaîne""" """Nettoie et tronque une chaîne"""
if value is None or str(value).lower() in ('none', 'null', ''): if value is None or str(value).lower() in ("none", "null", ""):
return "" return ""
return str(value)[:max_len].strip() return str(value)[:max_len].strip()
@ -37,7 +38,7 @@ def _try_set_attribute(obj, attr_name, value, variants=None):
variants = [attr_name] variants = [attr_name]
else: else:
variants = [attr_name] + variants variants = [attr_name] + variants
for variant in variants: for variant in variants:
try: try:
if hasattr(obj, variant): if hasattr(obj, variant):
@ -45,7 +46,7 @@ def _try_set_attribute(obj, attr_name, value, variants=None):
return True return True
except Exception as e: except Exception as e:
logger.debug(f" {variant} échec: {str(e)[:50]}") logger.debug(f" {variant} échec: {str(e)[:50]}")
return False return False
@ -77,12 +78,13 @@ def _get_type_libelle(type_doc: int) -> str:
return f"Type {type_doc}" return f"Type {type_doc}"
def _convertir_type_pour_sql(type_doc: int) -> int: def _convertir_type_pour_sql(type_doc: int) -> int:
"""COM → SQL : 0, 10, 20, 30... → 0, 1, 2, 3...""" """COM → SQL : 0, 10, 20, 30... → 0, 1, 2, 3..."""
mapping = {0: 0, 10: 1, 20: 2, 30: 3, 40: 4, 50: 5, 60: 6} mapping = {0: 0, 10: 1, 20: 2, 30: 3, 40: 4, 50: 5, 60: 6}
return mapping.get(type_doc, type_doc) return mapping.get(type_doc, type_doc)
def _convertir_type_depuis_sql(type_sql: int) -> int: def _convertir_type_depuis_sql(type_sql: int) -> int:
"""SQL → COM : 0, 1, 2, 3... → 0, 10, 20, 30...""" """SQL → COM : 0, 1, 2, 3... → 0, 10, 20, 30..."""
mapping = {0: 0, 1: 10, 2: 20, 3: 30, 4: 40, 5: 50, 6: 60} mapping = {0: 0, 1: 10, 2: 20, 3: 30, 4: 40, 5: 50, 6: 60}
@ -96,16 +98,17 @@ def _normaliser_type_document(type_doc: int) -> int:
return type_doc return type_doc
mapping_normalisation = { mapping_normalisation = {
1: 10, # Commande 1: 10,
2: 20, # Préparation 2: 20,
3: 30, # BL 3: 30,
4: 40, # Retour 4: 40,
5: 50, # Avoir 5: 50,
6: 60, # Facture 6: 60,
} }
return mapping_normalisation.get(type_doc, type_doc) return mapping_normalisation.get(type_doc, type_doc)
def normaliser_date(valeur): def normaliser_date(valeur):
if isinstance(valeur, str): if isinstance(valeur, str):
try: try:
@ -121,7 +124,8 @@ def normaliser_date(valeur):
else: else:
return datetime.now() return datetime.now()
__all__ = [ __all__ = [
"_clean_str", "_clean_str",
"_safe_strip", "_safe_strip",
@ -131,5 +135,5 @@ __all__ = [
"_normaliser_type_document", "_normaliser_type_document",
"_convertir_type_depuis_sql", "_convertir_type_depuis_sql",
"_convertir_type_pour_sql", "_convertir_type_pour_sql",
"normaliser_date" "normaliser_date",
] ]

View file

@ -4,27 +4,32 @@ from utils.functions.functions import _safe_strip
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _contact_to_dict(contact, numero_client=None, contact_numero=None, n_contact=None) -> Dict:
def _contact_to_dict(
contact, numero_client=None, contact_numero=None, n_contact=None
) -> Dict:
try: try:
civilite_code = getattr(contact, "Civilite", None) civilite_code = getattr(contact, "Civilite", None)
civilite_map = {0: "M.", 1: "Mme", 2: "Mlle", 3: "Société"} civilite_map = {0: "M.", 1: "Mme", 2: "Mlle", 3: "Société"}
civilite = civilite_map.get(civilite_code) if civilite_code is not None else None civilite = (
civilite_map.get(civilite_code) if civilite_code is not None else None
)
telephone = None telephone = None
portable = None portable = None
telecopie = None telecopie = None
email = None email = None
if hasattr(contact, 'Telecom'): if hasattr(contact, "Telecom"):
try: try:
telecom = contact.Telecom telecom = contact.Telecom
telephone = _safe_strip(getattr(telecom, "Telephone", None)) telephone = _safe_strip(getattr(telecom, "Telephone", None))
portable = _safe_strip(getattr(telecom, "Portable", None)) portable = _safe_strip(getattr(telecom, "Portable", None))
telecopie = _safe_strip(getattr(telecom, "Telecopie", None)) telecopie = _safe_strip(getattr(telecom, "Telecopie", None))
email = _safe_strip(getattr(telecom, "EMail", None)) email = _safe_strip(getattr(telecom, "EMail", None))
except: except Exception:
pass pass
return { return {
"numero": numero_client, "numero": numero_client,
"contact_numero": contact_numero, "contact_numero": contact_numero,
@ -46,16 +51,19 @@ def _contact_to_dict(contact, numero_client=None, contact_numero=None, n_contact
logger.warning(f"Erreur conversion contact: {e}") logger.warning(f"Erreur conversion contact: {e}")
return {} return {}
def _row_to_contact_dict(row) -> Dict: def _row_to_contact_dict(row) -> Dict:
"""Convertit une ligne SQL en dictionnaire contact""" """Convertit une ligne SQL en dictionnaire contact"""
civilite_code = row.CT_Civilite civilite_code = row.CT_Civilite
civilite_map = {0: "M.", 1: "Mme", 2: "Mlle", 3: "Société"} civilite_map = {0: "M.", 1: "Mme", 2: "Mlle", 3: "Société"}
return { return {
"numero": _safe_strip(row.CT_Num), "numero": _safe_strip(row.CT_Num),
"contact_numero": row.CT_No, "contact_numero": row.CT_No,
"n_contact": row.N_Contact, "n_contact": row.N_Contact,
"civilite": civilite_map.get(civilite_code) if civilite_code is not None else None, "civilite": civilite_map.get(civilite_code)
if civilite_code is not None
else None,
"nom": _safe_strip(row.CT_Nom), "nom": _safe_strip(row.CT_Nom),
"prenom": _safe_strip(row.CT_Prenom), "prenom": _safe_strip(row.CT_Prenom),
"fonction": _safe_strip(row.CT_Fonction), "fonction": _safe_strip(row.CT_Fonction),
@ -69,6 +77,7 @@ def _row_to_contact_dict(row) -> Dict:
"skype": _safe_strip(row.CT_Skype), "skype": _safe_strip(row.CT_Skype),
} }
def _row_to_tiers_dict(row) -> Dict: def _row_to_tiers_dict(row) -> Dict:
"""Convertit une ligne SQL en dictionnaire tiers (factorisation DRY)""" """Convertit une ligne SQL en dictionnaire tiers (factorisation DRY)"""
return { return {
@ -81,7 +90,6 @@ def _row_to_tiers_dict(row) -> Dict:
"siret": _safe_strip(row.CT_Siret), "siret": _safe_strip(row.CT_Siret),
"tva_intra": _safe_strip(row.CT_Identifiant), "tva_intra": _safe_strip(row.CT_Identifiant),
"code_naf": _safe_strip(row.CT_Ape), "code_naf": _safe_strip(row.CT_Ape),
"contact": _safe_strip(row.CT_Contact), "contact": _safe_strip(row.CT_Contact),
"adresse": _safe_strip(row.CT_Adresse), "adresse": _safe_strip(row.CT_Adresse),
"complement": _safe_strip(row.CT_Complement), "complement": _safe_strip(row.CT_Complement),
@ -89,19 +97,16 @@ def _row_to_tiers_dict(row) -> Dict:
"ville": _safe_strip(row.CT_Ville), "ville": _safe_strip(row.CT_Ville),
"region": _safe_strip(row.CT_CodeRegion), "region": _safe_strip(row.CT_CodeRegion),
"pays": _safe_strip(row.CT_Pays), "pays": _safe_strip(row.CT_Pays),
"telephone": _safe_strip(row.CT_Telephone), "telephone": _safe_strip(row.CT_Telephone),
"telecopie": _safe_strip(row.CT_Telecopie), "telecopie": _safe_strip(row.CT_Telecopie),
"email": _safe_strip(row.CT_EMail), "email": _safe_strip(row.CT_EMail),
"site_web": _safe_strip(row.CT_Site), "site_web": _safe_strip(row.CT_Site),
"facebook": _safe_strip(row.CT_Facebook), "facebook": _safe_strip(row.CT_Facebook),
"linkedin": _safe_strip(row.CT_LinkedIn), "linkedin": _safe_strip(row.CT_LinkedIn),
"taux01": row.CT_Taux01, "taux01": row.CT_Taux01,
"taux02": row.CT_Taux02, "taux02": row.CT_Taux02,
"taux03": row.CT_Taux03, "taux03": row.CT_Taux03,
"taux04": row.CT_Taux04, "taux04": row.CT_Taux04,
"statistique01": _safe_strip(row.CT_Statistique01), "statistique01": _safe_strip(row.CT_Statistique01),
"statistique02": _safe_strip(row.CT_Statistique02), "statistique02": _safe_strip(row.CT_Statistique02),
"statistique03": _safe_strip(row.CT_Statistique03), "statistique03": _safe_strip(row.CT_Statistique03),
@ -112,12 +117,10 @@ def _row_to_tiers_dict(row) -> Dict:
"statistique08": _safe_strip(row.CT_Statistique08), "statistique08": _safe_strip(row.CT_Statistique08),
"statistique09": _safe_strip(row.CT_Statistique09), "statistique09": _safe_strip(row.CT_Statistique09),
"statistique10": _safe_strip(row.CT_Statistique10), "statistique10": _safe_strip(row.CT_Statistique10),
"encours_autorise": row.CT_Encours, "encours_autorise": row.CT_Encours,
"assurance_credit": row.CT_Assurance, "assurance_credit": row.CT_Assurance,
"langue": row.CT_Langue, "langue": row.CT_Langue,
"commercial_code": row.CO_No, "commercial_code": row.CO_No,
"lettrage_auto": (row.CT_Lettrage == 1), "lettrage_auto": (row.CT_Lettrage == 1),
"est_actif": (row.CT_Sommeil == 0), "est_actif": (row.CT_Sommeil == 0),
"type_facture": row.CT_Facture, "type_facture": row.CT_Facture,
@ -129,16 +132,12 @@ def _row_to_tiers_dict(row) -> Dict:
"exclure_relance": (row.CT_NotRappel == 1), "exclure_relance": (row.CT_NotRappel == 1),
"exclure_penalites": (row.CT_NotPenal == 1), "exclure_penalites": (row.CT_NotPenal == 1),
"bon_a_payer": row.CT_BonAPayer, "bon_a_payer": row.CT_BonAPayer,
"priorite_livraison": row.CT_PrioriteLivr, "priorite_livraison": row.CT_PrioriteLivr,
"livraison_partielle": row.CT_LivrPartielle, "livraison_partielle": row.CT_LivrPartielle,
"delai_transport": row.CT_DelaiTransport, "delai_transport": row.CT_DelaiTransport,
"delai_appro": row.CT_DelaiAppro, "delai_appro": row.CT_DelaiAppro,
"commentaire": _safe_strip(row.CT_Commentaire), "commentaire": _safe_strip(row.CT_Commentaire),
"section_analytique": _safe_strip(row.CA_Num), "section_analytique": _safe_strip(row.CA_Num),
"mode_reglement_code": row.MR_No, "mode_reglement_code": row.MR_No,
"surveillance_active": (row.CT_Surveillance == 1), "surveillance_active": (row.CT_Surveillance == 1),
"coface": _safe_strip(row.CT_Coface), "coface": _safe_strip(row.CT_Coface),
@ -149,14 +148,14 @@ def _row_to_tiers_dict(row) -> Dict:
"sv_objet_maj": _safe_strip(row.CT_SvObjetMaj), "sv_objet_maj": _safe_strip(row.CT_SvObjetMaj),
"sv_chiffre_affaires": row.CT_SvCA, "sv_chiffre_affaires": row.CT_SvCA,
"sv_resultat": row.CT_SvResultat, "sv_resultat": row.CT_SvResultat,
"compte_general": _safe_strip(row.CG_NumPrinc), "compte_general": _safe_strip(row.CG_NumPrinc),
"categorie_tarif": row.N_CatTarif, "categorie_tarif": row.N_CatTarif,
"categorie_compta": row.N_CatCompta, "categorie_compta": row.N_CatCompta,
} }
__all__ = [ __all__ = [
"_contact_to_dict", "_contact_to_dict",
"_row_to_contact_dict", "_row_to_contact_dict",
"_row_to_tiers_dict", "_row_to_tiers_dict",
] ]

View file

@ -3,7 +3,7 @@ from utils.functions.functions import (
_convertir_type_depuis_sql, _convertir_type_depuis_sql,
_convertir_type_pour_sql, _convertir_type_pour_sql,
_normaliser_type_document, _normaliser_type_document,
_get_type_libelle _get_type_libelle,
) )
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -23,7 +23,8 @@ def _verifier_devis_non_transforme(numero: str, doc, cursor):
statut_actuel = getattr(doc, "DO_Statut", 0) statut_actuel = getattr(doc, "DO_Statut", 0)
if statut_actuel == 5: if statut_actuel == 5:
raise ValueError(f" Devis {numero} déjà transformé (statut=5)") raise ValueError(f" Devis {numero} déjà transformé (statut=5)")
def verifier_si_deja_transforme_sql(numero_source, cursor, type_source): def verifier_si_deja_transforme_sql(numero_source, cursor, type_source):
"""Version corrigée avec normalisation des types""" """Version corrigée avec normalisation des types"""
logger.info( logger.info(
@ -57,7 +58,6 @@ def verifier_si_deja_transforme_sql(numero_source, cursor, type_source):
return {"deja_transforme": False, "documents_cibles": []} return {"deja_transforme": False, "documents_cibles": []}
try: try:
query = f""" query = f"""
SELECT DISTINCT SELECT DISTINCT
dc.DO_Piece, dc.DO_Piece,
@ -81,8 +81,8 @@ def verifier_si_deja_transforme_sql(numero_source, cursor, type_source):
doc = { doc = {
"numero": row.DO_Piece.strip() if row.DO_Piece else "", "numero": row.DO_Piece.strip() if row.DO_Piece else "",
"type": type_normalise, # ← TYPE NORMALISÉ "type": type_normalise,
"type_brut": type_brut, # Garder aussi le type original "type_brut": type_brut,
"type_libelle": _get_type_libelle(type_brut), "type_libelle": _get_type_libelle(type_brut),
"statut": int(row.DO_Statut) if row.DO_Statut else 0, "statut": int(row.DO_Statut) if row.DO_Statut else 0,
"nb_lignes": int(row.NbLignes) if row.NbLignes else 0, "nb_lignes": int(row.NbLignes) if row.NbLignes else 0,
@ -101,9 +101,7 @@ def verifier_si_deja_transforme_sql(numero_source, cursor, type_source):
f"[VERIF] Document {numero_source} a {len(documents_cibles)} transformation(s)" f"[VERIF] Document {numero_source} a {len(documents_cibles)} transformation(s)"
) )
else: else:
logger.info( logger.info(f"[VERIF] Document {numero_source} pas encore transformé")
f"[VERIF] Document {numero_source} pas encore transformé"
)
return { return {
"deja_transforme": deja_transforme, "deja_transforme": deja_transforme,
@ -114,21 +112,19 @@ def verifier_si_deja_transforme_sql(numero_source, cursor, type_source):
logger.error(f"[VERIF] Erreur vérification: {e}") logger.error(f"[VERIF] Erreur vérification: {e}")
return {"deja_transforme": False, "documents_cibles": []} return {"deja_transforme": False, "documents_cibles": []}
def peut_etre_transforme(cursor, numero_source, type_source, type_cible): def peut_etre_transforme(cursor, numero_source, type_source, type_cible):
"""Version corrigée avec normalisation""" """Version corrigée avec normalisation"""
type_source = _normaliser_type_document(type_source) type_source = _normaliser_type_document(type_source)
type_cible = _normaliser_type_document(type_cible) type_cible = _normaliser_type_document(type_cible)
logger.info( logger.info(
f"[VERIF_TRANSFO] {numero_source} " f"[VERIF_TRANSFO] {numero_source} (type {type_source}) → type {type_cible}"
f"(type {type_source}) → type {type_cible}"
) )
verif = verifier_si_deja_transforme_sql(cursor, numero_source, type_source) verif = verifier_si_deja_transforme_sql(cursor, numero_source, type_source)
docs_meme_type = [ docs_meme_type = [d for d in verif["documents_cibles"] if d["type"] == type_cible]
d for d in verif["documents_cibles"] if d["type"] == type_cible
]
if docs_meme_type: if docs_meme_type:
nums = [d["numero"] for d in docs_meme_type] nums = [d["numero"] for d in docs_meme_type]
@ -145,6 +141,7 @@ def peut_etre_transforme(cursor, numero_source, type_source, type_cible):
"documents_existants": [], "documents_existants": [],
} }
def lire_erreurs_sage(obj, nom_obj=""): def lire_erreurs_sage(obj, nom_obj=""):
erreurs = [] erreurs = []
try: try:
@ -154,7 +151,7 @@ def lire_erreurs_sage(obj, nom_obj=""):
nb_erreurs = 0 nb_erreurs = 0
try: try:
nb_erreurs = obj.Errors.Count nb_erreurs = obj.Errors.Count
except: except Exception:
return erreurs return erreurs
if nb_erreurs == 0: if nb_erreurs == 0:
@ -165,13 +162,13 @@ def lire_erreurs_sage(obj, nom_obj=""):
err = None err = None
try: try:
err = obj.Errors.Item(i) err = obj.Errors.Item(i)
except: except Exception:
try: try:
err = obj.Errors(i) err = obj.Errors(i)
except: except Exception:
try: try:
err = obj.Errors.Item(i - 1) err = obj.Errors.Item(i - 1)
except: except Exception:
pass pass
if err is not None: if err is not None:
@ -185,7 +182,7 @@ def lire_erreurs_sage(obj, nom_obj=""):
if val: if val:
description = str(val) description = str(val)
break break
except: except Exception:
pass pass
for attr in ["Field", "FieldName", "Champ", "Property"]: for attr in ["Field", "FieldName", "Champ", "Property"]:
@ -194,7 +191,7 @@ def lire_erreurs_sage(obj, nom_obj=""):
if val: if val:
field = str(val) field = str(val)
break break
except: except Exception:
pass pass
for attr in ["Number", "Code", "ErrorCode", "Numero"]: for attr in ["Number", "Code", "ErrorCode", "Numero"]:
@ -203,7 +200,7 @@ def lire_erreurs_sage(obj, nom_obj=""):
if val is not None: if val is not None:
number = str(val) number = str(val)
break break
except: except Exception:
pass pass
if description or field or number: if description or field or number:
@ -226,9 +223,10 @@ def lire_erreurs_sage(obj, nom_obj=""):
return erreurs return erreurs
__all__ = [ __all__ = [
"_verifier_devis_non_transforme", "_verifier_devis_non_transforme",
"verifier_si_deja_transforme_sql", "verifier_si_deja_transforme_sql",
"peut_etre_transforme", "peut_etre_transforme",
"lire_erreurs_sage" "lire_erreurs_sage",
] ]

View file

@ -3,6 +3,7 @@ import logging
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _cast_client(persist_obj): def _cast_client(persist_obj):
try: try:
obj = win32com.client.CastTo(persist_obj, "IBOClient3") obj = win32com.client.CastTo(persist_obj, "IBOClient3")
@ -11,7 +12,8 @@ def _cast_client(persist_obj):
except Exception as e: except Exception as e:
logger.debug(f" _cast_client échoue: {e}") logger.debug(f" _cast_client échoue: {e}")
return None return None
def _extraire_client(client_obj): def _extraire_client(client_obj):
try: try:
try: try:
@ -40,23 +42,23 @@ def _extraire_client(client_obj):
qualite_code = getattr(client_obj, "CT_Type", None) qualite_code = getattr(client_obj, "CT_Type", None)
qualite_map = { qualite_map = {
0: "CLI", # Client 0: "CLI",
1: "FOU", # Fournisseur 1: "FOU",
2: "CLIFOU", # Client + Fournisseur 2: "CLIFOU",
3: "SAL", # Salarié 3: "SAL",
4: "PRO", # Prospect 4: "PRO",
} }
data["qualite"] = qualite_map.get(qualite_code, "CLI") data["qualite"] = qualite_map.get(qualite_code, "CLI")
data["est_fournisseur"] = qualite_code in [1, 2] data["est_fournisseur"] = qualite_code in [1, 2]
except: except Exception:
data["qualite"] = "CLI" data["qualite"] = "CLI"
data["est_fournisseur"] = False data["est_fournisseur"] = False
try: try:
data["est_prospect"] = getattr(client_obj, "CT_Prospect", 0) == 1 data["est_prospect"] = getattr(client_obj, "CT_Prospect", 0) == 1
except: except Exception:
data["est_prospect"] = False data["est_prospect"] = False
if data["est_prospect"]: if data["est_prospect"]:
@ -72,7 +74,7 @@ def _extraire_client(client_obj):
sommeil = getattr(client_obj, "CT_Sommeil", 0) sommeil = getattr(client_obj, "CT_Sommeil", 0)
data["est_actif"] = sommeil == 0 data["est_actif"] = sommeil == 0
data["est_en_sommeil"] = sommeil == 1 data["est_en_sommeil"] = sommeil == 1
except: except Exception:
data["est_actif"] = True data["est_actif"] = True
data["est_en_sommeil"] = False data["est_en_sommeil"] = False
@ -81,24 +83,24 @@ def _extraire_client(client_obj):
data["forme_juridique"] = forme_juridique data["forme_juridique"] = forme_juridique
data["est_entreprise"] = bool(forme_juridique) data["est_entreprise"] = bool(forme_juridique)
data["est_particulier"] = not bool(forme_juridique) data["est_particulier"] = not bool(forme_juridique)
except: except Exception:
data["forme_juridique"] = "" data["forme_juridique"] = ""
data["est_entreprise"] = False data["est_entreprise"] = False
data["est_particulier"] = True data["est_particulier"] = True
try: try:
data["civilite"] = getattr(client_obj, "CT_Civilite", "").strip() data["civilite"] = getattr(client_obj, "CT_Civilite", "").strip()
except: except Exception:
data["civilite"] = "" data["civilite"] = ""
try: try:
data["nom"] = getattr(client_obj, "CT_Nom", "").strip() data["nom"] = getattr(client_obj, "CT_Nom", "").strip()
except: except Exception:
data["nom"] = "" data["nom"] = ""
try: try:
data["prenom"] = getattr(client_obj, "CT_Prenom", "").strip() data["prenom"] = getattr(client_obj, "CT_Prenom", "").strip()
except: except Exception:
data["prenom"] = "" data["prenom"] = ""
if data.get("nom") or data.get("prenom"): if data.get("nom") or data.get("prenom"):
@ -115,7 +117,7 @@ def _extraire_client(client_obj):
try: try:
data["contact"] = getattr(client_obj, "CT_Contact", "").strip() data["contact"] = getattr(client_obj, "CT_Contact", "").strip()
except: except Exception:
data["contact"] = "" data["contact"] = ""
try: try:
@ -123,36 +125,32 @@ def _extraire_client(client_obj):
if adresse_obj: if adresse_obj:
try: try:
data["adresse"] = getattr(adresse_obj, "Adresse", "").strip() data["adresse"] = getattr(adresse_obj, "Adresse", "").strip()
except: except Exception:
data["adresse"] = "" data["adresse"] = ""
try: try:
data["complement"] = getattr( data["complement"] = getattr(adresse_obj, "Complement", "").strip()
adresse_obj, "Complement", "" except Exception:
).strip()
except:
data["complement"] = "" data["complement"] = ""
try: try:
data["code_postal"] = getattr( data["code_postal"] = getattr(adresse_obj, "CodePostal", "").strip()
adresse_obj, "CodePostal", "" except Exception:
).strip()
except:
data["code_postal"] = "" data["code_postal"] = ""
try: try:
data["ville"] = getattr(adresse_obj, "Ville", "").strip() data["ville"] = getattr(adresse_obj, "Ville", "").strip()
except: except Exception:
data["ville"] = "" data["ville"] = ""
try: try:
data["region"] = getattr(adresse_obj, "Region", "").strip() data["region"] = getattr(adresse_obj, "Region", "").strip()
except: except Exception:
data["region"] = "" data["region"] = ""
try: try:
data["pays"] = getattr(adresse_obj, "Pays", "").strip() data["pays"] = getattr(adresse_obj, "Pays", "").strip()
except: except Exception:
data["pays"] = "" data["pays"] = ""
else: else:
data["adresse"] = "" data["adresse"] = ""
@ -175,22 +173,22 @@ def _extraire_client(client_obj):
if telecom: if telecom:
try: try:
data["telephone"] = getattr(telecom, "Telephone", "").strip() data["telephone"] = getattr(telecom, "Telephone", "").strip()
except: except Exception:
data["telephone"] = "" data["telephone"] = ""
try: try:
data["portable"] = getattr(telecom, "Portable", "").strip() data["portable"] = getattr(telecom, "Portable", "").strip()
except: except Exception:
data["portable"] = "" data["portable"] = ""
try: try:
data["telecopie"] = getattr(telecom, "Telecopie", "").strip() data["telecopie"] = getattr(telecom, "Telecopie", "").strip()
except: except Exception:
data["telecopie"] = "" data["telecopie"] = ""
try: try:
data["email"] = getattr(telecom, "EMail", "").strip() data["email"] = getattr(telecom, "EMail", "").strip()
except: except Exception:
data["email"] = "" data["email"] = ""
try: try:
@ -200,7 +198,7 @@ def _extraire_client(client_obj):
or getattr(telecom, "SiteWeb", "") or getattr(telecom, "SiteWeb", "")
) )
data["site_web"] = str(site).strip() if site else "" data["site_web"] = str(site).strip() if site else ""
except: except Exception:
data["site_web"] = "" data["site_web"] = ""
else: else:
data["telephone"] = "" data["telephone"] = ""
@ -218,17 +216,17 @@ def _extraire_client(client_obj):
try: try:
data["siret"] = getattr(client_obj, "CT_Siret", "").strip() data["siret"] = getattr(client_obj, "CT_Siret", "").strip()
except: except Exception:
data["siret"] = "" data["siret"] = ""
try: try:
data["siren"] = getattr(client_obj, "CT_Siren", "").strip() data["siren"] = getattr(client_obj, "CT_Siren", "").strip()
except: except Exception:
data["siren"] = "" data["siren"] = ""
try: try:
data["tva_intra"] = getattr(client_obj, "CT_Identifiant", "").strip() data["tva_intra"] = getattr(client_obj, "CT_Identifiant", "").strip()
except: except Exception:
data["tva_intra"] = "" data["tva_intra"] = ""
try: try:
@ -236,34 +234,34 @@ def _extraire_client(client_obj):
getattr(client_obj, "CT_CodeNAF", "").strip() getattr(client_obj, "CT_CodeNAF", "").strip()
or getattr(client_obj, "CT_APE", "").strip() or getattr(client_obj, "CT_APE", "").strip()
) )
except: except Exception:
data["code_naf"] = "" data["code_naf"] = ""
try: try:
data["secteur"] = getattr(client_obj, "CT_Secteur", "").strip() data["secteur"] = getattr(client_obj, "CT_Secteur", "").strip()
except: except Exception:
data["secteur"] = "" data["secteur"] = ""
try: try:
effectif = getattr(client_obj, "CT_Effectif", None) effectif = getattr(client_obj, "CT_Effectif", None)
data["effectif"] = int(effectif) if effectif is not None else None data["effectif"] = int(effectif) if effectif is not None else None
except: except Exception:
data["effectif"] = None data["effectif"] = None
try: try:
ca = getattr(client_obj, "CT_ChiffreAffaire", None) ca = getattr(client_obj, "CT_ChiffreAffaire", None)
data["ca_annuel"] = float(ca) if ca is not None else None data["ca_annuel"] = float(ca) if ca is not None else None
except: except Exception:
data["ca_annuel"] = None data["ca_annuel"] = None
try: try:
data["commercial_code"] = getattr(client_obj, "CO_No", "").strip() data["commercial_code"] = getattr(client_obj, "CO_No", "").strip()
except: except Exception:
try: try:
data["commercial_code"] = getattr( data["commercial_code"] = getattr(
client_obj, "CT_Commercial", "" client_obj, "CT_Commercial", ""
).strip() ).strip()
except: except Exception:
data["commercial_code"] = "" data["commercial_code"] = ""
if data.get("commercial_code"): if data.get("commercial_code"):
@ -276,48 +274,46 @@ def _extraire_client(client_obj):
).strip() ).strip()
else: else:
data["commercial_nom"] = "" data["commercial_nom"] = ""
except: except Exception:
data["commercial_nom"] = "" data["commercial_nom"] = ""
else: else:
data["commercial_nom"] = "" data["commercial_nom"] = ""
try: try:
data["categorie_tarifaire"] = getattr(client_obj, "N_CatTarif", None) data["categorie_tarifaire"] = getattr(client_obj, "N_CatTarif", None)
except: except Exception:
data["categorie_tarifaire"] = None data["categorie_tarifaire"] = None
try: try:
data["categorie_comptable"] = getattr(client_obj, "N_CatCompta", None) data["categorie_comptable"] = getattr(client_obj, "N_CatCompta", None)
except: except Exception:
data["categorie_comptable"] = None data["categorie_comptable"] = None
try: try:
data["encours_autorise"] = float(getattr(client_obj, "CT_Encours", 0.0)) data["encours_autorise"] = float(getattr(client_obj, "CT_Encours", 0.0))
except: except Exception:
data["encours_autorise"] = 0.0 data["encours_autorise"] = 0.0
try: try:
data["assurance_credit"] = float( data["assurance_credit"] = float(getattr(client_obj, "CT_Assurance", 0.0))
getattr(client_obj, "CT_Assurance", 0.0) except Exception:
)
except:
data["assurance_credit"] = 0.0 data["assurance_credit"] = 0.0
try: try:
data["compte_general"] = getattr(client_obj, "CG_Num", "").strip() data["compte_general"] = getattr(client_obj, "CG_Num", "").strip()
except: except Exception:
data["compte_general"] = "" data["compte_general"] = ""
try: try:
date_creation = getattr(client_obj, "CT_DateCreate", None) date_creation = getattr(client_obj, "CT_DateCreate", None)
data["date_creation"] = str(date_creation) if date_creation else "" data["date_creation"] = str(date_creation) if date_creation else ""
except: except Exception:
data["date_creation"] = "" data["date_creation"] = ""
try: try:
date_modif = getattr(client_obj, "CT_DateModif", None) date_modif = getattr(client_obj, "CT_DateModif", None)
data["date_modification"] = str(date_modif) if date_modif else "" data["date_modification"] = str(date_modif) if date_modif else ""
except: except Exception:
data["date_modification"] = "" data["date_modification"] = ""
return data return data
@ -325,8 +321,6 @@ def _extraire_client(client_obj):
except Exception as e: except Exception as e:
logger.error(f" ERREUR GLOBALE _extraire_client: {e}", exc_info=True) logger.error(f" ERREUR GLOBALE _extraire_client: {e}", exc_info=True)
return None return None
__all__ = [
"_extraire_client", __all__ = ["_extraire_client", "_cast_client"]
"_cast_client"
]

View file

@ -1,9 +1,9 @@
import win32com.client import win32com.client
import logging import logging
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _extraire_fournisseur_enrichi(fourn_obj): def _extraire_fournisseur_enrichi(fourn_obj):
try: try:
numero = getattr(fourn_obj, "CT_Num", "").strip() numero = getattr(fourn_obj, "CT_Num", "").strip()
@ -15,7 +15,7 @@ def _extraire_fournisseur_enrichi(fourn_obj):
data = { data = {
"numero": numero, "numero": numero,
"intitule": intitule, "intitule": intitule,
"type": 1, # Fournisseur "type": 1,
"est_fournisseur": True, "est_fournisseur": True,
} }
@ -23,7 +23,7 @@ def _extraire_fournisseur_enrichi(fourn_obj):
sommeil = getattr(fourn_obj, "CT_Sommeil", 0) sommeil = getattr(fourn_obj, "CT_Sommeil", 0)
data["est_actif"] = sommeil == 0 data["est_actif"] = sommeil == 0
data["en_sommeil"] = sommeil == 1 data["en_sommeil"] = sommeil == 1
except: except Exception:
data["est_actif"] = True data["est_actif"] = True
data["en_sommeil"] = False data["en_sommeil"] = False
@ -83,7 +83,7 @@ def _extraire_fournisseur_enrichi(fourn_obj):
or getattr(telecom_obj, "SiteWeb", "") or getattr(telecom_obj, "SiteWeb", "")
) )
data["site_web"] = str(site).strip() if site else "" data["site_web"] = str(site).strip() if site else ""
except: except Exception:
data["site_web"] = "" data["site_web"] = ""
else: else:
data["telephone"] = "" data["telephone"] = ""
@ -101,7 +101,7 @@ def _extraire_fournisseur_enrichi(fourn_obj):
try: try:
data["siret"] = getattr(fourn_obj, "CT_Siret", "").strip() data["siret"] = getattr(fourn_obj, "CT_Siret", "").strip()
except: except Exception:
data["siret"] = "" data["siret"] = ""
try: try:
@ -109,12 +109,12 @@ def _extraire_fournisseur_enrichi(fourn_obj):
data["siren"] = data["siret"][:9] data["siren"] = data["siret"][:9]
else: else:
data["siren"] = getattr(fourn_obj, "CT_Siren", "").strip() data["siren"] = getattr(fourn_obj, "CT_Siren", "").strip()
except: except Exception:
data["siren"] = "" data["siren"] = ""
try: try:
data["tva_intra"] = getattr(fourn_obj, "CT_Identifiant", "").strip() data["tva_intra"] = getattr(fourn_obj, "CT_Identifiant", "").strip()
except: except Exception:
data["tva_intra"] = "" data["tva_intra"] = ""
try: try:
@ -122,14 +122,14 @@ def _extraire_fournisseur_enrichi(fourn_obj):
getattr(fourn_obj, "CT_CodeNAF", "").strip() getattr(fourn_obj, "CT_CodeNAF", "").strip()
or getattr(fourn_obj, "CT_APE", "").strip() or getattr(fourn_obj, "CT_APE", "").strip()
) )
except: except Exception:
data["code_naf"] = "" data["code_naf"] = ""
try: try:
data["forme_juridique"] = getattr( data["forme_juridique"] = getattr(
fourn_obj, "CT_FormeJuridique", "" fourn_obj, "CT_FormeJuridique", ""
).strip() ).strip()
except: except Exception:
data["forme_juridique"] = "" data["forme_juridique"] = ""
try: try:
@ -137,7 +137,7 @@ def _extraire_fournisseur_enrichi(fourn_obj):
data["categorie_tarifaire"] = ( data["categorie_tarifaire"] = (
int(cat_tarif) if cat_tarif is not None else None int(cat_tarif) if cat_tarif is not None else None
) )
except: except Exception:
data["categorie_tarifaire"] = None data["categorie_tarifaire"] = None
try: try:
@ -145,7 +145,7 @@ def _extraire_fournisseur_enrichi(fourn_obj):
data["categorie_comptable"] = ( data["categorie_comptable"] = (
int(cat_compta) if cat_compta is not None else None int(cat_compta) if cat_compta is not None else None
) )
except: except Exception:
data["categorie_comptable"] = None data["categorie_comptable"] = None
try: try:
@ -162,11 +162,11 @@ def _extraire_fournisseur_enrichi(fourn_obj):
).strip() ).strip()
else: else:
data["conditions_reglement_libelle"] = "" data["conditions_reglement_libelle"] = ""
except: except Exception:
data["conditions_reglement_libelle"] = "" data["conditions_reglement_libelle"] = ""
else: else:
data["conditions_reglement_libelle"] = "" data["conditions_reglement_libelle"] = ""
except: except Exception:
data["conditions_reglement_code"] = "" data["conditions_reglement_code"] = ""
data["conditions_reglement_libelle"] = "" data["conditions_reglement_libelle"] = ""
@ -184,11 +184,11 @@ def _extraire_fournisseur_enrichi(fourn_obj):
).strip() ).strip()
else: else:
data["mode_reglement_libelle"] = "" data["mode_reglement_libelle"] = ""
except: except Exception:
data["mode_reglement_libelle"] = "" data["mode_reglement_libelle"] = ""
else: else:
data["mode_reglement_libelle"] = "" data["mode_reglement_libelle"] = ""
except: except Exception:
data["mode_reglement_code"] = "" data["mode_reglement_code"] = ""
data["mode_reglement_libelle"] = "" data["mode_reglement_libelle"] = ""
@ -199,53 +199,36 @@ def _extraire_fournisseur_enrichi(fourn_obj):
if factory_banque: if factory_banque:
index = 1 index = 1
while index <= 5: # Max 5 comptes bancaires while index <= 5:
try: try:
banque_persist = factory_banque.List(index) banque_persist = factory_banque.List(index)
if banque_persist is None: if banque_persist is None:
break break
banque = win32com.client.CastTo( banque = win32com.client.CastTo(banque_persist, "IBOBanque3")
banque_persist, "IBOBanque3"
)
banque.Read() banque.Read()
compte_bancaire = { compte_bancaire = {
"banque_nom": getattr( "banque_nom": getattr(banque, "BI_Intitule", "").strip(),
banque, "BI_Intitule", ""
).strip(),
"iban": getattr(banque, "RIB_Iban", "").strip(), "iban": getattr(banque, "RIB_Iban", "").strip(),
"bic": getattr(banque, "RIB_Bic", "").strip(), "bic": getattr(banque, "RIB_Bic", "").strip(),
"code_banque": getattr( "code_banque": getattr(banque, "RIB_Banque", "").strip(),
banque, "RIB_Banque", "" "code_guichet": getattr(banque, "RIB_Guichet", "").strip(),
).strip(), "numero_compte": getattr(banque, "RIB_Compte", "").strip(),
"code_guichet": getattr(
banque, "RIB_Guichet", ""
).strip(),
"numero_compte": getattr(
banque, "RIB_Compte", ""
).strip(),
"cle_rib": getattr(banque, "RIB_Cle", "").strip(), "cle_rib": getattr(banque, "RIB_Cle", "").strip(),
} }
if ( if compte_bancaire["iban"] or compte_bancaire["numero_compte"]:
compte_bancaire["iban"]
or compte_bancaire["numero_compte"]
):
data["coordonnees_bancaires"].append(compte_bancaire) data["coordonnees_bancaires"].append(compte_bancaire)
index += 1 index += 1
except: except Exception:
break break
except Exception as e: except Exception as e:
logger.debug( logger.debug(f"Erreur coordonnées bancaires fournisseur {numero}: {e}")
f"Erreur coordonnées bancaires fournisseur {numero}: {e}"
)
if data["coordonnees_bancaires"]: if data["coordonnees_bancaires"]:
data["iban_principal"] = data["coordonnees_bancaires"][0].get( data["iban_principal"] = data["coordonnees_bancaires"][0].get("iban", "")
"iban", ""
)
data["bic_principal"] = data["coordonnees_bancaires"][0].get("bic", "") data["bic_principal"] = data["coordonnees_bancaires"][0].get("bic", "")
else: else:
data["iban_principal"] = "" data["iban_principal"] = ""
@ -258,15 +241,13 @@ def _extraire_fournisseur_enrichi(fourn_obj):
if factory_contact: if factory_contact:
index = 1 index = 1
while index <= 20: # Max 20 contacts while index <= 20:
try: try:
contact_persist = factory_contact.List(index) contact_persist = factory_contact.List(index)
if contact_persist is None: if contact_persist is None:
break break
contact = win32com.client.CastTo( contact = win32com.client.CastTo(contact_persist, "IBOContact3")
contact_persist, "IBOContact3"
)
contact.Read() contact.Read()
contact_data = { contact_data = {
@ -274,14 +255,14 @@ def _extraire_fournisseur_enrichi(fourn_obj):
"prenom": getattr(contact, "CO_Prenom", "").strip(), "prenom": getattr(contact, "CO_Prenom", "").strip(),
"fonction": getattr(contact, "CO_Fonction", "").strip(), "fonction": getattr(contact, "CO_Fonction", "").strip(),
"service": getattr(contact, "CO_Service", "").strip(), "service": getattr(contact, "CO_Service", "").strip(),
"telephone": getattr( "telephone": getattr(contact, "CO_Telephone", "").strip(),
contact, "CO_Telephone", ""
).strip(),
"portable": getattr(contact, "CO_Portable", "").strip(), "portable": getattr(contact, "CO_Portable", "").strip(),
"email": getattr(contact, "CO_EMail", "").strip(), "email": getattr(contact, "CO_EMail", "").strip(),
} }
nom_complet = f"{contact_data['prenom']} {contact_data['nom']}".strip() nom_complet = (
f"{contact_data['prenom']} {contact_data['nom']}".strip()
)
if nom_complet: if nom_complet:
contact_data["nom_complet"] = nom_complet contact_data["nom_complet"] = nom_complet
else: else:
@ -291,7 +272,7 @@ def _extraire_fournisseur_enrichi(fourn_obj):
data["contacts"].append(contact_data) data["contacts"].append(contact_data)
index += 1 index += 1
except: except Exception:
break break
except Exception as e: except Exception as e:
logger.debug(f"Erreur contacts fournisseur {numero}: {e}") logger.debug(f"Erreur contacts fournisseur {numero}: {e}")
@ -305,29 +286,29 @@ def _extraire_fournisseur_enrichi(fourn_obj):
try: try:
data["encours_autorise"] = float(getattr(fourn_obj, "CT_Encours", 0.0)) data["encours_autorise"] = float(getattr(fourn_obj, "CT_Encours", 0.0))
except: except Exception:
data["encours_autorise"] = 0.0 data["encours_autorise"] = 0.0
try: try:
data["ca_annuel"] = float(getattr(fourn_obj, "CT_ChiffreAffaire", 0.0)) data["ca_annuel"] = float(getattr(fourn_obj, "CT_ChiffreAffaire", 0.0))
except: except Exception:
data["ca_annuel"] = 0.0 data["ca_annuel"] = 0.0
try: try:
data["compte_general"] = getattr(fourn_obj, "CG_Num", "").strip() data["compte_general"] = getattr(fourn_obj, "CG_Num", "").strip()
except: except Exception:
data["compte_general"] = "" data["compte_general"] = ""
try: try:
date_creation = getattr(fourn_obj, "CT_DateCreate", None) date_creation = getattr(fourn_obj, "CT_DateCreate", None)
data["date_creation"] = str(date_creation) if date_creation else "" data["date_creation"] = str(date_creation) if date_creation else ""
except: except Exception:
data["date_creation"] = "" data["date_creation"] = ""
try: try:
date_modif = getattr(fourn_obj, "CT_DateModif", None) date_modif = getattr(fourn_obj, "CT_DateModif", None)
data["date_modification"] = str(date_modif) if date_modif else "" data["date_modification"] = str(date_modif) if date_modif else ""
except: except Exception:
data["date_modification"] = "" data["date_modification"] = ""
return data return data
@ -378,6 +359,4 @@ def _extraire_fournisseur_enrichi(fourn_obj):
} }
__all__ = [ __all__ = ["_extraire_fournisseur_enrichi"]
"_extraire_fournisseur_enrichi"
]

View file

@ -50,6 +50,6 @@ def _build_tiers_select_query() -> str:
-- COMPTE GENERAL ET CATEGORIES (3) -- COMPTE GENERAL ET CATEGORIES (3)
CG_NumPrinc, N_CatTarif, N_CatCompta CG_NumPrinc, N_CatTarif, N_CatCompta
""" """
__all__ = [
"_build_tiers_select_query"
] __all__ = ["_build_tiers_select_query"]