log-center/app/gitea_client.py

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


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" if self.gitea_url else ""
        self.client = httpx.Client(timeout=30)

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

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

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

    def _get_api_url(self, pr_url: Optional[str] = None) -> str:
        """获取 API base URL，优先从 pr_url 解析，否则用 self.base_api_url"""
        if pr_url:
            match = re.search(r'(https?://[^/]+)/', pr_url)
            if match:
                return f"{match.group(1)}/api/v1"
        if self.base_api_url:
            return self.base_api_url
        raise ValueError("无法确定 Gitea API 地址：未配置 GITEA_URL 且 PR URL 中无法提取")

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

        Args:
            owner: 仓库所有者
            repo: 仓库名称
            pr_number: PR 编号
            api_base_url: 可选，API base URL（从 pr_url 解析得到）

        Returns:
            (是否成功, 消息)
        """
        base = api_base_url or self.base_api_url
        url = f"{base}/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:
            if e.response.status_code == 405:
                # PR 已经被合并或关闭，视为成功（期望的结果已达成）
                return True, "PR 已经被合并或关闭"
            error_msg = f"HTTP {e.response.status_code}"
            if 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 = "", api_base_url: str = None
    ) -> Tuple[bool, str]:
        """
        关闭 PR（可选添加评论说明原因）

        Args:
            owner: 仓库所有者
            repo: 仓库名称
            pr_number: PR 编号
            reason: 关闭原因（将作为评论添加）
            api_base_url: 可选，API base URL（从 pr_url 解析得到）

        Returns:
            (是否成功, 消息)
        """
        base = api_base_url or self.base_api_url

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

        # Step 2: 关闭 PR
        url = f"{base}/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, api_base_url: str = None
    ) -> Tuple[bool, str]:
        """
        添加 PR 评论

        Args:
            owner: 仓库所有者
            repo: 仓库名称
            pr_number: PR 编号
            comment: 评论内容
            api_base_url: 可选，API base URL

        Returns:
            (是否成功, 消息)
        """
        base = api_base_url or self.base_api_url
        # 注意: Gitea 中 PR 和 Issue 共用评论 API
        url = f"{base}/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 直接合并（从 URL 自动解析 Gitea 地址）"""
        try:
            base_url, owner, repo, pr_number = self._parse_pr_url(pr_url)
            api_base_url = f"{base_url}/api/v1"
            return self.merge_pr(owner, repo, pr_number, api_base_url=api_base_url)
        except Exception as e:
            return False, str(e)

    def close_pr_by_url(self, pr_url: str, reason: str = "") -> Tuple[bool, str]:
        """通过 PR URL 直接关闭（从 URL 自动解析 Gitea 地址）"""
        try:
            base_url, owner, repo, pr_number = self._parse_pr_url(pr_url)
            api_base_url = f"{base_url}/api/v1"
            return self.close_pr(owner, repo, pr_number, reason, api_base_url=api_base_url)
        except Exception as e:
            return False, str(e)