log-center/app/gitea_client.py

"""
Gitea API 客户端
"""
import httpx
import os
from typing import Tuple


class GiteaClient:
    """Gitea API 操作客户端"""

    def __init__(self, gitea_url: str = None, token: str = None):
        self.gitea_url = (gitea_url or os.getenv("GITEA_URL", "")).rstrip("/")
        self.token = token or os.getenv("GITEA_TOKEN", "")
        self.base_api_url = f"{self.gitea_url}/api/v1"
        self.client = httpx.Client(timeout=30)

    def _headers(self) -> dict:
        """生成请求头"""
        return {
            "Authorization": f"token {self.token}",
            "Content-Type": "application/json",
        }

    def _extract_owner_repo_from_pr_url(self, pr_url: str) -> Tuple[str, str, int]:
        """
        从 PR URL 提取 owner, repo, pr_number
        例如: https://gitea.airlabs.art/owner/repo/pulls/45
        """
        import re
        match = re.search(r'([^/]+)/([^/]+)/pulls/(\d+)', pr_url)
        if not match:
            raise ValueError(f"无法解析 PR URL: {pr_url}")

        owner, repo, pr_number = match.groups()
        return owner, repo, int(pr_number)

    def merge_pr(
        self, owner: str, repo: str, pr_number: int
    ) -> Tuple[bool, str]:
        """
        合并 PR

        Args:
            owner: 仓库所有者
            repo: 仓库名称
            pr_number: PR 编号

        Returns:
            (是否成功, 消息)
        """
        url = f"{self.base_api_url}/repos/{owner}/{repo}/pulls/{pr_number}/merge"
        payload = {
            "Do": "merge",  # merge / squash / rebase
            "MergeMessageField": f"Merge PR #{pr_number} (approved via Log Center)",
        }

        try:
            response = self.client.post(url, json=payload, headers=self._headers())
            response.raise_for_status()
            return True, "PR 合并成功"
        except httpx.HTTPStatusError as e:
            error_msg = f"HTTP {e.response.status_code}"
            if e.response.status_code == 405:
                error_msg += ": PR 已经合并或已关闭"
            elif e.response.status_code == 409:
                error_msg += ": 存在合并冲突"
            elif e.response.status_code == 403:
                error_msg += ": Token 权限不足"
            else:
                error_msg += f": {e.response.text[:200]}"
            return False, error_msg
        except Exception as e:
            return False, str(e)

    def close_pr(
        self, owner: str, repo: str, pr_number: int, reason: str = ""
    ) -> Tuple[bool, str]:
        """
        关闭 PR（可选添加评论说明原因）

        Args:
            owner: 仓库所有者
            repo: 仓库名称
            pr_number: PR 编号
            reason: 关闭原因（将作为评论添加）

        Returns:
            (是否成功, 消息)
        """
        # Step 1: 如果提供了原因，先添加评论
        if reason:
            comment_success, comment_msg = self.add_pr_comment(
                owner, repo, pr_number, f"## ❌ 修复被拒绝\n\n**原因:**\n{reason}"
            )
            if not comment_success:
                return False, f"添加评论失败: {comment_msg}"

        # Step 2: 关闭 PR
        url = f"{self.base_api_url}/repos/{owner}/{repo}/pulls/{pr_number}"
        payload = {"state": "closed"}

        try:
            response = self.client.patch(url, json=payload, headers=self._headers())
            response.raise_for_status()
            return True, "PR 已关闭"
        except httpx.HTTPStatusError as e:
            return False, f"HTTP {e.response.status_code}: {e.response.text[:200]}"
        except Exception as e:
            return False, str(e)

    def add_pr_comment(
        self, owner: str, repo: str, pr_number: int, comment: str
    ) -> Tuple[bool, str]:
        """
        添加 PR 评论

        Args:
            owner: 仓库所有者
            repo: 仓库名称
            pr_number: PR 编号
            comment: 评论内容

        Returns:
            (是否成功, 消息)
        """
        # 注意: Gitea 中 PR 和 Issue 共用评论 API
        url = f"{self.base_api_url}/repos/{owner}/{repo}/issues/{pr_number}/comments"
        payload = {"body": comment}

        try:
            response = self.client.post(url, json=payload, headers=self._headers())
            response.raise_for_status()
            return True, "评论添加成功"
        except Exception as e:
            return False, str(e)

    def merge_pr_by_url(self, pr_url: str) -> Tuple[bool, str]:
        """通过 PR URL 直接合并"""
        try:
            owner, repo, pr_number = self._extract_owner_repo_from_pr_url(pr_url)
            return self.merge_pr(owner, repo, pr_number)
        except Exception as e:
            return False, str(e)

    def close_pr_by_url(self, pr_url: str, reason: str = "") -> Tuple[bool, str]:
        """通过 PR URL 直接关闭"""
        try:
            owner, repo, pr_number = self._extract_owner_repo_from_pr_url(pr_url)
            return self.close_pr(owner, repo, pr_number, reason)
        except Exception as e:
            return False, str(e)