vent_deep_agent/api/skill_routes.py

# -*- coding: utf-8 -*-
"""
技能管理 API —— 查看/启用/禁用/上传技能

端点：
- GET  /api/skills              技能列表
- GET  /api/skills/{name}       查看技能内容（SKILL.md）
- POST /api/skills/{name}/toggle 启用/禁用
- POST /api/skills/upload        上传新技能（.zip）
"""

import json
import os
import re
import shutil
import zipfile
from pathlib import Path
from typing import Optional

from fastapi import APIRouter, File, UploadFile
from fastapi.responses import JSONResponse

router = APIRouter(prefix="/skills", tags=["技能管理"])

# 项目根目录
PROJECT_ROOT = str(Path(__file__).parent.parent)
SKILLS_DIR = os.path.join(PROJECT_ROOT, "skills")
DISABLED_FILE = os.path.join(PROJECT_ROOT, "skills_disabled.json")

# ── 技能 → 所属智能体映射 ──
SKILL_AGENT_MAP = {
    "click-interpret-tun":      ["点选解读 Agent"],
    "click-interpret-device":   ["点选解读 Agent"],
    "dialog-interpret":         ["通风对话助手"],
    "needq-calc":               ["通风对话助手"],
    "vent-plan-review-form":    ["形式审查 Agent"],
    "vent-plan-review-data":    ["数据一致性审查 Agent"],
    "vent-plan-review-calc":    ["计算核验审查 Agent"],
    "vent-plan-review-summary": ["汇总审查 Agent"],
}


def _get_disabled_set() -> set:
    """读取 skills_disabled.json，返回被禁用的技能名集合。"""
    if not os.path.exists(DISABLED_FILE):
        return set()
    try:
        with open(DISABLED_FILE, "r", encoding="utf-8") as f:
            data = json.load(f)
        return set(data.get("disabled", []))
    except (json.JSONDecodeError, KeyError):
        return set()


def _save_disabled_set(disabled: set):
    """保存被禁用的技能名集合到 skills_disabled.json。"""
    with open(DISABLED_FILE, "w", encoding="utf-8") as f:
        json.dump({"disabled": sorted(disabled)}, f, ensure_ascii=False, indent=2)


def _parse_skill_md(skill_dir: str) -> dict:
    """解析技能目录下的 SKILL.md，提取 frontmatter。

    Returns:
        {"name": str, "description": str} 或空字典
    """
    md_path = os.path.join(skill_dir, "SKILL.md")
    if not os.path.exists(md_path):
        return {}

    try:
        with open(md_path, "r", encoding="utf-8") as f:
            content = f.read()
    except Exception:
        return {}

    # 解析 YAML frontmatter（--- ... ---）
    match = re.match(r"^---\s*\n(.*?)\n---", content, re.DOTALL)
    if not match:
        return {"name": os.path.basename(skill_dir), "description": ""}

    frontmatter_text = match.group(1)

    # 简单解析 YAML（只取 name 和 description，不引入 pyyaml 依赖）
    result = {}
    # 解析 name: xxx
    name_match = re.search(r"^name:\s*(.+)$", frontmatter_text, re.MULTILINE)
    if name_match:
        result["name"] = name_match.group(1).strip().strip('"').strip("'")

    # 解析 description: |
    desc_match = re.search(r"^description:\s*\|\s*\n(.*?)(?=^\S|\Z)", frontmatter_text, re.MULTILINE | re.DOTALL)
    if desc_match:
        desc = desc_match.group(1).strip()
        # 去掉每行开头的缩进
        desc = "\n".join(line.strip() for line in desc.split("\n"))
        # 合并为单行（截取前 80 字符用于列表展示）
        desc = " ".join(desc.split())
        result["description"] = desc[:120] + ("..." if len(desc) > 120 else "")
    else:
        # 尝试单行 description
        desc_match = re.search(r"^description:\s*(.+)$", frontmatter_text, re.MULTILINE)
        if desc_match:
            d = desc_match.group(1).strip().strip('"').strip("'")
            result["description"] = d[:120] + ("..." if len(d) > 120 else "")

    return result


# ── GET /api/skills ──

@router.get("")
async def list_skills():
    """获取所有技能列表（含启用状态和所属智能体）。"""
    if not os.path.isdir(SKILLS_DIR):
        return {"skills": []}

    disabled_set = _get_disabled_set()
    skills = []

    for entry in sorted(os.listdir(SKILLS_DIR)):
        skill_path = os.path.join(SKILLS_DIR, entry)
        # 跳过非目录、隐藏目录、Python 缓存目录
        if not os.path.isdir(skill_path) or entry.startswith(".") or entry == "__pycache__":
            continue

        meta = _parse_skill_md(skill_path)
        name = meta.get("name", entry)
        description = meta.get("description", "")
        has_skill_md = os.path.exists(os.path.join(skill_path, "SKILL.md"))
        agents = SKILL_AGENT_MAP.get(entry, ["（未分配）"])

        skills.append({
            "name": entry,
            "display_name": name,
            "path": f"skills/{entry}",
            "description": description,
            "enabled": entry not in disabled_set,
            "agents": agents,
            "has_skill_md": has_skill_md,
        })

    return {"skills": skills}


# ── GET /api/skills/{name} ──

@router.get("/{name}")
async def get_skill_content(name: str):
    """获取指定技能的 SKILL.md 完整内容。"""
    skill_dir = os.path.join(SKILLS_DIR, name)
    md_path = os.path.join(skill_dir, "SKILL.md")

    if not os.path.isdir(skill_dir):
        return JSONResponse(status_code=404, content={"error": f"技能不存在: {name}"})

    if not os.path.exists(md_path):
        return JSONResponse(status_code=404, content={"error": f"技能 {name} 缺少 SKILL.md"})

    try:
        with open(md_path, "r", encoding="utf-8") as f:
            content = f.read()
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": f"读取失败: {str(e)}"})

    return {
        "name": name,
        "content": content,
    }


# ── POST /api/skills/{name}/toggle ──

@router.post("/{name}/toggle")
async def toggle_skill(name: str):
    """启用或禁用指定技能。"""
    skill_dir = os.path.join(SKILLS_DIR, name)
    if not os.path.isdir(skill_dir):
        return JSONResponse(status_code=404, content={"error": f"技能不存在: {name}"})

    disabled_set = _get_disabled_set()

    if name in disabled_set:
        disabled_set.discard(name)
        enabled = True
        msg = f"技能「{name}」已启用"
    else:
        disabled_set.add(name)
        enabled = False
        msg = f"技能「{name}」已禁用"

    _save_disabled_set(disabled_set)

    # 禁用技能时，通知 agent 缓存失效（下次请求 agent 会跳过已禁用的技能）
    if not enabled:
        try:
            from agents.vent_agent import invalidate_dialog_cache
            invalidate_dialog_cache()
        except Exception:
            pass
        try:
            from agents.review_agent import invalidate_cache
            invalidate_cache()
        except Exception:
            pass

    return {"name": name, "enabled": enabled, "message": msg}


# ── POST /api/skills/upload ──

@router.post("/upload")
async def upload_skill(file: UploadFile = File(...)):
    """上传新技能（.zip 压缩包）。

    要求：
    - 文件为 .zip 格式
    - 解压后必须包含 SKILL.md 文件
    - 技能名不能与已有技能重复
    """
    # 校验文件类型
    if not file.filename or not file.filename.lower().endswith(".zip"):
        return JSONResponse(status_code=400, content={"error": "仅支持 .zip 格式的技能包"})

    # 读取文件内容
    try:
        content = await file.read()
    except Exception as e:
        return JSONResponse(status_code=400, content={"error": f"读取文件失败: {str(e)}"})

    # 保存临时文件
    tmp_path = os.path.join(PROJECT_ROOT, "data", f"_tmp_skill_{file.filename}")
    os.makedirs(os.path.dirname(tmp_path), exist_ok=True)
    try:
        with open(tmp_path, "wb") as f:
            f.write(content)

        # 解压
        with zipfile.ZipFile(tmp_path, "r") as zf:
            # 安全检查：防止路径穿越攻击
            for member in zf.infolist():
                if member.filename.startswith("/") or ".." in member.filename:
                    return JSONResponse(status_code=400, content={"error": "技能包包含非法路径"})

            # 查找顶层目录名（技能名）
            names = zf.namelist()
            if not names:
                return JSONResponse(status_code=400, content={"error": "技能包为空"})

            # 取第一个条目的顶层目录作为技能名
            root_name = names[0].split("/")[0]
            if not root_name:
                return JSONResponse(status_code=400, content={"error": "技能包格式不正确"})

            # 检查 SKILL.md 是否存在
            has_skill_md = any(
                n == f"{root_name}/SKILL.md" or n.startswith(f"{root_name}/") and n.endswith("/SKILL.md")
                for n in names
            )
            if not has_skill_md:
                return JSONResponse(status_code=400, content={"error": "技能包缺少 SKILL.md 文件"})

            # 检查是否与已有技能重名
            target_dir = os.path.join(SKILLS_DIR, root_name)
            if os.path.exists(target_dir):
                return JSONResponse(status_code=409, content={"error": f"技能「{root_name}」已存在"})

            # 解压到 skills/ 目录
            zf.extractall(SKILLS_DIR)

    except zipfile.BadZipFile:
        return JSONResponse(status_code=400, content={"error": "文件不是有效的 .zip 压缩包"})
    except Exception as e:
        return JSONResponse(status_code=500, content={"error": f"解压失败: {str(e)}"})
    finally:
        # 清理临时文件
        if os.path.exists(tmp_path):
            try:
                os.unlink(tmp_path)
            except OSError:
                pass

    return {
        "name": root_name,
        "message": f"技能「{root_name}」上传成功",
    }