Agent/agent/tools/builtin/skill.py

"""
Skill 工具 - 按需加载 Skill 文件

Agent 可以调用此工具来加载特定的 skill 文档
"""

import importlib.util
import os
import shutil
import subprocess
from pathlib import Path
from typing import Optional

from agent.tools import tool, ToolResult
from agent.skill.skill_loader import SkillLoader

# 飞书 openclaw-lark 子模块 skills 根目录（整体搬迁时只改此处或环境变量）
_FEISHU_OPENCLAW_SKILLS_ROOT = os.path.join(
    os.getenv("FEISHU_OPENCLAW_ROOT", "./gateway/core/channels/feishu/openclaw-lark"),
    "skills",
)
_FEISHU_OPENCLAW_SKILL_NAMES = (
    "feishu-bitable",
    "feishu-calendar",
    "feishu-channel-rules",
    "feishu-create-doc",
    "feishu-fetch-doc",
    "feishu-im-read",
    "feishu-task",
    "feishu-troubleshoot",
    "feishu-update-doc",
)

# 默认 skills 目录（优先级：项目 skills > 框架 skills > 飞书 openclaw skills）
DEFAULT_SKILLS_DIRS = [
    os.getenv("SKILLS_DIR", "./skills"),
    "./agent/skill/skills",
    *(os.path.join(_FEISHU_OPENCLAW_SKILLS_ROOT, name) for name in _FEISHU_OPENCLAW_SKILL_NAMES),
]

# 默认单一目录（用于 list_skills）
DEFAULT_SKILLS_DIR = DEFAULT_SKILLS_DIRS[0]


def _browser_use_python_package_installed() -> bool:
    return importlib.util.find_spec("browser_use") is not None


def _browser_use_runtime_likely_ready() -> bool:
    """启发式：常见系统浏览器、Playwright 缓存或 browser-use 配置目录是否存在。"""
    try:
        for name in ("chromium", "chromium-browser", "google-chrome", "chrome"):
            if shutil.which(name):
                return True
        pw = Path.home() / ".cache" / "ms-playwright"
        if pw.is_dir() and any(pw.glob("chromium-*")):
            return True
        if (Path.home() / ".browser_use").is_dir():
            return True
    except OSError:
        pass
    return False


def _check_skill_setup(skill_name: str) -> Optional[str]:
    """
    检查 skill 的环境配置，返回缺失依赖的警告信息

    Args:
        skill_name: Skill 名称

    Returns:
        警告信息（如果有缺失的依赖），否则返回 None
    """
    # browser-use：仓库内从未存在 agent.skill.skills.browser_use.setup，旧逻辑会恒 ImportError 并被吞掉
    if skill_name in ["browser-use", "browser_use"]:
        pkg_ok = _browser_use_python_package_installed()
        runtime_ok = _browser_use_runtime_likely_ready()
        if not pkg_ok or not runtime_ok:
            warning = "\n⚠️ **Setup Required**\n\n"
            warning += "The following dependencies may be missing:\n\n"
            if not pkg_ok:
                warning += "- Python 包：`pip install browser-use`\n"
            if not runtime_ok:
                warning += "- 浏览器运行时：安装 Chromium（例如 `uvx browser-use install` 或 `playwright install chromium`）\n"
            warning += "\n若已安装仍提示缺失，可忽略本段（检测为启发式）。\n\n"
            return warning

    return None


@tool(
    description="加载指定的 skill 文档。Skills 提供领域知识和最佳实践指导。"
)
async def skill(
    skill_name: str,
    skills_dir: Optional[str] = None,
) -> ToolResult:
    """
    加载指定的 skill 文档

    Args:
        skill_name: Skill 名称（如 "browser-use", "error-handling"）
        skills_dir: Skills 目录路径（可选，默认按优先级查找）

    Returns:
        ToolResult: 包含 skill 的详细内容

    加载顺序：
    1. 如果指定 skills_dir，只在该目录查找
    2. 否则按优先级查找：./skills/ (项目) -> ./config/skills/ (框架)
    """
    # 确定要搜索的目录列表
    if skills_dir:
        search_paths = [Path(skills_dir)]
    else:
        search_paths = [Path(d) for d in DEFAULT_SKILLS_DIRS]

    # 在目录中查找 skill 文件
    skill_file = None
    found_in_dir = None

    for skills_path in search_paths:
        if not skills_path.exists():
            continue

        # 查找文件（支持 skill-name.md 或 skill_name.md）
        for ext in [".md"]:
            for name_format in [skill_name, skill_name.replace("-", "_"), skill_name.replace("_", "-")]:
                candidate = skills_path / f"{name_format}{ext}"
                if candidate.exists():
                    skill_file = candidate
                    found_in_dir = skills_path
                    break
            if skill_file:
                break

        if skill_file:
            break

    if not skill_file:
        # 列出所有可用的 skills
        available_skills = []
        for skills_path in search_paths:
            if skills_path.exists():
                available_skills.extend([f.stem for f in skills_path.glob("**/*.md")])

        return ToolResult(
            title=f"Skill '{skill_name}' 未找到",
            output=f"可用的 skills: {', '.join(set(available_skills))}\n\n"
                   f"查找路径: {', '.join([str(p) for p in search_paths])}",
            error=f"Skill not found: {skill_name}"
        )

    # 加载 skill
    try:
        loader = SkillLoader(str(skills_path))
        skill_obj = loader.load_file(skill_file)

        if not skill_obj:
            return ToolResult(
                title="加载失败",
                output=f"无法解析 skill 文件: {skill_file.name}",
                error="Failed to parse skill file"
            )

        # 格式化输出
        output = f"# {skill_obj.name}\n\n"
        output += f"**Category**: {skill_obj.category}\n\n"

        if skill_obj.description:
            output += f"## Description\n\n{skill_obj.description}\n\n"

        # 检查 skill 的环境配置
        setup_warning = _check_skill_setup(skill_name)
        if setup_warning:
            output += setup_warning

        if skill_obj.guidelines:
            output += f"## Guidelines\n\n"
            for i, guideline in enumerate(skill_obj.guidelines, 1):
                output += f"{i}. {guideline}\n"
            output += "\n"

        return ToolResult(
            title=f"Skill: {skill_obj.name}",
            output=output,
            long_term_memory=f"已加载 skill: {skill_obj.name} ({skill_obj.category}) from {found_in_dir}",
            include_output_only_once=True,  # skill 内容只展示一次
            metadata={
                "skill_name": skill_obj.name,
                "category": skill_obj.category,
                "scope": skill_obj.scope,
                "guidelines_count": len(skill_obj.guidelines),
                "loaded_from": str(found_in_dir)
            }
        )

    except Exception as e:
        return ToolResult(
            title="加载错误",
            output=f"加载 skill 时出错: {str(e)}",
            error=str(e)
        )


@tool(
    description="列出所有可用的 skills"
)
async def list_skills(
    skills_dir: Optional[str] = None,
) -> ToolResult:
    """
    列出所有可用的 skills

    Args:
        skills_dir: Skills 目录路径（可选）

    Returns:
        ToolResult: 包含所有 skills 的列表
    """
    skills_path = Path(skills_dir or DEFAULT_SKILLS_DIR)

    if not skills_path.exists():
        return ToolResult(
            title="Skills 目录不存在",
            output=f"找不到 skills 目录: {skills_path}",
            error=f"Directory not found: {skills_path}"
        )

    try:
        loader = SkillLoader(str(skills_path))
        skills = loader.load_all()

        if not skills:
            return ToolResult(
                title="没有可用的 Skills",
                output="skills 目录中没有找到任何 .md 文件",
                metadata={"count": 0}
            )

        # 按 category 分组
        by_category = {}
        for skill_obj in skills:
            category = skill_obj.category or "general"
            if category not in by_category:
                by_category[category] = []
            by_category[category].append(skill_obj)

        # 格式化输出
        output = f"# 可用的 Skills ({len(skills)} 个)\n\n"

        for category in sorted(by_category.keys()):
            output += f"## {category.title()}\n\n"
            for skill_obj in by_category[category]:
                skill_id = skill_obj.skill_id or skill_obj.name.lower().replace(' ', '-')
                output += f"- **{skill_obj.name}** (`{skill_id}`)\n"
                if skill_obj.description:
                    desc = skill_obj.description.split('\n')[0]  # 第一行
                    output += f"  {desc[:100]}{'...' if len(desc) > 100 else ''}\n"
            output += "\n"

        output += "\n使用 `skill` 工具加载具体的 skill：`skill(skill_name=\"browser-use\")`"

        return ToolResult(
            title=f"可用 Skills ({len(skills)} 个)",
            output=output,
            long_term_memory=f"找到 {len(skills)} 个可用 skills，分为 {len(by_category)} 个类别",
            include_output_only_once=True,
            metadata={
                "count": len(skills),
                "categories": list(by_category.keys()),
                "skills": [{"name": s.name, "category": s.category} for s in skills]
            }
        )

    except Exception as e:
        return ToolResult(
            title="列出 Skills 错误",
            output=f"列出 skills 时出错: {str(e)}",
            error=str(e)
        )