chore(gitea): add PR harness validator

2026-06-22 18:56:30 +09:00
parent 6c549b7bdc
commit 4c8048358a
1 changed files with 127 additions and 0 deletions
@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_BASE_URL = "http://kjh2064.synology.me:8418"
+DEFAULT_OWNER = "KimJaeHyun"
+DEFAULT_REPO = "myfinance"
+
+
+def _token() -> str:
+    # GITEA_TOKEN 환경 변수가 우선
+    token = os.environ.get("GITEA_TOKEN", "")
+    if not token.strip():
+        # 없으면 GITEA_TOKEN_HOME fallback
+        token = os.environ.get("GITEA_TOKEN_HOME", "")
+    if not token.strip():
+        raise SystemExit("GITEA_TOKEN 또는 GITEA_TOKEN_HOME 환경 변수가 설정되어 있지 않습니다.")
+    return token.strip()
+
+
+def _request_json(url: str, token: str, method: str = "GET", body: dict[str, Any] | None = None) -> tuple[int, str, Any]:
+    data = None
+    headers = {
+        "Authorization": f"token {token}",
+        "Accept": "application/json",
+    }
+    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=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) if raw else None
+        except Exception:
+            payload = raw
+        return exc.code, exc.reason or "", payload
+
+
+def _repo_base(args: argparse.Namespace) -> str:
+    return f"{args.base_url.rstrip('/')}/api/v1/repos/{args.owner}/{args.repo}"
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser(description="Create and Validate Gitea Pull Requests automatically.")
+    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("--head", required=True, help="Branch containing changes (e.g., feature/kis-token-db-cache)")
+    ap.add_argument("--base-branch", default="main", help="Target branch to merge changes into")
+    ap.add_argument("--title", default=None, help="PR Title")
+    ap.add_argument("--body", default="Automated PR generated by quant engine client.", help="PR Description")
+    args = ap.parse_args()
+
+    token = _token()
+    base = _repo_base(args)
+    
+    title = args.title or f"Merge {args.head} into {args.base_branch}"
+    
+    evidence: dict[str, Any] = {
+        "base_url": args.base_url,
+        "owner": args.owner,
+        "repo": args.repo,
+        "head": args.head,
+        "base_branch": args.base_branch,
+    }
+
+    # 1. 원격 저장소 접근 가능 여부 테스트 (Basic Auth/Token Check)
+    status, reason, repo_payload = _request_json(base, token)
+    evidence["repo_check_status"] = status
+    if status != 200:
+        print(json.dumps({"gate": "FAIL", "evidence": evidence, "error": "Repository inaccessible. Check credentials/token."}, ensure_ascii=False, indent=2))
+        return 1
+
+    # 2. 이미 동일한 Head -> Base PR이 열려 있는지 검사
+    status, reason, prs_payload = _request_json(f"{base}/pulls?state=open", token)
+    if status == 200 and isinstance(prs_payload, list):
+        for pr in prs_payload:
+            if pr.get("head", {}).get("ref") == args.head and pr.get("base", {}).get("ref") == args.base_branch:
+                evidence["pr_exists"] = True
+                evidence["pr_url"] = pr.get("html_url")
+                print(json.dumps({"gate": "PASS", "message": "PR already exists and is active.", "evidence": evidence}, ensure_ascii=False, indent=2))
+                return 0
+
+    # 3. 신규 Pull Request 작성 API 요청
+    pr_body = {
+        "title": title,
+        "body": args.body,
+        "head": args.head,
+        "base": args.base_branch,
+    }
+    
+    status, reason, pr_payload = _request_json(f"{base}/pulls", token, method="POST", body=pr_body)
+    evidence["create_status"] = status
+    evidence["create_reason"] = reason
+    
+    if status == 201:
+        evidence["pr_url"] = pr_payload.get("html_url")
+        result = {
+            "gate": "PASS",
+            "message": f"Successfully created Pull Request for branch: {args.head}",
+            "evidence": evidence
+        }
+        print(json.dumps(result, ensure_ascii=False, indent=2))
+        return 0
+    else:
+        # PR 생성 실패 (동일한 브랜치에 변경 사항이 없거나 권한 부족 등)
+        print(json.dumps({"gate": "FAIL", "evidence": evidence, "error": pr_payload}, ensure_ascii=False, indent=2))
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())