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="可选:指定用于知识提炼的大模型名称;留空使用默认配置")


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"
    model_name: 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"


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


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,
        )
    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,
        )
    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),
    }


@router.post("/ingest/threads/batch")
@require_auth
async def ingest_threads_batch(body: BatchIngestRequest, request: Request) -> dict[str, Any]:
    """Batch-sediment the caller's recent threads into knowledge notes."""
    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)
    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=get_effective_user_id(),
                skip_existing=body.skip_existing,
                model_name=body.model_name,
            )
            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 _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),
) -> 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),
        )
    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,
        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,
    source_type: str | None = None,
    status: str | None = None,
    limit: int = 20,
    offset: int = 0,
) -> dict[str, Any]:
    service = get_knowledge_service(request)
    limit = max(1, min(limit, 100))
    offset = max(0, offset)
    return await service.list_notes(keyword=keyword, tag=tag, source_type=source_type, status=status, 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)
    note = await service.update_note(
        note_id,
        title=body.title,
        content_md=body.content_md,
        summary=body.summary,
        tags=body.tags,
        status=body.status,
        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()


@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)