zcbot/main.py

"""zcbot 命令入口: web 服务 + db migration + 模型能力探测。

REPL 形态(原 `chat / tasks / export`)已撤(2026-05-18)—— 浏览器 dev SPA
(`/static/dev.html`)+ web `/v1/*` 路由全覆盖,本地命令行 REPL 维护双套
task 创建 / resume / 切换语义已经是冗余。所有交互走 `python main.py web`
起服务后浏览器操作。

装配 lib 在 `core/agent_builder.py`(原 main.py 内容,改名归位);
本文件仅做入口 + click 命令组。
"""
from __future__ import annotations

import sys

import click

from core.agent_builder import load_config
from core.paths import ROOT
from core.ui import make_console


@click.group()
def cli() -> None:
    """zcbot - 个人任务 agent"""


# ─────────────── DB migration(alembic 包装) ───────────────

@cli.group()
def db() -> None:
    """数据库管理 (alembic upgrade/downgrade/current)。需先 export ZCBOT_DB_URL。"""


def _alembic_cfg():
    from alembic.config import Config
    return Config(str(ROOT / "alembic.ini"))


def _run_alembic(fn, *args) -> None:
    """统一包一层友好出错(ZCBOT_DB_URL 未设置 / 连不上 → 简洁报错,不打 traceback)。"""
    try:
        fn(_alembic_cfg(), *args)
    except RuntimeError as e:
        click.echo(f"[err] {e}", err=True)
        sys.exit(2)
    except Exception as e:
        click.echo(f"[err] {type(e).__name__}: {e}", err=True)
        sys.exit(3)


@db.command("upgrade")
@click.argument("revision", default="head")
def db_upgrade(revision: str) -> None:
    """alembic upgrade <revision> (default head)."""
    from alembic import command
    _run_alembic(command.upgrade, revision)


@db.command("downgrade")
@click.argument("revision")
def db_downgrade(revision: str) -> None:
    """alembic downgrade <revision> (use -1 for one step, base for all)."""
    from alembic import command
    _run_alembic(command.downgrade, revision)


@db.command("current")
def db_current() -> None:
    """alembic current -- show currently applied revision."""
    from alembic import command
    _run_alembic(command.current)


# ─────────────── Capability probing ───────────────

@cli.command()
@click.option("--model", default=None, help="模型档案,如 deepseek_v4.flash 或 deepseek_v4.pro")
@click.option("--long-context", is_flag=True, help="加跑 needle-in-haystack(费 token,默认关)")
def probe(model: str, long_context: bool) -> None:
    """实测对账模型 yaml 声称的能力。会调用 LLM,有 API 开销。"""
    from rich.table import Table

    from core.capabilities import ModelCapabilities
    from core.llm import LLM
    from core.probe import probe_capabilities

    cfg = load_config()
    name = model or cfg["default_model"]

    console = make_console()
    try:
        caps = ModelCapabilities.load(name, ROOT / cfg["models_dir"])
    except Exception as e:
        console.print(f"[err]档案加载失败:[/err] {type(e).__name__}: {e}")
        sys.exit(1)

    console.print(
        f"[bold]probing[/bold] [accent]{caps.model_id}[/accent] (profile: {name}) "
        f"[muted]long-context={long_context}[/muted]\n"
    )

    try:
        llm = LLM(caps)
    except Exception as e:
        console.print(f"[err]LLM 构造失败:[/err] {type(e).__name__}: {e}")
        sys.exit(1)

    with console.status("[muted]running probes...[/muted]", spinner="dots"):
        report = probe_capabilities(caps, llm, include_long_context=long_context)

    tbl = Table(show_lines=False)
    tbl.add_column("capability", style="bold")
    tbl.add_column("declared")
    tbl.add_column("observed")
    tbl.add_column("status")
    tbl.add_column("detail")
    color = {"ok": "ok", "mismatch": "warn", "error": "err", "skip": "muted"}
    for r in report.results:
        c = color.get(r.status, "info")
        tbl.add_row(
            r.name,
            str(r.declared),
            str(r.observed),
            f"[{c}]{r.status}[/{c}]",
            r.detail,
        )
    console.print(tbl)

    if report.has_mismatch:
        console.print(
            "\n[warn]存在能力对账差异 —— 看 detail,必要时改 "
            f"config/models/{caps.family}.yaml[/warn]"
        )
        sys.exit(2)
    if any(r.status == "error" for r in report.results):
        console.print("\n[err]部分探测出错(见 detail)[/err]")
        sys.exit(3)
    console.print("\n[ok]全部能力声明与实测一致。[/ok]")


# ─────────────── Web 服务 ───────────────

@cli.command()
@click.option("--host", default="127.0.0.1", show_default=True,
              help="监听地址。本地形态默认 127.0.0.1,不对外暴露")
@click.option("--port", default=8765, show_default=True, type=int,
              help="监听端口")
@click.option("--reload/--no-reload", default=False,
              help="dev:文件改动自动重启(uvicorn 工厂模式)")
def web(host: str, port: int, reload: bool) -> None:
    """启动 Web 服务(JSON API + dev SPA)。Auth 需 PLATFORM_KEY / JWT_SECRET env。"""
    import uvicorn

    if reload:
        # reload 模式需要 import string + factory,uvicorn 才能监听文件
        uvicorn.run("web.app:create_app", host=host, port=port,
                    reload=True, factory=True, log_level="info")
    else:
        from web.app import create_app
        uvicorn.run(create_app(), host=host, port=port, log_level="info")


if __name__ == "__main__":
    cli()