Add gitea-manager skill for Gitea repository operations

Co-authored-by: monkeycode-ai <monkeycode-ai@chaitin.com>

gitea-manager/gitea_client.py

@@ -0,0 +1,423 @@
+#!/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()