154 lines
5.1 KiB
Python
154 lines
5.1 KiB
Python
"""
|
||
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:
|
||
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 = ""
|
||
) -> 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)
|