QuantEngineByItz/tools/gitea/harness.py

"""
tools/gitea/harness.py
Gitea API 하네스 CLI 진입점 - GiteaClient를 사용한 Actions 상태 조회

formula_id: GITEA_ACTIONS_HARNESS_V3

사용법:
  python tools/gitea/harness.py --help
  python tools/gitea/harness.py runs               # 최근 실행 목록
  python tools/gitea/harness.py run 2545           # 특정 run 상세
  python tools/gitea/harness.py workflows          # workflow 목록
  python tools/gitea/harness.py runners            # runner 상태
  python tools/gitea/harness.py secrets            # secrets 이름 목록
  python tools/gitea/harness.py vars               # variables 목록
  python tools/gitea/harness.py dispatch ci.yml    # workflow 트리거
  python tools/gitea/harness.py health             # 전체 상태 요약

토큰: GITEA_TOKEN_BAIK or GITEA_TOKEN_TAXBAIK 환경변수 자동 탐지
"""
from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

# tools/ 가 sys.path에 없을 때를 대비
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))

from tools.gitea.client import GiteaClient, GiteaApiError

ROOT = Path(__file__).resolve().parents[2]


def _out(data: object, out_path: Path | None = None) -> None:
    """JSON으로 출력하고, --out 지정 시 파일에도 저장"""
    text = json.dumps(data, ensure_ascii=False, indent=2)
    print(text)
    if out_path:
        out_path.parent.mkdir(parents=True, exist_ok=True)
        out_path.write_text(text, encoding="utf-8")


def cmd_runs(client: GiteaClient, args: argparse.Namespace) -> int:
    """최근 Actions 실행 목록"""
    runs = client.list_runs(limit=args.limit)
    result = {
        "formula_id": "GITEA_ACTIONS_HARNESS_V3",
        "command": "runs",
        "count": len(runs),
        "runs": [
            {
                "id": r.get("id"),
                "run_number": r.get("run_number"),
                "workflow": r.get("path", "").split("@")[0],
                "event": r.get("event"),
                "status": r.get("status"),
                "conclusion": r.get("conclusion"),
                "head_branch": r.get("head_branch"),
                "started_at": r.get("started_at"),
                "completed_at": r.get("completed_at"),
                "actor": (r.get("actor") or {}).get("login"),
                "display_title": r.get("display_title"),
            }
            for r in runs
        ],
    }
    _out(result, Path(ROOT / "Temp" / "gitea_harness_runs.json") if not args.no_save else None)
    return 0


def cmd_run(client: GiteaClient, args: argparse.Namespace) -> int:
    """특정 run 상세 + jobs"""
    run_id = args.run_id
    try:
        run = client.get_run(run_id)
    except GiteaApiError as e:
        print(json.dumps({"error": str(e), "status": e.status}), file=sys.stderr)
        return 1

    try:
        jobs = client.list_run_jobs(run_id)
    except GiteaApiError:
        jobs = []

    failed_steps = []
    for job in jobs:
        for step in (job.get("steps") or []):
            if step.get("conclusion") not in ("success", "skipped", None):
                failed_steps.append({
                    "job": job.get("name"),
                    "step": step.get("name"),
                    "conclusion": step.get("conclusion"),
                })

    result = {
        "formula_id": "GITEA_ACTIONS_HARNESS_V3",
        "command": "run",
        "run_id": run_id,
        "status": run.get("status"),
        "conclusion": run.get("conclusion"),
        "workflow": run.get("path", "").split("@")[0],
        "display_title": run.get("display_title"),
        "event": run.get("event"),
        "head_branch": run.get("head_branch"),
        "head_sha": run.get("head_sha"),
        "started_at": run.get("started_at"),
        "completed_at": run.get("completed_at"),
        "actor": (run.get("actor") or {}).get("login"),
        "jobs_count": len(jobs),
        "failed_steps": failed_steps,
        "gate": "PASS" if run.get("conclusion") == "success" else "FAIL",
    }
    _out(result, Path(ROOT / "Temp" / f"gitea_harness_run_{run_id}.json") if not args.no_save else None)
    return 0 if run.get("conclusion") == "success" else 1


def cmd_workflows(client: GiteaClient, args: argparse.Namespace) -> int:
    """workflow 목록"""
    try:
        wfs = client.list_workflows()
    except GiteaApiError as e:
        print(json.dumps({"error": str(e)}), file=sys.stderr)
        return 1

    result = {
        "formula_id": "GITEA_ACTIONS_HARNESS_V3",
        "command": "workflows",
        "count": len(wfs),
        "workflows": wfs,
    }
    _out(result, Path(ROOT / "Temp" / "gitea_harness_workflows.json") if not args.no_save else None)
    return 0


def cmd_runners(client: GiteaClient, args: argparse.Namespace) -> int:
    """runner 상태 목록"""
    try:
        runners = client.list_runners()
    except GiteaApiError as e:
        print(json.dumps({"error": str(e), "note": "admin token required"}), file=sys.stderr)
        return 1

    result = {
        "formula_id": "GITEA_ACTIONS_HARNESS_V3",
        "command": "runners",
        "count": len(runners),
        "runners": [
            {
                "id": r.get("id"),
                "name": r.get("name"),
                "status": r.get("status"),
                "labels": [lb.get("name") for lb in (r.get("labels") or [])],
                "os": r.get("os"),
                "version": r.get("version"),
            }
            for r in runners
        ],
    }
    _out(result, Path(ROOT / "Temp" / "gitea_harness_runners.json") if not args.no_save else None)
    return 0


def cmd_secrets(client: GiteaClient, args: argparse.Namespace) -> int:
    """secrets 이름 목록 (값은 반환되지 않음)"""
    try:
        secrets = client.list_secrets()
    except GiteaApiError as e:
        print(json.dumps({"error": str(e)}), file=sys.stderr)
        return 1

    result = {
        "formula_id": "GITEA_ACTIONS_HARNESS_V3",
        "command": "secrets",
        "count": len(secrets),
        "secret_names": [s.get("name") for s in secrets],
    }
    _out(result, Path(ROOT / "Temp" / "gitea_harness_secrets.json") if not args.no_save else None)
    return 0


def cmd_variables(client: GiteaClient, args: argparse.Namespace) -> int:
    """variables 목록"""
    try:
        variables = client.list_variables()
    except GiteaApiError as e:
        print(json.dumps({"error": str(e)}), file=sys.stderr)
        return 1

    result = {
        "formula_id": "GITEA_ACTIONS_HARNESS_V3",
        "command": "vars",
        "count": len(variables),
        "variables": variables,
    }
    _out(result, Path(ROOT / "Temp" / "gitea_harness_variables.json") if not args.no_save else None)
    return 0


def cmd_dispatch(client: GiteaClient, args: argparse.Namespace) -> int:
    """workflow dispatch 트리거"""
    try:
        client.dispatch_workflow(args.workflow_file, ref=args.ref)
        result = {
            "formula_id": "GITEA_ACTIONS_HARNESS_V3",
            "command": "dispatch",
            "workflow": args.workflow_file,
            "ref": args.ref,
            "gate": "PASS",
        }
        _out(result, None)
        return 0
    except GiteaApiError as e:
        print(json.dumps({"error": str(e), "gate": "FAIL"}), file=sys.stderr)
        return 1


def cmd_health(client: GiteaClient, args: argparse.Namespace) -> int:
    """전체 상태 요약 - repo + 최근 실행 + token 정보"""
    errors: list[str] = []

    # 1. repo 접근 확인
    try:
        repo = client.get_repo()
        repo_ok = True
    except GiteaApiError as e:
        repo_ok = False
        errors.append(f"repo: {e}")
        repo = {}

    # 2. 최근 실행 목록
    try:
        runs = client.list_runs(limit=5)
    except GiteaApiError as e:
        runs = []
        errors.append(f"runs: {e}")

    # 최근 ci.yml 결론
    ci_conclusion = None
    for r in runs:
        if "ci.yml" in (r.get("path") or ""):
            ci_conclusion = r.get("conclusion")
            break

    result = {
        "formula_id": "GITEA_ACTIONS_HARNESS_V3",
        "command": "health",
        "gate": "PASS" if not errors else "WARN",
        "client_summary": client.summary(),
        "repo_accessible": repo_ok,
        "repo_name": repo.get("full_name"),
        "recent_runs_count": len(runs),
        "latest_ci_conclusion": ci_conclusion,
        "recent_runs": [
            {
                "id": r.get("id"),
                "workflow": r.get("path", "").split("@")[0],
                "conclusion": r.get("conclusion"),
                "started_at": r.get("started_at"),
            }
            for r in runs
        ],
        "errors": errors,
    }
    _out(result, Path(ROOT / "Temp" / "gitea_harness_health.json") if not args.no_save else None)
    return 0 if not errors else 1


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Gitea API 하네스 CLI (formula_id: GITEA_ACTIONS_HARNESS_V3)",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Commands:
  runs                 최근 Actions 실행 목록
  run <run_id>         특정 run 상세 및 실패 스텝
  workflows            workflow 목록
  runners              runner 상태 (admin token 필요)
  secrets              secrets 이름 목록
  vars                 variables 목록
  dispatch <file.yml>  workflow dispatch 트리거
  health               전체 상태 요약

환경변수 (우선순위):
  GITEA_TOKEN_BAIK      사용자 지정 별칭
  GITEA_TOKEN_TAXBAIK   기존 표준
  GITEA_TOKEN           일반 fallback
  GITEA_TOKEN_HOME      레거시
        """,
    )
    ap.add_argument("--base-url", default="https://gitea.taxbaik.com")
    ap.add_argument("--owner", default="kjh2064")
    ap.add_argument("--repo", default="QuantEngineByItz")
    ap.add_argument("--token", default="", help="명시적 API 토큰 (없으면 환경변수 자동 탐지)")
    ap.add_argument("--limit", type=int, default=15, help="runs 명령의 최대 개수 (기본 15)")
    ap.add_argument("--no-save", action="store_true", help="Temp/ 결과 파일 저장 생략")

    # 공통 옵션을 서브파서에 상속시키는 parent 파서
    common = argparse.ArgumentParser(add_help=False)
    common.add_argument("--no-save", action="store_true", help="Temp/ 결과 파일 저장 생략")
    common.add_argument("--limit", type=int, default=15, help="목록 최대 개수 (기본 15)")

    sub = ap.add_subparsers(dest="command", required=True)

    runs_p = sub.add_parser("runs", parents=[common], help="최근 Actions 실행 목록")

    run_p = sub.add_parser("run", parents=[common], help="특정 run 상세")
    run_p.add_argument("run_id", type=int)

    sub.add_parser("workflows", parents=[common], help="workflow 목록")
    sub.add_parser("runners", parents=[common], help="runner 상태")
    sub.add_parser("secrets", parents=[common], help="secrets 이름 목록")
    sub.add_parser("vars", parents=[common], help="variables 목록")

    dispatch_p = sub.add_parser("dispatch", parents=[common], help="workflow dispatch 트리거")
    dispatch_p.add_argument("workflow_file", help="e.g. ci.yml")
    dispatch_p.add_argument("--ref", default="main", help="브랜치 (기본 main)")

    sub.add_parser("health", parents=[common], help="전체 상태 요약")

    args = ap.parse_args()

    client = GiteaClient(
        base_url=args.base_url,
        owner=args.owner,
        repo=args.repo,
        token=args.token,
    )

    dispatch_table = {
        "runs": cmd_runs,
        "run": cmd_run,
        "workflows": cmd_workflows,
        "runners": cmd_runners,
        "secrets": cmd_secrets,
        "vars": cmd_variables,
        "dispatch": cmd_dispatch,
        "health": cmd_health,
    }

    handler = dispatch_table.get(args.command)
    if handler is None:
        print(f"알 수 없는 명령: {args.command}", file=sys.stderr)
        return 1

    return handler(client, args)


if __name__ == "__main__":
    raise SystemExit(main())