diff --git a/sage_connector.py b/sage_connector.py index 481bb30..335c77d 100644 --- a/sage_connector.py +++ b/sage_connector.py @@ -48,9 +48,10 @@ from utils.functions.functions import ( ) from utils.functions.items_to_dict import ( - _contact_to_dict, - _row_to_contact_dict, - _row_to_tiers_dict, + contacts_to_dict, + contact_to_dict, + tiers_to_dict, + society_to_dict ) from utils.functions.sage_utilities import ( @@ -83,7 +84,13 @@ from utils.functions.data.create_doc import ( modifier_document_vente, ) -from utils.functions.items_to_dict import row_to_collaborateur_dict +from utils.functions.items_to_dict import collaborators_to_dict + +from utils.functions.society.societe_data import ( + get_societe_row, + add_logo, + build_exercices, +) logger = logging.getLogger(__name__) @@ -222,7 +229,7 @@ class SageConnector: fournisseurs = [] for row in rows: - fournisseur = _row_to_tiers_dict(row) + fournisseur = tiers_to_dict(row) fournisseur["contacts"] = _get_contacts(row.CT_Num, conn) fournisseurs.append(fournisseur) @@ -254,7 +261,7 @@ class SageConnector: if not row: return None - fournisseur = _row_to_tiers_dict(row) + fournisseur = tiers_to_dict(row) fournisseur["contacts"] = _get_contacts(row.CT_Num, conn) logger.info( @@ -736,7 +743,7 @@ class SageConnector: clients = [] for row in rows: - client = _row_to_tiers_dict(row) + client = tiers_to_dict(row) client["contacts"] = _get_contacts(row.CT_Num, conn) clients.append(client) @@ -766,7 +773,7 @@ class SageConnector: if not row: return None - client = _row_to_tiers_dict(row) + client = tiers_to_dict(row) client["contacts"] = _get_contacts(row.CT_Num, conn) logger.info(f"✓ SQL: Client {code_client} avec {len(client)} champs") @@ -1151,7 +1158,7 @@ class SageConnector: if not row: return None - return _row_to_contact_dict(row) + return contact_to_dict(row) except Exception as e: logger.error(f"Erreur obtention contact: {e}") @@ -2190,9 +2197,9 @@ class SageConnector: contact_dict = None if not contact_dict: - logger.info(" Stratégie 2: Lecture objet COM (_contact_to_dict)") + logger.info(" Stratégie 2: Lecture objet COM (contacts_to_dict)") try: - contact_dict = _contact_to_dict( + contact_dict = contacts_to_dict( contact, numero_client=numero_client, contact_numero=contact_no, @@ -2200,12 +2207,12 @@ class SageConnector: ) if contact_dict: logger.info( - f" _contact_to_dict réussi: {len(contact_dict)} champs" + f" contacts_to_dict réussi: {len(contact_dict)} champs" ) else: - logger.warning(" _contact_to_dict() retourne None/vide") + logger.warning(" contacts_to_dict() retourne None/vide") except Exception as e: - logger.error(f" Erreur _contact_to_dict: {e}", exc_info=True) + logger.error(f" Erreur contacts_to_dict: {e}", exc_info=True) contact_dict = None if not contact_dict: @@ -2748,16 +2755,16 @@ class SageConnector: if not contact_dict: try: - contact_dict = _contact_to_dict( + contact_dict = contacts_to_dict( contact, numero_client=numero, contact_numero=contact_numero, n_contact=None, ) if contact_dict: - logger.info(" _contact_to_dict reussi") + logger.info(" contacts_to_dict reussi") except Exception as e: - logger.warning(f" _contact_to_dict echoue: {e}") + logger.warning(f" contacts_to_dict echoue: {e}") if not contact_dict: logger.info(" Construction manuelle du retour") @@ -7274,7 +7281,7 @@ class SageConnector: tiers_list = [] for row in rows: - tiers = _row_to_tiers_dict(row) + tiers = tiers_to_dict(row) tiers["contacts"] = _get_contacts(row.CT_Num, conn) tiers_list.append(tiers) @@ -7306,7 +7313,7 @@ class SageConnector: if not row: return None - tiers = _row_to_tiers_dict(row) + tiers = tiers_to_dict(row) tiers["contacts"] = _get_contacts(row.CT_Num, conn) logger.info(f"✓ SQL: Tiers {code} lu avec succès") @@ -7350,7 +7357,7 @@ class SageConnector: rows = cursor.fetchall() # ⚠️⚠️⚠️ VÉRIFIE CETTE LIGNE ⚠️⚠️⚠️ - collaborateurs = [row_to_collaborateur_dict(row) for row in rows] + collaborateurs = [collaborators_to_dict(row) for row in rows] logger.info(f"✓ SQL: {len(collaborateurs)} collaborateurs") return collaborateurs @@ -7385,7 +7392,7 @@ class SageConnector: return None # ⚠️ UTILISER LA FONCTION DE CLASSE EXISTANTE - collaborateur = row_to_collaborateur_dict(row) + collaborateur = collaborators_to_dict(row) logger.info( f"✓ SQL: Collaborateur {numero} avec {len(collaborateur)} champs" @@ -7798,117 +7805,20 @@ class SageConnector: with self._get_sql_connection() as conn: cursor = conn.cursor() - # Requête pour récupérer toutes les colonnes importantes - query = """ - SELECT - D_RaisonSoc, D_NumDoss, D_Siret, D_Ape, D_Identifiant, - D_Adresse, D_Complement, D_CodePostal, D_Ville, - D_CodeRegion, D_Pays, - D_Telephone, D_Telecopie, D_EMail, D_EMailSoc, D_Site, - D_Capital, D_FormeJuridique, - D_DebutExo01, D_FinExo01, - D_DebutExo02, D_FinExo02, - D_DebutExo03, D_FinExo03, - D_DebutExo04, D_FinExo04, - D_DebutExo05, D_FinExo05, - N_DeviseCompte, N_DeviseEquival, - D_LgCg, D_LgAn, - D_RegimeFEC, - BM_Intitule, - cbMarq - FROM P_DOSSIER - """ - - cursor.execute(query) - row = cursor.fetchone() - + row = get_societe_row(cursor) if not row: logger.warning("Aucune donnée dans P_DOSSIER") return None - # Mapping des données - societe = { - # Identification - "raison_sociale": row.D_RaisonSoc.strip() - if row.D_RaisonSoc - else "", - "numero_dossier": row.D_NumDoss.strip() if row.D_NumDoss else "", - "siret": row.D_Siret.strip() if row.D_Siret else "", - "code_ape": row.D_Ape.strip() if row.D_Ape else "", - "numero_tva": row.D_Identifiant.strip() - if row.D_Identifiant - else "", - # Adresse - "adresse": row.D_Adresse.strip() if row.D_Adresse else "", - "complement_adresse": row.D_Complement.strip() - if row.D_Complement - else "", - "code_postal": row.D_CodePostal.strip() if row.D_CodePostal else "", - "ville": row.D_Ville.strip() if row.D_Ville else "", - "code_region": row.D_CodeRegion.strip() if row.D_CodeRegion else "", - "pays": row.D_Pays.strip() if row.D_Pays else "", - # Contacts - "telephone": row.D_Telephone.strip() if row.D_Telephone else "", - "telecopie": row.D_Telecopie.strip() if row.D_Telecopie else "", - "email": row.D_EMail.strip() if row.D_EMail else "", - "email_societe": row.D_EMailSoc.strip() if row.D_EMailSoc else "", - "site_web": row.D_Site.strip() if row.D_Site else "", - # Informations juridiques - "capital": float(row.D_Capital) if row.D_Capital else 0.0, - "forme_juridique": row.D_FormeJuridique.strip() - if row.D_FormeJuridique - else "", - # Exercices comptables (filtrer les exercices vides) - "exercices": [], - } - - # Ajouter uniquement les exercices valides (année > 1753) - exercices_potentiels = [ - (1, row.D_DebutExo01, row.D_FinExo01), - (2, row.D_DebutExo02, row.D_FinExo02), - (3, row.D_DebutExo03, row.D_FinExo03), - (4, row.D_DebutExo04, row.D_FinExo04), - (5, row.D_DebutExo05, row.D_FinExo05), - ] - - for numero, debut, fin in exercices_potentiels: - if debut and debut.year > 1753: - societe["exercices"].append( - { - "numero": numero, - "debut": debut.isoformat(), - "fin": fin.isoformat() - if fin and fin.year > 1753 - else None, - } - ) - - # Compléter les champs manquants - societe.update( - { - # Configuration - "devise_compte": row.N_DeviseCompte - if row.N_DeviseCompte - else 0, - "devise_equivalent": row.N_DeviseEquival - if row.N_DeviseEquival - else 0, - "longueur_compte_general": row.D_LgCg if row.D_LgCg else 0, - "longueur_compte_analytique": row.D_LgAn if row.D_LgAn else 0, - "regime_fec": row.D_RegimeFEC if row.D_RegimeFEC else 0, - # Autres - "base_modele": row.BM_Intitule.strip() - if row.BM_Intitule - else "", - "marqueur": row.cbMarq if row.cbMarq else 0, - } - ) + societe = society_to_dict(row) + societe["exercices"] = build_exercices(row) + add_logo(societe) logger.info( - f"✓ SQL: Informations société '{societe['raison_sociale']}' lues" + f"✓ Informations société '{societe['raison_sociale']}' lues" ) return societe except Exception as e: - logger.error(f"✗ Erreur SQL lecture P_DOSSIER: {e}", exc_info=True) + logger.error(f"✗ Erreur lecture P_DOSSIER: {e}", exc_info=True) raise RuntimeError(f"Erreur lecture informations société: {str(e)}") diff --git a/utils/functions/data/create_doc.py b/utils/functions/data/create_doc.py index e76072d..87e83e7 100644 --- a/utils/functions/data/create_doc.py +++ b/utils/functions/data/create_doc.py @@ -604,7 +604,7 @@ def modifier_document_vente( # 🔥 CONFIGURATION SPÉCIFIQUE FACTURE (avant lignes) if type_document == TypeDocumentVente.FACTURE: - _configurer_facture(doc) + _configurer_facture(self, doc) nouvelles_lignes = doc_data["lignes"] nb_nouvelles = len(nouvelles_lignes) diff --git a/utils/functions/items_to_dict.py b/utils/functions/items_to_dict.py index af14e56..0ff6ddf 100644 --- a/utils/functions/items_to_dict.py +++ b/utils/functions/items_to_dict.py @@ -5,7 +5,7 @@ from utils.functions.functions import _safe_strip logger = logging.getLogger(__name__) -def _contact_to_dict( +def contacts_to_dict( contact, numero_client=None, contact_numero=None, n_contact=None ) -> Dict: try: @@ -52,7 +52,7 @@ def _contact_to_dict( return {} -def _row_to_contact_dict(row) -> Dict: +def contact_to_dict(row) -> Dict: """Convertit une ligne SQL en dictionnaire contact""" civilite_code = row.CT_Civilite civilite_map = {0: "M.", 1: "Mme", 2: "Mlle", 3: "Société"} @@ -78,7 +78,7 @@ def _row_to_contact_dict(row) -> Dict: } -def _row_to_collaborateur_dict(row) -> Optional[dict]: +def _collaborators_to_dict(row) -> Optional[dict]: """Convertit une ligne SQL en dictionnaire collaborateur""" # Vérifier si le collaborateur existe if not hasattr(row, "Collab_CO_No") or row.Collab_CO_No is None: @@ -123,11 +123,11 @@ def _row_to_collaborateur_dict(row) -> Optional[dict]: } -def row_to_collaborateur_dict(row): +def collaborators_to_dict(row): """Convertit une ligne SQL en dictionnaire collaborateur""" return { "numero": row.CO_No, - "nom": _safe_strip(row.CO_Nom), # ⚠️ Utiliser _safe_strip + "nom": _safe_strip(row.CO_Nom), "prenom": _safe_strip(row.CO_Prenom), "fonction": _safe_strip(row.CO_Fonction), "adresse": _safe_strip(row.CO_Adresse), @@ -154,7 +154,7 @@ def row_to_collaborateur_dict(row): } -def _row_to_tiers_dict(row) -> dict: +def tiers_to_dict(row) -> dict: """Convertit une ligne SQL en dictionnaire tiers""" tiers = { # IDENTIFICATION @@ -203,7 +203,7 @@ def _row_to_tiers_dict(row) -> dict: "assurance_credit": row.CT_Assurance, "langue": row.CT_Langue, "commercial_code": row.CO_No, - "commercial": _row_to_collaborateur_dict(row), + "commercial": _collaborators_to_dict(row), # FACTURATION "lettrage_auto": (row.CT_Lettrage == 1), "est_actif": (row.CT_Sommeil == 0), @@ -245,9 +245,44 @@ def _row_to_tiers_dict(row) -> dict: return tiers +def society_to_dict(row) -> dict: + if not row: + return None + + return { + "raison_sociale": _safe_strip(row.D_RaisonSoc), + "numero_dossier": _safe_strip(row.D_NumDoss), + "siret": _safe_strip(row.D_Siret), + "code_ape": _safe_strip(row.D_Ape), + "numero_tva": _safe_strip(row.D_Identifiant), + "adresse": _safe_strip(row.D_Adresse), + "complement_adresse": _safe_strip(row.D_Complement), + "code_postal": _safe_strip(row.D_CodePostal), + "ville": _safe_strip(row.D_Ville), + "code_region": _safe_strip(row.D_CodeRegion), + "pays": _safe_strip(row.D_Pays), + "telephone": _safe_strip(row.D_Telephone), + "telecopie": _safe_strip(row.D_Telecopie), + "email": _safe_strip(row.D_EMail), + "email_societe": _safe_strip(row.D_EMailSoc), + "site_web": _safe_strip(row.D_Site), + "capital": float(row.D_Capital) if row.D_Capital else 0.0, + "forme_juridique": _safe_strip(row.D_FormeJuridique), + "devise_compte": row.N_DeviseCompte or 0, + "devise_equivalent": row.N_DeviseEquival or 0, + "longueur_compte_general": row.D_LgCg or 0, + "longueur_compte_analytique": row.D_LgAn or 0, + "regime_fec": row.D_RegimeFEC or 0, + "base_modele": _safe_strip(row.BM_Intitule), + "marqueur": row.cbMarq or 0, + "_logo_path": _safe_strip(row.D_Logo), + } + + __all__ = [ - "_contact_to_dict", - "_row_to_contact_dict", - "_row_to_tiers_dict", - "row_to_collaborateur_dict" + "contacts_to_dict", + "contact_to_dict", + "tiers_to_dict", + "collaborators_to_dict", + "society_to_dict" ] diff --git a/utils/functions/society/societe_data.py b/utils/functions/society/societe_data.py new file mode 100644 index 0000000..cc0c5b8 --- /dev/null +++ b/utils/functions/society/societe_data.py @@ -0,0 +1,89 @@ +from pathlib import Path +import base64 +import logging + +logger = logging.getLogger(__name__) + + +def get_societe_row(cursor): + """Récupère la ligne P_DOSSIER""" + query = """ + SELECT + D_RaisonSoc, D_NumDoss, D_Siret, D_Ape, D_Identifiant, + D_Adresse, D_Complement, D_CodePostal, D_Ville, + D_CodeRegion, D_Pays, + D_Telephone, D_Telecopie, D_EMail, D_EMailSoc, D_Site, + D_Capital, D_FormeJuridique, + D_DebutExo01, D_FinExo01, + D_DebutExo02, D_FinExo02, + D_DebutExo03, D_FinExo03, + D_DebutExo04, D_FinExo04, + D_DebutExo05, D_FinExo05, + N_DeviseCompte, N_DeviseEquival, + D_LgCg, D_LgAn, + D_RegimeFEC, + BM_Intitule, + cbMarq, + D_Logo + FROM P_DOSSIER + """ + + cursor.execute(query) + return cursor.fetchone() + + +def build_exercices(row) -> list: + """Construit la liste des exercices""" + exercices_data = [ + (1, row.D_DebutExo01, row.D_FinExo01), + (2, row.D_DebutExo02, row.D_FinExo02), + (3, row.D_DebutExo03, row.D_FinExo03), + (4, row.D_DebutExo04, row.D_FinExo04), + (5, row.D_DebutExo05, row.D_FinExo05), + ] + + exercices = [] + for numero, debut, fin in exercices_data: + if debut and debut.year > 1753: + exercices.append( + { + "numero": numero, + "debut": debut.isoformat(), + "fin": fin.isoformat() if fin and fin.year > 1753 else None, + } + ) + + return exercices + + +def add_logo(societe_dict: dict) -> None: + """Ajoute le logo en base64 au dict""" + logo_path = societe_dict.pop("_logo_path", None) + + if logo_path and Path(logo_path).exists(): + try: + ext = Path(logo_path).suffix.lower() + content_type = { + ".png": "image/png", + ".jpg": "image/jpeg", + ".jpeg": "image/jpeg", + ".bmp": "image/bmp", + ".gif": "image/gif", + }.get(ext, "image/png") + + with open(logo_path, "rb") as f: + societe_dict["logo_base64"] = base64.b64encode(f.read()).decode("utf-8") + societe_dict["logo_content_type"] = content_type + return + except Exception as e: + logger.warning(f"Erreur conversion logo: {e}") + + societe_dict["logo_base64"] = None + societe_dict["logo_content_type"] = None + + +__all__ = [ + "get_societe_row", + "build_exercices", + "add_logo", +] diff --git a/utils/tiers/contacts/contacts.py b/utils/tiers/contacts/contacts.py index 412e1e9..9795bd6 100644 --- a/utils/tiers/contacts/contacts.py +++ b/utils/tiers/contacts/contacts.py @@ -1,5 +1,5 @@ from typing import Dict, Optional -from utils.functions.items_to_dict import _row_to_contact_dict +from utils.functions.items_to_dict import contact_to_dict import logging from utils.functions.functions import _safe_strip @@ -38,7 +38,7 @@ def _get_contacts(numero: str, conn) -> list: contacts = [] for row in rows: - contact = _row_to_contact_dict(row) + contact = contact_to_dict(row) if nom_contact_defaut: nom_complet = f"{contact.get('prenom', '')} {contact['nom']}".strip()