Spaces:
Running
Running
Claude Opus 5.5
feat(ontology): module serieux (vue 360, fraude, coherence, terminal IA fiable)
c6c39e9 | """ | |
| KG DATA — coeur metier (pur, sans Streamlit) du module Ontology | |
| ================================================================ | |
| Tout ce que le module Ontology affiche ou transmet a l'agent IA part d'ici : | |
| - consolider_prets() : une seule ligne par pret (Prets_Master + derniere | |
| version Prets_Update), pour ne plus compter deux fois | |
| un pret modifie. | |
| - extraire_triplets() : triplets sujet/predicat/objet pilotes par la feuille | |
| Ontology, avec classes canoniques et identifiants | |
| normalises (telephone, email, piece d'identite...). | |
| - controles_coherence() : references orphelines, doublons d'ID, chaine des soldes | |
| de remboursement, statuts incoherents. | |
| - identifiants_partages() : meme telephone / email / piece d'identite chez des | |
| personnes differentes (fraude, prete-noms). | |
| - reseaux_garantie() : auto-garantie, garantie croisee, garant multi-emprunteurs. | |
| - rechercher_personnes() / fiche_personne() / fiche_texte() : vue 360 d'un client | |
| ou d'un garant. | |
| Les fonctions recoivent les feuilles deja lues ({nom_feuille: DataFrame}) : elles | |
| sont testables sans Google Sheets. | |
| """ | |
| import math | |
| import re | |
| import unicodedata | |
| import pandas as pd | |
| try: | |
| from rapidfuzz import fuzz as _fuzz | |
| except ImportError: # dependance declaree dans requirements.txt ; repli tolerant | |
| _fuzz = None | |
| # Feuilles des personnes : classe, colonne ID. | |
| PERSONNES = {"Clients_KYC": ("Client", "ID_Client"), "Garants_KYC": ("Garant", "ID_Garant")} | |
| COL_NOM = "Nom_Complet" | |
| FEUILLE_PRETS, FEUILLE_VERSIONS, FEUILLE_REMB = "Prets_Master", "Prets_Update", "Remboursements" | |
| # Colonnes d'identifiants comparees entre personnes, et nombre de personnes distinctes | |
| # tolere par valeur (adresse : colocation / famille). | |
| TYPES_IDENTIFIANTS = ("Telephone", "Email", "ID_Officiel", "Numero_Fiscal", "Reseau_sociaux", "Adresse") | |
| SEUILS_PARTAGE = {"Adresse": 3} | |
| # Valeurs jamais transmises en clair a un service externe (LLM, Jev) ni a l'export RDF. | |
| COLONNES_SENSIBLES = ("ID_Officiel", "Numero_Fiscal", "Telephone", "Email", "Reference_Externe", "Reseau_sociaux") | |
| _VALEURS_BIDON = {"", "NA", "N/A", "NAN", "NONE", "NULL", "AUCUN", "AUCUNE", "NEANT", "0", "00", "-", "X", "XX", "XXX"} | |
| GRAVITES = ("Erreur", "Avertissement", "Info") | |
| COLONNES_CONTROLES = ["Gravité", "Contrôle", "Feuille", "Référence", "Détail"] | |
| TOLERANCE_XOF = 1.0 | |
| SEUIL_GARANT_MULTI = 3 | |
| # ============================================================================ | |
| # NORMALISATION | |
| # ============================================================================ | |
| def vide(valeur) -> bool: | |
| if valeur is None: | |
| return True | |
| if isinstance(valeur, float) and math.isnan(valeur): | |
| return True | |
| return str(valeur).strip().lower() in ("", "nan", "none", "null", "n/a", "nat") | |
| def sans_accents(texte) -> str: | |
| return "".join(c for c in unicodedata.normalize("NFKD", str(texte)) if not unicodedata.combining(c)) | |
| def cle_texte(texte) -> str: | |
| """Cle de comparaison : minuscules, sans accents ni ponctuation, espaces simples.""" | |
| if vide(texte): | |
| return "" | |
| return re.sub(r"[^a-z0-9]+", " ", sans_accents(texte).lower()).strip() | |
| def nombre(valeur) -> float: | |
| """Montant lu depuis Sheets (deja numerise ou texte '12 990', '1,5', '12990 XOF'). NaN si illisible.""" | |
| if vide(valeur): | |
| return float("nan") | |
| if isinstance(valeur, (int, float)): | |
| return float(valeur) | |
| texte = str(valeur).replace("XOF", "").replace("FCFA", "").replace("\xa0", "").replace(" ", "").replace(" ", "") | |
| if "," in texte and "." in texte: | |
| texte = texte.replace(",", "") | |
| else: | |
| texte = texte.replace(",", ".") | |
| try: | |
| return float(texte) | |
| except ValueError: | |
| return float("nan") | |
| def id_entite(valeur) -> str: | |
| return "" if vide(valeur) else str(valeur).strip().upper() | |
| def normaliser_identifiant(type_id, valeur) -> str: | |
| """Forme comparable d'un identifiant ; '' si absent ou valeur de remplissage.""" | |
| if vide(valeur): | |
| return "" | |
| brut = str(valeur).strip() | |
| if re.fullmatch(r"\d+\.0", brut): # entier numerise par Sheets | |
| brut = brut[:-2] | |
| if type_id == "Telephone": | |
| chiffres = re.sub(r"\D", "", brut) | |
| if len(chiffres) < 8: | |
| return "" | |
| norme = chiffres[-9:] | |
| elif type_id == "Email": | |
| norme = brut.lower() | |
| if "@" not in norme: | |
| return "" | |
| elif type_id in ("ID_Officiel", "Numero_Fiscal", "Reference_Externe"): | |
| norme = re.sub(r"[^0-9A-Za-z]", "", brut).upper() | |
| else: | |
| norme = cle_texte(brut) | |
| return "" if norme.upper() in _VALEURS_BIDON or len(set(norme)) <= 1 else norme | |
| def masquer(type_id, valeur) -> str: | |
| """Valeur affichable/transmissible sans exposer l'identifiant complet.""" | |
| if vide(valeur): | |
| return "" | |
| brut = str(valeur).strip() | |
| if type_id == "Email": | |
| nom, _, domaine = brut.partition("@") | |
| return f"{nom[:1]}***@{domaine}" if domaine else "***" | |
| return "••••" + brut[-4:] if len(brut) > 4 else "••••" | |
| def proteger_proprietes(props: dict) -> dict: | |
| """Copie de `props` ou chaque colonne sensible est masquee.""" | |
| return {k: (masquer(k, v) if k in COLONNES_SENSIBLES else v) for k, v in props.items()} | |
| def texte_sur(valeur, longueur=300) -> str: | |
| """Texte destine a un rendu HTML/JS tiers (infobulles du graphe) : sans balise possible.""" | |
| return str(valeur).replace("<", "‹").replace(">", "›")[:longueur] | |
| def _controle(gravite, controle, feuille, reference, detail): | |
| return {"Gravité": gravite, "Contrôle": controle, "Feuille": feuille, "Référence": reference, "Détail": detail} | |
| def _df(feuilles, nom): | |
| df = feuilles.get(nom) | |
| return df if isinstance(df, pd.DataFrame) else pd.DataFrame() | |
| # ============================================================================ | |
| # PRETS : UNE LIGNE PAR PRET | |
| # ============================================================================ | |
| def consolider_prets(master: pd.DataFrame, versions: pd.DataFrame): | |
| """Une ligne par pret racine (ID_Pret de Prets_Master), portant les valeurs de sa | |
| derniere version Prets_Update. Ajoute ID_Pret_Courant, Nb_Versions. | |
| Retourne (DataFrame, info) ; info = {prets_consolides, versions_orphelines, | |
| montants_versions: {racine: [Montant_Total de chaque version]}}.""" | |
| info = {"prets_consolides": 0, "versions_orphelines": [], "montants_versions": {}} | |
| if master is None or master.empty or "ID_Pret" not in master.columns: | |
| return (master.copy() if isinstance(master, pd.DataFrame) else pd.DataFrame()), info | |
| par_racine = {} | |
| if versions is not None and not versions.empty and "ID_Pret" in versions.columns: | |
| v = versions.copy() | |
| v["_racine"] = v["ID_Pret"].map(id_entite) | |
| v["_ordre"] = range(len(v)) | |
| v["_version"] = pd.to_numeric(v["Version"], errors="coerce").fillna(0) if "Version" in v.columns else 0 | |
| v = v[v["_racine"] != ""].sort_values(["_racine", "_version", "_ordre"]) | |
| par_racine = {racine: grp for racine, grp in v.groupby("_racine", sort=False)} | |
| lignes, racines = [], set() | |
| for rec in master.to_dict("records"): | |
| racine = id_entite(rec.get("ID_Pret")) | |
| if not racine: | |
| continue | |
| racines.add(racine) | |
| rec["ID_Pret"] = racine | |
| montants = [nombre(rec.get("Montant_Total"))] | |
| grp = par_racine.get(racine) | |
| if grp is not None: | |
| for col, val in grp.iloc[-1].to_dict().items(): | |
| if col.startswith("_") or col == "ID_Pret" or vide(val): | |
| continue | |
| if col in rec: | |
| rec[col] = val | |
| rec["ID_Pret_Courant"] = id_entite(grp.iloc[-1].get("ID_Pret_Updated")) or racine | |
| rec["Nb_Versions"] = 1 + len(grp) | |
| montants += [nombre(x) for x in grp.get("Montant_Total", pd.Series(dtype=object))] | |
| info["prets_consolides"] += 1 | |
| else: | |
| rec["ID_Pret_Courant"] = racine | |
| rec["Nb_Versions"] = 1 | |
| info["montants_versions"][racine] = [m for m in montants if not math.isnan(m)] | |
| lignes.append(rec) | |
| for racine, grp in par_racine.items(): | |
| if racine not in racines: | |
| info["versions_orphelines"].append(racine) | |
| return pd.DataFrame(lignes), info | |
| def preparer_feuilles(feuilles: dict): | |
| """Remplace Prets_Master par sa version consolidee. Retourne (feuilles, info).""" | |
| sortie = dict(feuilles) | |
| prets, info = consolider_prets(_df(feuilles, FEUILLE_PRETS), _df(feuilles, FEUILLE_VERSIONS)) | |
| if FEUILLE_PRETS in feuilles: | |
| sortie[FEUILLE_PRETS] = prets | |
| return sortie, info | |
| # ============================================================================ | |
| # TRIPLETS (PILOTES PAR LA FEUILLE ONTOLOGY) | |
| # ============================================================================ | |
| COLONNES_REGLES = ["Sheet", "SubjectCol", "SubjectClass", "Predicate", "ObjectType", "ObjectColOrConcept"] | |
| def normaliser_regles(regles: pd.DataFrame) -> pd.DataFrame: | |
| """Regles exploitables (colonnes nettoyees) ; les lignes incompletes sont ecartees.""" | |
| if regles is None or regles.empty: | |
| return pd.DataFrame(columns=COLONNES_REGLES) | |
| r = regles.copy() | |
| r.columns = [str(c).strip() for c in r.columns] | |
| for col in COLONNES_REGLES: | |
| if col not in r.columns: | |
| r[col] = "" | |
| r[col] = r[col].map(lambda x: "" if vide(x) else str(x).strip()) | |
| r["ObjectType"] = r["ObjectType"].str.lower() | |
| # Noms de classe et de predicat en identifiants ASCII : ce sont aussi des noms RDF | |
| # ("Prêt" et "Pret" designent la meme classe ; la premiere casse rencontree est retenue). | |
| r["SubjectClass"] = r["SubjectClass"].map(nom_rdf) | |
| r["Predicate"] = r["Predicate"].map(nom_rdf) | |
| r = r[(r["Sheet"] != "") & (r["SubjectCol"] != "") & (r["SubjectClass"] != "") & (r["Predicate"] != "")] | |
| canon = {} | |
| for cls in r["SubjectClass"]: | |
| canon.setdefault(cls.lower(), cls) | |
| r["SubjectClass"] = r["SubjectClass"].map(lambda c: canon[c.lower()]) | |
| return r.reset_index(drop=True) | |
| def nom_rdf(nom) -> str: | |
| """Identifiant RDF sur : sans accents, [A-Za-z0-9_] uniquement.""" | |
| return re.sub(r"[^A-Za-z0-9_]+", "_", sans_accents(str(nom)).strip()).strip("_") | |
| def classe_objet(colonne, classes) -> str: | |
| """Classe de l'objet d'une relation : 'ID_Pret' -> 'Pret' SEULEMENT si 'Pret' est une | |
| classe connue ; sinon la colonne elle-meme ('ID_Officiel' reste 'ID_Officiel').""" | |
| col = nom_rdf(colonne) | |
| canon = {c.lower(): c for c in classes} | |
| if col.upper().startswith("ID_") and col[3:].lower() in canon: | |
| return canon[col[3:].lower()] | |
| return canon.get(col.lower(), col) | |
| def extraire_triplets(regles: pd.DataFrame, feuilles: dict): | |
| """Triplets a partir des regles Ontology et des feuilles PREPAREES (preparer_feuilles). | |
| Retourne (DataFrame[subject, predicate, object, object_type, sheet, colonne, subject_props], | |
| info = {classes, regles_ignorees: [{regle, raison}]}).""" | |
| regles = normaliser_regles(regles) | |
| classes = list(dict.fromkeys(regles["SubjectClass"])) | |
| ignorees, triplets = [], [] | |
| consolide = FEUILLE_PRETS in feuilles and FEUILLE_VERSIONS in feuilles | |
| for feuille, rs in regles.groupby("Sheet", sort=False): | |
| libelles = [f"{feuille}.{r['SubjectCol']} -> {r['Predicate']}" for r in rs.to_dict("records")] | |
| if consolide and feuille == FEUILLE_VERSIONS: | |
| ignorees += [{"regle": l, "raison": "versions fusionnees dans Prets_Master (un noeud par pret)"} for l in libelles] | |
| continue | |
| df = _df(feuilles, feuille) | |
| if df.empty: | |
| ignorees += [{"regle": l, "raison": "feuille absente ou vide"} for l in libelles] | |
| continue | |
| cols = set(df.columns) | |
| valides = [] | |
| for r in rs.to_dict("records"): | |
| libelle = f"{feuille}.{r['SubjectCol']} -> {r['Predicate']}" | |
| if r["SubjectCol"] not in cols: | |
| ignorees.append({"regle": libelle, "raison": f"colonne sujet '{r['SubjectCol']}' absente"}) | |
| elif r["ObjectType"] not in ("relation", "data_property"): | |
| ignorees.append({"regle": libelle, "raison": f"ObjectType '{r['ObjectType']}' inconnu"}) | |
| elif r["ObjectColOrConcept"] not in cols: | |
| ignorees.append({"regle": libelle, "raison": f"colonne objet '{r['ObjectColOrConcept']}' absente"}) | |
| else: | |
| r["_classe_objet"] = classe_objet(r["ObjectColOrConcept"], classes) | |
| valides.append(r) | |
| if not valides: | |
| continue | |
| for row in df.to_dict("records"): | |
| props = {k: v for k, v in row.items() if not vide(v) and not str(k).startswith("_")} | |
| for r in valides: | |
| s_val = id_entite(row.get(r["SubjectCol"])) | |
| if not s_val: | |
| continue | |
| oval = row.get(r["ObjectColOrConcept"]) | |
| if vide(oval): | |
| continue | |
| base = {"subject": f"{r['SubjectClass']}:{s_val}", "predicate": r["Predicate"], | |
| "sheet": feuille, "colonne": r["ObjectColOrConcept"], "subject_props": props} | |
| if r["ObjectType"] == "relation": | |
| ocls = r["_classe_objet"] | |
| if ocls in classes: | |
| cible = id_entite(oval) | |
| else: | |
| cible = normaliser_identifiant(ocls, oval) | |
| if cible: | |
| triplets.append({**base, "object": f"{ocls}:{cible}", "object_type": "entity"}) | |
| else: | |
| triplets.append({**base, "object": str(oval).strip(), "object_type": "literal"}) | |
| colonnes = ["subject", "predicate", "object", "object_type", "sheet", "colonne", "subject_props"] | |
| return pd.DataFrame(triplets, columns=colonnes), {"classes": classes, "regles_ignorees": ignorees} | |
| # ============================================================================ | |
| # CONTROLES DE COHERENCE | |
| # ============================================================================ | |
| def _colonne_id_transaction(df): | |
| for col in df.columns: | |
| if str(col).upper().startswith("ID_TR"): | |
| return col | |
| return None | |
| def controles_coherence(feuilles: dict, info_prets: dict) -> pd.DataFrame: | |
| """Anomalies de donnees, triees par gravite. `feuilles` = sortie de preparer_feuilles.""" | |
| out = [] | |
| clients, garants = _df(feuilles, "Clients_KYC"), _df(feuilles, "Garants_KYC") | |
| prets, remb = _df(feuilles, FEUILLE_PRETS), _df(feuilles, FEUILLE_REMB) | |
| # 1. Doublons d'identifiants | |
| cibles = [("Clients_KYC", "ID_Client"), ("Garants_KYC", "ID_Garant"), (FEUILLE_PRETS, "ID_Pret")] | |
| col_trx = _colonne_id_transaction(remb) if not remb.empty else None | |
| if col_trx: | |
| cibles.append((FEUILLE_REMB, col_trx)) | |
| for feuille, col in cibles: | |
| df = _df(feuilles, feuille) | |
| if df.empty or col not in df.columns: | |
| continue | |
| ids = df[col].map(id_entite) | |
| for valeur, n in ids[ids != ""].value_counts().items(): | |
| if n > 1: | |
| out.append(_controle("Erreur", "Identifiant en double", feuille, valeur, f"{col} present {n} fois.")) | |
| ids_clients = set(clients["ID_Client"].map(id_entite)) if "ID_Client" in clients.columns else None | |
| ids_garants = set(garants["ID_Garant"].map(id_entite)) if "ID_Garant" in garants.columns else None | |
| # 2. References des prets | |
| prets_par_id = {} | |
| if not prets.empty and "ID_Pret" in prets.columns: | |
| for p in prets.to_dict("records"): | |
| pid = id_entite(p.get("ID_Pret")) | |
| prets_par_id[pid] = p | |
| cli, gar = id_entite(p.get("ID_Client")), id_entite(p.get("ID_Garant")) | |
| if ids_clients is not None and (not cli or cli not in ids_clients): | |
| out.append(_controle("Erreur", "Prêt sans client connu", FEUILLE_PRETS, pid, | |
| f"ID_Client '{cli or 'vide'}' introuvable dans Clients_KYC.")) | |
| if ids_garants is not None and gar and gar not in ids_garants: | |
| out.append(_controle("Erreur", "Garant introuvable", FEUILLE_PRETS, pid, | |
| f"ID_Garant '{gar}' introuvable dans Garants_KYC.")) | |
| statut = str(p.get("Statut", "")).strip().upper() | |
| if statut == "UPDATED": | |
| out.append(_controle("Erreur", "Version de prêt manquante", FEUILLE_PRETS, pid, | |
| "Statut UPDATED mais aucune version dans Prets_Update.")) | |
| for racine in info_prets.get("versions_orphelines", []): | |
| out.append(_controle("Erreur", "Version orpheline", FEUILLE_VERSIONS, racine, | |
| "Version de prêt sans ligne d'origine dans Prets_Master.")) | |
| # 3. Remboursements : references et chaine des soldes | |
| if not remb.empty and "ID_Pret" in remb.columns: | |
| r = remb.copy() | |
| r["_pret"] = r["ID_Pret"].map(id_entite) | |
| r["_ordre"] = range(len(r)) | |
| for pid, grp in r.groupby("_pret", sort=False): | |
| if not pid: | |
| continue | |
| pret = prets_par_id.get(pid) | |
| if prets_par_id and pret is None: | |
| out.append(_controle("Erreur", "Paiement sur prêt inconnu", FEUILLE_REMB, pid, | |
| f"{len(grp)} paiement(s) rattaché(s) à un prêt absent de Prets_Master.")) | |
| if pret is not None and "ID_Client" in grp.columns: | |
| autres = {id_entite(c) for c in grp["ID_Client"]} - {"", id_entite(pret.get("ID_Client"))} | |
| if autres: | |
| out.append(_controle("Avertissement", "Client du paiement différent du prêt", FEUILLE_REMB, pid, | |
| f"Paiements enregistrés au nom de {', '.join(sorted(autres))}.")) | |
| out += _controles_soldes(pid, grp.sort_values("_ordre"), pret, info_prets, col_trx) | |
| ordre = {g: i for i, g in enumerate(GRAVITES)} | |
| df = pd.DataFrame(out, columns=COLONNES_CONTROLES) | |
| if not df.empty: | |
| df = df.sort_values(["Gravité", "Contrôle", "Référence"], key=lambda s: s.map(ordre) if s.name == "Gravité" else s) | |
| return df.reset_index(drop=True) | |
| def _controles_soldes(pid, grp, pret, info_prets, col_trx): | |
| """Regles d'ecriture de repayments.py / AnalyseRepayment.calculer_soldes : | |
| Solde_Avant(1er) = Montant_Total ; Solde_Apres = max(0, Solde_Avant - (Principal + Interets)) ; | |
| Solde_Avant(n) = Solde_Apres(n-1). Les penalites ne reduisent pas le solde.""" | |
| out = [] | |
| requis = {"Solde_Avant", "Solde_Apres"} | |
| if not requis.issubset(grp.columns): | |
| return out | |
| modifie = pret is not None and int(nombre(pret.get("Nb_Versions", 1)) or 1) > 1 | |
| precedent = None | |
| for i, ligne in enumerate(grp.to_dict("records")): | |
| ref = f"{pid} / {id_entite(ligne.get(col_trx)) or 'ligne ' + str(i + 1)}" if col_trx else pid | |
| avant, apres = nombre(ligne.get("Solde_Avant")), nombre(ligne.get("Solde_Apres")) | |
| if math.isnan(avant) or math.isnan(apres): | |
| out.append(_controle("Avertissement", "Solde illisible", FEUILLE_REMB, ref, "Solde_Avant ou Solde_Apres vide.")) | |
| precedent = None | |
| continue | |
| dette = sum(0.0 if math.isnan(x) else x for x in (nombre(ligne.get("Montant_Principal")), nombre(ligne.get("Montant_Interets")))) | |
| if {"Montant_Principal", "Montant_Interets"}.issubset(grp.columns): | |
| attendu = max(0.0, avant - dette) | |
| if abs(attendu - apres) > TOLERANCE_XOF: | |
| out.append(_controle("Erreur", "Solde après incohérent", FEUILLE_REMB, ref, | |
| f"Solde_Avant {avant:,.0f} − (principal + intérêts) {dette:,.0f} = {attendu:,.0f}, " | |
| f"mais Solde_Apres = {apres:,.0f}.".replace(",", " "))) | |
| if i == 0 and pret is not None: | |
| montants = info_prets.get("montants_versions", {}).get(pid) or [nombre(pret.get("Montant_Total"))] | |
| montants = [m for m in montants if not math.isnan(m)] | |
| if montants and all(abs(avant - m) > TOLERANCE_XOF for m in montants): | |
| connus = " / ".join(f"{m:,.0f}" for m in montants).replace(",", " ") | |
| out.append(_controle("Erreur", "Solde de départ ≠ montant du prêt", FEUILLE_REMB, ref, | |
| f"1er Solde_Avant = {avant:,.0f} ; Montant_Total du prêt = {connus}.".replace(",", " "))) | |
| if precedent is not None and abs(avant - precedent) > TOLERANCE_XOF: | |
| out.append(_controle("Avertissement" if modifie else "Erreur", "Rupture de la chaîne des soldes", FEUILLE_REMB, ref, | |
| f"Solde_Avant {avant:,.0f} ≠ Solde_Apres précédent {precedent:,.0f}".replace(",", " ") | |
| + (" (prêt modifié entre-temps)." if modifie else "."))) | |
| precedent = apres | |
| if pret is not None and precedent is not None: | |
| statut = str(pret.get("Statut", "")).strip().upper() | |
| if statut == "TERMINE" and precedent > TOLERANCE_XOF: | |
| out.append(_controle("Erreur", "Prêt terminé avec solde restant", FEUILLE_PRETS, pid, | |
| f"Statut TERMINE mais dernier Solde_Apres = {precedent:,.0f} XOF.".replace(",", " "))) | |
| elif statut == "ACTIF" and precedent <= TOLERANCE_XOF: | |
| out.append(_controle("Avertissement", "Prêt soldé encore actif", FEUILLE_PRETS, pid, | |
| "Dernier Solde_Apres = 0 mais statut ACTIF.")) | |
| return out | |
| # ============================================================================ | |
| # PERSONNES, IDENTIFIANTS PARTAGES, RESEAUX DE GARANTIE | |
| # ============================================================================ | |
| def personnes(feuilles: dict) -> list: | |
| """Clients et garants : [{classe, id, nom, cle_nom, id_off, cle, ligne}] ; `cle` identifie | |
| l'etre humain (meme piece d'identite = meme personne, sinon meme nom, sinon l'ID).""" | |
| sortie = [] | |
| for feuille, (classe, col_id) in PERSONNES.items(): | |
| df = _df(feuilles, feuille) | |
| if df.empty or col_id not in df.columns: | |
| continue | |
| for ligne in df.to_dict("records"): | |
| pid = id_entite(ligne.get(col_id)) | |
| if not pid: | |
| continue | |
| nom = "" if vide(ligne.get(COL_NOM)) else str(ligne.get(COL_NOM)).strip() | |
| id_off = normaliser_identifiant("ID_Officiel", ligne.get("ID_Officiel")) | |
| cle = f"OFF:{id_off}" if id_off else (f"NOM:{cle_texte(nom)}" if cle_texte(nom) else f"{classe}:{pid}") | |
| sortie.append({"classe": classe, "id": pid, "nom": nom, "cle_nom": cle_texte(nom), | |
| "id_off": id_off, "cle": cle, "ligne": ligne}) | |
| return sortie | |
| def _libelle(p): | |
| return f"{p['classe']} {p['id']}" + (f" ({p['nom']})" if p["nom"] else "") | |
| def identifiants_partages(feuilles: dict) -> pd.DataFrame: | |
| """Un identifiant porte par plus de personnes distinctes que le seuil tolere. | |
| Pour la piece d'identite, deux personnes sont distinctes si leurs noms different ; | |
| pour les autres identifiants, si leur piece d'identite (sinon leur nom) differe.""" | |
| pers = personnes(feuilles) | |
| lignes = [] | |
| for type_id in TYPES_IDENTIFIANTS: | |
| groupes = {} | |
| for p in pers: | |
| valeur = p["ligne"].get(type_id) | |
| norme = normaliser_identifiant(type_id, valeur) | |
| if norme: | |
| groupes.setdefault(norme, []).append((p, valeur)) | |
| seuil = SEUILS_PARTAGE.get(type_id, 1) | |
| for norme, membres in groupes.items(): | |
| if type_id == "ID_Officiel": | |
| distincts = {(p["cle_nom"] or f"{p['classe']}:{p['id']}") for p, _ in membres} | |
| else: | |
| distincts = {p["cle"] for p, _ in membres} | |
| if len(distincts) > seuil: | |
| lignes.append({ | |
| "Type": type_id, "Valeur": masquer(type_id, membres[0][1]), "_norme": norme, | |
| "Nb personnes": len(distincts), "Seuil": seuil, | |
| "Dossiers": ", ".join(sorted({_libelle(p) for p, _ in membres})), | |
| "_ids": sorted({f"{p['classe']}:{p['id']}" for p, _ in membres}), | |
| }) | |
| colonnes = ["Type", "Valeur", "Nb personnes", "Seuil", "Dossiers", "_norme", "_ids"] | |
| return pd.DataFrame(lignes, columns=colonnes).sort_values(["Nb personnes", "Type"], ascending=[False, True]).reset_index(drop=True) \ | |
| if lignes else pd.DataFrame(columns=colonnes) | |
| def reseaux_garantie(feuilles: dict) -> pd.DataFrame: | |
| """Auto-garantie, garantie croisee (A garantit B et B garantit A), garant de nombreux emprunteurs.""" | |
| colonnes = ["Gravité", "Signal", "Détail", "_ids"] | |
| prets = _df(feuilles, FEUILLE_PRETS) | |
| if prets.empty or not {"ID_Client", "ID_Garant"}.issubset(prets.columns): | |
| return pd.DataFrame(columns=colonnes) | |
| pers = personnes(feuilles) | |
| par_id = {(p["classe"], p["id"]): p for p in pers} | |
| liens = {} # (cle emprunteur, cle garant) -> [prets] | |
| for p in prets.to_dict("records"): | |
| emp = par_id.get(("Client", id_entite(p.get("ID_Client")))) | |
| gar = par_id.get(("Garant", id_entite(p.get("ID_Garant")))) | |
| if emp and gar: | |
| liens.setdefault((emp["cle"], gar["cle"]), []).append((id_entite(p.get("ID_Pret")), emp, gar)) | |
| lignes, vus = [], set() | |
| emprunteurs_par_garant = {} | |
| for (c_emp, c_gar), items in liens.items(): | |
| _, emp, gar = items[0] | |
| ids_prets = sorted(i[0] for i in items) | |
| if c_emp == c_gar: | |
| lignes.append({"Gravité": "Erreur", "Signal": "Auto-garantie", | |
| "Détail": f"{_libelle(emp)} se porte garant de son propre prêt via {_libelle(gar)} ({', '.join(ids_prets)}).", | |
| "_ids": [f"Client:{emp['id']}", f"Garant:{gar['id']}"]}) | |
| emprunteurs_par_garant.setdefault(c_gar, {})[c_emp] = (emp, gar) | |
| inverse = (c_gar, c_emp) | |
| if inverse in liens and frozenset((c_emp, c_gar)) not in vus and c_emp != c_gar: | |
| vus.add(frozenset((c_emp, c_gar))) | |
| _, emp2, gar2 = liens[inverse][0] | |
| lignes.append({"Gravité": "Avertissement", "Signal": "Garantie croisée", | |
| "Détail": f"{_libelle(emp)} et {_libelle(emp2)} se garantissent mutuellement.", | |
| "_ids": [f"Client:{emp['id']}", f"Client:{emp2['id']}", f"Garant:{gar['id']}", f"Garant:{gar2['id']}"]}) | |
| for c_gar, emps in emprunteurs_par_garant.items(): | |
| if len(emps) >= SEUIL_GARANT_MULTI: | |
| gar = next(iter(emps.values()))[1] | |
| noms = ", ".join(sorted(_libelle(e) for e, _ in emps.values())) | |
| lignes.append({"Gravité": "Avertissement", "Signal": "Garant multi-emprunteurs", | |
| "Détail": f"{_libelle(gar)} garantit {len(emps)} emprunteurs : {noms}.", | |
| "_ids": [f"Garant:{gar['id']}"] + [f"Client:{e['id']}" for e, _ in emps.values()]}) | |
| return pd.DataFrame(lignes, columns=colonnes) | |
| # ============================================================================ | |
| # VUE 360 | |
| # ============================================================================ | |
| def rechercher_personnes(feuilles: dict, recherche: str, limite: int = 10) -> list: | |
| """Clients/garants correspondant a un ID exact ou a un nom (tolerant aux fautes).""" | |
| requete = str(recherche or "").strip() | |
| if not requete: | |
| return [] | |
| pers = personnes(feuilles) | |
| exact = [p for p in pers if p["id"] == requete.upper()] | |
| if exact: | |
| return exact | |
| cle = cle_texte(requete) | |
| if not cle: | |
| return [] | |
| scores = [] | |
| for p in pers: | |
| if not p["cle_nom"]: | |
| continue | |
| if cle in p["cle_nom"] or all(m in p["cle_nom"].split() for m in cle.split()): | |
| score = 100 | |
| elif _fuzz is not None: | |
| score = max(_fuzz.token_set_ratio(cle, p["cle_nom"]), _fuzz.partial_ratio(cle, p["cle_nom"])) | |
| else: | |
| score = 0 | |
| if score >= 85: | |
| scores.append((score, p)) | |
| scores.sort(key=lambda x: (-x[0], x[1]["nom"])) | |
| return [p for _, p in scores[:limite]] | |
| def fiche_personne(feuilles: dict, personne: dict, partages: pd.DataFrame = None, | |
| reseaux: pd.DataFrame = None, controles: pd.DataFrame = None) -> dict: | |
| """Vue 360 d'une personne (toutes ses identites client/garant, meme etre humain).""" | |
| pers = personnes(feuilles) | |
| identites = [p for p in pers if p["cle"] == personne["cle"]] or [personne] | |
| ids_client = {p["id"] for p in identites if p["classe"] == "Client"} | |
| ids_garant = {p["id"] for p in identites if p["classe"] == "Garant"} | |
| noeuds = {f"{p['classe']}:{p['id']}" for p in identites} | |
| prets = _df(feuilles, FEUILLE_PRETS) | |
| remb = _df(feuilles, FEUILLE_REMB) | |
| clients_par_id = {p["id"]: p for p in pers if p["classe"] == "Client"} | |
| garants_par_id = {p["id"]: p for p in pers if p["classe"] == "Garant"} | |
| def _resume_pret(p, role): | |
| pid = id_entite(p.get("ID_Pret")) | |
| paiements = remb[remb["ID_Pret"].map(id_entite) == pid] if not remb.empty and "ID_Pret" in remb.columns else pd.DataFrame() | |
| verse = paiements["Montant_Verse"].map(nombre).sum() if "Montant_Verse" in paiements.columns else float("nan") | |
| dernier = nombre(paiements["Solde_Apres"].iloc[-1]) if "Solde_Apres" in paiements.columns and len(paiements) else float("nan") | |
| retards = int((paiements["Jours_Retard"].map(nombre) > 0).sum()) if "Jours_Retard" in paiements.columns else 0 | |
| emp = clients_par_id.get(id_entite(p.get("ID_Client"))) | |
| gar = garants_par_id.get(id_entite(p.get("ID_Garant"))) | |
| return { | |
| "Rôle": role, "Prêt": pid, "Version courante": p.get("ID_Pret_Courant", pid), | |
| "Statut": str(p.get("Statut", "")).strip(), "Déblocage": p.get("Date_Deblocage", ""), | |
| "Capital": nombre(p.get("Montant_Capital")), "Montant total": nombre(p.get("Montant_Total")), | |
| "Versé": verse, "Solde restant": dernier, "Paiements": len(paiements), "En retard": retards, | |
| "Emprunteur": _libelle(emp) if emp else id_entite(p.get("ID_Client")), | |
| "Garant": _libelle(gar) if gar else (id_entite(p.get("ID_Garant")) or "—"), | |
| } | |
| lignes_prets = [] | |
| if not prets.empty: | |
| for p in prets.to_dict("records"): | |
| if id_entite(p.get("ID_Client")) in ids_client: | |
| lignes_prets.append(_resume_pret(p, "Emprunteur")) | |
| if id_entite(p.get("ID_Garant")) in ids_garant: | |
| lignes_prets.append(_resume_pret(p, "Garant")) | |
| ids_prets = {l["Prêt"] for l in lignes_prets} | |
| paiements = pd.DataFrame() | |
| if not remb.empty and "ID_Pret" in remb.columns and ids_prets: | |
| paiements = remb[remb["ID_Pret"].map(id_entite).isin(ids_prets)].copy() | |
| def _filtre_ids(df): | |
| if df is None or df.empty or "_ids" not in df.columns: | |
| return pd.DataFrame() | |
| return df[df["_ids"].map(lambda ids: bool(noeuds & set(ids)))] | |
| controles_lies = pd.DataFrame() | |
| if controles is not None and not controles.empty: | |
| motifs = ids_prets | ids_client | ids_garant | |
| controles_lies = controles[controles["Référence"].map(lambda r: any(m and m in str(r) for m in motifs))] | |
| principal = identites[0] | |
| identite = proteger_proprietes({k: v for k, v in principal["ligne"].items() if not vide(v)}) | |
| return { | |
| "nom": principal["nom"], "identites": [_libelle(p) for p in identites], "noeuds": sorted(noeuds), | |
| "identite": identite, "prets": pd.DataFrame(lignes_prets), "paiements": paiements, | |
| "identifiants_partages": _filtre_ids(partages), "reseaux": _filtre_ids(reseaux), | |
| "controles": controles_lies, | |
| } | |
| def _xof(v): | |
| return "—" if v is None or (isinstance(v, float) and math.isnan(v)) else f"{v:,.0f} XOF".replace(",", " ") | |
| def fiche_texte(fiche: dict) -> str: | |
| """Resume texte (donnees sensibles masquees) d'une fiche 360, pour l'agent IA.""" | |
| lignes = [f"PERSONNE : {fiche['nom'] or '?'}", f"Identités : {', '.join(fiche['identites'])}"] | |
| for k in ("Telephone", "Email", "ID_Officiel", "Profession", "Secteur_Activite", "Ville", "Adresse"): | |
| if k in fiche["identite"]: | |
| lignes.append(f"{k} : {fiche['identite'][k]}") | |
| prets = fiche["prets"] | |
| if prets.empty: | |
| lignes.append("Aucun prêt (ni comme emprunteur, ni comme garant).") | |
| else: | |
| lignes.append(f"PRÊTS ({len(prets)}) :") | |
| for p in prets.to_dict("records"): | |
| lignes.append( | |
| f"- {p['Prêt']} [{p['Rôle']}] statut {p['Statut'] or '?'}, capital {_xof(p['Capital'])}, " | |
| f"total dû {_xof(p['Montant total'])}, versé {_xof(p['Versé'])}, solde {_xof(p['Solde restant'])}, " | |
| f"{p['Paiements']} paiement(s) dont {p['En retard']} en retard ; emprunteur {p['Emprunteur']}, garant {p['Garant']}") | |
| for titre, cle, champ in (("IDENTIFIANTS PARTAGÉS", "identifiants_partages", None), ("RÉSEAUX DE GARANTIE", "reseaux", "Détail")): | |
| df = fiche[cle] | |
| if not df.empty: | |
| lignes.append(f"{titre} :") | |
| for r in df.to_dict("records"): | |
| lignes.append(f"- {r[champ]}" if champ else f"- {r['Type']} {r['Valeur']} partagé par {r['Dossiers']}") | |
| if not fiche["controles"].empty: | |
| lignes.append("ANOMALIES DE DONNÉES :") | |
| for r in fiche["controles"].head(10).to_dict("records"): | |
| lignes.append(f"- [{r['Gravité']}] {r['Contrôle']} — {r['Référence']} : {r['Détail']}") | |
| return "\n".join(lignes) | |