yanghuajun336
ab5e59e999
- 新增 setup.sh 配置脚本 - 新增 api-examples.sh API 函数库 - 新增 gitea-cli.py Python CLI 工具 - 完善文档结构 (1640 行) - 添加实际用例和工作流示例 - 增强故障排除和最佳实践指南
341 lines
13 KiB
Python
Executable File
341 lines
13 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Gitea CLI 工具
|
||
提供命令行界面访问 Gitea API
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import json
|
||
import argparse
|
||
import requests
|
||
from typing import Optional, Dict, Any, List
|
||
|
||
|
||
class GiteaClient:
|
||
"""Gitea API 客户端"""
|
||
|
||
def __init__(self, url: str = None, token: str = None):
|
||
self.url = url or os.environ.get("GITEA_URL", "https://gitea.com")
|
||
self.token = token or os.environ.get("GITEA_TOKEN")
|
||
|
||
if not self.token:
|
||
print("错误: 未设置 GITEA_TOKEN 环境变量")
|
||
print("请设置: export GITEA_TOKEN='your_token_here'")
|
||
sys.exit(1)
|
||
|
||
# 移除URL末尾的斜杠
|
||
self.url = self.url.rstrip("/")
|
||
self.session = requests.Session()
|
||
self.session.headers.update(
|
||
{
|
||
"Authorization": f"token {self.token}",
|
||
"Accept": "application/json",
|
||
"Content-Type": "application/json",
|
||
}
|
||
)
|
||
|
||
def _request(self, method: str, endpoint: str, data: Dict = None) -> Dict:
|
||
"""发送API请求"""
|
||
url = f"{self.url}/api/v1/{endpoint}"
|
||
|
||
try:
|
||
if method.upper() == "GET":
|
||
response = self.session.get(url, params=data)
|
||
elif method.upper() == "POST":
|
||
response = self.session.post(url, json=data)
|
||
elif method.upper() == "PUT":
|
||
response = self.session.put(url, json=data)
|
||
elif method.upper() == "PATCH":
|
||
response = self.session.patch(url, json=data)
|
||
elif method.upper() == "DELETE":
|
||
response = self.session.delete(url)
|
||
else:
|
||
raise ValueError(f"不支持的HTTP方法: {method}")
|
||
|
||
response.raise_for_status()
|
||
|
||
# 对于204 No Content响应，返回空字典
|
||
if response.status_code == 204:
|
||
return {}
|
||
|
||
return response.json()
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
print(f"API请求失败: {e}")
|
||
if hasattr(e, "response") and e.response is not None:
|
||
print(f"状态码: {e.response.status_code}")
|
||
try:
|
||
error_data = e.response.json()
|
||
print(f"错误信息: {error_data.get('message', '未知错误')}")
|
||
except:
|
||
print(f"响应内容: {e.response.text[:200]}")
|
||
sys.exit(1)
|
||
|
||
def get_current_user(self) -> Dict:
|
||
"""获取当前用户信息"""
|
||
return self._request("GET", "user")
|
||
|
||
def list_repos(self, page: int = 1, limit: int = 20) -> List[Dict]:
|
||
"""列出用户仓库"""
|
||
params = {"page": page, "limit": limit}
|
||
return self._request("GET", "user/repos", params)
|
||
|
||
def create_repo(
|
||
self, name: str, description: str = "", private: bool = False
|
||
) -> Dict:
|
||
"""创建仓库"""
|
||
data = {
|
||
"name": name,
|
||
"description": description,
|
||
"private": private,
|
||
"auto_init": True,
|
||
}
|
||
return self._request("POST", "user/repos", data)
|
||
|
||
def get_repo(self, owner: str, repo: str) -> Dict:
|
||
"""获取仓库信息"""
|
||
return self._request("GET", f"repos/{owner}/{repo}")
|
||
|
||
def create_pull_request(
|
||
self,
|
||
owner: str,
|
||
repo: str,
|
||
title: str,
|
||
body: str,
|
||
head: str,
|
||
base: str = "main",
|
||
) -> Dict:
|
||
"""创建Pull Request"""
|
||
data = {"title": title, "body": body, "head": head, "base": base}
|
||
return self._request("POST", f"repos/{owner}/{repo}/pulls", data)
|
||
|
||
def list_pull_requests(
|
||
self, owner: str, repo: str, state: str = "open"
|
||
) -> List[Dict]:
|
||
"""列出Pull Requests"""
|
||
params = {"state": state}
|
||
return self._request("GET", f"repos/{owner}/{repo}/pulls", params)
|
||
|
||
def merge_pull_request(self, owner: str, repo: str, pr_number: int) -> Dict:
|
||
"""合并Pull Request"""
|
||
data = {"Do": "merge"}
|
||
return self._request(
|
||
"POST", f"repos/{owner}/{repo}/pulls/{pr_number}/merge", data
|
||
)
|
||
|
||
def create_issue(
|
||
self,
|
||
owner: str,
|
||
repo: str,
|
||
title: str,
|
||
body: str = "",
|
||
labels: List[str] = None,
|
||
) -> Dict:
|
||
"""创建Issue"""
|
||
data = {"title": title}
|
||
if body:
|
||
data["body"] = body
|
||
if labels:
|
||
data["labels"] = labels
|
||
return self._request("POST", f"repos/{owner}/{repo}/issues", data)
|
||
|
||
def create_branch(
|
||
self, owner: str, repo: str, new_branch: str, from_branch: str = "main"
|
||
) -> Dict:
|
||
"""创建分支"""
|
||
data = {"new_branch_name": new_branch, "old_branch_name": from_branch}
|
||
return self._request("POST", f"repos/{owner}/{repo}/branches", data)
|
||
|
||
def create_tag(
|
||
self,
|
||
owner: str,
|
||
repo: str,
|
||
tag_name: str,
|
||
message: str = "",
|
||
target: str = "main",
|
||
) -> Dict:
|
||
"""创建标签"""
|
||
data = {
|
||
"tag_name": tag_name,
|
||
"message": message or f"Release {tag_name}",
|
||
"target": target,
|
||
}
|
||
return self._request("POST", f"repos/{owner}/{repo}/tags", data)
|
||
|
||
def search_repos(self, query: str, limit: int = 10) -> Dict:
|
||
"""搜索仓库"""
|
||
params = {"q": query, "limit": limit}
|
||
return self._request("GET", "repos/search", params)
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="Gitea CLI 工具")
|
||
subparsers = parser.add_subparsers(dest="command", help="可用命令")
|
||
|
||
# 用户命令
|
||
user_parser = subparsers.add_parser("user", help="用户相关操作")
|
||
user_parser.add_argument(
|
||
"action", choices=["info", "repos"], help="info: 获取用户信息, repos: 列出仓库"
|
||
)
|
||
user_parser.add_argument("--page", type=int, default=1, help="页码")
|
||
user_parser.add_argument("--limit", type=int, default=20, help="每页数量")
|
||
|
||
# 仓库命令
|
||
repo_parser = subparsers.add_parser("repo", help="仓库相关操作")
|
||
repo_parser.add_argument("action", choices=["create", "get", "search"])
|
||
repo_parser.add_argument("--name", help="仓库名称 (create时使用)")
|
||
repo_parser.add_argument("--description", default="", help="仓库描述")
|
||
repo_parser.add_argument("--private", action="store_true", help="是否私有")
|
||
repo_parser.add_argument("--owner", help="仓库所有者 (get/search时使用)")
|
||
repo_parser.add_argument("--repo", help="仓库名称 (get时使用)")
|
||
repo_parser.add_argument("--query", help="搜索关键词")
|
||
repo_parser.add_argument("--limit", type=int, default=10, help="搜索结果数量")
|
||
|
||
# PR命令
|
||
pr_parser = subparsers.add_parser("pr", help="Pull Request 相关操作")
|
||
pr_parser.add_argument("action", choices=["create", "list", "merge"])
|
||
pr_parser.add_argument("--owner", required=True, help="仓库所有者")
|
||
pr_parser.add_argument("--repo", required=True, help="仓库名称")
|
||
pr_parser.add_argument("--title", help="PR标题 (create时使用)")
|
||
pr_parser.add_argument("--body", default="", help="PR描述")
|
||
pr_parser.add_argument("--head", help="源分支")
|
||
pr_parser.add_argument("--base", default="main", help="目标分支")
|
||
pr_parser.add_argument("--number", type=int, help="PR编号 (merge时使用)")
|
||
pr_parser.add_argument(
|
||
"--state",
|
||
default="open",
|
||
choices=["open", "closed", "all"],
|
||
help="PR状态 (list时使用)",
|
||
)
|
||
|
||
# Issue命令
|
||
issue_parser = subparsers.add_parser("issue", help="Issue 相关操作")
|
||
issue_parser.add_argument("action", choices=["create"])
|
||
issue_parser.add_argument("--owner", required=True, help="仓库所有者")
|
||
issue_parser.add_argument("--repo", required=True, help="仓库名称")
|
||
issue_parser.add_argument("--title", required=True, help="Issue标题")
|
||
issue_parser.add_argument("--body", default="", help="Issue描述")
|
||
issue_parser.add_argument("--labels", help="标签 (逗号分隔)")
|
||
|
||
# 分支命令
|
||
branch_parser = subparsers.add_parser("branch", help="分支相关操作")
|
||
branch_parser.add_argument("action", choices=["create"])
|
||
branch_parser.add_argument("--owner", required=True, help="仓库所有者")
|
||
branch_parser.add_argument("--repo", required=True, help="仓库名称")
|
||
branch_parser.add_argument("--name", required=True, help="新分支名称")
|
||
branch_parser.add_argument("--from-branch", default="main", help="源分支")
|
||
|
||
# 标签命令
|
||
tag_parser = subparsers.add_parser("tag", help="标签相关操作")
|
||
tag_parser.add_argument("action", choices=["create"])
|
||
tag_parser.add_argument("--owner", required=True, help="仓库所有者")
|
||
tag_parser.add_argument("--repo", required=True, help="仓库名称")
|
||
tag_parser.add_argument("--name", required=True, help="标签名称")
|
||
tag_parser.add_argument("--message", default="", help="标签消息")
|
||
tag_parser.add_argument("--target", default="main", help="目标分支/提交")
|
||
|
||
# 配置命令
|
||
config_parser = subparsers.add_parser("config", help="配置相关操作")
|
||
config_parser.add_argument("action", choices=["test"], help="测试连接")
|
||
|
||
args = parser.parse_args()
|
||
|
||
if not args.command:
|
||
parser.print_help()
|
||
sys.exit(1)
|
||
|
||
# 创建客户端
|
||
client = GiteaClient()
|
||
|
||
# 处理命令
|
||
try:
|
||
if args.command == "user":
|
||
if args.action == "info":
|
||
user = client.get_current_user()
|
||
print(json.dumps(user, indent=2, ensure_ascii=False))
|
||
elif args.action == "repos":
|
||
repos = client.list_repos(args.page, args.limit)
|
||
for repo in repos:
|
||
print(f"{repo['full_name']}: {repo.get('description', '')}")
|
||
|
||
elif args.command == "repo":
|
||
if args.action == "create":
|
||
if not args.name:
|
||
print("错误: --name 参数是必需的")
|
||
sys.exit(1)
|
||
repo = client.create_repo(args.name, args.description, args.private)
|
||
print(f"仓库创建成功: {repo['html_url']}")
|
||
|
||
elif args.action == "get":
|
||
if not args.owner or not args.repo:
|
||
print("错误: --owner 和 --repo 参数是必需的")
|
||
sys.exit(1)
|
||
repo = client.get_repo(args.owner, args.repo)
|
||
print(json.dumps(repo, indent=2, ensure_ascii=False))
|
||
|
||
elif args.action == "search":
|
||
if not args.query:
|
||
print("错误: --query 参数是必需的")
|
||
sys.exit(1)
|
||
result = client.search_repos(args.query, args.limit)
|
||
for repo in result.get("data", []):
|
||
print(f"{repo['full_name']} - ★{repo.get('stars_count', 0)}")
|
||
|
||
elif args.command == "pr":
|
||
if args.action == "create":
|
||
if not all([args.title, args.head]):
|
||
print("错误: --title 和 --head 参数是必需的")
|
||
sys.exit(1)
|
||
pr = client.create_pull_request(
|
||
args.owner, args.repo, args.title, args.body, args.head, args.base
|
||
)
|
||
print(f"PR创建成功: #{pr['number']} - {pr['html_url']}")
|
||
|
||
elif args.action == "list":
|
||
prs = client.list_pull_requests(args.owner, args.repo, args.state)
|
||
for pr in prs:
|
||
print(f"#{pr['number']}: {pr['title']} ({pr['state']})")
|
||
|
||
elif args.action == "merge":
|
||
if not args.number:
|
||
print("错误: --number 参数是必需的")
|
||
sys.exit(1)
|
||
result = client.merge_pull_request(args.owner, args.repo, args.number)
|
||
print(f"PR #{args.number} 合并成功")
|
||
|
||
elif args.command == "issue":
|
||
if args.action == "create":
|
||
labels = args.labels.split(",") if args.labels else None
|
||
issue = client.create_issue(
|
||
args.owner, args.repo, args.title, args.body, labels
|
||
)
|
||
print(f"Issue创建成功: #{issue['number']} - {issue['html_url']}")
|
||
|
||
elif args.command == "branch":
|
||
if args.action == "create":
|
||
branch = client.create_branch(
|
||
args.owner, args.repo, args.name, args.from_branch
|
||
)
|
||
print(f"分支创建成功: {args.name}")
|
||
|
||
elif args.command == "tag":
|
||
if args.action == "create":
|
||
tag = client.create_tag(
|
||
args.owner, args.repo, args.name, args.message, args.target
|
||
)
|
||
print(f"标签创建成功: {args.name}")
|
||
|
||
elif args.command == "config":
|
||
if args.action == "test":
|
||
user = client.get_current_user()
|
||
print(f"连接成功! 用户: {user.get('login', user.get('username'))}")
|
||
|
||
except Exception as e:
|
||
print(f"错误: {e}")
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|