from typing import Dict, Optional import logging from utils.functions.functions import _safe_strip logger = logging.getLogger(__name__) def _contact_to_dict( contact, numero_client=None, contact_numero=None, n_contact=None ) -> Dict: try: civilite_code = getattr(contact, "Civilite", None) 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 ) telephone = None portable = None telecopie = None email = None if hasattr(contact, "Telecom"): try: telecom = contact.Telecom telephone = _safe_strip(getattr(telecom, "Telephone", None)) portable = _safe_strip(getattr(telecom, "Portable", None)) telecopie = _safe_strip(getattr(telecom, "Telecopie", None)) email = _safe_strip(getattr(telecom, "EMail", None)) except Exception: pass return { "numero": numero_client, "contact_numero": contact_numero, "n_contact": n_contact or contact_numero, "civilite": civilite, "nom": _safe_strip(getattr(contact, "Nom", None)), "prenom": _safe_strip(getattr(contact, "Prenom", None)), "fonction": _safe_strip(getattr(contact, "Fonction", None)), "service_code": getattr(contact, "ServiceContact", None), "telephone": telephone, "portable": portable, "telecopie": telecopie, "email": email, "facebook": _safe_strip(getattr(contact, "Facebook", None)), "linkedin": _safe_strip(getattr(contact, "LinkedIn", None)), "skype": _safe_strip(getattr(contact, "Skype", None)), } except Exception as e: logger.warning(f"Erreur conversion contact: {e}") return {} def _row_to_contact_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é"} return { "numero": _safe_strip(row.CT_Num), "contact_numero": row.CT_No, "n_contact": row.N_Contact, "civilite": civilite_map.get(civilite_code) if civilite_code is not None else None, "nom": _safe_strip(row.CT_Nom), "prenom": _safe_strip(row.CT_Prenom), "fonction": _safe_strip(row.CT_Fonction), "service_code": row.N_Service, "telephone": _safe_strip(row.CT_Telephone), "portable": _safe_strip(row.CT_TelPortable), "telecopie": _safe_strip(row.CT_Telecopie), "email": _safe_strip(row.CT_EMail), "facebook": _safe_strip(row.CT_Facebook), "linkedin": _safe_strip(row.CT_LinkedIn), "skype": _safe_strip(row.CT_Skype), } def _row_to_collaborateur_dict(row) -> Optional[dict]: """Convertit une ligne SQL en dictionnaire collaborateur""" # Vérifier si le collaborateur existe if not hasattr(row, "Collab_CO_No") or row.Collab_CO_No is None: return None return { "numero": row.Collab_CO_No, "nom": _safe_strip(row.Collab_CO_Nom), "prenom": _safe_strip(row.Collab_CO_Prenom), "fonction": _safe_strip(row.Collab_CO_Fonction), "adresse": _safe_strip(row.Collab_CO_Adresse), "complement": _safe_strip(row.Collab_CO_Complement), "code_postal": _safe_strip(row.Collab_CO_CodePostal), "ville": _safe_strip(row.Collab_CO_Ville), "region": _safe_strip(row.Collab_CO_CodeRegion), "pays": _safe_strip(row.Collab_CO_Pays), "service": _safe_strip(row.Collab_CO_Service), "est_vendeur": (row.Collab_CO_Vendeur == 1) if row.Collab_CO_Vendeur is not None else None, "est_caissier": (row.Collab_CO_Caissier == 1) if row.Collab_CO_Caissier is not None else None, "est_acheteur": (row.Collab_CO_Acheteur == 1) if row.Collab_CO_Acheteur is not None else None, "telephone": _safe_strip(row.Collab_CO_Telephone), "telecopie": _safe_strip(row.Collab_CO_Telecopie), "email": _safe_strip(row.Collab_CO_EMail), "tel_portable": _safe_strip(row.Collab_CO_TelPortable), "matricule": _safe_strip(row.Collab_CO_Matricule), "facebook": _safe_strip(row.Collab_CO_Facebook), "linkedin": _safe_strip(row.Collab_CO_LinkedIn), "skype": _safe_strip(row.Collab_CO_Skype), "est_actif": (row.Collab_CO_Sommeil == 0) if row.Collab_CO_Sommeil is not None else None, "est_chef_ventes": (row.Collab_CO_ChefVentes == 1) if row.Collab_CO_ChefVentes is not None else None, "chef_ventes_numero": row.Collab_CO_NoChefVentes, } def _row_to_tiers_dict(row) -> dict: """Convertit une ligne SQL en dictionnaire tiers""" tiers = { # IDENTIFICATION "numero": _safe_strip(row.CT_Num), "intitule": _safe_strip(row.CT_Intitule), "type_tiers": row.CT_Type, "qualite": _safe_strip(row.CT_Qualite), "classement": _safe_strip(row.CT_Classement), "raccourci": _safe_strip(row.CT_Raccourci), "siret": _safe_strip(row.CT_Siret), "tva_intra": _safe_strip(row.CT_Identifiant), "code_naf": _safe_strip(row.CT_Ape), # ADRESSE "contact": _safe_strip(row.CT_Contact), "adresse": _safe_strip(row.CT_Adresse), "complement": _safe_strip(row.CT_Complement), "code_postal": _safe_strip(row.CT_CodePostal), "ville": _safe_strip(row.CT_Ville), "region": _safe_strip(row.CT_CodeRegion), "pays": _safe_strip(row.CT_Pays), # TELECOM "telephone": _safe_strip(row.CT_Telephone), "telecopie": _safe_strip(row.CT_Telecopie), "email": _safe_strip(row.CT_EMail), "site_web": _safe_strip(row.CT_Site), "facebook": _safe_strip(row.CT_Facebook), "linkedin": _safe_strip(row.CT_LinkedIn), # TAUX "taux01": row.CT_Taux01, "taux02": row.CT_Taux02, "taux03": row.CT_Taux03, "taux04": row.CT_Taux04, # STATISTIQUES "statistique01": _safe_strip(row.CT_Statistique01), "statistique02": _safe_strip(row.CT_Statistique02), "statistique03": _safe_strip(row.CT_Statistique03), "statistique04": _safe_strip(row.CT_Statistique04), "statistique05": _safe_strip(row.CT_Statistique05), "statistique06": _safe_strip(row.CT_Statistique06), "statistique07": _safe_strip(row.CT_Statistique07), "statistique08": _safe_strip(row.CT_Statistique08), "statistique09": _safe_strip(row.CT_Statistique09), "statistique10": _safe_strip(row.CT_Statistique10), # COMMERCIAL "encours_autorise": row.CT_Encours, "assurance_credit": row.CT_Assurance, "langue": row.CT_Langue, "commercial_code": row.CO_No, "commercial": _row_to_collaborateur_dict(row), # FACTURATION "lettrage_auto": (row.CT_Lettrage == 1), "est_actif": (row.CT_Sommeil == 0), "type_facture": row.CT_Facture, "est_prospect": (row.CT_Prospect == 1), "bl_en_facture": row.CT_BLFact, "saut_page": row.CT_Saut, "validation_echeance": row.CT_ValidEch, "controle_encours": row.CT_ControlEnc, "exclure_relance": (row.CT_NotRappel == 1), "exclure_penalites": (row.CT_NotPenal == 1), "bon_a_payer": row.CT_BonAPayer, # LOGISTIQUE "priorite_livraison": row.CT_PrioriteLivr, "livraison_partielle": row.CT_LivrPartielle, "delai_transport": row.CT_DelaiTransport, "delai_appro": row.CT_DelaiAppro, # COMMENTAIRE "commentaire": _safe_strip(row.CT_Commentaire), # ANALYTIQUE "section_analytique": _safe_strip(row.CA_Num), # ORGANISATION / SURVEILLANCE "mode_reglement_code": row.MR_No, "surveillance_active": (row.CT_Surveillance == 1), "coface": _safe_strip(row.CT_Coface), "forme_juridique": _safe_strip(row.CT_SvFormeJuri), "effectif": _safe_strip(row.CT_SvEffectif), "sv_regularite": _safe_strip(row.CT_SvRegul), "sv_cotation": _safe_strip(row.CT_SvCotation), "sv_objet_maj": _safe_strip(row.CT_SvObjetMaj), "sv_chiffre_affaires": row.CT_SvCA, "sv_resultat": row.CT_SvResultat, # COMPTE GENERAL ET CATEGORIES "compte_general": _safe_strip(row.CG_NumPrinc), "categorie_tarif": row.N_CatTarif, "categorie_compta": row.N_CatCompta, } return tiers __all__ = [ "_contact_to_dict", "_row_to_contact_dict", "_row_to_tiers_dict", ]