QuantEngineByItz/tools/inspect_gitea_actions_run_v2.py

#!/usr/bin/env python3
"""
Gitea Actions Run Detail Inspector v2
AGENTS.md 지침 준수: Gitea API를 하네스로 조회. 직접 서버 접근 금지.
formula_id: GITEA_ACTIONS_RUN_DETAIL_V2

토큰 우선순위 (높은 것부터):
  1. --token CLI 인자
  2. GITEA_TOKEN_BAIK  (사용자 지정 별칭)
  3. GITEA_TOKEN_TAXBAIK  (기존 표준)
  4. GITEA_TOKEN  (일반 fallback)
  5. GITEA_TOKEN_HOME  (레거시)
"""
from __future__ import annotations

import argparse
import json
import os
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_BASE_URL = "https://gitea.taxbaik.com"
DEFAULT_OWNER = "kjh2064"
DEFAULT_REPO = "QuantEngineByItz"


def _get_token(cli_token: str = "") -> str:
    """토큰 우선순위: CLI → GITEA_TOKEN_BAIK → GITEA_TOKEN_TAXBAIK → GITEA_TOKEN → GITEA_TOKEN_HOME"""
    if cli_token and cli_token.strip():
        return cli_token.strip()
    for env_key in ("GITEA_TOKEN_BAIK", "GITEA_TOKEN_TAXBAIK", "GITEA_TOKEN", "GITEA_TOKEN_HOME"):
        val = os.environ.get(env_key, "").strip()
        if val:
            return val
    return ""


def _request_json(url: str, token: str = "") -> tuple[int, str, Any]:
    headers = {
        "Accept": "application/json",
        "User-Agent": "QuantEngine-Harness/2.0",
    }
    if token:
        headers["Authorization"] = f"token {token}"
    req = urllib.request.Request(url, headers=headers, method="GET")
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            raw = resp.read().decode("utf-8", errors="replace")
            payload = json.loads(raw) if raw else None
            return resp.status, resp.reason, payload
    except urllib.error.HTTPError as exc:
        raw = exc.read().decode("utf-8", errors="replace")
        try:
            payload = json.loads(raw)
        except Exception:
            payload = raw
        return exc.code, exc.reason or "", payload


def _summary_run(run: dict) -> dict:
    return {
        "id": run.get("id"),
        "run_number": run.get("run_number"),
        "display_title": run.get("display_title"),
        "workflow_path": run.get("path"),
        "event": run.get("event"),
        "status": run.get("status"),
        "conclusion": run.get("conclusion"),
        "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"),
    }


def main() -> int:
    ap = argparse.ArgumentParser(description="Gitea Actions Run Detail Inspector v2")
    ap.add_argument("--base-url", default=DEFAULT_BASE_URL)
    ap.add_argument("--owner", default=DEFAULT_OWNER)
    ap.add_argument("--repo", default=DEFAULT_REPO)
    ap.add_argument("--run-id", type=int, default=2545, help="Gitea Actions run ID")
    ap.add_argument("--token", default="", help="Gitea API Personal Access Token")
    ap.add_argument("--all-runs", action="store_true", help="List recent N runs")
    ap.add_argument("--limit", type=int, default=10, help="Number of recent runs to list when --all-runs")
    args = ap.parse_args()

    token = _get_token(args.token)
    base = f"{args.base_url.rstrip('/')}/api/v1/repos/{args.owner}/{args.repo}"
    errors: list[str] = []
    evidence: dict[str, Any] = {
        "owner": args.owner,
        "repo": args.repo,
        "target_run_id": args.run_id,
        "token_provided": bool(token),
    }

    # 1. Target run details
    run_url = f"{base}/actions/runs/{args.run_id}"
    s1, _, run_payload = _request_json(run_url, token=token)
    evidence["target_run_http_status"] = s1

    if s1 != 200:
        errors.append(f"Run #{args.run_id} fetch failed: HTTP {s1}")
        target_run_summary = None
        jobs_data = None
    else:
        target_run_summary = _summary_run(run_payload)

        # 2. Fetch jobs for this run
        jobs_url = f"{base}/actions/runs/{args.run_id}/jobs"
        s2, _, jobs_payload = _request_json(jobs_url, token=token)
        evidence["jobs_http_status"] = s2

        if s2 == 200 and isinstance(jobs_payload, dict):
            raw_jobs = jobs_payload.get("workflow_jobs") or []
            jobs_data = []
            for job in raw_jobs:
                steps = job.get("steps") or []
                failed_steps = [
                    {
                        "step_number": st.get("number"),
                        "name": st.get("name"),
                        "conclusion": st.get("conclusion"),
                        "started_at": st.get("started_at"),
                        "completed_at": st.get("completed_at"),
                    }
                    for st in steps
                    if st.get("conclusion") not in ("success", "skipped", None)
                ]
                jobs_data.append({
                    "job_id": job.get("id"),
                    "job_name": job.get("name"),
                    "status": job.get("status"),
                    "conclusion": job.get("conclusion"),
                    "started_at": job.get("started_at"),
                    "completed_at": job.get("completed_at"),
                    "runner_name": job.get("runner_name"),
                    "total_steps": len(steps),
                    "failed_steps": failed_steps,
                })
        else:
            jobs_data = None
            if s2 != 200:
                errors.append(f"Jobs fetch failed: HTTP {s2}")

    # 3. Recent runs list (optional)
    recent_runs = None
    if args.all_runs:
        runs_url = f"{base}/actions/runs?limit={args.limit}"
        s3, _, runs_payload = _request_json(runs_url, token=token)
        evidence["runs_list_http_status"] = s3
        if s3 == 200 and isinstance(runs_payload, dict):
            raw_runs = runs_payload.get("workflow_runs") or []
            recent_runs = [_summary_run(r) for r in raw_runs]
        else:
            errors.append(f"Runs list fetch failed: HTTP {s3}")

    result = {
        "formula_id": "GITEA_ACTIONS_RUN_DETAIL_V2",
        "gate": "PASS" if not errors else "FAIL",
        "errors": errors,
        "evidence": evidence,
        "run_summary": target_run_summary,
        "jobs": jobs_data,
        "recent_runs": recent_runs,
    }

    out = ROOT / "Temp" / f"gitea_actions_run_{args.run_id}_detail_v2.json"
    out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0 if not errors else 1


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