context-continuity/trilium_api.py

"""
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",
]

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