Master
943acbc573
Durcit la convention de nommage des projets (dérive constatée : 'Sliding Automation', 'code_versioning'... au lieu des formes canoniques). - trilium_api.py : projets_canoniques() lit le référentiel = valeurs du label projet sur les notes de type=projet (source unique, pas de constante en dur). Note-projet CodeVersioning créée (manquait). - mcp_server.py : _valider_projet() branché dans les 6 tools de création (add_decision/history/backlog, new_conversation, create_entite, add_skill). Refuse un projet non canonique (suggestion si faute) ou inconnu (renvoi au processus de création de projet). Ne verrouille pas si référentiel illisible. - lint_audit.py : VAL-nommage aligné sur le référentiel (attrape casse, espace ET snake_case ; l'ancien 'contient un espace' ratait code_versioning). - Données : 79 notes ré-étiquetées vers les 3 formes canoniques. Quality by design : l'erreur de nommage devient impossible à l'écriture, le Lint n'est plus que le filet de sécurité.
293 lines
12 KiB
Python
293 lines
12 KiB
Python
"""
|
|
trilium_api.py — Wrapper trilium-py pour TriliumNext 0.95+
|
|
Tous les autres scripts importent depuis ce module uniquement.
|
|
"""
|
|
import secrets
|
|
import os
|
|
from typing import Optional
|
|
from dotenv import load_dotenv
|
|
from trilium_py.client import ETAPI
|
|
|
|
load_dotenv()
|
|
|
|
_TRILIUM_URL = os.getenv("TRILIUM_URL", "http://localhost:4292")
|
|
_TRILIUM_TOKEN = os.getenv("TRILIUM_TOKEN", "")
|
|
|
|
ea: Optional[ETAPI] = None
|
|
|
|
def check_api():
|
|
global ea
|
|
try:
|
|
ea = ETAPI(_TRILIUM_URL, _TRILIUM_TOKEN)
|
|
result = ea.get_note("root")
|
|
if not result or "noteId" not in result:
|
|
raise ConnectionError("Réponse inattendue")
|
|
return True
|
|
except Exception as e:
|
|
raise SystemExit(
|
|
f"\n❌ Trilium inaccessible sur {_TRILIUM_URL}"
|
|
f"\n Détail : {e}"
|
|
"\n sudo docker ps | grep trilium"
|
|
"\n sudo docker start trilium\n"
|
|
)
|
|
|
|
def _ea() -> ETAPI:
|
|
global ea
|
|
if ea is None:
|
|
ea = ETAPI(_TRILIUM_URL, _TRILIUM_TOKEN)
|
|
return ea
|
|
|
|
def _note_id(result: dict) -> str:
|
|
"""Extrait le noteId quel que soit le format retourné par trilium-py."""
|
|
if "note" in result:
|
|
return result["note"]["noteId"]
|
|
if "noteId" in result:
|
|
return result["noteId"]
|
|
raise ValueError(f"Format inattendu : {list(result.keys())}")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Notes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def create_note(parent_id: str, title: str, content: str = " ",
|
|
note_type: str = "text") -> dict:
|
|
return _ea().create_note(
|
|
parentNoteId=parent_id, title=title,
|
|
type=note_type, content=content
|
|
)
|
|
|
|
def get_note_id(result: dict) -> str:
|
|
"""Helper public pour extraire le noteId d'un résultat create_note."""
|
|
return _note_id(result)
|
|
|
|
def get_note(note_id: str) -> dict:
|
|
return _ea().get_note(note_id)
|
|
|
|
def get_note_content(note_id: str) -> str:
|
|
result = _ea().get_note_content(note_id)
|
|
if isinstance(result, bytes):
|
|
return result.decode("utf-8")
|
|
return result or ""
|
|
|
|
def update_note_content(note_id: str, content: str):
|
|
_ea().update_note_content(note_id, content)
|
|
|
|
def get_children(note_id: str) -> list:
|
|
note = _ea().get_note(note_id)
|
|
return [{"noteId": cid} for cid in note.get("childNoteIds", [])]
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Recherche
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def search_notes(query: str, limit: int = None) -> list:
|
|
# limit=None : pas de plafond (ETAPI renvoie tout). Un limit explicite plafonne volontairement.
|
|
if limit is None:
|
|
result = _ea().search_note(search=query)
|
|
else:
|
|
result = _ea().search_note(search=query, limit=limit)
|
|
return result.get("results", []) if isinstance(result, dict) else []
|
|
|
|
def search_by_label(label: str, value: str = "", limit: int = None) -> list:
|
|
query = f"#{label}" if not value else f"#{label}={value}"
|
|
return search_notes(query, limit)
|
|
|
|
def find_note_by_title(title: str, parent_id: str = "") -> Optional[str]:
|
|
for note in search_notes(title):
|
|
if note.get("title", "").strip().lower() == title.strip().lower():
|
|
if not parent_id:
|
|
return note["noteId"]
|
|
detail = get_note(note["noteId"])
|
|
if parent_id in detail.get("parentNoteIds", []):
|
|
return note["noteId"]
|
|
return None
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Attributs
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def set_label(note_id: str, name: str, value: str = ""):
|
|
"""Cree ou met a jour un label via patch_attribute natif (pas d empilement)."""
|
|
ea = _ea()
|
|
note = get_note(note_id)
|
|
existants = [a for a in note.get("attributes", [])
|
|
if a.get("type") == "label" and a.get("name") == name]
|
|
if existants:
|
|
ea.patch_attribute(attributeId=existants[0].get("attributeId"), value=value)
|
|
else:
|
|
ea.create_attribute(attributeId=secrets.token_hex(6), noteId=note_id, type="label",
|
|
name=name, value=value, isInheritable=False)
|
|
|
|
def get_label_value(note_id: str, name: str) -> Optional[str]:
|
|
note = get_note(note_id)
|
|
for a in note.get("attributes", []):
|
|
if a.get("type") == "label" and a.get("name") == name:
|
|
return a.get("value", "")
|
|
try:
|
|
attrs = _ea().get_note_attributes(note_id)
|
|
if isinstance(attrs, list):
|
|
for a in attrs:
|
|
if a.get("type") == "label" and a.get("name") == name:
|
|
return a.get("value", "")
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
TYPES_SYSTEME = ["backlogItem","decision","historiqueItem","conversation","termeGlossaire","projet","contexteReprise","skill","documentation"]
|
|
|
|
def delete_note_safe(note_id: str):
|
|
"""Supprime une note SEULEMENT si elle porte un label type connu du systeme.
|
|
Garde-fou contre la suppression accidentelle de dossiers/notes hors systeme.
|
|
Retourne (True, message) ou (False, message)."""
|
|
note = get_note(note_id)
|
|
if not note:
|
|
return (False, "Note introuvable")
|
|
if not _est_editable(note):
|
|
return (False, "Refus : note structurelle protegee (sans label projet)")
|
|
titre = note.get("title", "?")
|
|
ok = _ea().delete_note(note_id)
|
|
if ok:
|
|
return (True, "Note supprimee : %s (%s)" % (titre, note_id))
|
|
return (False, "Echec suppression cote Trilium")
|
|
|
|
|
|
def move_note_safe(note_id, ancien_parent_id, nouveau_parent_id):
|
|
"""Deplace une note d un parent vers un autre (gere les clones : ne touche
|
|
que la branche depuis ancien_parent_id). Garde-fou leger : verifie l existence
|
|
de la note et du nouveau parent. Retourne (True, message) ou (False, message)."""
|
|
ea = _ea()
|
|
note = get_note(note_id)
|
|
if not note:
|
|
return (False, "Note introuvable: %s" % note_id)
|
|
if not get_note(nouveau_parent_id):
|
|
return (False, "Nouveau parent introuvable: %s" % nouveau_parent_id)
|
|
# Retrouver la branche reelle depuis l ancien parent
|
|
cible_branche = None
|
|
for bid in note.get("parentBranchIds", []):
|
|
if bid.startswith(ancien_parent_id + "_"):
|
|
cible_branche = bid
|
|
break
|
|
if not cible_branche:
|
|
return (False, "Aucune branche depuis le parent %s" % ancien_parent_id)
|
|
# Creer la nouvelle branche puis supprimer l ancienne
|
|
try:
|
|
ea.create_branch(noteId=note_id, parentNoteId=nouveau_parent_id,
|
|
prefix="", isExpanded=False, notePosition=10)
|
|
except Exception as e:
|
|
return (False, "Echec create_branch: %s" % e)
|
|
try:
|
|
ea.delete_branch(cible_branche)
|
|
except Exception as e:
|
|
return (False, "Branche creee mais echec suppression ancienne (%s): %s" % (cible_branche, e))
|
|
return (True, "Note %s deplacee de %s vers %s" % (note_id, ancien_parent_id, nouveau_parent_id))
|
|
|
|
|
|
# Relations autorisees par l ontologie (anti-drift). Le controle domaine/portee
|
|
# fin est delegue au Lint ; ici on bloque seulement les noms hors ontologie.
|
|
RELATIONS_ONTOLOGIE = [
|
|
"concerneProjet", "meneePar", "priseDans", "produitDans", "synthetise",
|
|
"destineA", "hebergeSur", "implementePar", "dependDe", "aVersion",
|
|
"documentePar", "decrit", "impacte", "contraintPar", "revise", "invalide",
|
|
"reference", "illustre", "provient",
|
|
]
|
|
|
|
# Types d'entites canoniques de l'ontologie (source unique cote code).
|
|
# NE contient PAS 'container' (= dossiers structurels, hors couche connaissance).
|
|
# Cible a terme : lire cette liste depuis les notes d'ontologie (dossier
|
|
# Modeles & Ontologie) plutot que la maintenir ici. En attendant, source unique
|
|
# partagee par le Lint (VAL-type) et create_entite.
|
|
TYPES_CANONIQUES = [
|
|
"projet", "conversation", "backlogItem", "decision", "historiqueItem",
|
|
"concept", "contexteReprise", "skill", "documentation",
|
|
"composantLogiciel", "service", "infrastructure", "outil",
|
|
"contrainte", "principe", "convention", "methode", "version",
|
|
]
|
|
|
|
|
|
def projets_canoniques():
|
|
"""Referentiel des noms de projets valides = ensemble des valeurs du label
|
|
projet portees par les notes de type=projet (dossier Projets).
|
|
Source unique : la note-projet se declare elle-meme via son label projet.
|
|
Lecture directe (pas de cache) : un projet cree est immediatement reconnu.
|
|
Retourne un set de chaines. Set vide si aucune note-projet (a traiter comme
|
|
'ne pas bloquer' par les appelants, pour ne pas se verrouiller)."""
|
|
noms = set()
|
|
try:
|
|
for n in search_by_label("type", "projet"):
|
|
v = get_label_value(n["noteId"], "projet")
|
|
if v:
|
|
noms.add(v)
|
|
except Exception:
|
|
pass
|
|
return noms
|
|
|
|
def add_relation_safe(source_id, nom, cible_id):
|
|
"""Cree une relation (object property) entre deux notes.
|
|
Garde-fou Option B : nom dans l ontologie + source/cible existent.
|
|
Idempotent : ne recree pas une relation identique deja presente.
|
|
Retourne (True, message) ou (False, message)."""
|
|
if nom not in RELATIONS_ONTOLOGIE:
|
|
return (False, "Relation hors ontologie: %s (voir RELATIONS_ONTOLOGIE)" % nom)
|
|
src = get_note(source_id)
|
|
if not src:
|
|
return (False, "Note source introuvable: %s" % source_id)
|
|
if not get_note(cible_id):
|
|
return (False, "Note cible introuvable: %s" % cible_id)
|
|
# Idempotence : relation identique deja la ?
|
|
for a in src.get("attributes", []):
|
|
if a.get("type") == "relation" and a.get("name") == nom and a.get("value") == cible_id:
|
|
return (True, "Relation deja presente: %s ~%s %s" % (source_id, nom, cible_id))
|
|
try:
|
|
_ea().create_attribute(attributeId=secrets.token_hex(6), noteId=source_id, type="relation",
|
|
name=nom, value=cible_id, isInheritable=False)
|
|
except Exception as e:
|
|
return (False, "Echec creation relation: %s" % e)
|
|
return (True, "Relation creee: %s ~%s %s" % (source_id, nom, cible_id))
|
|
|
|
|
|
def remove_relation_safe(source_id, nom, cible_id):
|
|
"""Retire la relation source ~nom-> cible si elle existe (via son attributeId).
|
|
Idempotent : si absente, ne fait rien et le signale.
|
|
Retourne (True, message) ou (False, message)."""
|
|
src = get_note(source_id)
|
|
if not src:
|
|
return (False, "Note source introuvable: %s" % source_id)
|
|
aid = None
|
|
for a in src.get("attributes", []):
|
|
if a.get("type") == "relation" and a.get("name") == nom and a.get("value") == cible_id:
|
|
aid = a.get("attributeId")
|
|
break
|
|
if aid is None:
|
|
return (True, "Relation absente (rien a retirer): %s ~%s %s" % (source_id, nom, cible_id))
|
|
try:
|
|
_ea().delete_attribute(aid)
|
|
except Exception as e:
|
|
return (False, "Echec retrait relation: %s" % e)
|
|
return (True, "Relation retiree: %s ~%s %s" % (source_id, nom, cible_id))
|
|
|
|
|
|
def move_relation_safe(source_id, nom, ancienne_cible, nouvelle_cible):
|
|
"""Reconnecte une relation : cree source ~nom-> nouvelle_cible puis retire
|
|
source ~nom-> ancienne_cible. Geste de fusion en une operation.
|
|
Retourne (True, message) ou (False, message)."""
|
|
ok_add, msg_add = add_relation_safe(source_id, nom, nouvelle_cible)
|
|
if not ok_add:
|
|
return (False, "Echec (creation nouvelle cible) : %s" % msg_add)
|
|
ok_rm, msg_rm = remove_relation_safe(source_id, nom, ancienne_cible)
|
|
if not ok_rm:
|
|
return (False, "Nouvelle cible creee MAIS echec retrait ancienne : %s" % msg_rm)
|
|
return (True, "Relation deplacee: %s ~%s de %s vers %s" % (source_id, nom, ancienne_cible, nouvelle_cible))
|
|
|
|
|
|
def _est_editable(note):
|
|
"""Nouveau critere de garde-fou : une note est editable/supprimable si elle
|
|
porte un label projet (note de projet, propriete du projet), ou si son type
|
|
est 'projet' lui-meme. Sinon (ontologie, skills universels, dossiers
|
|
structurels) elle est protegee par defaut. Remplace TYPES_SYSTEME (liste
|
|
figee, desynchronisee de l ontologie a chaque nouvelle classe)."""
|
|
labels = {a.get("name"): a.get("value") for a in note.get("attributes", []) if a.get("type") == "label"}
|
|
if labels.get("type") == "projet":
|
|
return True
|
|
return bool(labels.get("projet"))
|