"""Virtual Persona CRUD API — /v1/virtual-personas。"""

from __future__ import annotations

from pathlib import Path
from typing import TYPE_CHECKING, Any

from fastapi import APIRouter, File, Header, HTTPException, Request, Response, UploadFile
from pydantic import BaseModel, Field
from fastapi.responses import FileResponse

from niannian.http.core_auth import parse_user_role, require_baseline_owner
from niannian.soul.search_providers import build_search_provider_from_yaml
from niannian.soul.web_learning import WebLearningService
from niannian.token.budget import budget_allows, load_token_budget
from niannian.tenant.context import _session_header, resolve_request_context
from niannian.virtual_persona.assembly import build_character_assembly_block
from niannian.virtual_persona.enrich import enrich_virtual_persona
from niannian.virtual_persona.md_editor import MdStaleError, normalize_md_file, read_vp_md, write_vp_md
from niannian.virtual_persona.profile_md import merge_profile_sections
from niannian.virtual_persona.pack import patch_pack_fields, read_pack
from niannian.virtual_persona.pack import patch_pack_fields, read_pack
from niannian.virtual_persona.stage import StageTransitionError, transition_stage
from niannian.virtual_persona.store import VirtualPersonaStore, VirtualPersonaValidationError
from niannian.virtual_persona.sync import sync_virtual_persona_to_cloud

if TYPE_CHECKING:
    from niannian.client import NianNian

VP_READ_ROLES = frozenset({"owner", "authorized_corrector", "guest"})
VP_WRITE_ROLES = frozenset({"owner", "authorized_corrector"})


class CreateVirtualPersonaBody(BaseModel):
    id: str = Field(..., min_length=1, max_length=64)
    label: str = Field(..., min_length=1)


class PatchVirtualPersonaBody(BaseModel):
    profile_md: str | None = None
    profile_sections: dict[str, str] | None = None
    label: str | None = None
    storage_tier: str | None = None
    stage: str | None = None


class PatchVirtualPersonaPackBody(BaseModel):
    stage: str | None = None
    equipped_skills: list[str] | None = None


class EnrichVirtualPersonaBody(BaseModel):
    query: str | None = None
    urls: list[str] = Field(default_factory=list)
    max_results: int = Field(default=5, ge=1, le=10)
    mock: bool = False
    fetch_avatar: bool = True


class StickerSearchBody(BaseModel):
    query: str | None = None
    meme_keywords: list[str] | None = None
    max_results: int = Field(default=8, ge=1, le=16)
    mock: bool = False


class StickerImportFromUrlBody(BaseModel):
    url: str = Field(..., min_length=1)
    label: str | None = None
    slug: str | None = None
    triggers: list[str] = Field(default_factory=list)
    mock: bool = False


class StickerGenerateBody(BaseModel):
    prompts: list[str] = Field(default_factory=list)
    count: int = Field(default=4, ge=1, le=8)


class PerfectDistillExportOptions(BaseModel):
    include_full_l2: bool = False
    include_stickers: bool = False
    include_proactive_profile: bool = False
    include_deep_learn_runs: bool = False


class PerfectDistillBody(BaseModel):
    export_options: PerfectDistillExportOptions = Field(default_factory=PerfectDistillExportOptions)
    write: bool = False


class ImportSourceBody(BaseModel):
    url: str = Field(..., min_length=1)
    kind: str = Field(default="web")
    merge_into: str = Field(default="sources")
    respect_locked_sections: bool = True


class PutVpMdBody(BaseModel):
    content: str = ""


def _web_learning_yaml(nn: NianNian, *, mock: bool = False) -> dict:
    svc = WebLearningService(nn._subject_store, nn._config, home=nn._home)
    yaml_cfg = svc.load_yaml_config()
    if mock:
        yaml_cfg = {**yaml_cfg, "mock_search": {**dict(yaml_cfg.get("mock_search") or {}), "enabled": True}}
    return yaml_cfg


def _require_pinn_cloud_sync(request: Request, nn: NianNian) -> str:
    require_baseline_owner(request)
    if _session_header(request):
        raise HTTPException(
            status_code=403,
            detail={"code": "vp_sync_forbidden", "message": "tenant session cannot cloud_sync virtual personas"},
        )
    user_id, _role = _require_vp_capability(request, nn)
    if user_id != "pinn":
        raise HTTPException(
            status_code=403,
            detail={"code": "vp_sync_forbidden", "message": "virtual persona cloud_sync is pinn-only"},
        )
    return user_id


def _require_vp_capability(request: Request, nn: NianNian) -> tuple[str, str]:
    ctx = resolve_request_context(request, nn._config, home=nn._home)
    token = _session_header(request)
    if token:
        if "virtual_persona.local" not in ctx.capabilities:
            raise HTTPException(
                status_code=403,
                detail={"code": "vp_forbidden", "message": "virtual persona not available on this profile"},
            )
        return ctx.user_id, ctx.role
    role = parse_user_role(request)
    return ctx.user_id, role


def _require_vp_read(request: Request, nn: NianNian) -> tuple[str, str]:
    user_id, role = _require_vp_capability(request, nn)
    if _session_header(request):
        return user_id, role
    if role not in VP_READ_ROLES:
        raise HTTPException(
            status_code=403,
            detail={"code": "vp_forbidden", "message": "无权读取虚拟人物"},
        )
    return user_id, role


def _require_vp_write(request: Request, nn: NianNian) -> tuple[str, str]:
    user_id, role = _require_vp_capability(request, nn)
    if _session_header(request):
        return user_id, role
    if role not in VP_WRITE_ROLES:
        raise HTTPException(
            status_code=403,
            detail={"code": "vp_forbidden", "message": "guest 不可修改虚拟人物"},
        )
    return user_id, role


def create_virtual_persona_router(nn: NianNian) -> APIRouter:
    router = APIRouter(prefix="/v1/virtual-personas", tags=["virtual-persona"])

    @router.get("")
    def list_virtual_personas(request: Request) -> dict[str, Any]:
        user_id, _role = _require_vp_read(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        return {"user_id": user_id, "personas": store.list_personas()}

    @router.post("", status_code=201)
    def create_virtual_persona(request: Request, body: CreateVirtualPersonaBody) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            row = store.create(body.id, label=body.label)
        except FileExistsError as exc:
            raise HTTPException(
                status_code=409,
                detail={"code": "vp_exists", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        except VirtualPersonaValidationError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_profile", "message": str(exc)},
            ) from exc
        row["assembly_block"] = build_character_assembly_block(row.get("profile_md") or "")
        return row

    @router.post("/import", status_code=201)
    async def import_virtual_persona_bundle(
        request: Request,
        bundle: UploadFile = File(...),
    ) -> dict[str, str]:
        user_id, _role = _require_vp_write(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        data = await bundle.read()
        from niannian.virtual_persona.import_bundle import ImportBundleError, import_bundle

        try:
            return import_bundle(store, data)
        except ImportBundleError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_import_bundle", "message": str(exc)},
            ) from exc
        except VirtualPersonaValidationError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_profile", "message": str(exc)},
            ) from exc
        except FileExistsError as exc:
            raise HTTPException(
                status_code=409,
                detail={"code": "vp_exists", "message": str(exc)},
            ) from exc

    @router.get("/{vp_id}")
    def get_virtual_persona(request: Request, vp_id: str) -> dict[str, Any]:
        user_id, _role = _require_vp_read(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            row = store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        row["assembly_block"] = build_character_assembly_block(row.get("profile_md") or "")
        return row

    @router.patch("/{vp_id}")
    def patch_virtual_persona(
        request: Request,
        vp_id: str,
        body: PatchVirtualPersonaBody,
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        profile_md = body.profile_md
        if body.profile_sections is not None:
            current = store.get(vp_id)
            profile_md = merge_profile_sections(
                current.get("profile_md") or "",
                body.profile_sections,
                only_empty=False,
            )
        try:
            row = store.update(
                vp_id,
                profile_md=profile_md,
                label=body.label,
                storage_tier=body.storage_tier,
            )
            if body.stage is not None:
                transition_stage(store, vp_id, body.stage)
                row = store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except StageTransitionError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_stage_transition", "message": str(exc)},
            ) from exc
        except VirtualPersonaValidationError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_profile", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        row["assembly_block"] = build_character_assembly_block(row.get("profile_md") or "")
        return row

    @router.get("/{vp_id}/pack")
    def get_virtual_persona_pack(request: Request, vp_id: str) -> dict[str, Any]:
        user_id, _role = _require_vp_read(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.get(vp_id)
            pack = read_pack(store, vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        except VirtualPersonaValidationError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_pack", "message": str(exc)},
            ) from exc
        return {"vp_id": vp_id, "pack": pack}

    @router.patch("/{vp_id}/pack")
    def patch_virtual_persona_pack(
        request: Request,
        vp_id: str,
        body: PatchVirtualPersonaPackBody,
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        if body.stage is None and body.equipped_skills is None:
            raise HTTPException(
                status_code=400,
                detail={"code": "empty_patch", "message": "stage or equipped_skills required"},
            )
        try:
            result: dict[str, Any] | None = None
            if body.stage is not None:
                result = transition_stage(store, vp_id, body.stage)
            if body.equipped_skills is not None:
                pack = patch_pack_fields(store, vp_id, equipped_skills=body.equipped_skills)
                if result is None:
                    result = {
                        "vp_id": vp_id,
                        "previous_stage": pack["stage"],
                        "stage": pack["stage"],
                        "pack": pack,
                    }
                else:
                    result["pack"] = pack
            assert result is not None
            return result
        except StageTransitionError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_stage_transition", "message": str(exc)},
            ) from exc
        except VirtualPersonaValidationError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_profile", "message": str(exc)},
            ) from exc

    @router.delete("/{vp_id}", status_code=204)
    def delete_virtual_persona(request: Request, vp_id: str) -> None:
        user_id, _role = _require_vp_write(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.delete(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        return None

    @router.get("/{vp_id}/avatar")
    def get_virtual_persona_avatar(request: Request, vp_id: str) -> FileResponse:
        user_id, _role = _require_vp_read(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            if not store.has_avatar(vp_id):
                raise HTTPException(
                    status_code=404,
                    detail={"code": "vp_avatar_not_found", "message": "no avatar"},
                )
            path = store.avatar_path(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        return FileResponse(path, media_type="image/png")

    @router.put("/{vp_id}/avatar")
    async def put_virtual_persona_avatar(
        request: Request,
        vp_id: str,
        file: UploadFile = File(...),
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        data = await file.read()
        try:
            store.save_avatar(vp_id, data, kind="upload")
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_avatar", "message": str(exc)},
            ) from exc
        except RuntimeError as exc:
            raise HTTPException(
                status_code=500,
                detail={"code": "avatar_unavailable", "message": str(exc)},
            ) from exc
        return {"ok": True, "has_avatar": True, "vp_id": vp_id}

    @router.delete("/{vp_id}/avatar")
    def delete_virtual_persona_avatar(request: Request, vp_id: str) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.delete_avatar(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        return {"ok": True, "has_avatar": False, "vp_id": vp_id}

    @router.get("/{vp_id}/stickers")
    def list_virtual_persona_stickers(request: Request, vp_id: str) -> dict[str, Any]:
        user_id, _role = _require_vp_read(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        from niannian.virtual_persona.stickers import list_vp_stickers
        from niannian.virtual_persona.sticker_generate import is_image_gen_enabled

        rows = list_vp_stickers(store, vp_id)
        stickers = [
            {
                "slug": row["slug"],
                "label": row["label"],
                "file": Path(str(row["path"])).name,
                "source": row["source"],
            }
            for row in rows
        ]
        return {
            "vp_id": vp_id,
            "sticker_count": len(stickers),
            "stickers": stickers,
            "image_gen_enabled": is_image_gen_enabled(nn._config),
        }

    @router.post("/{vp_id}/stickers/search")
    def search_virtual_persona_stickers(
        request: Request,
        vp_id: str,
        body: StickerSearchBody,
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        budget = load_token_budget(nn._home, user_id)
        if not budget_allows(budget, "virtual_persona_sticker_search_enabled"):
            raise HTTPException(
                status_code=403,
                detail={
                    "code": "token_budget_vp_sticker_search_disabled",
                    "message": "VP 贴纸联网查找已在 Token 页关闭",
                },
            )
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            row = store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc

        from niannian.virtual_persona.sticker_search import (
            build_sticker_search_query,
            resolve_meme_keywords,
            search_sticker_candidates,
            sticker_search_config,
        )

        cfg = sticker_search_config(nn._config)
        meme_kw = resolve_meme_keywords(
            store,
            vp_id,
            profile_md=str(row.get("profile_md") or ""),
            override=body.meme_keywords,
        )
        label = str(row.get("label") or vp_id)
        query = build_sticker_search_query(
            label,
            query=body.query,
            meme_keywords=meme_kw,
            meme_hints=cfg["search_meme_hints"],
        )
        yaml_cfg = _web_learning_yaml(nn, mock=body.mock)
        search = build_search_provider_from_yaml(yaml_cfg)
        svc = WebLearningService(nn._subject_store, nn._config, home=nn._home)
        allowlist = svc.allowlist_patterns()
        max_results = min(int(body.max_results), int(cfg["max_stickers_per_run"]))
        result = search_sticker_candidates(
            search=search,
            query=query,
            allowlist=allowlist,
            max_results=max_results,
        )
        return {
            "ok": True,
            "vp_id": vp_id,
            "query": query,
            "meme_keywords": meme_kw,
            **result,
        }

    @router.post("/{vp_id}/stickers/import-from-url")
    def import_virtual_persona_sticker_from_url(
        request: Request,
        vp_id: str,
        body: StickerImportFromUrlBody,
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        budget = load_token_budget(nn._home, user_id)
        if not budget_allows(budget, "virtual_persona_sticker_search_enabled"):
            raise HTTPException(
                status_code=403,
                detail={
                    "code": "token_budget_vp_sticker_search_disabled",
                    "message": "VP 贴纸联网查找已在 Token 页关闭",
                },
            )
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc

        from niannian.virtual_persona.sticker_search import (
            StickerSearchError,
            import_vp_sticker_from_url,
            sticker_search_config,
        )

        cfg = sticker_search_config(nn._config)
        svc = WebLearningService(nn._subject_store, nn._config, home=nn._home)
        allowlist = svc.allowlist_patterns()
        try:
            return import_vp_sticker_from_url(
                store,
                vp_id,
                body.url,
                allowlist=allowlist,
                label=body.label,
                slug=body.slug,
                triggers=body.triggers,
                max_bytes=int(cfg["max_image_bytes"]),
            )
        except StickerSearchError as exc:
            code = exc.code
            status = 400
            if code in {"url_not_allowed", "invalid_image_url"}:
                status = 403
            raise HTTPException(
                status_code=status,
                detail={"code": code, "message": exc.message},
            ) from exc

    @router.post("/{vp_id}/stickers/generate")
    def generate_virtual_persona_stickers(
        request: Request,
        vp_id: str,
        body: StickerGenerateBody,
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        from niannian.virtual_persona.sticker_generate import generate_vp_stickers, is_image_gen_enabled

        if not is_image_gen_enabled(nn._config):
            raise HTTPException(
                status_code=503,
                detail={
                    "code": "image_gen_disabled",
                    "message": "未配置生图 API，可使用联网查找或上传",
                },
            )
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            row = store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        label = str(row.get("label") or vp_id)
        return generate_vp_stickers(
            store,
            vp_id,
            prompts=body.prompts,
            count=body.count,
            label=label,
        )

    @router.post("/{vp_id}/enrich")
    def enrich_virtual_persona_route(
        request: Request,
        vp_id: str,
        body: EnrichVirtualPersonaBody,
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        budget = load_token_budget(nn._home, user_id)
        if not budget_allows(budget, "virtual_persona_enrich_enabled"):
            raise HTTPException(
                status_code=403,
                detail={"code": "token_budget_vp_enrich_disabled", "message": "VP 联网 enrich 已在 Token 页关闭"},
            )
        fetch_avatar = body.fetch_avatar and budget_allows(budget, "virtual_persona_avatar_fetch_enabled")
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            row = store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        query = (body.query or row.get("label") or vp_id).strip()
        yaml_cfg = _web_learning_yaml(nn, mock=body.mock)
        search = build_search_provider_from_yaml(yaml_cfg)
        try:
            result = enrich_virtual_persona(
                store,
                vp_id,
                search=search,
                query=query,
                max_results=body.max_results,
                fetch_avatar=fetch_avatar,
            )
        except VirtualPersonaValidationError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_profile", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        result["assembly_block"] = build_character_assembly_block(result.get("profile_preview") or "")
        return result

    @router.post("/{vp_id}/perfect-distill")
    def perfect_distill_virtual_persona(
        request: Request,
        vp_id: str,
        body: PerfectDistillBody,
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        from niannian.virtual_persona.perfect_distill import (
            PerfectDistillError,
            build_perfect_distill_preview,
        )

        try:
            result = build_perfect_distill_preview(
                store,
                vp_id,
                body.export_options.model_dump(),
            )
            if body.write:
                from niannian.virtual_persona.export_bundle import (
                    bundle_id_from_path,
                    write_export_bundle,
                )

                bundle_dir = write_export_bundle(
                    store,
                    vp_id,
                    body.export_options.model_dump(),
                    preview=result["preview"],
                )
                result["bundle_id"] = bundle_id_from_path(bundle_dir)
            return result
        except PerfectDistillError as exc:
            raise HTTPException(
                status_code=409,
                detail={"code": "perfect_distill_forbidden", "message": str(exc)},
            ) from exc

    @router.get("/{vp_id}/exports")
    def list_virtual_persona_exports(request: Request, vp_id: str) -> dict[str, Any]:
        user_id, _role = _require_vp_read(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        from niannian.virtual_persona.export_bundle import list_export_bundles

        return {"vp_id": vp_id, "exports": list_export_bundles(store, vp_id)}

    @router.post("/{vp_id}/import-source")
    def import_virtual_persona_source(
        request: Request,
        vp_id: str,
        body: ImportSourceBody,
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc

        from niannian.virtual_persona.import_source import ImportSourceError, run_import_source

        svc = WebLearningService(nn._subject_store, nn._config, home=nn._home)
        allowlist = svc.allowlist_patterns()
        try:
            return run_import_source(
                store,
                vp_id,
                url=body.url,
                kind=body.kind,
                merge_into=body.merge_into,
                respect_locked_sections=body.respect_locked_sections,
                allowlist=allowlist,
            )
        except ImportSourceError as exc:
            if exc.code == "import_source_unsupported":
                raise HTTPException(
                    status_code=501,
                    detail={
                        "code": exc.code,
                        "message": exc.message,
                        "status": exc.status or "unsupported",
                        "hint": exc.hint or "待视频解析 AI",
                        "url": body.url.strip(),
                    },
                ) from exc
            status = 403 if exc.code in {"url_not_allowed", "invalid_url", "invalid_image_url"} else 400
            raise HTTPException(
                status_code=status,
                detail={"code": exc.code, "message": exc.message},
            ) from exc

    def _vp_md_response(
        *,
        vp_id: str,
        file: str,
        content: str,
        etag: str,
        response: Response | None = None,
    ) -> dict[str, Any]:
        if response is not None:
            response.headers["ETag"] = etag
        return {"vp_id": vp_id, "file": file, "content": content, "etag": etag}

    def _get_vp_md(
        request: Request,
        vp_id: str,
        file_key: str,
        response: Response,
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_read(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        try:
            file = normalize_md_file(file_key)
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_md_file", "message": str(exc)},
            ) from exc
        content, etag = read_vp_md(store, vp_id, file)
        return _vp_md_response(vp_id=vp_id, file=file, content=content, etag=etag, response=response)

    def _put_vp_md(
        request: Request,
        vp_id: str,
        file_key: str,
        body: PutVpMdBody,
        if_match: str | None,
        response: Response,
    ) -> dict[str, Any]:
        user_id, _role = _require_vp_write(request, nn)
        store = VirtualPersonaStore(nn._home, user_id)
        try:
            store.get(vp_id)
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_vp_id", "message": str(exc)},
            ) from exc
        try:
            file = normalize_md_file(file_key)
        except ValueError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_md_file", "message": str(exc)},
            ) from exc
        try:
            content, etag = write_vp_md(
                store,
                vp_id,
                file,
                body.content,
                if_match=if_match,
            )
        except MdStaleError as exc:
            raise HTTPException(
                status_code=409,
                detail={
                    "code": "md_stale",
                    "message": "file changed since last read; refresh and retry",
                    "etag": exc.current_etag,
                },
            ) from exc
        except VirtualPersonaValidationError as exc:
            raise HTTPException(
                status_code=400,
                detail={"code": "invalid_profile", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=428,
                detail={"code": "precondition_required", "message": str(exc)},
            ) from exc
        return _vp_md_response(vp_id=vp_id, file=file, content=content, etag=etag, response=response)

    @router.get("/{vp_id}/profile-md")
    def get_virtual_persona_profile_md(request: Request, vp_id: str, response: Response) -> dict[str, Any]:
        return _get_vp_md(request, vp_id, "profile-md", response)

    @router.put("/{vp_id}/profile-md")
    def put_virtual_persona_profile_md(
        request: Request,
        vp_id: str,
        body: PutVpMdBody,
        response: Response,
        if_match: str | None = Header(default=None, alias="If-Match"),
    ) -> dict[str, Any]:
        return _put_vp_md(request, vp_id, "profile-md", body, if_match, response)

    @router.get("/{vp_id}/assembly-master")
    def get_virtual_persona_assembly_master(request: Request, vp_id: str, response: Response) -> dict[str, Any]:
        return _get_vp_md(request, vp_id, "assembly-master", response)

    @router.put("/{vp_id}/assembly-master")
    def put_virtual_persona_assembly_master(
        request: Request,
        vp_id: str,
        body: PutVpMdBody,
        response: Response,
        if_match: str | None = Header(default=None, alias="If-Match"),
    ) -> dict[str, Any]:
        return _put_vp_md(request, vp_id, "assembly-master", body, if_match, response)

    @router.get("/{vp_id}/md/{file}")
    def get_virtual_persona_md(request: Request, vp_id: str, file: str, response: Response) -> dict[str, Any]:
        return _get_vp_md(request, vp_id, file, response)

    @router.put("/{vp_id}/md/{file}")
    def put_virtual_persona_md(
        request: Request,
        vp_id: str,
        file: str,
        body: PutVpMdBody,
        response: Response,
        if_match: str | None = Header(default=None, alias="If-Match"),
    ) -> dict[str, Any]:
        return _put_vp_md(request, vp_id, file, body, if_match, response)

    @router.post("/{vp_id}/cloud-sync")
    def cloud_sync_virtual_persona(request: Request, vp_id: str) -> dict[str, Any]:
        _require_pinn_cloud_sync(request, nn)
        try:
            return sync_virtual_persona_to_cloud(nn._home, vp_id, user_id="pinn")
        except FileNotFoundError as exc:
            raise HTTPException(
                status_code=404,
                detail={"code": "vp_not_found", "message": str(exc)},
            ) from exc
        except ValueError as exc:
            raise HTTPException(
                status_code=403,
                detail={"code": "vp_sync_forbidden", "message": str(exc)},
            ) from exc

    return router
