""" Gitea 版本控制服務 """ from typing import Optional, List, Dict, Any import requests from app.core.config import settings class GiteaService: """Gitea API 服務""" def __init__(self): self.base_url = settings.GITEA_URL self.user = settings.GITEA_USER self.token = settings.GITEA_TOKEN self.headers = { "Authorization": f"token {self.token}", "Content-Type": "application/json", } def _request( self, method: str, endpoint: str, data: Optional[dict] = None, ) -> Dict[str, Any]: """ 發送 API 請求 Args: method: HTTP 方法 endpoint: API 端點 data: 請求資料 Returns: API 回應 """ url = f"{self.base_url}/api/v1{endpoint}" try: response = requests.request( method=method, url=url, headers=self.headers, json=data, timeout=30, ) if response.status_code == 204: return {"success": True} return response.json() except Exception as e: return {"error": str(e)} def get_user(self) -> Dict[str, Any]: """取得當前使用者資訊""" return self._request("GET", "/user") def list_repos(self) -> List[Dict[str, Any]]: """列出使用者的所有 Repo""" return self._request("GET", f"/users/{self.user}/repos") def get_repo(self, repo_name: str) -> Dict[str, Any]: """ 取得 Repo 資訊 Args: repo_name: Repo 名稱 """ return self._request("GET", f"/repos/{self.user}/{repo_name}") def create_repo( self, name: str, description: str = "", private: bool = False, auto_init: bool = True, default_branch: str = "main", ) -> Dict[str, Any]: """ 建立新的 Repo Args: name: Repo 名稱 description: 描述 private: 是否為私有 auto_init: 是否自動初始化 (建立 README) default_branch: 預設分支名稱 Returns: 建立結果 """ data = { "name": name, "description": description, "private": private, "auto_init": auto_init, "default_branch": default_branch, } return self._request("POST", "/user/repos", data) def delete_repo(self, repo_name: str) -> Dict[str, Any]: """ 刪除 Repo Args: repo_name: Repo 名稱 """ return self._request("DELETE", f"/repos/{self.user}/{repo_name}") def create_file( self, repo_name: str, file_path: str, content: str, message: str = "Add file", branch: str = "main", ) -> Dict[str, Any]: """ 在 Repo 中建立檔案 Args: repo_name: Repo 名稱 file_path: 檔案路徑 content: 檔案內容 message: Commit 訊息 branch: 分支名稱 """ import base64 data = { "content": base64.b64encode(content.encode()).decode(), "message": message, "branch": branch, } return self._request( "POST", f"/repos/{self.user}/{repo_name}/contents/{file_path}", data ) def get_file(self, repo_name: str, file_path: str) -> Dict[str, Any]: """ 取得檔案內容 Args: repo_name: Repo 名稱 file_path: 檔案路徑 """ return self._request( "GET", f"/repos/{self.user}/{repo_name}/contents/{file_path}" ) def list_branches(self, repo_name: str) -> List[Dict[str, Any]]: """ 列出分支 Args: repo_name: Repo 名稱 """ return self._request("GET", f"/repos/{self.user}/{repo_name}/branches") def list_commits( self, repo_name: str, branch: str = "main", limit: int = 10 ) -> List[Dict[str, Any]]: """ 列出 Commits Args: repo_name: Repo 名稱 branch: 分支名稱 limit: 數量限制 """ return self._request( "GET", f"/repos/{self.user}/{repo_name}/commits?sha={branch}&limit={limit}" ) # 單例 gitea_service = GiteaService() def create_kpi_management_repo() -> Dict[str, Any]: """ 建立 KPI Management Repo Returns: 建立結果 """ result = gitea_service.create_repo( name="KPI-management", description="KPI 管理系統 - 員工績效考核管理平台", private=False, auto_init=True, default_branch="main", ) return result