AgentEvalTool/backend/cli/target.py

"""CLI commands for evaluation target management."""

import asyncio
import json
from pathlib import Path
from typing import Optional

import typer
from agenteval.channels.factory import ChannelFactory
from agenteval.models import ChannelType, EvalTarget, PlatformType, TargetStatus
from agenteval.storage.repository import TargetRepository
from rich.console import Console
from rich.table import Table

app = typer.Typer(help="评测对象管理")
console = Console()


def _load_config_file(path: Path) -> dict:
    text = path.read_text(encoding="utf-8")
    return json.loads(text)


@app.command("list", help="列出所有评测对象")
def list_targets() -> None:
    repo = TargetRepository()
    targets = repo.list_all()
    if not targets:
        console.print("暂无评测对象")
        return

    table = Table(show_header=True, header_style="bold")
    table.add_column("ID")
    table.add_column("名称")
    table.add_column("平台")
    table.add_column("通道类型")
    table.add_column("状态")
    table.add_column("创建时间")

    for t in targets:
        table.add_row(
            t.id or "",
            t.name,
            t.platform.value,
            t.channel_type.value,
            t.status.value,
            str(t.created_at)[:19] if t.created_at else "",
        )
    console.print(table)


@app.command("add", help="添加评测对象")
def add_target(
    name: str = typer.Option(..., help="评测对象名称"),
    config: Path = typer.Option(..., help="通道配置文件路径 (JSON)"),
    description: str = typer.Option("", help="描述"),
    platform: PlatformType = typer.Option(PlatformType.AI_DIGITAL_EMPLOYEE, help="平台类型"),
    channel_type: ChannelType = typer.Option(ChannelType.TUTU_API, help="通道类型"),
) -> None:
    cfg = _load_config_file(config)
    target = EvalTarget(
        name=name,
        description=description,
        platform=platform,
        channel_type=channel_type,
        channel_config=cfg,
        status=TargetStatus.ACTIVE,
    )
    created = TargetRepository().create(target)
    console.print(f"已创建评测对象: {created.id} - {created.name}")


@app.command("get", help="查看评测对象详情")
def get_target(target_id: str) -> None:
    target = TargetRepository().get(target_id)
    if not target:
        console.print(f"评测对象不存在: {target_id}", style="red")
        raise typer.Exit(1)

    console.print_json(data=target.model_dump())


@app.command("update", help="更新评测对象")
def update_target(
    target_id: str,
    config: Path = typer.Option(..., help="通道配置文件路径 (JSON)"),
    name: Optional[str] = typer.Option(None, help="名称"),
    description: Optional[str] = typer.Option(None, help="描述"),
) -> None:
    repo = TargetRepository()
    target = repo.get(target_id)
    if not target:
        console.print(f"评测对象不存在: {target_id}", style="red")
        raise typer.Exit(1)

    cfg = _load_config_file(config)
    target.channel_config = cfg
    if name is not None:
        target.name = name
    if description is not None:
        target.description = description

    updated = repo.update(target)
    if updated:
        console.print(f"已更新评测对象: {updated.id}")
    else:
        console.print("更新失败", style="red")
        raise typer.Exit(1)


@app.command("remove", help="删除评测对象")
def remove_target(target_id: str) -> None:
    if TargetRepository().delete(target_id):
        console.print(f"已删除评测对象: {target_id}")
    else:
        console.print(f"评测对象不存在: {target_id}", style="red")
        raise typer.Exit(1)


@app.command("test", help="检测评测对象通道连通性")
def test_target(target_id: str) -> None:
    repo = TargetRepository()
    target = repo.get(target_id)
    if not target:
        console.print(f"评测对象不存在: {target_id}", style="red")
        raise typer.Exit(1)

    channel = ChannelFactory.create(target)
    health = asyncio.run(channel.health_check())
    if health.ok:
        console.print(f"通道正常: {health.message}", style="green")
    else:
        console.print(f"通道异常: {health.message}", style="red")
        raise typer.Exit(1)