from typing import Dict, Optional from utils.functions.items_to_dict import _row_to_contact_dict import logging from utils.functions.functions import _safe_strip logger = logging.getLogger(__name__) def _get_contacts(numero: str, conn) -> list: try: cursor = conn.cursor() query = """ SELECT CT_Num, CT_No, N_Contact, CT_Civilite, CT_Nom, CT_Prenom, CT_Fonction, N_Service, CT_Telephone, CT_TelPortable, CT_Telecopie, CT_EMail, CT_Facebook, CT_LinkedIn, CT_Skype FROM F_CONTACTT WHERE CT_Num = ? ORDER BY N_Contact, CT_Nom, CT_Prenom """ cursor.execute(query, [numero]) rows = cursor.fetchall() query_client = """ SELECT CT_Contact FROM F_COMPTET WHERE CT_Num = ? """ cursor.execute(query_client, [numero]) client_row = cursor.fetchone() nom_contact_defaut = None if client_row: nom_contact_defaut = _safe_strip(client_row.CT_Contact) contacts = [] for row in rows: contact = _row_to_contact_dict(row) if nom_contact_defaut: nom_complet = f"{contact.get('prenom', '')} {contact['nom']}".strip() contact["est_defaut"] = ( nom_complet == nom_contact_defaut or contact['nom'] == nom_contact_defaut ) else: contact["est_defaut"] = False contacts.append(contact) return contacts except Exception as e: logger.warning(f" Impossible de récupérer contacts pour {numero}: {e}") return [] def _chercher_contact_en_base( conn, numero_client: str, nom: str, prenom: Optional[str] = None ) -> Optional[Dict]: try: cursor = conn.cursor() if prenom: query = """ SELECT TOP 1 CT_No, N_Contact FROM F_CONTACTT WHERE CT_Num = ? AND LTRIM(RTRIM(CT_Nom)) = ? AND LTRIM(RTRIM(CT_Prenom)) = ? ORDER BY CT_No DESC """ cursor.execute(query, (numero_client.upper(), nom.strip(), prenom.strip())) else: query = """ SELECT TOP 1 CT_No, N_Contact FROM F_CONTACTT WHERE CT_Num = ? AND LTRIM(RTRIM(CT_Nom)) = ? AND (CT_Prenom IS NULL OR LTRIM(RTRIM(CT_Prenom)) = '') ORDER BY CT_No DESC """ cursor.execute(query, (numero_client.upper(), nom.strip())) row = cursor.fetchone() if row: return { "contact_numero": row.CT_No, "n_contact": row.N_Contact } return None except Exception as e: logger.warning(f"Erreur recherche contact en base: {e}") return None def _lire_contact_depuis_base( conn, numero_client: str, contact_no: int ) -> Optional[Dict]: try: cursor = conn.cursor() query = """ SELECT CT_Num, CT_No, N_Contact, CT_Civilite, CT_Nom, CT_Prenom, CT_Fonction, N_Service, CT_Telephone, CT_TelPortable, CT_Telecopie, CT_EMail, CT_Facebook, CT_LinkedIn, CT_Skype FROM F_CONTACTT WHERE CT_Num = ? AND CT_No = ? """ logger.info(f" Execution SQL: CT_Num='{numero_client.upper()}', CT_No={contact_no}") cursor.execute(query, (numero_client.upper(), contact_no)) row = cursor.fetchone() if not row: logger.warning(f" Aucune ligne retournée pour CT_Num={numero_client.upper()}, CT_No={contact_no}") return None logger.info(f" Ligne SQL trouvée: Nom={row.CT_Nom}, Prenom={row.CT_Prenom}") civilite_map = {0: "M.", 1: "Mme", 2: "Mlle", 3: "Société"} result = { "numero": _safe_strip(row.CT_Num), "contact_numero": row.CT_No, "n_contact": row.N_Contact, "civilite": civilite_map.get(row.CT_Civilite, 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), "est_defaut": False, } logger.info(f" Dict construit: numero={result['numero']}, nom={result['nom']}") return result except Exception as e: logger.error(f" Exception dans : {e}", exc_info=True) return None __all__ = [ "_get_contacts", "_chercher_contact_en_base", "_lire_contact_depuis_base" ]