Initial commit: KPI Management System Backend
Features: - FastAPI backend with JWT authentication - MySQL database with SQLAlchemy ORM - KPI workflow: draft → pending → approved → evaluation → completed - Ollama LLM API integration for AI features - Gitea API integration for version control - Complete API endpoints for KPI, dashboard, notifications Tables: KPI_D_* prefix naming convention 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
198
app/services/gitea_service.py
Normal file
198
app/services/gitea_service.py
Normal file
@@ -0,0 +1,198 @@
|
||||
"""
|
||||
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
|
||||
Reference in New Issue
Block a user