SupplyAgent/supply_agent/logging/cli.py

"""CLI for visualizing agent run logs."""

from __future__ import annotations

import argparse
import webbrowser
from pathlib import Path

from supply_agent.logging.visualize import generate_visualization

# supply_agent/logging/cli.py → project root
_PROJECT_ROOT = Path(__file__).resolve().parents[2]


def _has_run_logs(directory: Path) -> bool:
    if not directory.is_dir():
        return False
    return any(directory.glob("run_*.jsonl")) or any(directory.glob("run_*.log"))


def _resolve_logs_dir(logs_dir: str | Path) -> Path:
    """
    Resolve logs directory.

    Relative paths prefer the SupplyAgent project root (stable regardless of
    CWD). Falls back to CWD only when that location actually contains run logs.
    """
    path = Path(logs_dir)
    if path.is_absolute():
        return path.resolve()

    project_candidate = (_PROJECT_ROOT / path).resolve()
    cwd_candidate = (Path.cwd() / path).resolve()

    if _has_run_logs(project_candidate):
        return project_candidate
    if _has_run_logs(cwd_candidate):
        return cwd_candidate
    if project_candidate.is_dir():
        return project_candidate
    if cwd_candidate.is_dir():
        return cwd_candidate
    return project_candidate


def _resolve_run_path(run: str | Path) -> Path:
    path = Path(run)
    if path.is_absolute():
        return path.resolve()
    cwd_candidate = (Path.cwd() / path).resolve()
    if cwd_candidate.exists():
        return cwd_candidate
    project_candidate = (_PROJECT_ROOT / path).resolve()
    if project_candidate.exists():
        return project_candidate
    # Stem / missing suffix: try under project logs/
    logs = _resolve_logs_dir("logs")
    for suffix in (".jsonl", ".log", ""):
        candidate = logs / f"{path.name}{suffix}" if suffix else logs / path.name
        if candidate.exists():
            return candidate
    return project_candidate


def _find_latest_run(logs_dir: Path) -> Path:
    runs = _list_all_runs(logs_dir)
    if not runs:
        raise FileNotFoundError(f"No run_*.jsonl / run_*.log found in {logs_dir}")
    return runs[-1]


def _list_all_runs(logs_dir: Path) -> list[Path]:
    """
    List all unique agent runs in logs_dir.

    Prefers ``.jsonl`` over ``.log`` when both exist for the same stem.
    Sorted by mtime ascending (oldest first).
    """
    logs_dir = _resolve_logs_dir(logs_dir)
    if not logs_dir.is_dir():
        raise FileNotFoundError(f"Logs directory not found: {logs_dir}")

    by_stem: dict[str, Path] = {}
    for path in logs_dir.glob("run_*.jsonl"):
        by_stem[path.stem] = path
    for path in logs_dir.glob("run_*.log"):
        by_stem.setdefault(path.stem, path)

    return sorted(by_stem.values(), key=lambda p: p.stat().st_mtime)


def main(argv: list[str] | None = None) -> int:
    """
    Generate HTML visualizations for agent run logs.

    Parameters
    ----------
    argv:
        Optional CLI args. Use ``main()`` or ``main([])`` from Python to
        visualize all runs under the project ``logs/`` directory (CWD-independent).
    """
    parser = argparse.ArgumentParser(
        description="将 Agent 运行日志生成为可视化 HTML 页面",
    )
    parser.add_argument(
        "run",
        nargs="?",
        help="日志路径（.jsonl / .log / 不带后缀的 run id）；省略则处理 logs 目录下全部运行",
    )
    parser.add_argument(
        "-o",
        "--output",
        help="输出 HTML 路径（仅单文件模式有效；默认与日志同目录同名 .html）",
    )
    parser.add_argument(
        "--logs-dir",
        default="logs",
        help="日志目录（默认项目根下 logs；无参数时批量生成，或配合 --latest）",
    )
    parser.add_argument(
        "--latest",
        action="store_true",
        help="仅可视化 logs 目录中最新一次运行",
    )
    parser.add_argument(
        "--open",
        action="store_true",
        help="生成后在浏览器中打开（批量模式打开最新一份）",
    )
    # 直接 main() / main(None) 视为无参数批量，避免 IDE / 其它入口的 argv 干扰
    args = parser.parse_args([] if argv is None else argv)

    logs_dir = _resolve_logs_dir(args.logs_dir)

    if args.latest:
        run_paths = [_find_latest_run(logs_dir)]
    elif args.run:
        run_paths = [_resolve_run_path(args.run)]
    else:
        try:
            run_paths = _list_all_runs(logs_dir)
        except FileNotFoundError as exc:
            print(exc)
            return 1
        if not run_paths:
            print(f"未找到可可视化的日志: {logs_dir}/run_*.jsonl|*.log")
            return 1
        if args.output:
            parser.error("批量生成时不能指定 --output，请省略 -o 或指定单个 run 路径")

    outputs: list[Path] = []
    errors = 0
    for run_path in run_paths:
        try:
            out = generate_visualization(
                run_path,
                args.output if len(run_paths) == 1 else None,
            )
            outputs.append(out)
            print(f"已生成可视化页面: {out}")
        except Exception as exc:
            errors += 1
            print(f"跳过 {run_path}: {exc}")

    if not outputs:
        return 1

    if args.open:
        webbrowser.open(outputs[-1].resolve().as_uri())

    if len(outputs) > 1:
        print(f"共生成 {len(outputs)} 个页面" + (f"，失败 {errors} 个" if errors else ""))

    return 1 if errors else 0


if __name__ == "__main__":
    import sys

    raise SystemExit(main(sys.argv[1:]))