""" 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"