Generalized status permutation, reviewed some create and delocalized functions

main.py

@@ -260,16 +260,16 @@ def devis_list(
         raise HTTPException(500, str(e))
 
 
-@app.post("/sage/devis/statut", dependencies=[Depends(verify_token)])
+@app.post("/sage/document/statut", dependencies=[Depends(verify_token)])
-def changer_statut_devis_endpoint(numero: str, nouveau_statut: int):
+def changer_statut_document(numero: str, type_doc: int, nouveau_statut: int):
-    """Change le statut d'un devis"""
+    """Change le statut d'un document"""
     try:
         with sage._com_context(), sage._lock_com:
             factory = sage.cial.FactoryDocumentVente
-            persist = factory.ReadPiece(0, numero)
+            persist = factory.ReadPiece(type_doc, numero)
 
             if not persist:
-                raise HTTPException(404, f"Devis {numero} introuvable")
+                raise HTTPException(404, f"Document {numero} introuvable")
 
             doc = win32com.client.CastTo(persist, "IBODocumentVente3")
             doc.Read()
@@ -278,7 +278,7 @@ def changer_statut_devis_endpoint(numero: str, nouveau_statut: int):
             doc.DO_Statut = nouveau_statut
             doc.Write()
 
-            logger.info(f" Statut devis {numero}: {statut_actuel} → {nouveau_statut}")
+            logger.info(f" Statut document {numero}: {statut_actuel} → {nouveau_statut}")
 
             return {
                 "success": True,

utils/documents/devis/devis_check.py

@@ -28,7 +28,7 @@ def _rechercher_devis_dans_liste(numero_devis, factory_doc):
 
     return None
       
-def _recuperer_numero_devis(self, process, doc):
+def _recuperer_numero_devis(process, doc):
     """Récupère le numéro du devis créé via plusieurs méthodes."""
     numero_devis = None
 
@@ -55,66 +55,7 @@ def _recuperer_numero_devis(self, process, doc):
 
     return numero_devis
 
-def _relire_devis(self, numero_devis, devis_data, forcer_brouillon):
-    """Relit le devis créé et extrait les informations finales."""
-    factory_doc = self.cial.FactoryDocumentVente
-    persist_reread = factory_doc.ReadPiece(0, numero_devis)
-
-    if not persist_reread:
-        logger.debug("ReadPiece échoué, recherche dans List()...")
-        persist_reread = _rechercher_devis_dans_liste(
-            numero_devis, factory_doc
-        )
-
-    if persist_reread:
-        doc_final = win32com.client.CastTo(persist_reread, "IBODocumentVente3")
-        doc_final.Read()
-
-        total_ht = float(getattr(doc_final, "DO_TotalHT", 0.0))
-        total_ttc = float(getattr(doc_final, "DO_TotalTTC", 0.0))
-        statut_final = getattr(doc_final, "DO_Statut", 0)
-        reference_final = getattr(doc_final, "DO_Ref", "")
-
-        date_livraison_final = None
-
-        try:
-            date_livr = getattr(doc_final, "DO_DateLivr", None)
-            if date_livr:
-                date_livraison_final = date_livr.strftime("%Y-%m-%d")
-        except:
-            pass
-    else:
-        total_calcule = sum(
-            l.get("montant_ligne_ht", 0) for l in devis_data["lignes"]
-        )
-        total_ht = total_calcule
-        total_ttc = round(total_calcule * 1.20, 2)
-        statut_final = 0 if forcer_brouillon else 2
-        reference_final = devis_data.get("reference", "")
-        date_livraison_final = devis_data.get("date_livraison")
-
-    logger.info(f" Total HT: {total_ht}€")
-    logger.info(f" Total TTC: {total_ttc}€")
-    logger.info(f" Statut final: {statut_final}")
-    if reference_final:
-        logger.info(f" Référence: {reference_final}")
-    if date_livraison_final:
-        logger.info(f" Date livraison: {date_livraison_final}")
-
-    return {
-        "numero_devis": numero_devis,
-        "total_ht": total_ht,
-        "total_ttc": total_ttc,
-        "nb_lignes": len(devis_data["lignes"]),
-        "client_code": devis_data["client"]["code"],
-        "date_devis": str(devis_data.get("date_devis", "")),
-        "date_livraison": date_livraison_final,
-        "reference": reference_final,
-        "statut": statut_final,
-    }
-
-
 __all__ = [
     "_recuperer_numero_devis",
-    "_relire_devis"
+    "_rechercher_devis_dans_liste"
 ]

utils/functions/sage_utilities.py

@@ -114,7 +114,7 @@ def verifier_si_deja_transforme_sql(numero_source, cursor, type_source):
         logger.error(f"[VERIF] Erreur vérification: {e}")
         return {"deja_transforme": False, "documents_cibles": []}
 
-def peut_etre_transforme(numero_source, type_source, type_cible):
+def peut_etre_transforme(cursor, numero_source, type_source, type_cible):
     """Version corrigée avec normalisation"""
     type_source = _normaliser_type_document(type_source)
     type_cible = _normaliser_type_document(type_cible)
@@ -124,7 +124,7 @@ def peut_etre_transforme(numero_source, type_source, type_cible):
         f"(type {type_source}) → type {type_cible}"
     )
 
-    verif = verifier_si_deja_transforme_sql(numero_source, type_source)
+    verif = verifier_si_deja_transforme_sql(cursor, numero_source, type_source)
 
     docs_meme_type = [
         d for d in verif["documents_cibles"] if d["type"] == type_cible

utils/tiers/contacts/contacts.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Optional, Any
+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
@@ -57,6 +57,112 @@ def _get_contacts_client(numero: str, conn) -> list:
         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_client"
+    "_get_contacts_client",
+    "_chercher_contact_en_base",
+    "_lire_contact_depuis_base"
 ]