diff --git a/AGENTS.md b/AGENTS.md index d42d8bec..31b8f5a8 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -84,6 +84,12 @@ - `tools/validate_qualitative_sell_strategy_pipeline_v1.py`: qualitative sell validator. - `tools/validate_gitea_secrets_contract_v1.py`: Gitea secrets validator. - `tools/validate_gitea_ci_workflow_lint_v1.py`: CI workflow lint validator for recurring service-binding mistakes. +- `tools/validate_gitea_pr_harness_v1.py`: Gitea PR 생성/조회 하네스. +- `tools/validate_gitea_token_home_v1.py`: Gitea 토큰 유효성 검증용 하네스. +- `tools/gitea/`: **Gitea API 하네스 패키지** (단일 권위). 토큰 우선순위: `GITEA_TOKEN_BAIK` → `GITEA_TOKEN_TAXBAIK` → `GITEA_TOKEN` → `GITEA_TOKEN_HOME`. + - `tools/gitea/client.py`: `GiteaClient` - SOLID SRP 기반 Gitea REST API v1 클라이언트 (runs/jobs/secrets/vars/runners/PR/releases 지원). + - `tools/gitea/harness.py`: CLI 하네스 진입점. `python tools/gitea/harness.py health|runs|run |secrets|vars|workflows|runners|dispatch ` 형식으로 사용. + - `tools/inspect_gitea_actions_run_v1.py` / `v2.py`: 구 하네스 (레거시, `tools/gitea/harness.py run `으로 대체). - `tools/validate_snapshot_admin_web_v1.py`: snapshot admin smoke validator. - `tests/parity/test_price_qty_parity_v1.py`: price/qty parity. - `tests/parity/test_score_parity_v1.py`: timing score parity. diff --git a/tools/gitea/__init__.py b/tools/gitea/__init__.py new file mode 100644 index 00000000..c83d4166 --- /dev/null +++ b/tools/gitea/__init__.py @@ -0,0 +1,20 @@ +""" +tools/gitea/__init__.py +Gitea API 공통 클라이언트 패키지 - QuantEngine 하네스 표준 + +토큰 우선순위: + 1. GITEA_TOKEN_BAIK (사용자 지정 별칭) + 2. GITEA_TOKEN_TAXBAIK (기존 표준, 현재 환경 보유) + 3. GITEA_TOKEN (일반 fallback) + 4. GITEA_TOKEN_HOME (레거시) + +사용법: + from tools.gitea import GiteaClient + client = GiteaClient() # 환경변수 자동 탐지 + run = client.get_run(2545) + runs = client.list_runs(limit=10) +""" + +from .client import GiteaClient, GiteaApiError + +__all__ = ["GiteaClient", "GiteaApiError"] diff --git a/tools/gitea/client.py b/tools/gitea/client.py new file mode 100644 index 00000000..34165cda --- /dev/null +++ b/tools/gitea/client.py @@ -0,0 +1,315 @@ +""" +tools/gitea/client.py +Gitea API 공통 클라이언트 - SOLID SRP 준수, 하나의 책임: Gitea REST API 통신 + +formula_id: GITEA_API_CLIENT_V1 +""" +from __future__ import annotations + +import json +import os +import urllib.error +import urllib.request +from typing import Any + +DEFAULT_BASE_URL = "https://gitea.taxbaik.com" +DEFAULT_OWNER = "kjh2064" +DEFAULT_REPO = "QuantEngineByItz" + +# 토큰 환경변수 우선순위 (높은 것부터) +_TOKEN_ENV_PRIORITY = ( + "GITEA_TOKEN_BAIK", # 사용자 지정 별칭 + "GITEA_TOKEN_TAXBAIK", # 기존 표준 (현재 환경 보유) + "GITEA_TOKEN", # 일반 fallback + "GITEA_TOKEN_HOME", # 레거시 +) + + +class GiteaApiError(Exception): + """Gitea API 호출 오류""" + def __init__(self, status: int, reason: str, body: Any) -> None: + self.status = status + self.reason = reason + self.body = body + super().__init__(f"HTTP {status} {reason}: {body}") + + +def _resolve_token(explicit: str = "") -> str: + """ + 토큰 해석 우선순위: + 1. explicit 인자 (CLI --token 등) + 2. GITEA_TOKEN_BAIK + 3. GITEA_TOKEN_TAXBAIK + 4. GITEA_TOKEN + 5. GITEA_TOKEN_HOME + """ + if explicit and explicit.strip(): + return explicit.strip() + for key in _TOKEN_ENV_PRIORITY: + val = os.environ.get(key, "").strip() + if val: + return val + return "" + + +class GiteaClient: + """ + Gitea REST API v1 클라이언트 + + 사용 예시: + client = GiteaClient() # 환경변수 자동 탐지 + client = GiteaClient(token="my_token") # 명시적 토큰 + client = GiteaClient(base_url="https://my.gitea") # 다른 인스턴스 + + 모든 메서드는 dict/list를 반환하며, HTTP 오류 시 GiteaApiError를 발생시킵니다. + """ + + def __init__( + self, + base_url: str = DEFAULT_BASE_URL, + owner: str = DEFAULT_OWNER, + repo: str = DEFAULT_REPO, + token: str = "", + timeout: int = 30, + ) -> None: + self.base_url = base_url.rstrip("/") + self.owner = owner + self.repo = repo + self.token = _resolve_token(token) + self.timeout = timeout + self._api = f"{self.base_url}/api/v1" + self._repo_url = f"{self._api}/repos/{self.owner}/{self.repo}" + + # ------------------------------------------------------------------ + # 내부 HTTP 레이어 + # ------------------------------------------------------------------ + + def _request( + self, + url: str, + method: str = "GET", + body: dict | None = None, + ) -> Any: + """HTTP 요청 → JSON 반환. 오류 시 GiteaApiError 발생.""" + headers: dict[str, str] = { + "Accept": "application/json", + "User-Agent": "QuantEngine-Gitea-Client/1.0", + } + if self.token: + headers["Authorization"] = f"token {self.token}" + + data: bytes | None = None + if body is not None: + headers["Content-Type"] = "application/json" + data = json.dumps(body).encode("utf-8") + + req = urllib.request.Request(url, data=data, headers=headers, method=method) + try: + with urllib.request.urlopen(req, timeout=self.timeout) as resp: + raw = resp.read().decode("utf-8", errors="replace") + return json.loads(raw) if raw else None + except urllib.error.HTTPError as exc: + raw = exc.read().decode("utf-8", errors="replace") + try: + payload = json.loads(raw) + except Exception: + payload = raw + raise GiteaApiError(exc.code, exc.reason or "", payload) + + # ------------------------------------------------------------------ + # Repository + # ------------------------------------------------------------------ + + def get_repo(self) -> dict: + """저장소 정보 조회""" + return self._request(self._repo_url) + + # ------------------------------------------------------------------ + # Actions Runs + # ------------------------------------------------------------------ + + def list_runs(self, limit: int = 20, page: int = 1) -> list[dict]: + """ + 최근 Actions 실행 목록 조회 + + Args: + limit: 반환할 최대 개수 (기본 20) + page: 페이지 번호 (기본 1) + + Returns: + workflow_run 목록 (list[dict]) + """ + url = f"{self._repo_url}/actions/runs?limit={limit}&page={page}" + payload = self._request(url) + if isinstance(payload, dict): + return payload.get("workflow_runs") or [] + return [] + + def get_run(self, run_id: int) -> dict: + """ + 특정 Actions 실행 상세 조회 + + Args: + run_id: Gitea Actions run ID + + Returns: + workflow_run dict + """ + url = f"{self._repo_url}/actions/runs/{run_id}" + return self._request(url) + + def list_run_jobs(self, run_id: int) -> list[dict]: + """ + 특정 실행의 job 목록 조회 + + Args: + run_id: Gitea Actions run ID + + Returns: + workflow_job 목록 (list[dict]) + """ + url = f"{self._repo_url}/actions/runs/{run_id}/jobs" + payload = self._request(url) + if isinstance(payload, dict): + return payload.get("workflow_jobs") or [] + return [] + + def list_workflows(self) -> list[dict]: + """ + 저장소 workflow 목록 조회 + + Returns: + workflow 목록 (list[dict]) + """ + url = f"{self._repo_url}/actions/workflows" + payload = self._request(url) + if isinstance(payload, dict): + return payload.get("workflows") or [] + return [] + + def dispatch_workflow(self, workflow_id: str, ref: str = "main", inputs: dict | None = None) -> None: + """ + workflow_dispatch 이벤트 트리거 + + Args: + workflow_id: workflow 파일명 (e.g. 'ci.yml') + ref: 브랜치명 (기본 'main') + inputs: workflow_dispatch inputs dict + """ + url = f"{self._repo_url}/actions/workflows/{workflow_id}/dispatches" + body: dict[str, Any] = {"ref": ref} + if inputs: + body["inputs"] = inputs + self._request(url, method="POST", body=body) + + # ------------------------------------------------------------------ + # Pull Requests + # ------------------------------------------------------------------ + + def list_prs(self, state: str = "open") -> list[dict]: + """ + PR 목록 조회 + + Args: + state: 'open' | 'closed' | 'all' + + Returns: + PR 목록 (list[dict]) + """ + url = f"{self._repo_url}/pulls?state={state}" + payload = self._request(url) + return payload if isinstance(payload, list) else [] + + def create_pr( + self, + title: str, + head: str, + base: str = "main", + body: str = "", + ) -> dict: + """ + PR 생성 + + Args: + title: PR 제목 + head: 소스 브랜치 + base: 타겟 브랜치 (기본 'main') + body: PR 본문 + + Returns: + 생성된 PR dict + """ + url = f"{self._repo_url}/pulls" + return self._request(url, method="POST", body={ + "title": title, + "head": head, + "base": base, + "body": body, + }) + + # ------------------------------------------------------------------ + # Releases + # ------------------------------------------------------------------ + + def get_latest_release(self) -> dict: + """최신 릴리스 조회""" + url = f"{self._repo_url}/releases/latest" + return self._request(url) + + def list_releases(self, limit: int = 10) -> list[dict]: + """릴리스 목록 조회""" + url = f"{self._repo_url}/releases?limit={limit}" + payload = self._request(url) + return payload if isinstance(payload, list) else [] + + # ------------------------------------------------------------------ + # Secrets / Variables (관리용) + # ------------------------------------------------------------------ + + def list_secrets(self) -> list[dict]: + """저장소 secrets 이름 목록 (값은 반환되지 않음)""" + url = f"{self._repo_url}/actions/secrets" + payload = self._request(url) + return payload if isinstance(payload, list) else [] + + def list_variables(self) -> list[dict]: + """저장소 actions variables 목록""" + url = f"{self._repo_url}/actions/variables" + payload = self._request(url) + if isinstance(payload, dict): + return payload.get("variables") or [] + return [] + + # ------------------------------------------------------------------ + # Runners + # ------------------------------------------------------------------ + + def list_runners(self) -> list[dict]: + """저장소에 연결된 Actions runner 목록""" + url = f"{self._api}/admin/runners" + payload = self._request(url) + if isinstance(payload, dict): + return payload.get("runners") or [] + return [] + + # ------------------------------------------------------------------ + # Utility + # ------------------------------------------------------------------ + + def summary(self) -> dict: + """클라이언트 설정 요약 (토큰값 비노출)""" + return { + "base_url": self.base_url, + "owner": self.owner, + "repo": self.repo, + "token_source": self._token_source(), + "token_chars": len(self.token), + } + + def _token_source(self) -> str: + """어느 환경변수에서 토큰을 탐지했는지 반환""" + for key in _TOKEN_ENV_PRIORITY: + val = os.environ.get(key, "").strip() + if val and val == self.token: + return key + return "explicit_arg" if self.token else "none" diff --git a/tools/gitea/harness.py b/tools/gitea/harness.py new file mode 100644 index 00000000..29d827ef --- /dev/null +++ b/tools/gitea/harness.py @@ -0,0 +1,350 @@ +""" +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 상세 및 실패 스텝 + workflows workflow 목록 + runners runner 상태 (admin token 필요) + secrets secrets 이름 목록 + vars variables 목록 + dispatch 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())