128 lines
4.8 KiB
Python
128 lines
4.8 KiB
Python
#!/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())
|