""" 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 = 50) -> list: 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 = 50) -> 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") type_val = None for a in note.get("attributes", []): if a.get("type") == "label" and a.get("name") == "type": type_val = a.get("value") break if type_val not in TYPES_SYSTEME: return (False, "Refus : note sans type systeme connu (type=%s)" % type_val) 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", ] 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))