#!/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()