deerflow-code/offline-backend-20260512/backend/app/gateway/routers/knowledge.py

"""Knowledge base API (``/api/knowledge``).

Sediments threads / searches into an obsidian-wiki compatible vault, mirrors
metadata into the database, and serves listing, detail, edit, delete, search
and graph export. The knowledge base is **global / shared** — no per-user
isolation; writes record ``created_by`` / ``updated_by`` for audit.

All obsidian-wiki specifics live in :mod:`deerflow.knowledge`; this router only
adapts HTTP ↔ the :class:`~deerflow.knowledge.service.KnowledgeService`.
"""

from __future__ import annotations

import logging
import mimetypes
import re
import tempfile
from pathlib import Path
from typing import Any

from fastapi import APIRouter, BackgroundTasks, File, Form, HTTPException, Request, UploadFile
from fastapi.responses import FileResponse
from pydantic import BaseModel, Field

from app.gateway.authz import require_auth, require_permission
from app.gateway.deps import get_checkpointer, get_knowledge_service, get_local_provider, get_thread_store
from app.gateway.utils import sanitize_log_param
from deerflow.knowledge.service import EmptyThreadError
from deerflow.runtime import serialize_channel_values
from deerflow.runtime.references import build_reference_batches
from deerflow.runtime.user_context import get_effective_user_id
from deerflow.utils.file_conversion import CONVERTIBLE_EXTENSIONS, convert_file_to_markdown

# Plain-text/markdown files are read directly; binary documents are converted.
_TEXT_EXTENSIONS = {".md", ".markdown", ".txt", ".text", ""}
# Bound the in-memory read so a knowledge import can't OOM the gateway.
_MAX_IMPORT_BYTES = 25 * 1024 * 1024

logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api/knowledge", tags=["knowledge"])


# --------------------------------------------------------------------------- #
# Request / response models
# --------------------------------------------------------------------------- #
class IngestThreadRequest(BaseModel):
    mode: str = Field(default="summary", description="summary | full")
    title: str | None = None
    status: str = Field(default="approved", description="draft | approved")
    tags: list[str] = Field(default_factory=list)
    include_sources: bool = True
    background: bool = Field(
        default=False,
        description="When true, sediment the thread asynchronously in a background task and return immediately.",
    )
    model_name: str | None = Field(default=None, description="可选:指定用于知识提炼的大模型名称;留空使用默认配置")
    template_id: str | None = Field(default=None, description="可选:抽取提示词模板 id;留空使用对话默认模板")
    folder: str | None = Field(default=None, description="可选:沉淀结果归入的目录(斜杠分隔路径)")


class BatchIngestRequest(BaseModel):
    limit: int = Field(default=100, ge=1, le=500)
    offset: int = Field(default=0, ge=0)
    status: str = "approved"
    include_sources: bool = True
    skip_existing: bool = True
    mode: str = "summary"
    background: bool = Field(
        default=True,
        description="When true (default), the per-thread extraction loop runs in a background task and the endpoint returns immediately ({queued:true}). Batch capture of many threads is far too slow to finish inside one HTTP request — running it inline makes nginx time out (504), which the frontend surfaces as a generic 'request failed'.",
    )
    model_name: str | None = Field(default=None, description="可选:指定用于知识提炼的大模型名称;留空使用默认配置")
    template_id: str | None = Field(default=None, description="可选:抽取提示词模板 id;留空使用对话默认模板")
    folder: str | None = Field(default=None, description="可选:沉淀结果归入的目录(斜杠分隔路径)")


class SearchResultItem(BaseModel):
    title: str | None = None
    url: str | None = None
    snippet: str | None = None
    tool_name: str | None = None
    score: float | None = None


class IngestSearchRequest(BaseModel):
    query: str
    results: list[SearchResultItem] = Field(default_factory=list)
    thread_id: str | None = None
    message_id: str | None = None
    title: str | None = None
    tags: list[str] = Field(default_factory=list)
    status: str = "approved"


class CreateNoteRequest(BaseModel):
    title: str
    content_md: str
    summary: str | None = None
    tags: list[str] = Field(default_factory=list)
    status: str = "approved"
    folder: str | None = None


class UpdateNoteRequest(BaseModel):
    title: str | None = None
    content_md: str | None = None
    summary: str | None = None
    tags: list[str] | None = None
    status: str | None = None
    # Present (even as null) ⇒ (re)assign the note's directory; absent ⇒ leave as-is.
    folder: str | None = None


class CreateFolderRequest(BaseModel):
    path: str = Field(description="目录全路径,斜杠分隔,如 投研/行业")


class RenameFolderRequest(BaseModel):
    path: str = Field(description="新的目录全路径")


class ExtractTemplateRequest(BaseModel):
    name: str
    system_prompt: str
    scope: str = Field(default="both", description="thread | document | both")
    description: str | None = None
    is_default: bool = False
    enabled: bool = True


class UpdateExtractTemplateRequest(BaseModel):
    name: str | None = None
    system_prompt: str | None = None
    scope: str | None = None
    description: str | None = None
    is_default: bool | None = None
    enabled: bool | None = None


class SearchRequest(BaseModel):
    query: str
    mode: str = Field(default="keyword", description="keyword | vector | hybrid (vector/hybrid fall back to keyword in v1)")
    limit: int = Field(default=8, ge=1, le=50)


class ResolveWikilinksRequest(BaseModel):
    targets: list[str] = Field(default_factory=list, description="`[[wikilink]]` targets to resolve to note ids / vault pages.")


# --------------------------------------------------------------------------- #
# Helpers
# --------------------------------------------------------------------------- #
async def _read_thread_values(checkpointer, thread_id: str) -> dict[str, Any]:
    """Return serialized channel values for a thread, or raise 404/500."""
    config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
    try:
        checkpoint_tuple = await checkpointer.aget_tuple(config)
    except Exception:
        logger.exception("Failed to read thread %s for knowledge ingest", sanitize_log_param(thread_id))
        raise HTTPException(status_code=500, detail="Failed to read thread state")
    if checkpoint_tuple is None:
        raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
    checkpoint = getattr(checkpoint_tuple, "checkpoint", {}) or {}
    return serialize_channel_values(checkpoint.get("channel_values", {}) or {})


async def _enrich_actor_names(note: dict[str, Any]) -> None:
    """Resolve created_by/updated_by user ids to a human label (email).

    Adds ``created_by_name`` / ``updated_by_name`` to the note dict so the UI can
    show a person instead of a raw UUID. Best-effort: unresolved ids fall back to
    ``None`` and the frontend then shows a shortened id.
    """
    provider = get_local_provider()
    cache: dict[str, str | None] = {}

    async def name_of(uid: str | None) -> str | None:
        if not uid:
            return None
        if uid not in cache:
            try:
                user = await provider.get_user(uid)
                cache[uid] = user.email if user else None
            except Exception:
                cache[uid] = None
        return cache[uid]

    note["created_by_name"] = await name_of(note.get("created_by"))
    note["updated_by_name"] = await name_of(note.get("updated_by"))


def _messages_and_meta(values: dict[str, Any]) -> tuple[list[dict[str, Any]], str | None, list[dict[str, Any]]]:
    messages = values.get("messages")
    if not isinstance(messages, list):
        messages = []
    title = values.get("title") if isinstance(values.get("title"), str) else None
    batches = build_reference_batches(messages) if messages else []
    return messages, title, batches


async def _ingest_thread_background(service, checkpointer, thread_id: str, body: IngestThreadRequest, created_by: str | None) -> None:
    """Sediment a thread into a note without an HTTP context.

    Runs after the response is sent (FastAPI ``BackgroundTasks``); ``created_by``
    must be captured by the request handler since the user contextvar is gone
    by the time this executes. All failures are swallowed (logged) — the user
    has already been told the save was queued.
    """
    try:
        values = await _read_thread_values(checkpointer, thread_id)
        messages, title, batches = _messages_and_meta(values)
        if not messages:
            logger.warning("Background knowledge ingest skipped: thread %s has no messages", sanitize_log_param(thread_id))
            return
        await service.capture_thread(
            thread_id=thread_id,
            messages=messages,
            thread_title=title,
            mode=body.mode,
            title=body.title,
            status=body.status,
            tags=body.tags,
            include_sources=body.include_sources,
            reference_batches=batches,
            created_by=created_by,
            model_name=body.model_name,
            template_id=body.template_id,
            folder=body.folder,
        )
    except EmptyThreadError:
        logger.warning("Background knowledge ingest skipped: thread %s has no sediment-able content", sanitize_log_param(thread_id))
    except Exception:
        logger.exception("Background knowledge ingest failed for thread %s", sanitize_log_param(thread_id))


# --------------------------------------------------------------------------- #
# Ingest
# --------------------------------------------------------------------------- #
@router.post("/ingest/thread/{thread_id}")
@require_auth
@require_permission("threads", "read", owner_check=True)
async def ingest_thread(thread_id: str, body: IngestThreadRequest, request: Request, background_tasks: BackgroundTasks) -> dict[str, Any]:
    """Sediment a single thread into a knowledge note.

    With ``background=true`` the heavy capture (read checkpoint → extract →
    write vault + DB) is deferred to a background task and the endpoint returns
    ``{"queued": true}`` immediately, so the user's "保存" click never blocks.
    """
    service = get_knowledge_service(request)
    checkpointer = get_checkpointer(request)
    if body.background:
        # Capture the user id now — the contextvar is gone once the task runs.
        background_tasks.add_task(
            _ingest_thread_background,
            service,
            checkpointer,
            thread_id,
            body,
            get_effective_user_id(),
        )
        return {"queued": True}
    values = await _read_thread_values(checkpointer, thread_id)
    messages, title, batches = _messages_and_meta(values)
    if not messages:
        raise HTTPException(status_code=400, detail="Thread has no messages to sediment")
    try:
        result = await service.capture_thread(
            thread_id=thread_id,
            messages=messages,
            thread_title=title,
            mode=body.mode,
            title=body.title,
            status=body.status,
            tags=body.tags,
            include_sources=body.include_sources,
            reference_batches=batches,
            created_by=get_effective_user_id(),
            model_name=body.model_name,
            template_id=body.template_id,
            folder=body.folder,
        )
    except EmptyThreadError:
        raise HTTPException(status_code=400, detail="Thread has no sediment-able content")
    return {
        "note": result["note"],
        "created": result["created"],
        "notes": result.get("notes", [result["note"]]),
        "llm_fallback": result.get("llm_fallback", False),
    }


async def _run_batch_ingest(service, checkpointer, threads: list[dict[str, Any]], body: BatchIngestRequest, created_by: str | None) -> dict[str, Any]:
    """Run the per-thread sediment loop and tally the outcome.

    Shared by the inline and background paths; ``created_by`` is passed in
    explicitly because the user contextvar is gone once a background task runs.
    Each thread's transcript is truncated by the extractor (``max_input_chars``)
    before the LLM call, so a single huge conversation can't blow the context.
    """
    created = skipped = failed = 0
    # Notes that landed via the rule-based fallback because the chosen model
    # produced nothing usable (timeout / API error / non-JSON). Surfaced so the
    # caller can tell the user "X 条因大模型异常改用了规则抽取".
    llm_fallback = 0
    items: list[dict[str, Any]] = []
    for meta in threads:
        thread_id = meta.get("thread_id")
        if not thread_id:
            continue
        try:
            values = await _read_thread_values(checkpointer, thread_id)
            messages, title, batches = _messages_and_meta(values)
            if not messages:
                skipped += 1
                items.append({"thread_id": thread_id, "note_id": None, "status": "skipped"})
                continue
            result = await service.capture_thread(
                thread_id=thread_id,
                messages=messages,
                thread_title=title or meta.get("display_name"),
                mode=body.mode,
                status=body.status,
                include_sources=body.include_sources,
                reference_batches=batches,
                created_by=created_by,
                skip_existing=body.skip_existing,
                model_name=body.model_name,
                template_id=body.template_id,
                folder=body.folder,
            )
            if result.get("llm_fallback"):
                llm_fallback += 1
            if result["created"]:
                created += 1
                items.append({"thread_id": thread_id, "note_id": result["note"]["id"], "status": "created", "llm_fallback": bool(result.get("llm_fallback"))})
            else:
                skipped += 1
                items.append({"thread_id": thread_id, "note_id": result["note"]["id"], "status": "skipped"})
        except EmptyThreadError:
            skipped += 1
            items.append({"thread_id": thread_id, "note_id": None, "status": "skipped"})
        except HTTPException:
            failed += 1
            items.append({"thread_id": thread_id, "note_id": None, "status": "failed"})
        except Exception:
            logger.exception("Batch ingest failed for thread %s", sanitize_log_param(str(thread_id)))
            failed += 1
            items.append({"thread_id": thread_id, "note_id": None, "status": "failed"})

    return {"ok": True, "processed": len(items), "created": created, "skipped": skipped, "failed": failed, "llm_fallback": llm_fallback, "items": items}


async def _ingest_threads_batch_background(service, checkpointer, threads: list[dict[str, Any]], body: BatchIngestRequest, created_by: str | None) -> None:
    """Background variant of the batch loop (failures swallowed/logged)."""
    try:
        result = await _run_batch_ingest(service, checkpointer, threads, body, created_by)
        logger.info(
            "Background batch ingest finished: processed=%d created=%d skipped=%d failed=%d llm_fallback=%d",
            result["processed"], result["created"], result["skipped"], result["failed"], result["llm_fallback"],
        )
    except Exception:
        logger.exception("Background batch ingest failed")


@router.post("/ingest/threads/batch")
@require_auth
async def ingest_threads_batch(body: BatchIngestRequest, request: Request, background_tasks: BackgroundTasks) -> dict[str, Any]:
    """Batch-sediment the caller's recent threads into knowledge notes.

    Defaults to ``background=true``: the (slow) per-thread extraction runs in a
    background task and the endpoint returns ``{"queued": true, "processed": N}``
    immediately, so the request never outlives the nginx gateway timeout. Pass
    ``background=false`` to block until every thread is processed (small batches /
    scripts that want the full per-thread tally).
    """
    service = get_knowledge_service(request)
    checkpointer = get_checkpointer(request)
    thread_store = get_thread_store(request)

    threads = await thread_store.search(limit=body.limit, offset=body.offset)
    if body.background:
        # Capture the user id now — the contextvar is gone once the task runs.
        background_tasks.add_task(
            _ingest_threads_batch_background,
            service,
            checkpointer,
            threads,
            body,
            get_effective_user_id(),
        )
        return {"ok": True, "queued": True, "processed": len(threads), "created": 0, "skipped": 0, "failed": 0, "llm_fallback": 0, "items": []}

    return await _run_batch_ingest(service, checkpointer, threads, body, get_effective_user_id())


async def _extract_upload_text(file: UploadFile) -> tuple[str, str]:
    """Read an uploaded file and return ``(filename, markdown_text)``.

    Plain-text/markdown is decoded directly; PDF/Office documents are converted
    to Markdown via the shared converter. Raises ``HTTPException`` on empty /
    oversized / unsupported / unparseable files.
    """
    name = file.filename or "document"
    ext = Path(name).suffix.lower()
    raw = await file.read()
    if not raw:
        raise HTTPException(status_code=400, detail="文件为空")
    if len(raw) > _MAX_IMPORT_BYTES:
        raise HTTPException(status_code=413, detail=f"文件过大，最大支持 {_MAX_IMPORT_BYTES // (1024 * 1024)}MB")

    if ext in _TEXT_EXTENSIONS:
        text = raw.decode("utf-8", errors="ignore")
    elif ext in CONVERTIBLE_EXTENSIONS:
        # Convert in an isolated temp dir; a fixed safe name avoids path issues
        # while preserving the extension so the converter detects the type.
        with tempfile.TemporaryDirectory() as tmp:
            tmp_path = Path(tmp) / f"import{ext}"
            tmp_path.write_bytes(raw)
            md_path = await convert_file_to_markdown(tmp_path)
            if md_path is None:
                raise HTTPException(status_code=422, detail=f"无法解析文件：{name}")
            text = md_path.read_text(encoding="utf-8", errors="ignore")
    else:
        raise HTTPException(status_code=415, detail=f"不支持的文件类型：{ext or '未知'}（支持 md/txt/pdf/word/ppt/excel）")

    if not text.strip():
        raise HTTPException(status_code=422, detail="文件内容为空或无法提取文本")
    return name, text


@router.post("/ingest/file")
@require_auth
async def ingest_file(
    request: Request,
    file: UploadFile = File(...),
    title: str | None = Form(default=None),
    tags: str | None = Form(default=None),
    status: str = Form(default="approved"),
    model_name: str | None = Form(default=None),
    template_id: str | None = Form(default=None),
    folder: str | None = Form(default=None),
) -> dict[str, Any]:
    """Import an uploaded file into the knowledge base as wiki note(s).

    Converts the file to Markdown, distills it via the wiki-capture LLM pipeline
    (same as thread capture), and writes the resulting declarative note(s) into
    the shared vault. ``tags`` is a comma/space-separated string.
    """
    service = get_knowledge_service(request)
    name, text = await _extract_upload_text(file)
    tag_list = [t for t in re.split(r"[,，\s]+", tags or "") if t]
    try:
        result = await service.capture_file(
            content_md=text,
            filename=name,
            title=(title or None),
            tags=tag_list,
            status=status,
            created_by=get_effective_user_id(),
            model_name=(model_name or None),
            template_id=(template_id or None),
            folder=(folder or None),
        )
    except EmptyThreadError:
        raise HTTPException(status_code=422, detail="文件内容无可沉淀的知识")
    return {
        "note": result["note"],
        "created": result["created"],
        "notes": result.get("notes", [result["note"]]),
        "llm_fallback": result.get("llm_fallback", False),
    }


@router.post("/ingest/search")
@require_auth
async def ingest_search(body: IngestSearchRequest, request: Request) -> dict[str, Any]:
    """Sediment search/tool results as a reference note."""
    service = get_knowledge_service(request)
    note = await service.capture_search(
        query=body.query,
        results=[r.model_dump() for r in body.results],
        thread_id=body.thread_id,
        message_id=body.message_id,
        title=body.title,
        tags=body.tags,
        status=body.status,
        created_by=get_effective_user_id(),
    )
    return {"note": note}


# --------------------------------------------------------------------------- #
# CRUD
# --------------------------------------------------------------------------- #
@router.post("/notes")
@require_auth
async def create_note(body: CreateNoteRequest, request: Request) -> dict[str, Any]:
    service = get_knowledge_service(request)
    note = await service.create_manual_note(
        title=body.title,
        content_md=body.content_md,
        summary=body.summary,
        tags=body.tags,
        status=body.status,
        folder=body.folder,
        created_by=get_effective_user_id(),
    )
    return {"note": note}


@router.get("/notes")
@require_auth
async def list_notes(
    request: Request,
    keyword: str | None = None,
    tag: str | None = None,
    entity: str | None = None,
    source_type: str | None = None,
    status: str | None = None,
    folder: str | None = None,
    limit: int = 20,
    offset: int = 0,
) -> dict[str, Any]:
    service = get_knowledge_service(request)
    limit = max(1, min(limit, 1000))
    offset = max(0, offset)
    return await service.list_notes(keyword=keyword, tag=tag, entity=entity, source_type=source_type, status=status, folder=folder, limit=limit, offset=offset)


@router.get("/notes/{note_id}")
@require_auth
async def get_note(note_id: str, request: Request) -> dict[str, Any]:
    service = get_knowledge_service(request)
    note = await service.get_note(note_id)
    if note is None:
        raise HTTPException(status_code=404, detail="Knowledge note not found")
    await _enrich_actor_names(note)
    return note


@router.put("/notes/{note_id}")
@require_auth
async def update_note(note_id: str, body: UpdateNoteRequest, request: Request) -> dict[str, Any]:
    service = get_knowledge_service(request)
    # ``folder`` present in the request body (even as null) ⇒ (re)assign the
    # directory; absent ⇒ leave it untouched.
    folder_provided = "folder" in body.model_fields_set
    note = await service.update_note(
        note_id,
        title=body.title,
        content_md=body.content_md,
        summary=body.summary,
        tags=body.tags,
        status=body.status,
        folder=body.folder,
        set_folder=folder_provided,
        updated_by=get_effective_user_id(),
    )
    if note is None:
        raise HTTPException(status_code=404, detail="Knowledge note not found")
    return note


@router.delete("/notes/{note_id}")
@require_auth
async def delete_note(note_id: str, request: Request, hard: bool = False) -> dict[str, Any]:
    service = get_knowledge_service(request)
    ok = await service.delete_note(note_id, updated_by=get_effective_user_id(), hard=hard)
    if not ok:
        raise HTTPException(status_code=404, detail="Knowledge note not found")
    return {"ok": True, "note_id": note_id}


# --------------------------------------------------------------------------- #
# Search & graph
# --------------------------------------------------------------------------- #
@router.post("/search")
@require_auth
async def search(body: SearchRequest, request: Request) -> dict[str, Any]:
    service = get_knowledge_service(request)
    items = await service.search(body.query, mode=body.mode, limit=body.limit)
    return {"items": items}


# --------------------------------------------------------------------------- #
# Wikilinks ([[...]] resolution + vault meta-page viewer)
# --------------------------------------------------------------------------- #
@router.post("/resolve")
@require_auth
async def resolve_wikilinks(body: ResolveWikilinksRequest, request: Request) -> dict[str, Any]:
    """Resolve ``[[wikilink]]`` targets to navigable destinations.

    Each target resolves to a knowledge note (``kind=note``), a vault meta/hub
    page (``kind=vault``), or nothing (``kind=missing``). Powers clickable
    wikilinks in the knowledge UI.
    """
    service = get_knowledge_service(request)
    return {"resolved": await service.resolve_wikilinks(body.targets)}


@router.get("/notes/{note_id}/wikilinks")
@require_auth
async def resolve_note_wikilinks(note_id: str, request: Request) -> dict[str, Any]:
    """Resolve every ``[[wikilink]]`` found in a note's body in one call."""
    service = get_knowledge_service(request)
    resolved = await service.resolve_note_wikilinks(note_id)
    if resolved is None:
        raise HTTPException(status_code=404, detail="Knowledge note not found")
    return {"resolved": resolved}


@router.get("/vault-page")
@require_auth
async def get_vault_page(request: Request, path: str) -> dict[str, Any]:
    """Return a raw vault Markdown meta/hub page (no DB note) by its path."""
    service = get_knowledge_service(request)
    page = service.read_vault_page(path)
    if page is None:
        raise HTTPException(status_code=404, detail="Vault page not found")
    return page


class MergeRequest(BaseModel):
    primary_id: str
    duplicate_ids: list[str] = Field(default_factory=list)


@router.get("/duplicates")
@require_auth
async def list_duplicates(request: Request) -> dict[str, Any]:
    """Return suggested duplicate groups (phase 3, heuristic)."""
    service = get_knowledge_service(request)
    return {"groups": await service.find_duplicates()}


@router.post("/merge")
@require_auth
async def merge_notes(body: MergeRequest, request: Request) -> dict[str, Any]:
    """Merge duplicate notes into a primary note (sources moved, dups archived)."""
    service = get_knowledge_service(request)
    note = await service.merge_notes(body.primary_id, body.duplicate_ids, updated_by=get_effective_user_id())
    if note is None:
        raise HTTPException(status_code=404, detail="Primary note not found")
    return {"note": note}


@router.post("/reindex")
@require_auth
async def reindex_embeddings(request: Request) -> dict[str, Any]:
    """Rebuild the vector index for all notes (maintenance; no-op if embeddings off)."""
    service = get_knowledge_service(request)
    return await service.reindex_all()


@router.get("/notes/{note_id}/versions")
@require_auth
async def list_note_versions(note_id: str, request: Request) -> dict[str, Any]:
    """Return the edit history (pre-update snapshots) for a note."""
    service = get_knowledge_service(request)
    return {"items": await service.list_versions(note_id)}


@router.get("/graph")
@require_auth
async def get_graph(request: Request) -> dict[str, Any]:
    """Return product knowledge-graph data (notes + entities + relations)."""
    service = get_knowledge_service(request)
    return await service.get_graph()


class GraphBlacklistAddRequest(BaseModel):
    labels: list[str] = Field(default_factory=list)


@router.get("/graph/blacklist")
@require_auth
async def list_graph_blacklist(request: Request) -> dict[str, Any]:
    """Return blacklisted graph node labels (hidden from the graph)."""
    service = get_knowledge_service(request)
    return {"items": await service.list_graph_blacklist()}


@router.post("/graph/blacklist")
@require_auth
async def add_graph_blacklist(body: GraphBlacklistAddRequest, request: Request) -> dict[str, Any]:
    """Blacklist node labels (entity names / note titles); they disappear from the graph."""
    service = get_knowledge_service(request)
    added = await service.add_graph_blacklist(body.labels, created_by=get_effective_user_id())
    return {"added": added, "items": await service.list_graph_blacklist()}


@router.delete("/graph/blacklist/{entry_id}")
@require_auth
async def remove_graph_blacklist(entry_id: str, request: Request) -> dict[str, Any]:
    """Remove a blacklist entry so the node shows up in the graph again."""
    service = get_knowledge_service(request)
    if not await service.remove_graph_blacklist(entry_id):
        raise HTTPException(status_code=404, detail="Blacklist entry not found")
    return {"ok": True}


@router.post("/export/graph")
@require_auth
async def export_graph(request: Request) -> dict[str, Any]:
    service = get_knowledge_service(request)
    result = await service.export_graph()
    base = "/api/knowledge/export/files"
    return {
        "ok": True,
        "files": {
            "graph_json": f"{base}/graph.json",
            "graph_html": f"{base}/graph.html",
            "graph_graphml": f"{base}/graph.graphml",
            "cypher": f"{base}/cypher.txt",
        },
        "stats": result["stats"],
    }


@router.get("/export/files/{filename}")
@require_auth
async def export_file(filename: str, request: Request) -> FileResponse:
    service = get_knowledge_service(request)
    path = service.export_file_path(filename)
    if path is None:
        # Lazily build the export on first access so the iframe never 404s.
        await service.export_graph()
        path = service.export_file_path(filename)
    if path is None:
        raise HTTPException(status_code=404, detail="Export file not found")
    media_type = mimetypes.guess_type(filename)[0] or "application/octet-stream"
    return FileResponse(path, media_type=media_type)


# --------------------------------------------------------------------------- #
# Folders (user-managed directories)
# --------------------------------------------------------------------------- #
@router.get("/folders")
@require_auth
async def list_folders(request: Request) -> dict[str, Any]:
    """List every managed knowledge directory (for the directory tree + picker)."""
    service = get_knowledge_service(request)
    return {"items": await service.list_folders()}


@router.post("/folders")
@require_auth
async def create_folder(body: CreateFolderRequest, request: Request) -> dict[str, Any]:
    """Create a directory (parents auto-created). Notes are filed into it later."""
    service = get_knowledge_service(request)
    try:
        folder = await service.create_folder(body.path, created_by=get_effective_user_id())
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    return {"folder": folder}


@router.put("/folders/{folder_id}")
@require_auth
async def rename_folder(folder_id: str, body: RenameFolderRequest, request: Request) -> dict[str, Any]:
    """Rename/move a directory; descendants + filed notes follow the new prefix."""
    service = get_knowledge_service(request)
    try:
        folder = await service.rename_folder(folder_id, body.path)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    if folder is None:
        raise HTTPException(status_code=404, detail="Folder not found")
    return {"folder": folder}


@router.delete("/folders/{folder_id}")
@require_auth
async def delete_folder(folder_id: str, request: Request, reassign_to: str | None = None) -> dict[str, Any]:
    """Delete a directory (and sub-dirs); notes inside move to ``reassign_to`` or unfiled."""
    service = get_knowledge_service(request)
    if not await service.delete_folder(folder_id, reassign_to=reassign_to):
        raise HTTPException(status_code=404, detail="Folder not found")
    return {"ok": True}


# --------------------------------------------------------------------------- #
# Extraction prompt templates
# --------------------------------------------------------------------------- #
@router.get("/extract-templates")
@require_auth
async def list_extract_templates(request: Request, scope: str | None = None, enabled_only: bool = False) -> dict[str, Any]:
    """List editable extraction-prompt templates (optionally filtered by scope)."""
    service = get_knowledge_service(request)
    return {"items": await service.list_extract_templates(scope=scope, enabled_only=enabled_only)}


@router.get("/extract-templates/{template_id}")
@require_auth
async def get_extract_template(template_id: str, request: Request) -> dict[str, Any]:
    service = get_knowledge_service(request)
    tpl = await service.get_extract_template(template_id)
    if tpl is None:
        raise HTTPException(status_code=404, detail="Template not found")
    return {"template": tpl}


@router.post("/extract-templates")
@require_auth
async def create_extract_template(body: ExtractTemplateRequest, request: Request) -> dict[str, Any]:
    service = get_knowledge_service(request)
    if await service.template_name_taken(body.name):
        raise HTTPException(status_code=409, detail="同名模板已存在")
    tpl = await service.create_extract_template(
        name=body.name,
        system_prompt=body.system_prompt,
        scope=body.scope,
        description=body.description,
        is_default=body.is_default,
        enabled=body.enabled,
        created_by=get_effective_user_id(),
    )
    return {"template": tpl}


@router.put("/extract-templates/{template_id}")
@require_auth
async def update_extract_template(template_id: str, body: UpdateExtractTemplateRequest, request: Request) -> dict[str, Any]:
    service = get_knowledge_service(request)
    if body.name is not None and await service.template_name_taken(body.name, exclude_id=template_id):
        raise HTTPException(status_code=409, detail="同名模板已存在")
    tpl = await service.update_extract_template(
        template_id,
        name=body.name,
        system_prompt=body.system_prompt,
        scope=body.scope,
        description=body.description,
        is_default=body.is_default,
        enabled=body.enabled,
        updated_by=get_effective_user_id(),
    )
    if tpl is None:
        raise HTTPException(status_code=404, detail="Template not found")
    return {"template": tpl}


@router.delete("/extract-templates/{template_id}")
@require_auth
async def delete_extract_template(template_id: str, request: Request) -> dict[str, Any]:
    service = get_knowledge_service(request)
    if not await service.delete_extract_template(template_id):
        raise HTTPException(status_code=404, detail="Template not found")
    return {"ok": True}