variet-agent/integrations/gitea_client.py

"""Gitea API 클라이언트.

git.variet.net API를 통해 레포지토리, 브랜치, PR, 커밋을 관리합니다.
"""

import logging
from typing import Optional

import httpx

import config

logger = logging.getLogger("variet.gitea")


class GiteaClient:
    """Gitea REST API 비동기 클라이언트."""

    def __init__(
        self,
        base_url: str = None,
        token: str = None,
        repo: str = None,
    ):
        self.base_url = (base_url or config.GITEA_URL).rstrip("/")
        self.api_url = f"{self.base_url}/api/v1"
        self.token = token or config.GITEA_TOKEN
        self.repo = repo or config.GITEA_REPO
        self.headers = {
            "Authorization": f"token {self.token}",
            "Content-Type": "application/json",
        }

    async def _get(self, path: str, params: dict = None) -> dict | list:
        async with httpx.AsyncClient(timeout=30) as client:
            resp = await client.get(
                f"{self.api_url}{path}",
                headers=self.headers,
                params=params,
            )
            resp.raise_for_status()
            return resp.json()

    async def _post(self, path: str, data: dict = None) -> dict:
        async with httpx.AsyncClient(timeout=30) as client:
            resp = await client.post(
                f"{self.api_url}{path}",
                headers=self.headers,
                json=data,
            )
            resp.raise_for_status()
            return resp.json()

    # === Commits ===

    async def get_commits(self, limit: int = 10, branch: str = "main") -> list[dict]:
        """최근 커밋 조회."""
        commits = await self._get(
            f"/repos/{self.repo}/commits",
            params={"limit": limit, "sha": branch},
        )
        return [
            {
                "sha": c["sha"][:7],
                "message": c["commit"]["message"].split("\n")[0],
                "author": c["commit"]["author"]["name"],
                "date": c["commit"]["author"]["date"],
            }
            for c in commits
        ]

    # === Branches ===

    async def create_branch(self, name: str, from_branch: str = "main") -> dict:
        """새 브랜치 생성."""
        result = await self._post(
            f"/repos/{self.repo}/branches",
            data={"new_branch_name": name, "old_branch_name": from_branch},
        )
        logger.info(f"브랜치 생성: {name} (from {from_branch})")
        return result

    async def list_branches(self) -> list[str]:
        """브랜치 목록."""
        branches = await self._get(f"/repos/{self.repo}/branches")
        return [b["name"] for b in branches]

    # === Pull Requests ===

    async def create_pr(
        self,
        title: str,
        head: str,
        base: str = "main",
        body: str = "",
    ) -> dict:
        """PR 생성."""
        result = await self._post(
            f"/repos/{self.repo}/pulls",
            data={
                "title": title,
                "head": head,
                "base": base,
                "body": body,
            },
        )
        pr_number = result.get("number")
        logger.info(f"PR #{pr_number} 생성: {title}")
        return result

    async def list_prs(self, state: str = "open") -> list[dict]:
        """PR 목록 조회."""
        prs = await self._get(
            f"/repos/{self.repo}/pulls",
            params={"state": state},
        )
        return [
            {
                "number": p["number"],
                "title": p["title"],
                "state": p["state"],
                "user": p["user"]["login"],
            }
            for p in prs
        ]

    async def merge_pr(self, pr_number: int, merge_type: str = "merge") -> dict:
        """PR 머지."""
        result = await self._post(
            f"/repos/{self.repo}/pulls/{pr_number}/merge",
            data={"Do": merge_type},
        )
        logger.info(f"PR #{pr_number} 머지 완료")
        return result

    # === Issues ===

    async def list_issues(self, state: str = "open") -> list[dict]:
        """이슈 목록 조회."""
        issues = await self._get(
            f"/repos/{self.repo}/issues",
            params={"state": state, "type": "issues"},
        )
        return [
            {
                "number": i["number"],
                "title": i["title"],
                "state": i["state"],
            }
            for i in issues
        ]

    async def create_issue(self, title: str, body: str = "") -> dict:
        """이슈 생성."""
        result = await self._post(
            f"/repos/{self.repo}/issues",
            data={"title": title, "body": body},
        )
        logger.info(f"이슈 #{result.get('number')} 생성: {title}")
        return result

    # === Commit Status (CI) ===

    async def get_commit_status(self, sha: str) -> list[dict]:
        """커밋의 CI 상태 조회."""
        statuses = await self._get(
            f"/repos/{self.repo}/statuses/{sha}",
        )
        return [
            {
                "status": s["status"],
                "description": s.get("description", ""),
                "context": s.get("context", ""),
                "target_url": s.get("target_url", ""),
            }
            for s in statuses
        ]

    async def get_combined_status(self, sha: str) -> dict:
        """커밋의 통합 CI 상태 조회."""
        return await self._get(
            f"/repos/{self.repo}/commits/{sha}/status",
        )