""" 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"))