context-continuity/lint_audit.py

#!/usr/bin/env python3
"""
lint_audit.py - Moteur d'audit (Lint) du knowledge graph Context Continuity.

ETAGE 1 : LECTURE SEULE. Detecte et rapporte des anomalies ; ne corrige RIEN
et n'ecrit RIEN dans Trilium (rapport texte : stdout + fichier horodate dans
lint_reports/).

ORGANISATION PAR DATA QUALITY DIMENSIONS (referentiel autoporteur, extensible).
Chaque controle declare la dimension qu'il sert ; le rapport est groupe par
dimension. Ajouter un controle = decorer une fonction avec @check(dimension,...).
Ajouter une dimension = l'ajouter a DIMENSIONS. Le rapport suit automatiquement.

Chaque anomalie porte aussi un bac d'action (prepare l'etage 2) :
  AUTO      : correction mecanique sure (accent, casse, enum non ambigue)
  SIGNALER  : deterministe mais sans correction evidente -> decision humaine
  JUGEMENT  : demande du contexte metier -> revue par l'agent / l'humain
A l'etage 1, AUCUNE correction n'est appliquee, quel que soit le bac.

Usage (sur GrosseBertha, dans le venv) :
    cd ~/App/Context_continuity && source venv/bin/activate
    python3 lint_audit.py
"""

import os
import json
import unicodedata
from datetime import datetime, timezone
from typing import List, Dict, Optional

from trilium_api import (
    search_by_label, get_note, get_note_content,
    RELATIONS_ONTOLOGIE, TYPES_CANONIQUES, projets_canoniques,
)

BASE = os.path.expanduser("~/App/Context_continuity")
IDS = json.load(open(os.path.join(BASE, "trilium_ids.json")))


# ==========================================================================
# 1. LA TAXONOMIE : data quality dimensions (autoporteur)
# ==========================================================================

# Ordre = ordre d'affichage dans le rapport. Definitions = celles de Bastien
# (Data Governance), alignees sur les dimensions DAMA standard.
DIMENSIONS = {
    "completeness": "Les champs obligatoires sont remplis ET les entites attendues sont toutes saisies (exhaustivite).",
    "accuracy":     "Le niveau de la donnee reflete la realite (bon rangement, contenu conforme au type, statut a jour).",
    "consistency":  "Coherence dans le temps et a travers les entites (vocabulaire courant, informations repetees a jour partout).",
    "validity":     "Les valeurs et relations respectent les regles en vigueur (enums, conventions de nommage, ontologie des relations).",
    "uniqueness":   "Absence de doublons.",
    "freshness":    "Les entites ne portent pas un etat perime (anciennete anormale, non-cloture).",
}

# Controles PREVUS mais NON deterministes : delegues a l'agent reviseur (etage 2).
# Rendus visibles pour l'honnetete de la couverture.
CHECKS_DELEGUES = [
    {"dimension": "completeness", "code": "COMP-exhaustivite",
     "libelle": "Entites manquantes vs realite du projet (ex : backlog incomplet)",
     "raison": "Non detectable en deterministe : requiert la comparaison avec la realite (code, conversations). -> agent reviseur."},
    {"dimension": "accuracy", "code": "ACC-contenu-type",
     "libelle": "Le contenu redige d'une note reflete-t-il bien son type",
     "raison": "Requiert un jugement semantique. -> agent reviseur."},
    {"dimension": "consistency", "code": "CONS-chiffres-croises",
     "libelle": "Une information repetee (ex : nombre de tools) est-elle a jour dans toutes les notes",
     "raison": "Partiellement couvert par CONS-vocab ; la verification exhaustive des chiffres croises -> agent reviseur."},
]


# ==========================================================================
# 2. REFERENTIEL : ce que le systeme considere comme valide
# ==========================================================================

ENUMS = {
    ("projet", "statut"): {"actif", "en-pause", "archive"},
    ("conversation", "statut"): {"en-cours", "clos"},
    ("backlogItem", "statut"): {"a faire", "en cours", "bloque", "fait", "abandonne"},
    ("backlogItem", "priorite"): {"haute", "moyenne", "basse"},
    ("decision", "statut"): {"active", "revisee", "annulee"},
    ("historiqueItem", "typeHistorique"): {
        "Fait etabli", "Test effectue", "Hypothese invalidee", "Contrainte decouverte",
    },
    ("historiqueItem", "encoreValide"): {"true", "false"},
}

LABELS_OBLIGATOIRES = {
    "projet": ["projet", "statut"],
    "conversation": ["projet", "llm", "statut", "date"],
    "backlogItem": ["projet", "statut", "priorite"],
    "decision": ["projet", "statut"],
    "historiqueItem": ["projet", "typeHistorique", "encoreValide"],
    "concept": ["projet"],
    "contexteReprise": ["projet", "llmCible", "version"],
    "composantLogiciel": ["projet"],
    "service": ["projet"],
    "infrastructure": ["projet"],
    "outil": ["projet"],
    "contrainte": ["projet"],
    "principe": ["projet"],
    "convention": ["projet"],
    "methode": ["projet"],
    "version": ["projet"],
}

DOSSIER_ATTENDU = {
    "backlogItem": "Backlog",
    "decision": "Decisions",
    "historiqueItem": "Historique",
    "conversation": "Conversations",
    "concept": "Concepts",
    "contexteReprise": "ContextesReprise",
    "projet": "Projets",
}

TYPES_METIER = list(LABELS_OBLIGATOIRES.keys()) + ["skill", "documentation"]

TYPES_DEPRECIES = {"termeGlossaire"}
MOTS_DEPRECIES = ["type systeme", "type système", "termeGlossaire"]

JOURS_STALE_ENCOURS = 90
JOURS_STALE_CONVO = 30

# Domaine -> portee (sous-ensemble ROBUSTE ; classes mappees aux types reels).
# domaine None = seule la portee est verifiee. Table volontairement partielle :
# uniquement les relations dont le mapping classe-ontologie -> type est certain.
RELATION_DP = {
    "impacte":        ({"decision", "historiqueItem"}, {"composantLogiciel", "service"}),
    "revise":         ({"decision"}, {"decision"}),
    "dependDe":       ({"composantLogiciel", "service"}, {"composantLogiciel", "service", "outil"}),
    "aVersion":       ({"composantLogiciel", "service"}, {"version"}),
    "implementePar":  ({"service"}, {"composantLogiciel"}),
    "contraintPar":   ({"decision", "composantLogiciel"}, {"contrainte", "principe"}),
    "illustre":       ({"concept", "composantLogiciel", "service"}, {"concept"}),
    "documentePar":   ({"service", "outil", "methode", "composantLogiciel", "historiqueItem", "decision"}, {"skill", "documentation"}),
    "concerneProjet": (None, {"projet"}),
}


# ==========================================================================
# 3. REGISTRE DES CONTROLES (decorateur autoporteur)
# ==========================================================================

CHECKS: List[dict] = []      # {dimension, code, libelle, fn}
_CTX = {"dimension": None}   # dimension du check en cours d'execution


def check(dimension: str, code: str, libelle: str):
    """Enregistre une fonction de controle sous une dimension de qualite."""
    def deco(fn):
        CHECKS.append({"dimension": dimension, "code": code, "libelle": libelle, "fn": fn})
        return fn
    return deco


# ==========================================================================
# 4. UTILITAIRES + CHARGEMENT (une passe, cache)
# ==========================================================================

def sans_accents(s: str) -> str:
    return "".join(c for c in unicodedata.normalize("NFD", s)
                   if unicodedata.category(c) != "Mn")


NOTES: Dict[str, dict] = {}
BY_TYPE: Dict[str, List[str]] = {}


def labels_of(note: dict) -> Dict[str, str]:
    return {a.get("name"): a.get("value", "")
            for a in note.get("attributes", []) if a.get("type") == "label"}


def relations_of(note: dict) -> List[Dict[str, str]]:
    return [{"name": a.get("name"), "value": a.get("value", "")}
            for a in note.get("attributes", []) if a.get("type") == "relation"]


def type_of(note: dict) -> Optional[str]:
    return labels_of(note).get("type")


def age_jours(note: dict) -> Optional[int]:
    dm = note.get("utcDateModified") or note.get("dateModified")
    if not dm:
        return None
    for fmt in ("%Y-%m-%d %H:%M:%S.%fZ", "%Y-%m-%d %H:%M:%SZ",
                "%Y-%m-%dT%H:%M:%S.%fZ", "%Y-%m-%d %H:%M:%S.%f"):
        try:
            d = datetime.strptime(dm, fmt).replace(tzinfo=timezone.utc)
            return (datetime.now(timezone.utc) - d).days
        except Exception:
            continue
    return None


def charger():
    """Charge TOUTES les notes portant un label type (recherche #type globale),
    quel que soit le type - y compris les types invalides/inconnus, pour que
    VAL-type puisse les detecter. Indexe par la valeur reelle du label type."""
    try:
        res = search_by_label("type", "")   # #type sans valeur = toutes les notes typees
    except Exception as e:
        print("  ! echec recherche #type : %s" % e)
        res = []
    for r in res:
        nid = r.get("noteId")
        if not nid or nid in NOTES:
            continue
        try:
            note = get_note(nid)
        except Exception:
            continue
        NOTES[nid] = note
        tv = type_of(note) or "(sans type)"
        BY_TYPE.setdefault(tv, []).append(nid)
    print("  Charge : %d notes sur %d valeurs de type" % (len(NOTES), len(BY_TYPE)))


# ==========================================================================
# 5. COLLECTE DES ANOMALIES
# ==========================================================================

ANOMALIES: List[dict] = []


def signaler(code, bac, note_id, titre, detail, suggestion=""):
    """Enregistre une anomalie ; la dimension est heritee du check courant."""
    ANOMALIES.append({
        "dimension": _CTX["dimension"], "code": code, "bac": bac,
        "note_id": note_id, "titre": titre, "detail": detail, "suggestion": suggestion,
    })


# ==========================================================================
# 6. LES CONTROLES, GROUPES PAR DIMENSION
# ==========================================================================

# ---- COMPLETENESS --------------------------------------------------------

@check("completeness", "COMP-labels", "Labels obligatoires presents selon le type")
def c_labels_obligatoires():
    for typ, obligatoires in LABELS_OBLIGATOIRES.items():
        for nid in BY_TYPE.get(typ, []):
            note = NOTES[nid]
            labels = labels_of(note)
            manquants = [l for l in obligatoires if not labels.get(l)]
            if manquants:
                signaler("COMP-labels", "SIGNALER", nid, note.get("title", "?"),
                         "type=%s : labels manquants %s" % (typ, manquants),
                         "ajouter les labels manquants")


# ---- ACCURACY ------------------------------------------------------------

@check("accuracy", "ACC-rangement", "Note rangee dans le dossier attendu de son type")
def a_rangement():
    for typ, cle_dossier in DOSSIER_ATTENDU.items():
        dossier_id = IDS.get(cle_dossier)
        if not dossier_id:
            continue
        for nid in BY_TYPE.get(typ, []):
            note = NOTES[nid]
            parents = note.get("parentNoteIds", [])
            if parents and dossier_id not in parents:
                signaler("ACC-rangement", "JUGEMENT", nid, note.get("title", "?"),
                         "type=%s hors du dossier %s (parents=%s)" % (typ, cle_dossier, parents),
                         "verifier le rangement (clone legitime possible)")


@check("accuracy", "ACC-decision", "Statut des decisions actives reflete la realite (v1 vs v2)")
def a_decisions_obsoletes():
    SEUIL = 20
    par_projet = {}
    for nid in BY_TYPE.get("decision", []):
        labels = labels_of(NOTES[nid])
        if labels.get("statut") == "active":
            par_projet.setdefault(labels.get("projet", "?"), []).append(nid)
    for proj, ids in par_projet.items():
        if len(ids) >= SEUIL:
            signaler("ACC-decision", "JUGEMENT", ids[0], "(projet %s)" % proj,
                     "%d decisions actives sur %s : possibles v1 non revisees" % (len(ids), proj),
                     "revue : tisser ~revise, passer les v1 obsoletes en revisee")


# ---- CONSISTENCY ---------------------------------------------------------

@check("consistency", "CONS-vocab", "Vocabulaire courant (types et termes non deprecies)")
def cons_vocabulaire():
    # Types deprecies
    for typ in TYPES_DEPRECIES:
        for nid in BY_TYPE.get(typ, []):
            signaler("CONS-type", "SIGNALER", nid, NOTES[nid].get("title", "?"),
                     "type deprecie : %s" % typ, "migrer vers concept")
    # Mots deprecies dans le contenu des skills / documentation
    for typ in ("skill", "documentation"):
        for nid in BY_TYPE.get(typ, []):
            try:
                contenu = get_note_content(nid)
            except Exception:
                continue
            trouves = [m for m in MOTS_DEPRECIES if m in contenu]
            if trouves:
                signaler("CONS-vocab", "JUGEMENT", nid, NOTES[nid].get("title", "?"),
                         "mots deprecies dans le contenu : %s" % trouves,
                         "reformuler (faux positif possible si mention explicative legitime)")


# ---- VALIDITY ------------------------------------------------------------

@check("validity", "VAL-enum", "Valeurs de labels conformes aux enumerations permises")
def v_enums():
    for (typ, label), valides in ENUMS.items():
        for nid in BY_TYPE.get(typ, []):
            note = NOTES[nid]
            val = labels_of(note).get(label)
            if val is None or val in valides:
                continue
            titre = note.get("title", "?")
            cible = None
            for v in valides:
                if sans_accents(val.lower()) == sans_accents(v.lower()):
                    cible = v
                    break
            if cible is not None:
                signaler("VAL-enum", "AUTO", nid, titre,
                         "%s=%r (casse/accent)" % (label, val), "corriger en %r" % cible)
            else:
                signaler("VAL-enum", "SIGNALER", nid, titre,
                         "%s=%r hors valeurs permises %s" % (label, val, sorted(valides)),
                         "corriger manuellement")


@check("validity", "VAL-type", "Label type appartient a la liste canonique de l ontologie")
def v_type():
    for nid, note in NOTES.items():
        tv = type_of(note)
        if tv is None or tv == "container" or tv in TYPES_CANONIQUES:
            continue  # container = dossier structurel, hors perimetre
        titre = note.get("title", "?")
        cible = None
        for t in TYPES_CANONIQUES:
            if sans_accents(tv.lower()) == sans_accents(t.lower()):
                cible = t
                break
        if cible is not None:
            signaler("VAL-type", "AUTO", nid, titre,
                     "type=%r (casse)" % tv, "corriger en %r" % cible)
        else:
            signaler("VAL-type", "SIGNALER", nid, titre,
                     "type=%r hors liste canonique" % tv, "corriger ou ajouter a l ontologie")


@check("validity", "VAL-nommage", "Label projet appartient au referentiel (notes de type projet)")
def v_nommage():
    canon = projets_canoniques()
    if not canon:
        return  # referentiel illisible : ne pas produire de faux positifs
    import unicodedata
    def norm(s):
        s = "".join(ch for ch in unicodedata.normalize("NFD", s) if unicodedata.category(ch) != "Mn")
        return s.lower().replace(" ", "").replace("_", "").replace("-", "")
    for nid, note in NOTES.items():
        val = labels_of(note).get("projet")
        if not val or val in canon:
            continue
        # Chercher un canonique proche (faute de casse/espace/accent) -> AUTO
        cible = None
        for cand in canon:
            if norm(cand) == norm(val):
                cible = cand
                break
        if cible is not None:
            signaler("VAL-nommage", "AUTO", nid, note.get("title", "?"),
                     "projet=%r non canonique" % val, "corriger en %r" % cible)
        else:
            signaler("VAL-nommage", "SIGNALER", nid, note.get("title", "?"),
                     "projet=%r hors referentiel" % val,
                     "corriger ou creer la note de type projet correspondante")


@check("validity", "VAL-relation", "Relations valides (ontologie, cible existante, domaine/portee)")
def v_relations():
    for nid, note in NOTES.items():
        typ_src = type_of(note)
        titre = note.get("title", "?")
        for rel in relations_of(note):
            nom, cible = rel["name"], rel["value"]
            if nom not in RELATIONS_ONTOLOGIE:
                signaler("VAL-rel-nom", "SIGNALER", nid, titre,
                         "relation ~%s hors ontologie" % nom, "renommer ou retirer")
                continue
            cible_note = NOTES.get(cible)
            if cible_note is None:
                try:
                    cible_note = get_note(cible)
                except Exception:
                    cible_note = None
            if not cible_note:
                signaler("VAL-rel-cible", "SIGNALER", nid, titre,
                         "~%s pointe vers une note absente (%s)" % (nom, cible),
                         "corriger ou retirer la relation")
                continue
            dp = RELATION_DP.get(nom)
            if dp:
                dom, por = dp
                typ_cible = type_of(cible_note)
                if dom is not None and typ_src not in dom:
                    signaler("VAL-rel-domaine", "JUGEMENT", nid, titre,
                             "~%s : source type=%s hors domaine %s" % (nom, typ_src, sorted(dom)),
                             "verifier la pertinence")
                if por is not None and typ_cible not in por:
                    signaler("VAL-rel-portee", "JUGEMENT", nid, titre,
                             "~%s : cible type=%s hors portee %s" % (nom, typ_cible, sorted(por)),
                             "verifier la pertinence")


# ---- UNIQUENESS ----------------------------------------------------------

@check("uniqueness", "UNIQ-doublon", "Absence de notes identiques (titre + type + projet)")
def u_doublons():
    vus = {}
    for nid, note in NOTES.items():
        cle = ((note.get("title") or "").strip().lower(),
               type_of(note), labels_of(note).get("projet", ""))
        vus.setdefault(cle, []).append(nid)
    for (titre, typ, proj), ids in vus.items():
        if len(ids) > 1:
            signaler("UNIQ-doublon", "SIGNALER", ids[0], titre or "(sans titre)",
                     "%d notes identiques (type=%s projet=%s) : %s" % (len(ids), typ, proj, ids),
                     "fusionner ou supprimer les doublons")


# ---- FRESHNESS -----------------------------------------------------------

@check("freshness", "FRESH-staleness", "Entites sans etat perime (anciennete, non-cloture)")
def f_staleness():
    for nid in BY_TYPE.get("backlogItem", []):
        note = NOTES[nid]
        if labels_of(note).get("statut") == "en cours":
            age = age_jours(note)
            if age is not None and age > JOURS_STALE_ENCOURS:
                signaler("FRESH-backlog", "JUGEMENT", nid, note.get("title", "?"),
                         "backlog 'en cours' inchange depuis %d j" % age, "verifier si toujours actif")
    for nid in BY_TYPE.get("conversation", []):
        note = NOTES[nid]
        if labels_of(note).get("statut") == "en-cours":
            age = age_jours(note)
            if age is not None and age > JOURS_STALE_CONVO:
                signaler("FRESH-conversation", "JUGEMENT", nid, note.get("title", "?"),
                         "conversation 'en-cours' non cloturee depuis %d j" % age, "cloturer si terminee")


# ==========================================================================
# 7. ORCHESTRATION + RAPPORT (groupes par dimension)
# ==========================================================================

def executer():
    print("Chargement des notes...")
    charger()
    print("Execution des controles par dimension...")
    ordre = {d: i for i, d in enumerate(DIMENSIONS)}
    for chk in sorted(CHECKS, key=lambda c: ordre.get(c["dimension"], 99)):
        _CTX["dimension"] = chk["dimension"]
        try:
            chk["fn"]()
        except Exception as e:
            print("  ! controle %s (%s) a echoue : %s" % (chk["code"], chk["dimension"], e))


def rapport() -> str:
    date = datetime.now().strftime("%Y-%m-%d %H:%M")
    L = ["# RAPPORT LINT - Context Continuity - %s" % date,
         "# ETAGE 1 (lecture seule) - groupe par data quality dimension", ""]

    # Inventaire
    L.append("## Inventaire")
    L.append("- Notes chargees : %d" % len(NOTES))
    for t in sorted(BY_TYPE):
        if BY_TYPE[t]:
            L.append("  - %s : %d" % (t, len(BY_TYPE[t])))
    L.append("")

    # Synthese
    par_dim = {d: 0 for d in DIMENSIONS}
    par_bac = {"AUTO": 0, "SIGNALER": 0, "JUGEMENT": 0}
    for a in ANOMALIES:
        par_dim[a["dimension"]] = par_dim.get(a["dimension"], 0) + 1
        par_bac[a["bac"]] = par_bac.get(a["bac"], 0) + 1
    L.append("## Synthese : %d anomalies" % len(ANOMALIES))
    L.append("  Par dimension : " + ", ".join("%s=%d" % (d, par_dim[d]) for d in DIMENSIONS))
    L.append("  Par bac d'action : " + ", ".join("%s=%d" % (b, par_bac[b]) for b in ("AUTO", "SIGNALER", "JUGEMENT")))
    L.append("")

    # Detail par dimension
    for dim, definition in DIMENSIONS.items():
        anos = [a for a in ANOMALIES if a["dimension"] == dim]
        actifs = [c for c in CHECKS if c["dimension"] == dim]
        delegues = [c for c in CHECKS_DELEGUES if c["dimension"] == dim]
        L.append("## %s (%d anomalie%s)" % (dim.upper(), len(anos), "s" if len(anos) != 1 else ""))
        L.append("  %s" % definition)
        L.append("  Controles deterministes : " + ", ".join(c["code"] for c in actifs))
        for d in delegues:
            L.append("  [delegue agent] %s : %s" % (d["code"], d["libelle"]))
        if anos:
            L.append("")
            for a in anos:
                L.append("  - [%s | %s] %s (%s)" % (a["code"], a["bac"], a["titre"], a["note_id"]))
                L.append("      %s" % a["detail"])
                if a["suggestion"]:
                    L.append("      -> %s" % a["suggestion"])
        L.append("")
    return "\n".join(L)


def main():
    executer()
    txt = rapport()
    print("\n" + txt)
    rep_dir = os.path.join(BASE, "lint_reports")
    os.makedirs(rep_dir, exist_ok=True)
    chemin = os.path.join(rep_dir, "lint_%s.txt" % datetime.now().strftime("%Y%m%d_%H%M%S"))
    with open(chemin, "w", encoding="utf-8") as f:
        f.write(txt)
    print("\nRapport ecrit : %s" % chemin)
    if len(ANOMALIES) > 100:
        print("\n!!! ALERTE : %d anomalies (>100). Investiguer avant toute correction." % len(ANOMALIES))


if __name__ == "__main__":
    main()