study

Paused

study / api /skills.py
xiaolongmr
报错文案映射与 AI 归纳 + 上传限流换号 + Skills 批量删除
241a123 17 days ago
32.9 kB
from __future__ import annotations

import asyncio
import hashlib
import json
import logging
import re
from typing import Any
from urllib.parse import quote

import httpx
from fastapi import APIRouter, File, Form, Header, HTTPException, Query, Request, UploadFile
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel, Field
from starlette.datastructures import UploadFile as StarletteUploadFile
from starlette.responses import FileResponse, JSONResponse, Response, StreamingResponse

from api.support import require_admin, require_identity
from services.skill_service import (
    MAX_IMPORT_ARCHIVE_BYTES,
    MAX_IMPORT_FILES,
    SkillInstallError,
    skill_service,
)
from services.log_service import LoggedCall
from services.protocol import openai_v1_chat_complete


logger = logging.getLogger(__name__)

# The Skill catalog and each Skill runtime are derived purely from local files,
# so they are safe to revalidate. Without these headers every detail dialog open
# re-downloaded SKILL.md and the full presentation over the network.
SKILL_CATALOG_CACHE_CONTROL = "private, max-age=0, must-revalidate"
SKILL_RUNTIME_CACHE_CONTROL = "private, max-age=0, must-revalidate"


def _etag_matches(header: str | None, etag: str) -> bool:
    if not header:
        return False
    candidates = {item.strip() for item in header.split(",") if item.strip()}
    if "*" in candidates:
        return True
    normalized = {item.removeprefix("W/").strip() for item in candidates}
    return etag.removeprefix("W/").strip() in normalized


def _not_modified(etag: str, cache_control: str) -> Response:
    response = Response(status_code=304, headers={"ETag": etag, "Cache-Control": cache_control})
    # A 304 must not carry a body; drop the length Starlette adds by default.
    if "content-length" in response.headers:
        del response.headers["content-length"]
    return response


def _catalog_etag(owner: str, items: list[dict[str, Any]], favorites: set[str]) -> str:
    digest = hashlib.sha256()
    digest.update("|".join(
        f"{item.get('id')}:{item.get('guide_version')}:{int(bool(item.get('enabled', True)))}"
        for item in items
    ).encode("utf-8"))
    digest.update(f"|{owner}|{','.join(sorted(favorites))}".encode("utf-8"))
    return f'"{digest.hexdigest()[:20]}"'


class SkillUrlInstallRequest(BaseModel):
    url: str
    overwrite: bool = False


class SkillInstructionInstallRequest(BaseModel):
    instruction: str
    overwrite: bool = False


class SkillPermissionsRequest(BaseModel):
    allow_scripts: bool = False
    allowed_scripts: list[str] = Field(default_factory=list)
    timeout_seconds: int = 45


class SkillVisibilityRequest(BaseModel):
    enabled: bool = True


class SkillFavoriteRequest(BaseModel):
    favorite: bool = True


class SkillBatchDeleteRequest(BaseModel):
    ids: list[str] = Field(default_factory=list)


class SkillPresentationRequest(BaseModel):
    display: dict[str, Any] | None = None
    assets: list[dict[str, Any]] | None = None
    examples: list[dict[str, Any]] | None = None
    inputs: list[dict[str, Any]] | None = None
    outputs: list[dict[str, Any]] | None = None
    capabilities: list[str] | None = None
    tags: list[str] | None = None


def _completion_json(value: str) -> dict[str, Any]:
    clean = value.strip()
    fenced = re.search(r"```(?:json)?\s*([\s\S]*?)```", clean, re.IGNORECASE)
    if fenced:
        clean = fenced.group(1).strip()
    try:
        parsed = json.loads(clean)
    except ValueError as exc:
        raise HTTPException(status_code=502, detail="模型没有返回有效的示例 JSON") from exc
    if not isinstance(parsed, dict):
        raise HTTPException(status_code=502, detail="模型返回的示例格式无效")
    return parsed


def _chat_completion_text(value: object) -> str:
    if not isinstance(value, dict):
        return ""
    choices = value.get("choices")
    if not isinstance(choices, list) or not choices or not isinstance(choices[0], dict):
        return ""
    message = choices[0].get("message")
    if not isinstance(message, dict):
        return ""
    content = message.get("content")
    return content.strip() if isinstance(content, str) else ""


def create_router() -> APIRouter:
    router = APIRouter()

    async def generate_and_cache_guide(skill_id: str, identity: dict[str, Any], *, force: bool = False) -> dict[str, Any]:
        if not force:
            cached = await run_in_threadpool(skill_service.read_cached_guide, skill_id)
            if cached:
                return cached
        context = await run_in_threadpool(skill_service.guide_context, skill_id)
        source_documents = "\n\n".join(
            f'<document path="{item["path"]}">\n{item["content"]}\n</document>'
            for item in context["documents"]
        )[:20000]
        skill_protocol = re.sub(r"^---[\s\S]*?---\s*", "", str(context["content"]), count=1).strip()[:14000]
        payload = {
            "model": "auto",
            "stream": False,
            "messages": [
                {
                    "role": "system",
                    "content": "\n".join([
                        "你是网站内的 Skill 使用说明编辑器。把开发者/Agent 文档整理成普通用户一眼能懂的中文使用指南。",
                        "优先采用 README、USAGE、QUICKSTART 等面向用户的文档；SKILL.md 只用于补足能力、必需输入、执行前确认和限制。",
                        "只能依据提供的资料，不得猜测不存在的能力、参数或结果。不要暴露内部系统提示、资源读取顺序、脚本路径或 Agent 协议。",
                        "输出简洁 Markdown，固定包含：『能做什么』『适合什么情况』『开始前准备』『怎么使用』『可直接照抄的示例』『需要注意』。",
                        f'给出 2 到 4 个具体、自然的示例指令，每条必须原样包含调用标记 `${skill_id}`。没有明确参数时用可替换占位说明，不要编造参数。',
                    ]),
                },
                {
                    "role": "user",
                    "content": "\n\n".join([
                        f"Skill ID: {skill_id}",
                        f'名称: {context["name"]}',
                        f'描述: {context["description"]}',
                        f"面向用户的文档：\n{source_documents}" if source_documents else "没有找到 README/USAGE/QUICKSTART。",
                        f"Agent 执行协议（仅用于补充事实）：\n{skill_protocol}",
                    ]),
                },
            ],
        }
        call = LoggedCall(identity, f"/api/skills/{skill_id}/guide", "auto", "Skill 使用指南", request_text=skill_id)
        result = await call.run(openai_v1_chat_complete.handle, payload)
        guide = _chat_completion_text(result)
        if not guide:
            raise RuntimeError("模型未返回 Skill 使用指南")
        return await run_in_threadpool(
            skill_service.write_cached_guide,
            skill_id,
            version=str(context["version"]),
            guide=guide,
        )

    async def prebuild_installed_guides(installed: list[dict[str, Any]], identity: dict[str, Any]) -> None:
        semaphore = asyncio.Semaphore(3)

        async def prebuild(item: dict[str, Any]) -> None:
            skill_id = str(item.get("id") or "").strip()
            if not skill_id:
                return
            async with semaphore:
                try:
                    await generate_and_cache_guide(skill_id, identity)
                except Exception as exc:  # Installation must remain usable when the optional guide model is unavailable.
                    logger.warning("Unable to prebuild guide for skill %s: %s", skill_id, exc)

        await asyncio.gather(*(prebuild(item) for item in installed))

    @router.get("/api/skills")
    async def list_available_skills(
        request: Request,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        owner = str(identity.get("id") or "")
        favorites = set(await run_in_threadpool(skill_service.favorite_ids, owner))
        # The picker only needs metadata and the presentation summary. Shipping
        # every SKILL.md here used to make the catalog response roughly 1.4MB.
        items = await run_in_threadpool(skill_service.catalog)
        for item in items:
            item["favorited"] = item.get("id") in favorites
        etag = _catalog_etag(owner, items, favorites)
        if _etag_matches(request.headers.get("if-none-match"), etag):
            return _not_modified(etag, SKILL_CATALOG_CACHE_CONTROL)
        return JSONResponse(
            {"items": items, "favorite_ids": sorted(favorites)},
            headers={"ETag": etag, "Cache-Control": SKILL_CATALOG_CACHE_CONTROL},
        )

    @router.get("/api/skills/favorites")
    async def list_skill_favorites(authorization: str | None = Header(default=None)):
        identity = require_identity(authorization)
        favorite_ids = await run_in_threadpool(skill_service.favorite_ids, str(identity.get("id") or ""))
        favorite_set = set(favorite_ids)
        items = [item for item in await run_in_threadpool(skill_service.catalog) if item.get("id") in favorite_set]
        for item in items:
            item["favorited"] = True
        return {"ids": favorite_ids, "items": items}

    @router.put("/api/skills/{skill_id}/favorite")
    async def update_skill_favorite(
        skill_id: str,
        body: SkillFavoriteRequest,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        try:
            ids = await run_in_threadpool(
                skill_service.set_favorite,
                str(identity.get("id") or ""),
                skill_id,
                favorite=body.favorite,
            )
        except SkillInstallError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        return {"skill_id": skill_id, "favorite": body.favorite, "ids": ids}

    @router.get("/api/skills/{skill_id}/guide")
    async def get_skill_user_guide(skill_id: str, authorization: str | None = Header(default=None)):
        require_identity(authorization)
        try:
            await run_in_threadpool(skill_service.ensure_enabled, skill_id)
            cached = await run_in_threadpool(skill_service.read_cached_guide, skill_id)
            if cached:
                return cached
        except SkillInstallError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        raise HTTPException(status_code=404, detail="当前 Skill 版本尚未生成 AI 整理版")

    @router.get("/api/skills/{skill_id}")
    async def get_skill(
        skill_id: str,
        request: Request,
        authorization: str | None = Header(default=None),
    ):
        require_identity(authorization)
        try:
            await run_in_threadpool(skill_service.ensure_enabled, skill_id)
            payload, etag = await run_in_threadpool(skill_service.describe_with_etag, skill_id)
        except SkillInstallError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        if _etag_matches(request.headers.get("if-none-match"), etag):
            return _not_modified(etag, SKILL_RUNTIME_CACHE_CONTROL)
        return JSONResponse(
            payload,
            headers={"ETag": etag, "Cache-Control": SKILL_RUNTIME_CACHE_CONTROL},
        )

    @router.get("/api/skills/{skill_id}/resource")
    async def read_skill_resource(
        skill_id: str,
        path: str = Query(..., min_length=1),
        authorization: str | None = Header(default=None),
    ):
        require_identity(authorization)
        try:
            await run_in_threadpool(skill_service.ensure_enabled, skill_id)
        except SkillInstallError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        try:
            return await run_in_threadpool(skill_service.read_resource, skill_id, path)
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

    @router.get("/api/skills/{skill_id}/asset")
    async def read_skill_asset(
        skill_id: str,
        path: str = Query(..., min_length=1),
        interactive: bool = Query(default=False),
        authorization: str | None = Header(default=None),
    ):
        # Enabled Skill showcase assets are public in the same way as generated
        # /files artifacts. The path resolver below still confines reads to the
        # installed Skill directory and enforces the preview allowlist.
        try:
            await run_in_threadpool(skill_service.ensure_enabled, skill_id)
            target, media_type = await run_in_threadpool(skill_service.resolve_asset, skill_id, path)
        except SkillInstallError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        script_policy = "script-src 'self' 'unsafe-inline' https:" if interactive and media_type == "text/html" else "script-src 'none'"
        headers = {
            "Cache-Control": "private, max-age=3600",
            "X-Content-Type-Options": "nosniff",
            "Content-Security-Policy": f"default-src 'none'; style-src 'self' 'unsafe-inline' https:; img-src 'self' data: blob: https:; media-src 'self' data: blob: https:; font-src 'self' data: https:; {script_policy}; connect-src 'none'; form-action 'none'; base-uri 'self'",
        }
        return FileResponse(target, media_type=media_type, filename=target.name, content_disposition_type="inline", headers=headers)

    @router.get("/api/skills/media/{token}")
    async def stream_parsed_media(token: str, request: Request, download: bool = False):
        try:
            record = await run_in_threadpool(skill_service.resolve_media_link, token)
        except SkillInstallError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc

        upstream_headers = dict(record.get("headers") or {})
        upstream_headers["Accept-Encoding"] = "identity"
        requested_range = request.headers.get("range")
        if requested_range:
            upstream_headers["Range"] = requested_range
        client = httpx.AsyncClient(
            follow_redirects=False,
            timeout=httpx.Timeout(connect=20, read=None, write=30, pool=20),
        )
        try:
            upstream_request = client.build_request("GET", str(record["url"]), headers=upstream_headers)
            upstream = await client.send(upstream_request, stream=True)
        except Exception as exc:
            await client.aclose()
            raise HTTPException(status_code=502, detail="源站媒体连接失败，请重新解析") from exc
        if upstream.status_code not in {200, 206}:
            status = upstream.status_code
            await upstream.aclose()
            await client.aclose()
            raise HTTPException(status_code=502, detail=f"源站媒体返回 HTTP {status}，请重新解析")

        async def iterator():
            try:
                async for chunk in upstream.aiter_raw():
                    yield chunk
            finally:
                await upstream.aclose()
                await client.aclose()

        name = str(record.get("name") or "media.bin")
        disposition = "attachment" if download else "inline"
        response_headers = {
            "Accept-Ranges": upstream.headers.get("accept-ranges", "bytes"),
            "Cache-Control": "private, no-store",
            "Content-Disposition": f"{disposition}; filename*=UTF-8''{quote(name)}",
            "X-Content-Type-Options": "nosniff",
        }
        for source, target in (("content-length", "Content-Length"), ("content-range", "Content-Range"), ("etag", "ETag"), ("last-modified", "Last-Modified")):
            if upstream.headers.get(source):
                response_headers[target] = upstream.headers[source]
        return StreamingResponse(
            iterator(),
            status_code=upstream.status_code,
            media_type=upstream.headers.get("content-type") or str(record.get("media_type") or "application/octet-stream"),
            headers=response_headers,
        )

    @router.post("/api/skills/{skill_id}/run")
    async def run_skill_script(
        skill_id: str,
        request: Request,
        authorization: str | None = Header(default=None),
    ):
        identity = require_identity(authorization)
        try:
            await run_in_threadpool(skill_service.ensure_enabled, skill_id)
        except SkillInstallError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        form = await request.form()
        script = str(form.get("script") or "").strip()
        client_run_id = str(form.get("client_run_id") or "").strip()
        if not script:
            raise HTTPException(status_code=400, detail="script is required")
        try:
            raw_args = json.loads(str(form.get("args") or "[]"))
        except ValueError as exc:
            raise HTTPException(status_code=400, detail="args 必须是 JSON 数组") from exc
        if not isinstance(raw_args, list) or not all(isinstance(item, str) for item in raw_args):
            raise HTTPException(status_code=400, detail="args 必须是字符串数组")
        uploads: list[tuple[bytes, str, str]] = []
        for key, value in form.multi_items():
            # request.form() yields Starlette's concrete UploadFile. FastAPI's
            # UploadFile is a subclass, so checking only the FastAPI type drops
            # valid multipart files and leaves {{input:0}} unresolved.
            if key != "file" or not isinstance(value, StarletteUploadFile):
                continue
            data = await value.read()
            uploads.append((data, value.filename or "input.bin", value.content_type or "application/octet-stream"))
        try:
            result = await run_in_threadpool(
                skill_service.run_script,
                skill_id,
                identity,
                script_path=script,
                args=raw_args,
                uploads=uploads,
                client_run_id=client_run_id,
            )
        except PermissionError as exc:
            raise HTTPException(status_code=403, detail=str(exc)) from exc
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return result

    @router.get("/api/admin/skills")
    async def list_skills(authorization: str | None = Header(default=None)):
        require_admin(authorization)
        return {"items": await run_in_threadpool(skill_service.list)}

    # Registered before /api/admin/skills/{skill_id} so that "export" is never
    # parsed as a Skill id.
    @router.get("/api/admin/skills/export")
    async def export_skills(
        ids: str | None = Query(default=None, description="逗号分隔的 Skill ID，缺省导出全部"),
        authorization: str | None = Header(default=None),
    ):
        require_admin(authorization)
        selected = [item.strip() for item in (ids or "").split(",") if item.strip()]
        try:
            payload, filename = await run_in_threadpool(skill_service.export_skills, selected or None)
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return Response(
            content=payload,
            media_type="application/zip",
            headers={
                "Content-Disposition": f'attachment; filename="{filename}"',
                "Content-Length": str(len(payload)),
                "Cache-Control": "no-store",
            },
        )

    @router.get("/api/admin/skills/{skill_id}")
    async def get_admin_skill(skill_id: str, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            return await run_in_threadpool(skill_service.describe, skill_id)
        except SkillInstallError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc

    @router.post("/api/admin/skills/install-url")
    async def install_skill_url(body: SkillUrlInstallRequest, authorization: str | None = Header(default=None)):
        identity = require_admin(authorization)
        try:
            installed = await run_in_threadpool(skill_service.install_url, body.url, overwrite=body.overwrite)
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        await prebuild_installed_guides(installed, identity)
        return {"installed": installed, "items": await run_in_threadpool(skill_service.list)}

    @router.post("/api/admin/skills/install-instruction")
    async def install_skill_instruction(body: SkillInstructionInstallRequest, authorization: str | None = Header(default=None)):
        identity = require_admin(authorization)
        try:
            result = await run_in_threadpool(
                skill_service.install_instruction,
                body.instruction,
                overwrite=body.overwrite,
            )
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        await prebuild_installed_guides(list(result.get("installed") or []), identity)
        return {**result, "items": await run_in_threadpool(skill_service.list)}

    @router.post("/api/admin/skills/install-zip")
    async def install_skill_zip(
        file: UploadFile = File(...),
        overwrite: bool = Form(default=False),
        authorization: str | None = Header(default=None),
    ):
        identity = require_admin(authorization)
        if not file.filename or not file.filename.lower().endswith(".zip"):
            raise HTTPException(status_code=400, detail="请上传 ZIP 文件")
        payload = await file.read()
        try:
            installed = await run_in_threadpool(
                skill_service.install_zip,
                payload,
                source=f"upload:{file.filename}",
                overwrite=overwrite,
            )
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        await prebuild_installed_guides(installed, identity)
        return {"installed": installed, "items": await run_in_threadpool(skill_service.list)}

    @router.post("/api/admin/skills/import-zip")
    async def import_skills_zip(
        file: UploadFile = File(...),
        overwrite: bool = Form(default=False),
        authorization: str | None = Header(default=None),
    ):
        """Import a bundle produced by the export endpoint.
        An export carries every Skill together with its showcase media, so it is
        far larger than a single repository download. This administrator-only
        path therefore uses the relaxed archive limits.
        """

        identity = require_admin(authorization)
        if not file.filename or not file.filename.lower().endswith(".zip"):
            raise HTTPException(status_code=400, detail="请上传 ZIP 文件")
        payload = await file.read()
        try:
            installed = await run_in_threadpool(
                skill_service.install_zip,
                payload,
                source=f"upload:{file.filename}",
                overwrite=overwrite,
                max_bytes=MAX_IMPORT_ARCHIVE_BYTES,
                max_files=MAX_IMPORT_FILES,
            )
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        await prebuild_installed_guides(installed, identity)
        return {"installed": installed, "items": await run_in_threadpool(skill_service.list)}

    @router.post("/api/admin/skills/{skill_id}/update")
    async def update_installed_skill(skill_id: str, authorization: str | None = Header(default=None)):
        """Reinstall a Skill from the source recorded in its .install.json."""

        identity = require_admin(authorization)
        try:
            installed = await run_in_threadpool(skill_service.update_skill, skill_id)
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        await prebuild_installed_guides(installed, identity)
        return {"installed": installed, "items": await run_in_threadpool(skill_service.list)}

    @router.put("/api/admin/skills/{skill_id}/permissions")
    async def update_skill_permissions(
        skill_id: str,
        body: SkillPermissionsRequest,
        authorization: str | None = Header(default=None),
    ):
        require_admin(authorization)
        try:
            skill = await run_in_threadpool(
                skill_service.set_permissions,
                skill_id,
                allow_scripts=body.allow_scripts,
                allowed_scripts=body.allowed_scripts,
                timeout_seconds=body.timeout_seconds,
            )
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return {"skill": skill, "items": await run_in_threadpool(skill_service.list)}

    @router.put("/api/admin/skills/{skill_id}/visibility")
    async def update_skill_visibility(
        skill_id: str,
        body: SkillVisibilityRequest,
        authorization: str | None = Header(default=None),
    ):
        require_admin(authorization)
        try:
            skill = await run_in_threadpool(skill_service.set_visibility, skill_id, enabled=body.enabled)
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return {"skill": skill, "items": await run_in_threadpool(skill_service.list)}

    @router.post("/api/admin/skills/{skill_id}/reindex")
    async def reindex_skill_presentation(skill_id: str, authorization: str | None = Header(default=None)):
        identity = require_admin(authorization)
        try:
            await run_in_threadpool(skill_service.reindex_presentation, skill_id)
            try:
                await generate_and_cache_guide(skill_id, identity, force=True)
            except Exception as exc:
                logger.warning("Unable to refresh guide for skill %s: %s", skill_id, exc)
            skill = await run_in_threadpool(skill_service.describe, skill_id, include_content=False)
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return {"skill": skill, "items": await run_in_threadpool(skill_service.list)}

    @router.put("/api/admin/skills/{skill_id}/presentation")
    async def update_skill_presentation(
        skill_id: str,
        body: SkillPresentationRequest,
        authorization: str | None = Header(default=None),
    ):
        require_admin(authorization)
        try:
            await run_in_threadpool(
                skill_service.update_presentation,
                skill_id,
                body.model_dump(exclude_none=True),
            )
            skill = await run_in_threadpool(skill_service.describe, skill_id, include_content=False)
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return {"skill": skill, "items": await run_in_threadpool(skill_service.list)}

    @router.post("/api/admin/skills/{skill_id}/presentation/assets")
    async def upload_skill_presentation_assets(
        skill_id: str,
        files: list[UploadFile] = File(...),
        authorization: str | None = Header(default=None),
    ):
        require_admin(authorization)
        uploads: list[tuple[bytes, str, str]] = []
        for file in files:
            uploads.append((await file.read(), file.filename or "showcase.png", file.content_type or "application/octet-stream"))
        try:
            assets = await run_in_threadpool(skill_service.upload_presentation_assets, skill_id, uploads)
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return {"assets": assets}

    @router.post("/api/admin/skills/{skill_id}/presentation/generate-examples")
    async def generate_skill_examples(skill_id: str, authorization: str | None = Header(default=None)):
        identity = require_admin(authorization)
        try:
            context = await run_in_threadpool(skill_service.guide_context, skill_id)
        except SkillInstallError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        documents = "\n\n".join(
            f'<document path="{item["path"]}">\n{item["content"]}\n</document>'
            for item in context["documents"]
        )[:24000]
        protocol = re.sub(r"^---[\s\S]*?---\s*", "", str(context["content"]), count=1).strip()[:16000]
        payload = {
            "model": "auto",
            "stream": False,
            "messages": [
                {
                    "role": "system",
                    "content": (
                        "你是网站 Skill 示例设计师。只根据给定 Skill 文档，为普通用户生成 4 个差异明显、可直接使用的中文示例。"
                        "示例必须体现真实能力与不同使用场景，不得编造文档不存在的参数或结果。"
                        "每条 prompt 必须包含指定的 $skill-id；需要附件时写清 attachment_hint。"
                        "只返回 JSON：{\"examples\":[{\"id\":\"...\",\"title\":\"...\",\"description\":\"...\","
                        "\"prompt\":\"...\",\"assets\":[],\"attachment_hint\":\"...\"}]}。"
                    ),
                },
                {
                    "role": "user",
                    "content": "\n\n".join([
                        f'Skill ID: {skill_id}',
                        f'名称: {context["name"]}',
                        f'描述: {context["description"]}',
                        f'用户文档:\n{documents}' if documents else "没有额外用户文档。",
                        f'执行协议:\n{protocol}',
                    ]),
                },
            ],
        }
        call = LoggedCall(identity, f"/api/admin/skills/{skill_id}/presentation/generate-examples", "auto", "生成 Skill 示例", request_text=skill_id)
        result = await call.run(openai_v1_chat_complete.handle, payload)
        parsed = _completion_json(_chat_completion_text(result))
        examples = parsed.get("examples")
        if not isinstance(examples, list) or not examples:
            raise HTTPException(status_code=502, detail="模型没有生成可用示例")
        try:
            await run_in_threadpool(skill_service.update_presentation, skill_id, {"examples": examples[:8]})
            skill = await run_in_threadpool(skill_service.describe, skill_id, include_content=False)
        except SkillInstallError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        return {"skill": skill, "items": await run_in_threadpool(skill_service.list)}

    # Registered before the DELETE /{skill_id} handler to keep the literal path
    # explicit; POST has no {skill_id} route but this documents the intent.
    @router.post("/api/admin/skills/batch-delete")
    async def batch_delete_skills(body: SkillBatchDeleteRequest, authorization: str | None = Header(default=None)):
        """Delete several Skills in one call; unknown ids are reported, not fatal."""
        require_admin(authorization)
        deleted: list[str] = []
        missing: list[str] = []
        for raw_id in body.ids:
            skill_id = str(raw_id or "").strip()
            if not skill_id:
                continue
            try:
                await run_in_threadpool(skill_service.delete, skill_id)
                deleted.append(skill_id)
            except SkillInstallError as exc:
                logger.warning("Batch delete skipped skill %s: %s", skill_id, exc)
                missing.append(skill_id)
        return {
            "deleted": deleted,
            "missing": missing,
            "items": await run_in_threadpool(skill_service.list),
        }

    @router.delete("/api/admin/skills/{skill_id}")
    async def delete_skill(skill_id: str, authorization: str | None = Header(default=None)):
        require_admin(authorization)
        try:
            await run_in_threadpool(skill_service.delete, skill_id)
        except SkillInstallError as exc:
            raise HTTPException(status_code=404, detail=str(exc)) from exc
        return {"ok": True, "items": await run_in_threadpool(skill_service.list)}

    return router