"""
api_context.py — API FastAPI pour Context Continuity
Expose les objets Trilium (Projet, Backlog, Decision, Historique,
Conversation, Glossaire, ContexteReprise) via REST.
Usage :
uvicorn api_context:app --host 127.0.0.1 --port 5000
Doc : http://localhost:5000/api/docs
"""
import os
import re
from typing import Optional, List
from datetime import datetime
from dotenv import load_dotenv
from fastapi import FastAPI, Depends, HTTPException, Header
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field
load_dotenv()
API_KEY_CLAUDE = os.getenv("API_KEY_CLAUDE", "")
API_KEY_LECHAT = os.getenv("API_KEY_LECHAT", "")
VALID_KEYS = {API_KEY_CLAUDE: "Claude", API_KEY_LECHAT: "LeChat"}
app = FastAPI(
title="Context Continuity API",
description="API REST pour la gestion du contexte multi-LLM via Trilium",
version="1.0.0",
docs_url="/api/docs",
redoc_url="/api/redoc",
openapi_url="/api/openapi.json",
)
# ---------------------------------------------------------------------------
# Auth
# ---------------------------------------------------------------------------
def get_llm(authorization: str = Header(...)) -> str:
"""Valide la clé API et retourne le nom du LLM appelant."""
key = authorization.replace("Bearer ", "").strip()
if key not in VALID_KEYS:
raise HTTPException(status_code=401, detail="Clé API invalide")
return VALID_KEYS[key]
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def strip_html(text: str) -> str:
text = re.sub(r"
", "\n", text)
text = re.sub(r"<[^>]+>", "", text)
return text.replace("&", "&").replace("<", "<").replace(">", ">").strip()
def note_to_dict(note: dict, with_content: bool = False) -> dict:
from trilium_api import get_note, get_note_content, get_label_value
nid = note.get("noteId", note.get("id", ""))
detail = get_note(nid)
attrs = detail.get("attributes", [])
labels = {a["name"]: a["value"] for a in attrs if a.get("type") == "label"}
result = {
"id": nid,
"title": detail.get("title", ""),
"labels": labels,
"created": detail.get("utcDateCreated", ""),
"modified": detail.get("utcDateModified", ""),
}
if with_content:
result["content"] = strip_html(get_note_content(nid))
return result
# ---------------------------------------------------------------------------
# Modèles Pydantic
# ---------------------------------------------------------------------------
class ProjetCreate(BaseModel):
nom: str = Field(..., description="Nom du projet (sans espaces)")
objectif: str = Field(..., description="Objectif en 1-2 phrases")
stack: Optional[str] = Field(None, description="Stack technique")
llm_repartition: Optional[str] = Field(None, description="Répartition LLMs")
class ProjetPatch(BaseModel):
statut: Optional[str] = Field(None, pattern="^(actif|en-pause|archive)$")
objectif: Optional[str] = None
stack: Optional[str] = None
class BacklogCreate(BaseModel):
projet: str
titre: str
priorite: str = Field("moyenne", pattern="^(haute|moyenne|basse)$")
class BacklogPatch(BaseModel):
statut: Optional[str] = Field(None, pattern="^(a faire|en cours|bloque|fait|abandonne)$")
priorite: Optional[str] = Field(None, pattern="^(haute|moyenne|basse)$")
class DecisionCreate(BaseModel):
projet: str
enonce: str
justification: Optional[str] = None
llm: Optional[str] = None
class DecisionPatch(BaseModel):
statut: Optional[str] = Field(None, pattern="^(active|revisee|annulee)$")
justification: Optional[str] = None
class HistoriqueCreate(BaseModel):
projet: str
enonce: str
type_historique: str = Field(..., alias="type",
pattern="^(Fait etabli|Test effectue|Hypothese invalidee|Contrainte decouverte)$")
detail: Optional[str] = None
model_config = {"populate_by_name": True}
class HistoriquePatch(BaseModel):
encore_valide: Optional[bool] = None
detail: Optional[str] = None
class ConversationCreate(BaseModel):
projet: str
titre: str
llm: str = Field(..., pattern="^(Claude Sonnet|Claude Opus|Le Chat Large|Le Chat Medium)$")
class ConversationPatch(BaseModel):
synthese_cloture: Optional[str] = Field(None, max_length=500)
statut: Optional[str] = Field(None, pattern="^(en-cours|clos)$")
class ConceptCreate(BaseModel):
projet: str
titre: str
definition: Optional[str] = None
source: Optional[str] = None
class ConceptPatch(BaseModel):
definition: Optional[str] = None
source: Optional[str] = None
class SkillCreate(BaseModel):
titre: str
contenu: str
portee: str = "projet" # "universel" | "reference-technique" | "projet" | autre libellé libre
projet: Optional[str] = None # optionnel si portee="universel"
class SkillPatch(BaseModel):
contenu: Optional[str] = None
portee: Optional[str] = None
# ---------------------------------------------------------------------------
# Routes — Health
# ---------------------------------------------------------------------------
@app.get("/api/health", tags=["Système"])
def health():
return {"status": "ok", "version": "1.0.0"}
# ---------------------------------------------------------------------------
# Routes — Projets
# ---------------------------------------------------------------------------
@app.get("/api/projets", tags=["Projets"])
def list_projets(llm: str = Depends(get_llm)):
from trilium_api import search_by_label
notes = search_by_label("type", "projet")
# Exclure la note racine Context Continuity
notes = [n for n in notes if n.get("title") != "Context Continuity"]
return [note_to_dict(n) for n in notes]
@app.get("/api/projets/{nom}", tags=["Projets"])
def get_projet(nom: str, llm: str = Depends(get_llm)):
from trilium_api import search_by_label
notes = search_by_label("projet", nom)
projets = [n for n in notes if n.get("title", "").replace(" ", "") == nom or
any(a.get("name") == "projet" and a.get("value") == nom
for a in n.get("attributes", []))]
# Chercher la note projet spécifique
from trilium_api import search_notes, get_note, get_label_value
all_projets = search_by_label("type", "projet")
for p in all_projets:
if get_label_value(p["noteId"], "projet") == nom:
return note_to_dict(p, with_content=True)
raise HTTPException(status_code=404, detail=f"Projet '{nom}' introuvable")
@app.post("/api/projets", tags=["Projets"], status_code=201)
def create_projet(body: ProjetCreate, llm: str = Depends(get_llm)):
import json
from trilium_api import create_note, get_note_id, set_label, update_note_content
with open(os.path.expanduser("~/App/Context_continuity/trilium_ids.json")) as f:
ids = json.load(f)
res = create_note(ids["Projets"], body.nom)
nid = get_note_id(res)
set_label(nid, "type", "projet")
set_label(nid, "projet", body.nom)
set_label(nid, "statut", "actif")
content = f"
Objectif : {body.objectif}
" if body.stack: content += f"Stack : {body.stack}
" if body.llm_repartition: content += f"LLM : {body.llm_repartition}
" update_note_content(nid, content) return {"id": nid, "nom": body.nom, "created_by": llm} @app.patch("/api/projets/{note_id}", tags=["Projets"]) def patch_projet(note_id: str, body: ProjetPatch, llm: str = Depends(get_llm)): from trilium_api import get_note, set_label, update_note_content, get_note_content try: get_note(note_id) except Exception: raise HTTPException(status_code=404, detail="Projet introuvable") if body.statut: set_label(note_id, "statut", body.statut) if body.objectif or body.stack: content = get_note_content(note_id) if body.objectif: content += f"Objectif mis à jour : {body.objectif}
" if body.stack: content += f"Stack : {body.stack}
" update_note_content(note_id, content) return {"id": note_id, "updated_by": llm} # --------------------------------------------------------------------------- # Routes — Backlog # --------------------------------------------------------------------------- @app.get("/api/backlog/{projet}", tags=["Backlog"]) def list_backlog(projet: str, llm: str = Depends(get_llm)): from trilium_api import search_by_label, get_label_value notes = search_by_label("type", "backlogItem") actifs = [n for n in notes if get_label_value(n["noteId"], "projet") == projet and get_label_value(n["noteId"], "statut") not in ("fait", "abandonne")] prio_ordre = {"haute": 0, "moyenne": 1, "basse": 2} actifs.sort(key=lambda n: prio_ordre.get( get_label_value(n["noteId"], "priorite") or "basse", 99)) return [note_to_dict(n) for n in actifs] @app.post("/api/backlog", tags=["Backlog"], status_code=201) def create_backlog(body: BacklogCreate, llm: str = Depends(get_llm)): import json from trilium_api import create_note, get_note_id, set_label with open(os.path.expanduser("~/App/Context_continuity/trilium_ids.json")) as f: ids = json.load(f) res = create_note(ids["Backlog"], body.titre) nid = get_note_id(res) set_label(nid, "type", "backlogItem") set_label(nid, "projet", body.projet) set_label(nid, "priorite", body.priorite) set_label(nid, "statut", "a faire") return {"id": nid, "titre": body.titre, "created_by": llm} @app.patch("/api/backlog/{note_id}", tags=["Backlog"]) def patch_backlog(note_id: str, body: BacklogPatch, llm: str = Depends(get_llm)): from trilium_api import get_note, set_label try: get_note(note_id) except Exception: raise HTTPException(status_code=404, detail="Backlog item introuvable") if body.statut: set_label(note_id, "statut", body.statut) if body.priorite: set_label(note_id, "priorite", body.priorite) return {"id": note_id, "updated_by": llm} # --------------------------------------------------------------------------- # Routes — Décisions # --------------------------------------------------------------------------- @app.get("/api/decisions/{projet}", tags=["Décisions"]) def list_decisions(projet: str, llm: str = Depends(get_llm)): from trilium_api import search_by_label, get_label_value notes = search_by_label("type", "decision") actives = [n for n in notes if get_label_value(n["noteId"], "projet") == projet and get_label_value(n["noteId"], "statut") == "active"] return [note_to_dict(n) for n in actives] @app.post("/api/decisions", tags=["Décisions"], status_code=201) def create_decision(body: DecisionCreate, llm: str = Depends(get_llm)): import json from trilium_api import create_note, get_note_id, set_label with open(os.path.expanduser("~/App/Context_continuity/trilium_ids.json")) as f: ids = json.load(f) res = create_note(ids["Decisions"], body.enonce, content=f"Justification : {body.justification or '-'}
") nid = get_note_id(res) set_label(nid, "type", "decision") set_label(nid, "projet", body.projet) set_label(nid, "statut", "active") set_label(nid, "llm", body.llm or llm) return {"id": nid, "enonce": body.enonce, "created_by": llm} @app.patch("/api/decisions/{note_id}", tags=["Décisions"]) def patch_decision(note_id: str, body: DecisionPatch, llm: str = Depends(get_llm)): from trilium_api import get_note, set_label, get_note_content, update_note_content try: get_note(note_id) except Exception: raise HTTPException(status_code=404, detail="Décision introuvable") if body.statut: set_label(note_id, "statut", body.statut) if body.justification: content = get_note_content(note_id) content += f"Révision ({llm}) : {body.justification}
" update_note_content(note_id, content) return {"id": note_id, "updated_by": llm} # --------------------------------------------------------------------------- # Routes — Historique # --------------------------------------------------------------------------- @app.get("/api/historique/{projet}", tags=["Historique"]) def list_historique(projet: str, llm: str = Depends(get_llm)): from trilium_api import search_by_label, get_label_value notes = search_by_label("type", "historiqueItem") valides = [n for n in notes if get_label_value(n["noteId"], "projet") == projet and get_label_value(n["noteId"], "encoreValide") == "true"] return [note_to_dict(n) for n in valides] @app.post("/api/historique", tags=["Historique"], status_code=201) def create_historique(body: HistoriqueCreate, llm: str = Depends(get_llm)): import json from trilium_api import create_note, get_note_id, set_label with open(os.path.expanduser("~/App/Context_continuity/trilium_ids.json")) as f: ids = json.load(f) type_h = body.type_historique res = create_note(ids["Historique"], body.enonce, content=f"Type : {type_h}
" f"Détail : {body.detail or '-'}
") nid = get_note_id(res) set_label(nid, "type", "historiqueItem") set_label(nid, "projet", body.projet) set_label(nid, "typeHistorique", type_h) set_label(nid, "encoreValide", "true") return {"id": nid, "enonce": body.enonce, "created_by": llm} @app.patch("/api/historique/{note_id}", tags=["Historique"]) def patch_historique(note_id: str, body: HistoriquePatch, llm: str = Depends(get_llm)): from trilium_api import get_note, set_label, get_note_content, update_note_content try: get_note(note_id) except Exception: raise HTTPException(status_code=404, detail="Entrée historique introuvable") if body.encore_valide is not None: set_label(note_id, "encoreValide", "true" if body.encore_valide else "false") if body.detail: content = get_note_content(note_id) content += f"Mise à jour ({llm}) : {body.detail}
" update_note_content(note_id, content) return {"id": note_id, "updated_by": llm} # --------------------------------------------------------------------------- # Routes — Conversations # --------------------------------------------------------------------------- @app.get("/api/conversations/{projet}", tags=["Conversations"]) def list_conversations(projet: str, llm: str = Depends(get_llm)): from trilium_api import search_by_label, get_label_value notes = search_by_label("type", "conversation") conv = [n for n in notes if get_label_value(n["noteId"], "projet") == projet] return [note_to_dict(n) for n in conv] @app.get("/api/conversations/{projet}/{note_id}", tags=["Conversations"]) def get_conversation(projet: str, note_id: str, llm: str = Depends(get_llm)): from trilium_api import get_note, get_label_value try: note = get_note(note_id) except Exception: raise HTTPException(status_code=404, detail="Conversation introuvable") return note_to_dict({"noteId": note_id}, with_content=True) @app.post("/api/conversations", tags=["Conversations"], status_code=201) def create_conversation(body: ConversationCreate, llm: str = Depends(get_llm)): import json from trilium_api import create_note, get_note_id, set_label with open(os.path.expanduser("~/App/Context_continuity/trilium_ids.json")) as f: ids = json.load(f) date = datetime.now().strftime("%Y-%m-%d %H:%M") titre = f"[{body.llm}] {date} — {body.titre}" content = (f"| Projet | {body.projet} |
| LLM | {body.llm} |
| Date | {date} |
| Synthèse de clôture | " f"À remplir en fin de session |
{body.synthese_cloture}
" update_note_content(note_id, content) set_label(note_id, "syntheseCloture", body.synthese_cloture[:200]) set_label(note_id, "statut", "clos") if body.statut: set_label(note_id, "statut", body.statut) return {"id": note_id, "updated_by": llm} # --------------------------------------------------------------------------- # Routes — Concepts # --------------------------------------------------------------------------- @app.get("/api/concepts/{projet}", tags=["Concepts"]) def list_concepts(projet: str, llm: str = Depends(get_llm)): from trilium_api import search_by_label, get_label_value notes = search_by_label("type", "concept") concepts = [n for n in notes if get_label_value(n["noteId"], "projet") == projet] return [note_to_dict(n, with_content=True) for n in concepts] @app.post("/api/concepts", tags=["Concepts"], status_code=201) def create_concept(body: ConceptCreate, llm: str = Depends(get_llm)): import json from trilium_api import create_note, get_note_id, set_label with open(os.path.expanduser("~/App/Context_continuity/trilium_ids.json")) as f: ids = json.load(f) content = f"Définition : {body.definition}
" if body.definition else " " res = create_note(ids["Concepts"], body.titre, content=content) nid = get_note_id(res) set_label(nid, "type", "concept") set_label(nid, "projet", body.projet) if body.definition: set_label(nid, "definition", body.definition) if body.source: set_label(nid, "source", body.source) return {"id": nid, "titre": body.titre, "created_by": llm} @app.patch("/api/concepts/{note_id}", tags=["Concepts"]) def patch_concept(note_id: str, body: ConceptPatch, llm: str = Depends(get_llm)): from trilium_api import get_note, set_label, get_note_content, update_note_content try: get_note(note_id) except Exception: raise HTTPException(status_code=404, detail="Concept introuvable") if body.definition: set_label(note_id, "definition", body.definition) content = get_note_content(note_id) content += f"Définition mise à jour ({llm}) : {body.definition}
" update_note_content(note_id, content) if body.synonymes_eviter: content = get_note_content(note_id) content += f"Synonymes à éviter : {body.synonymes_eviter}
" update_note_content(note_id, content) return {"id": note_id, "updated_by": llm} # --------------------------------------------------------------------------- # Routes — Contexte Reprise # --------------------------------------------------------------------------- @app.get("/api/contexte/{projet}", tags=["Contexte Reprise"]) def get_contexte(projet: str, llm_cible: str = "LLM", llm: str = Depends(get_llm)): """Génère et retourne le briefing de reprise en texte.""" from trilium_api import (search_by_label, get_label_value, get_note_content, create_note, get_note_id, set_label, get_note) import json, re as re2 decisions = search_by_label("type", "decision") decisions = [d for d in decisions if get_label_value(d["noteId"], "projet") == projet and get_label_value(d["noteId"], "statut") == "active"] historique = search_by_label("type", "historiqueItem") historique = [h for h in historique if get_label_value(h["noteId"], "projet") == projet and get_label_value(h["noteId"], "encoreValide") == "true"] backlog = search_by_label("type", "backlogItem") backlog = [b for b in backlog if get_label_value(b["noteId"], "projet") == projet and get_label_value(b["noteId"], "statut") not in ("fait", "abandonne")] glossaire = search_by_label("type", "concept") glossaire = [g for g in glossaire if get_label_value(g["noteId"], "projet") == projet and get_label_value(g["noteId"], "definition")] date = datetime.now().strftime("%d/%m/%Y") lines = [ f"# REPRISE DE CONTEXTE — {projet} — {date}", f"**LLM cible : {llm_cible}**", "", ] if decisions: lines += ["## Décisions actives (ne pas remettre en question)"] for d in decisions: lines.append(f"- {d.get('title', '?')}") lines.append("") if historique: lines += ["## Historique — déjà testé / établi"] for h in historique: type_h = get_label_value(h["noteId"], "typeHistorique") or "" lines.append(f"- [{type_h}] {h.get('title', '?')}") lines.append("") if glossaire: lines += ["## Concepts"] for g in glossaire: defn = get_label_value(g["noteId"], "definition") or "(voir note)" lines.append(f"- **{g.get('title','?')}** : {defn}") lines.append("") if backlog: lines += ["## Backlog actif"] prio_ordre = {"haute": 0, "moyenne": 1, "basse": 2} backlog.sort(key=lambda b: prio_ordre.get( get_label_value(b["noteId"], "priorite") or "basse", 99)) for b in backlog: prio = get_label_value(b["noteId"], "priorite") or "?" statut = get_label_value(b["noteId"], "statut") or "?" lines.append(f"- [{statut}] [{prio}] {b.get('title','?')}") lines.append("") lines += [ "---", "Confirme en 3 lignes : (a) objectif (b) prochaine action (c) incertitude", ] briefing = "\n".join(lines) tokens = max(1, int(len(briefing) / 4)) return {"projet": projet, "tokens": tokens, "briefing": briefing} @app.get("/api/contexte/{projet}/last", tags=["Contexte Reprise"]) def get_last_contexte(projet: str, llm: str = Depends(get_llm)): """Retourne le dernier contexte de reprise sauvegardé.""" from trilium_api import search_by_label, get_label_value, get_note_content notes = search_by_label("type", "contexteReprise") notes = [n for n in notes if get_label_value(n["noteId"], "projet") == projet] if not notes: raise HTTPException(status_code=404, detail=f"Aucun contexte de reprise pour '{projet}'") notes.sort(key=lambda n: int(get_label_value(n["noteId"], "version") or 0), reverse=True) derniere = notes[0] nid = derniere["noteId"] content = get_note_content(nid) texte = re.sub(r"{body.contenu}"
res = create_note(ids["Skills"], body.titre, content=content)
nid = get_note_id(res)
set_label(nid, "type", "skill")
set_label(nid, "portee", body.portee)
if body.projet:
set_label(nid, "projet", body.projet)
return {"id": nid, "titre": body.titre, "created_by": llm}
@app.patch("/api/skills/{note_id}", tags=["Skills"])
def patch_skill(note_id: str, body: SkillPatch, llm: str = Depends(get_llm)):
"""Met à jour le contenu et/ou la portee d'un skill existant."""
from trilium_api import get_note, set_label, update_note_content
try:
get_note(note_id)
except Exception:
raise HTTPException(status_code=404, detail="Skill introuvable")
if body.contenu is not None:
update_note_content(note_id, f"{body.contenu}")
if body.portee is not None:
set_label(note_id, "portee", body.portee)
return {"id": note_id, "updated_by": llm}
@app.delete("/api/notes/{note_id}", tags=["Système"])
def delete_note_endpoint(note_id: str, llm: str = Depends(get_llm)):
"""Supprime une note du systeme. Garde-fou : refuse les notes hors systeme."""
from trilium_api import delete_note_safe
ok, message = delete_note_safe(note_id)
if not ok:
raise HTTPException(status_code=400, detail=message)
return {"deleted": True, "message": message, "by": llm}
class MoveNoteBody(BaseModel):
ancien_parent_id: str
nouveau_parent_id: str
@app.post("/api/notes/{note_id}/move", tags=["Système"])
def move_note_endpoint(note_id: str, body: MoveNoteBody, llm: str = Depends(get_llm)):
"""Deplace une note d un parent vers un autre. Gere les clones."""
from trilium_api import move_note_safe
ok, message = move_note_safe(note_id, body.ancien_parent_id, body.nouveau_parent_id)
if not ok:
raise HTTPException(status_code=400, detail=message)
return {"moved": True, "message": message, "by": llm}
class AddRelationBody(BaseModel):
nom: str
cible_id: str
@app.post("/api/notes/{source_id}/relation", tags=["Système"])
def add_relation_endpoint(source_id: str, body: AddRelationBody, llm: str = Depends(get_llm)):
"""Cree une relation typee entre deux notes. Garde-fou : nom dans l ontologie."""
from trilium_api import add_relation_safe
ok, message = add_relation_safe(source_id, body.nom, body.cible_id)
if not ok:
raise HTTPException(status_code=400, detail=message)
return {"created": True, "message": message, "by": llm}