Master
7c56f125ad
- trilium_api.py : remove_relation_safe (retrait via attributeId, idempotent), move_relation_safe (reconnexion = add nouvelle puis remove ancienne, ordre sur) - mcp_server.py : 2 tools MCP exposes (22 tools au total) Comble le manque revele par la fusion render_engine.py : le systeme savait creer une relation mais pas la retirer ni la reconnecter. Rend triviales les corrections de mesusage et les fusions de hub. Teste de bout en bout.
274 lines
11 KiB
Python
274 lines
11 KiB
Python
"""
|
|
trilium_api.py — Wrapper trilium-py pour TriliumNext 0.95+
|
|
Tous les autres scripts importent depuis ce module uniquement.
|
|
"""
|
|
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(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 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(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"))
|