424 lines
18 KiB
Python
424 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
"""Gitea API Client - 封装 Gitea REST API 的所有操作"""
|
||
|
||
import os
|
||
import json
|
||
import requests
|
||
from typing import Optional, Dict, List, Any
|
||
|
||
|
||
class GiteaClient:
|
||
"""Gitea API 客户端"""
|
||
|
||
def __init__(self, url: Optional[str] = None, token: Optional[str] = None):
|
||
self.url = url or os.environ.get("GITEA_URL", "http://localhost:3000")
|
||
self.token = token or os.environ.get("GITEA_TOKEN", "")
|
||
self.session = requests.Session()
|
||
self.session.headers.update({
|
||
"Authorization": f"token {self.token}",
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json"
|
||
})
|
||
if not self.token:
|
||
raise ValueError("GITEA_TOKEN 环境变量未设置")
|
||
|
||
def _request(self, method: str, path: str, **kwargs) -> requests.Response:
|
||
"""发送 API 请求"""
|
||
url = f"{self.url.rstrip('/')}{path}"
|
||
response = self.session.request(method, url, **kwargs)
|
||
response.raise_for_status()
|
||
return response
|
||
|
||
# ==================== 用户操作 ====================
|
||
|
||
def get_current_user(self) -> Dict:
|
||
"""获取当前认证用户信息"""
|
||
return self._request("GET", "/user").json()
|
||
|
||
def get_user(self, username: str) -> Dict:
|
||
"""获取指定用户信息"""
|
||
return self._request("GET", f"/users/{username}").json()
|
||
|
||
def list_user_repos(self, username: str, page: int = 1, limit: int = 50) -> List[Dict]:
|
||
"""列出用户的所有仓库"""
|
||
return self._request("GET", f"/users/{username}/repos", params={"page": page, "limit": limit}).json()
|
||
|
||
def create_user(self, email: str, username: str, password: str, **kwargs) -> Dict:
|
||
"""创建用户（需要管理员权限）"""
|
||
data = {"email": email, "username": username, "password": password}
|
||
data.update(kwargs)
|
||
return self._request("POST", "/admin/users", json=data).json()
|
||
|
||
def delete_user(self, username: str) -> None:
|
||
"""删除用户（需要管理员权限）"""
|
||
self._request("DELETE", f"/admin/users/{username}")
|
||
|
||
# ==================== 组织操作 ====================
|
||
|
||
def list_orgs(self, page: int = 1, limit: int = 50) -> List[Dict]:
|
||
"""列出当前用户所属组织"""
|
||
return self._request("GET", "/user/orgs", params={"page": page, "limit": limit}).json()
|
||
|
||
def get_org(self, username: str) -> Dict:
|
||
"""获取组织信息"""
|
||
return self._request("GET", f"/orgs/{username}").json()
|
||
|
||
def create_org(self, username: str, email: str, **kwargs) -> Dict:
|
||
"""创建组织"""
|
||
data = {"username": username, "email": email}
|
||
data.update(kwargs)
|
||
return self._request("POST", "/admin/orgs", json=data).json()
|
||
|
||
def list_org_repos(self, org: str, page: int = 1, limit: int = 50) -> List[Dict]:
|
||
"""列出组织的所有仓库"""
|
||
return self._request("GET", f"/orgs/{org}/repos", params={"page": page, "limit": limit}).json()
|
||
|
||
# ==================== 仓库操作 ====================
|
||
|
||
def list_my_repos(self, page: int = 1, limit: int = 50) -> List[Dict]:
|
||
"""列出当前用户的仓库"""
|
||
return self._request("GET", "/user/repos", params={"page": page, "limit": limit}).json()
|
||
|
||
def get_repo(self, owner: str, repo: str) -> Dict:
|
||
"""获取仓库信息"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}").json()
|
||
|
||
def create_repo(self, name: str, description: str = "", private: bool = False,
|
||
auto_init: bool = True, **kwargs) -> Dict:
|
||
"""创建仓库"""
|
||
data = {
|
||
"name": name,
|
||
"description": description,
|
||
"private": private,
|
||
"auto_init": auto_init
|
||
}
|
||
data.update(kwargs)
|
||
return self._request("POST", "/user/repos", json=data).json()
|
||
|
||
def create_org_repo(self, org: str, name: str, description: str = "",
|
||
private: bool = False, auto_init: bool = True, **kwargs) -> Dict:
|
||
"""在组织下创建仓库"""
|
||
data = {
|
||
"name": name,
|
||
"description": description,
|
||
"private": private,
|
||
"auto_init": auto_init
|
||
}
|
||
data.update(kwargs)
|
||
return self._request("POST", f"/orgs/{org}/repos", json=data).json()
|
||
|
||
def delete_repo(self, owner: str, repo: str) -> None:
|
||
"""删除仓库"""
|
||
self._request("DELETE", f"/repos/{owner}/{repo}")
|
||
|
||
def get_repo_info(self, owner: str, repo: str) -> Dict:
|
||
"""获取仓库详细信息"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}").json()
|
||
|
||
def update_repo(self, owner: str, repo: str, **kwargs) -> Dict:
|
||
"""更新仓库设置"""
|
||
return self._request("PATCH", f"/repos/{owner}/{repo}", json=kwargs).json()
|
||
|
||
# ==================== 分支操作 ====================
|
||
|
||
def list_branches(self, owner: str, repo: str, page: int = 1, limit: int = 50) -> List[Dict]:
|
||
"""列出仓库分支"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/branches",
|
||
params={"page": page, "limit": limit}).json()
|
||
|
||
def get_branch(self, owner: str, repo: str, branch: str) -> Dict:
|
||
"""获取分支信息"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/branches/{branch}").json()
|
||
|
||
def create_branch(self, owner: str, repo: str, branch_name: str, from_branch: str) -> Dict:
|
||
"""创建分支"""
|
||
data = {
|
||
"new_branch_name": branch_name,
|
||
"from_branch": from_branch
|
||
}
|
||
return self._request("POST", f"/repos/{owner}/{repo}/branches", json=data).json()
|
||
|
||
def delete_branch(self, owner: str, repo: str, branch: str) -> None:
|
||
"""删除分支"""
|
||
self._request("DELETE", f"/repos/{owner}/{repo}/branches/{branch}")
|
||
|
||
def list_protected_branches(self, owner: str, repo: str) -> List[Dict]:
|
||
"""列出受保护分支"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/protected_branches").json()
|
||
|
||
def get_protected_branch(self, owner: str, repo: str, branch: str) -> Dict:
|
||
"""获取受保护分支规则"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/protected_branches/{branch}").json()
|
||
|
||
def set_protected_branch(self, owner: str, repo: str, branch: str, **kwargs) -> Dict:
|
||
"""设置分支保护规则"""
|
||
return self._request("PUT", f"/repos/{owner}/{repo}/protected_branches/{branch}", json=kwargs).json()
|
||
|
||
# ==================== 文件操作 ====================
|
||
|
||
def get_file_content(self, owner: str, repo: str, path: str, ref: str = "main") -> str:
|
||
"""获取文件内容（Base64 编码）"""
|
||
response = self._request("GET", f"/repos/{owner}/{repo}/raw/{path}", params={"ref": ref})
|
||
import base64
|
||
return base64.b64encode(response.content).decode('utf-8')
|
||
|
||
def get_file_info(self, owner: str, repo: str, path: str, ref: str = "main") -> Dict:
|
||
"""获取文件信息"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/contents/{path}", params={"ref": ref}).json()
|
||
|
||
def create_or_update_file(self, owner: str, repo: str, path: str, content: str,
|
||
message: str, branch: str, email: str, name: str) -> Dict:
|
||
"""创建或更新文件"""
|
||
import base64
|
||
data = {
|
||
"content": base64.b64encode(content.encode()).decode('utf-8'),
|
||
"message": message,
|
||
"branch": branch,
|
||
"author": {"email": email, "name": name},
|
||
"committer": {"email": email, "name": name}
|
||
}
|
||
return self._request("POST", f"/repos/{owner}/{repo}/contents/{path}", json=data).json()
|
||
|
||
def delete_file(self, owner: str, repo: str, path: str, message: str,
|
||
branch: str, sha: str, email: str, name: str) -> Dict:
|
||
"""删除文件"""
|
||
data = {
|
||
"message": message,
|
||
"branch": branch,
|
||
"sha": sha,
|
||
"author": {"email": email, "name": name},
|
||
"committer": {"email": email, "name": name}
|
||
}
|
||
return self._request("DELETE", f"/repos/{owner}/{repo}/contents/{path}", json=data).json()
|
||
|
||
def list_directory(self, owner: str, repo: str, path: str = "",
|
||
ref: str = "main", page: int = 1, limit: int = 100) -> List[Dict]:
|
||
"""列出目录内容"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/contents/{path}",
|
||
params={"ref": ref, "page": page, "limit": limit}).json()
|
||
|
||
# ==================== Issue 操作 ====================
|
||
|
||
def list_issues(self, owner: str, repo: str, state: str = "open",
|
||
page: int = 1, limit: int = 50, **kwargs) -> List[Dict]:
|
||
"""列出仓库的 Issue"""
|
||
params = {"state": state, "page": page, "limit": limit}
|
||
params.update(kwargs)
|
||
return self._request("GET", f"/repos/{owner}/{repo}/issues", params=params).json()
|
||
|
||
def get_issue(self, owner: str, repo: str, index: int) -> Dict:
|
||
"""获取 Issue 详情"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/issues/{index}").json()
|
||
|
||
def create_issue(self, owner: str, repo: str, title: str, body: str = "",
|
||
**kwargs) -> Dict:
|
||
"""创建 Issue"""
|
||
data = {"title": title, "body": body}
|
||
data.update(kwargs)
|
||
return self._request("POST", f"/repos/{owner}/{repo}/issues", json=data).json()
|
||
|
||
def update_issue(self, owner: str, repo: str, index: int, **kwargs) -> Dict:
|
||
"""更新 Issue"""
|
||
return self._request("PATCH", f"/repos/{owner}/{repo}/issues/{index}", json=kwargs).json()
|
||
|
||
def close_issue(self, owner: str, repo: str, index: int) -> Dict:
|
||
"""关闭 Issue"""
|
||
return self.update_issue(owner, repo, index, state="closed")
|
||
|
||
def list_issue_comments(self, owner: str, repo: str, issue_index: int,
|
||
page: int = 1, limit: int = 50) -> List[Dict]:
|
||
"""列出 Issue 的评论"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/issues/{issue_index}/comments",
|
||
params={"page": page, "limit": limit}).json()
|
||
|
||
def create_issue_comment(self, owner: str, repo: str, issue_index: int, body: str) -> Dict:
|
||
"""创建 Issue 评论"""
|
||
return self._request("POST", f"/repos/{owner}/{repo}/issues/{issue_index}/comments",
|
||
json={"body": body}).json()
|
||
|
||
# ==================== Pull Request 操作 ====================
|
||
|
||
def list_pulls(self, owner: str, repo: str, state: str = "open",
|
||
page: int = 1, limit: int = 50) -> List[Dict]:
|
||
"""列出 Pull Request"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/pulls",
|
||
params={"state": state, "page": page, "limit": limit}).json()
|
||
|
||
def get_pull(self, owner: str, repo: str, index: int) -> Dict:
|
||
"""获取 Pull Request 详情"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/pulls/{index}").json()
|
||
|
||
def create_pull(self, owner: str, repo: str, title: str, body: str,
|
||
head: str, base: str, **kwargs) -> Dict:
|
||
"""创建 Pull Request"""
|
||
data = {
|
||
"title": title,
|
||
"body": body,
|
||
"head": head,
|
||
"base": base
|
||
}
|
||
data.update(kwargs)
|
||
return self._request("POST", f"/repos/{owner}/{repo}/pulls", json=data).json()
|
||
|
||
def update_pull(self, owner: str, repo: str, index: int, **kwargs) -> Dict:
|
||
"""更新 Pull Request"""
|
||
return self._request("PATCH", f"/repos/{owner}/{repo}/pulls/{index}", json=kwargs).json()
|
||
|
||
def merge_pull(self, owner: str, repo: str, index: int,
|
||
merge_message: str = "", merge_style: str = "merge") -> Dict:
|
||
"""合并 Pull Request"""
|
||
data = {
|
||
"merge_commit_message": merge_message,
|
||
"merge_style": merge_style
|
||
}
|
||
return self._request("POST", f"/repos/{owner}/{repo}/pulls/{index}/merge", json=data).json()
|
||
|
||
def list_pull_comments(self, owner: str, repo: str, index: int) -> List[Dict]:
|
||
"""列出 Pull Request 的评论"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/issues/{index}/comments").json()
|
||
|
||
# ==================== Webhook 操作 ====================
|
||
|
||
def list_hooks(self, owner: str, repo: str) -> List[Dict]:
|
||
"""列出仓库的 Webhook"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/hooks").json()
|
||
|
||
def create_hook(self, owner: str, repo: str, config: Dict, events: List[str],
|
||
active: bool = True) -> Dict:
|
||
"""创建 Webhook"""
|
||
data = {
|
||
"type": "gitea",
|
||
"config": config,
|
||
"events": events,
|
||
"active": active
|
||
}
|
||
return self._request("POST", f"/repos/{owner}/{repo}/hooks", json=data).json()
|
||
|
||
def delete_hook(self, owner: str, repo: str, hook_id: int) -> None:
|
||
"""删除 Webhook"""
|
||
self._request("DELETE", f"/repos/{owner}/{repo}/hooks/{hook_id}")
|
||
|
||
# ==================== Deploy Key 操作 ====================
|
||
|
||
def list_deploy_keys(self, owner: str, repo: str) -> List[Dict]:
|
||
"""列出仓库的 Deploy Key"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/keys").json()
|
||
|
||
def create_deploy_key(self, owner: str, repo: str, title: str, key: str,
|
||
read_only: bool = True) -> Dict:
|
||
"""创建 Deploy Key"""
|
||
data = {
|
||
"title": title,
|
||
"key": key,
|
||
"read_only": read_only
|
||
}
|
||
return self._request("POST", f"/repos/{owner}/{repo}/keys", json=data).json()
|
||
|
||
def delete_deploy_key(self, owner: str, repo: str, key_id: int) -> None:
|
||
"""删除 Deploy Key"""
|
||
self._request("DELETE", f"/repos/{owner}/{repo}/keys/{key_id}")
|
||
|
||
# ==================== 标签操作 ====================
|
||
|
||
def list_tags(self, owner: str, repo: str, page: int = 1, limit: int = 50) -> List[Dict]:
|
||
"""列出仓库的标签"""
|
||
return self._request("GET", f"/repos/{owner}/{repo}/tags",
|
||
params={"page": page, "limit": limit}).json()
|
||
|
||
def create_tag(self, owner: str, repo: str, tag_name: str, target: str,
|
||
message: str = "", **kwargs) -> Dict:
|
||
"""创建标签"""
|
||
data = {
|
||
"tag_name": tag_name,
|
||
"target": target,
|
||
"message": message
|
||
}
|
||
data.update(kwargs)
|
||
return self._request("POST", f"/repos/{owner}/{repo}/tags", json=data).json()
|
||
|
||
# ==================== 搜索操作 ====================
|
||
|
||
def search_repos(self, query: str, page: int = 1, limit: int = 50, **kwargs) -> List[Dict]:
|
||
"""搜索仓库"""
|
||
params = {"q": query, "page": page, "limit": limit}
|
||
params.update(kwargs)
|
||
return self._request("GET", "/repos/search", params=params).json().get("data", [])
|
||
|
||
def search_issues(self, owner: str, repo: str, query: str,
|
||
page: int = 1, limit: int = 50) -> List[Dict]:
|
||
"""搜索 Issue"""
|
||
params = {"q": query, "page": page, "limit": limit}
|
||
return self._request("GET", f"/repos/{owner}/{repo}/issues/search", params=params).json()
|
||
|
||
# ==================== 团队操作 ====================
|
||
|
||
def list_org_teams(self, org: str) -> List[Dict]:
|
||
"""列出组织的团队"""
|
||
return self._request("GET", f"/orgs/{org}/teams").json()
|
||
|
||
def create_team(self, org: str, name: str, permission: str = "read", **kwargs) -> Dict:
|
||
"""创建团队"""
|
||
data = {"name": name, "permission": permission}
|
||
data.update(kwargs)
|
||
return self._request("POST", f"/orgs/{org}/teams", json=data).json()
|
||
|
||
def add_team_member(self, org: str, team: str, username: str) -> None:
|
||
"""添加团队成员"""
|
||
self._request("PUT", f"/orgs/{org}/teams/{team}/members/{username}")
|
||
|
||
def remove_team_member(self, org: str, team: str, username: str) -> None:
|
||
"""移除团队成员"""
|
||
self._request("DELETE", f"/orgs/{org}/teams/{team}/members/{username}")
|
||
|
||
def add_team_repo(self, org: str, team: str, owner: str, repo: str) -> None:
|
||
"""添加团队仓库"""
|
||
self._request("PUT", f"/orgs/{org}/teams/{team}/repos/{owner}/{repo}")
|
||
|
||
def remove_team_repo(self, org: str, team: str, owner: str, repo: str) -> None:
|
||
"""移除团队仓库"""
|
||
self._request("DELETE", f"/orgs/{org}/teams/{team}/repos/{owner}/{repo}")
|
||
|
||
|
||
def main():
|
||
"""命令行入口"""
|
||
import argparse
|
||
|
||
parser = argparse.ArgumentParser(description="Gitea API Client")
|
||
parser.add_argument("--url", default=None, help="Gitea URL")
|
||
parser.add_argument("--token", default=None, help="API Token")
|
||
parser.add_argument("command", help="要执行的命令")
|
||
parser.add_argument("args", nargs="*", help="命令参数")
|
||
|
||
args = parser.parse_args()
|
||
|
||
client = GiteaClient(url=args.url, token=args.token)
|
||
|
||
if args.command == "list-repos":
|
||
repos = client.list_my_repos()
|
||
for r in repos:
|
||
print(f"{r['full_name']}: {r.get('description', '')}")
|
||
|
||
elif args.command == "get-repo":
|
||
owner, repo = args.args[0].split("/")
|
||
r = client.get_repo(owner, repo)
|
||
print(json.dumps(r, indent=2))
|
||
|
||
elif args.command == "create-repo":
|
||
name = args.args[0]
|
||
description = args.args[1] if len(args.args) > 1 else ""
|
||
r = client.create_repo(name, description)
|
||
print(json.dumps(r, indent=2))
|
||
|
||
elif args.command == "delete-repo":
|
||
owner, repo = args.args[0].split("/")
|
||
client.delete_repo(owner, repo)
|
||
print(f"Repository {owner}/{repo} deleted")
|
||
|
||
else:
|
||
print(f"Unknown command: {args.command}")
|
||
parser.print_help()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|