plugins/browser/api.py

"""
浏览器MCP插件 API 路由（v3.0）。

端点：
- GET  /                API 根
- GET  /status          插件状态
- GET  /binary-check    Lightpanda 二进制探测
- GET  /tools           列出 28 个上游工具（含 schema）
- GET  /health          健康检查（二进制 + 子进程 + 漂移校验）
- POST /mcp/{tool}      同步调用 MCP 工具（前端用）
- POST /process/stop    停止 lightpanda 子进程
- GET  /process/logs    查看子进程 stderr 最近 N 行
"""

from __future__ import annotations

import logging
from typing import Any, Dict, Optional

from fastapi import APIRouter, HTTPException

from .browser.binary_locator import locate_lightpanda
from .browser.client import LightpandaClient
from .browser.tools_meta import TOOLS_META

logger = logging.getLogger(__name__)

router = APIRouter()

# 运行时注入的插件实例（由 plugin_loader 调用 set_plugin_instance 注入）
plugin = None


def set_plugin_instance(plugin_instance):
    """由系统调用，注入插件实例。"""
    global plugin
    plugin = plugin_instance


@router.get("/")
async def api_root():
    """API 根路径。"""
    return {"message": "浏览器MCP插件API", "backend": "lightpanda", "status": "运行中"}


@router.get("/status")
async def get_status():
    """获取插件状态。"""
    if plugin is None:
        return {
            "name": "browser",
            "enabled": False,
            "message": "插件未加载",
        }
    return plugin.get_status()


@router.get("/binary-check")
async def binary_check():
    """Lightpanda 二进制结构化探测（路径/版本/install_hints）。"""
    return locate_lightpanda().to_dict()


@router.get("/tools")
async def list_tools():
    """列出 28 个上游工具的元数据（含 input_schema，供前端动态表单）。"""
    return {
        "tools": [
            {
                "name": f"browser-{m.name}",
                "upstream_name": m.name,
                "title": m.title,
                "summary": m.summary,
                "description": m.description,
                "input_schema": m.input_schema,
                "read_only": m.read_only,
            }
            for m in TOOLS_META
        ],
        "total": len(TOOLS_META),
    }


@router.get("/health")
async def health():
    """健康检查：二进制可用性 + 子进程存活 + 上游工具集漂移校验。"""
    check = locate_lightpanda()
    client = LightpandaClient.instance()
    process_alive = False
    drift: Optional[Dict[str, Any]] = None
    try:
        process_alive = client.process_alive
    except Exception:
        process_alive = False

    # 子进程已启动时校验上游工具集与静态 meta 是否漂移
    cached = client.tools_cache
    if cached is not None:
        upstream_names = {t.get("name") for t in cached if isinstance(t, dict)}
        meta_names = {m.name for m in TOOLS_META}
        missing = sorted(meta_names - upstream_names)   # meta 有、上游无
        extra = sorted(upstream_names - meta_names)      # 上游有、meta 无
        if missing or extra:
            drift = {"missing_in_upstream": missing, "extra_in_upstream": extra}

    return {
        "binary_available": check.available,
        "binary_path": check.path,
        "binary_version": check.version,
        "process_alive": process_alive,
        "tools_cached": cached is not None,
        "tools_count": len(cached) if cached else 0,
        "drift": drift,
    }


@router.post("/mcp/{tool_name}")
async def call_mcp_tool(tool_name: str, body: Dict[str, Any]):
    """同步调用 MCP 工具；返回结构化结果（run_id + result/error_code）。

    前端轮询 run_log 时用返回的 run_id。
    """
    from . import mcp as mcp_module

    # tool_name 形如 "browser-goto" 或 "goto"，统一查 _mcp_tool_def
    func = None
    for name in dir(mcp_module):
        attr = getattr(mcp_module, name, None)
        if callable(attr) and hasattr(attr, "_mcp_tool_def"):
            defn = attr._mcp_tool_def
            registered = defn.get("name")
            if registered == tool_name or registered == f"browser-{tool_name}" or registered.endswith(f"-{tool_name}"):
                func = attr
                break
    if func is None:
        raise HTTPException(status_code=404, detail=f"未找到 MCP 工具: {tool_name}")

    try:
        # 工具签名: async def _tool(params: Optional[dict] = None)
        result = await func(body or {})
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"工具执行失败: {e}")

    return result


@router.post("/process/stop")
async def process_stop():
    """停止 lightpanda 子进程（下次工具调用会重新懒启动）。"""
    await LightpandaClient.instance().close()
    return {"success": True, "message": "lightpanda 子进程已停止"}


@router.get("/process/logs")
async def process_logs(tail: int = 50):
    """查看子进程 stderr 最近 N 行。"""
    from pathlib import Path

    log_path = Path("data/lightpanda") / "mcp.log"
    if not log_path.exists():
        return {"lines": [], "message": "日志文件不存在（子进程尚未启动过）"}
    try:
        data = log_path.read_bytes()
        text = data.decode("utf-8", errors="replace")
        lines = text.splitlines()[-tail:]
        return {"lines": lines, "log_path": str(log_path)}
    except OSError as e:
        raise HTTPException(status_code=500, detail=f"读取日志失败: {e}")