Hitonabi
152 lines
4.8 KiB
Python
152 lines
4.8 KiB
Python
"""Wissens-Vault-Reader: die nächtlichen Traum-Notizen (~/wissens-vault) als klickbares
|
|
Wiki in der Zentrale. Bewusst READ-ONLY — geschrieben wird der Vault nur vom Traum-Cron
|
|
(und via Auftragsbuch-Entscheidungen); hier wird nur gelesen und navigiert."""
|
|
|
|
import os
|
|
import re
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from pydantic import BaseModel
|
|
|
|
router = APIRouter(prefix="/api")
|
|
|
|
VAULT = Path(os.environ.get("MC_VAULT_DIR", "~/wissens-vault")).expanduser()
|
|
|
|
|
|
def _available() -> bool:
|
|
return VAULT.is_dir()
|
|
|
|
|
|
def _title_of(p: Path) -> str:
|
|
"""Erste nicht-leere Zeile (ohne Markdown-#) als Anzeigename."""
|
|
try:
|
|
with p.open(encoding="utf-8", errors="replace") as fh:
|
|
for line in fh:
|
|
s = line.strip()
|
|
if s:
|
|
return s.lstrip("# ").strip()[:160]
|
|
except OSError:
|
|
pass
|
|
return p.stem
|
|
|
|
|
|
class VaultFile(BaseModel):
|
|
path: str
|
|
name: str
|
|
dir: str
|
|
title: str
|
|
mtime: float
|
|
neu: bool
|
|
|
|
|
|
class WissenListResponse(BaseModel):
|
|
available: bool
|
|
files: list[VaultFile]
|
|
|
|
|
|
class WissenFileResponse(BaseModel):
|
|
path: str
|
|
content: str
|
|
mtime: float
|
|
|
|
|
|
@router.get("/wissen", response_model=WissenListResponse)
|
|
def list_vault() -> dict:
|
|
"""Alle Markdown-Notizen des Vaults (relativer Pfad, Titel, Ordner, Alter)."""
|
|
if not _available():
|
|
return {"available": False, "files": []}
|
|
files = []
|
|
now = time.time()
|
|
for p in sorted(VAULT.rglob("*.md")):
|
|
if ".git" in p.parts or "traeume" in p.parts:
|
|
continue
|
|
rel = p.relative_to(VAULT).as_posix()
|
|
st = p.stat()
|
|
files.append({
|
|
"path": rel,
|
|
"name": p.stem,
|
|
"dir": p.parent.relative_to(VAULT).as_posix() if p.parent != VAULT else "",
|
|
"title": _title_of(p),
|
|
"mtime": st.st_mtime,
|
|
"neu": (now - st.st_mtime) < 36 * 3600, # „neu seit gestern Nacht"
|
|
})
|
|
files.sort(key=lambda f: f["mtime"], reverse=True)
|
|
return {"available": True, "files": files}
|
|
|
|
|
|
@router.get("/wissen/datei", response_model=WissenFileResponse)
|
|
def read_file(pfad: str) -> dict:
|
|
"""Inhalt einer Vault-Notiz — Traversal hart geblockt (resolve + is_relative_to)."""
|
|
if not _available():
|
|
raise HTTPException(404, "Wissens-Vault liegt auf der Box (hier nicht verfügbar).")
|
|
try:
|
|
target = (VAULT / pfad).resolve()
|
|
if not target.is_relative_to(VAULT.resolve()) or target.suffix != ".md" or not target.is_file():
|
|
raise HTTPException(404, "Notiz nicht gefunden.")
|
|
return {"path": pfad, "content": target.read_text(encoding="utf-8", errors="replace")[:400_000],
|
|
"mtime": target.stat().st_mtime}
|
|
except HTTPException:
|
|
raise
|
|
except (ValueError, OSError):
|
|
raise HTTPException(404, "Notiz nicht lesbar.")
|
|
|
|
|
|
@router.get("/wissen/graph")
|
|
def graph_vault() -> dict:
|
|
"""Obsidian-artiger Graph des Vaults (Nodes = Dateien, Edges = Wiki-Links)."""
|
|
if not _available():
|
|
return {"nodes": [], "edges": []}
|
|
|
|
nodes = []
|
|
edges = []
|
|
# Für schnelle Link-Auflösung (case-insensitive Name -> relativer Pfad als ID)
|
|
name_to_id = {}
|
|
|
|
# 1. Alle Nodes sammeln
|
|
for p in VAULT.rglob("*.md"):
|
|
if ".git" in p.parts or "traeume" in p.parts:
|
|
continue
|
|
rel = p.relative_to(VAULT).as_posix()
|
|
name_to_id[p.stem.lower()] = rel
|
|
|
|
# Kategorie aus dem Ordner ableiten
|
|
cat = "knowledge"
|
|
parent = p.parent.name if p.parent != VAULT else ""
|
|
if parent == "traeume":
|
|
cat = "events"
|
|
elif parent == "muster":
|
|
cat = "rules"
|
|
elif parent == "skill-kandidaten":
|
|
cat = "identity"
|
|
|
|
nodes.append({
|
|
"id": rel,
|
|
"content": p.stem, # stem ist oft kürzer/prägnanter als der Titel
|
|
"category": cat,
|
|
"source": "wiki"
|
|
})
|
|
|
|
# 2. Edges extrahieren (Wiki-Links [[Name]])
|
|
link_rx = re.compile(r"\[\[([^\]]+)\]\]")
|
|
for n in nodes:
|
|
try:
|
|
target = VAULT / n["id"]
|
|
content = target.read_text(encoding="utf-8", errors="replace")
|
|
# Set, um mehrfache Links auf dieselbe Datei zu entduplizieren
|
|
found_links = {m.group(1).strip().lower() for m in link_rx.finditer(content)}
|
|
|
|
for link in found_links:
|
|
target_id = name_to_id.get(link)
|
|
if target_id and target_id != n["id"]:
|
|
edges.append({
|
|
"source": n["id"],
|
|
"target": target_id,
|
|
"weight": 1.0
|
|
})
|
|
except OSError:
|
|
pass
|
|
|
|
return {"nodes": nodes, "edges": edges}
|