Sage100-ws/utils/tiers/contacts/contacts.py

from typing import Dict, Optional
from utils.functions.items_to_dict import contact_to_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 = contact_to_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"
]