algorithm/backend/app/gitea/client.py

"""Gitea API客户端，用于与Gitea服务器通信"""

import requests
import json
import logging
from typing import Optional, Dict, Any, List

logger = logging.getLogger(__name__)


class GiteaClient:
    """Gitea API客户端类"""
    
    def __init__(self, server_url: str, access_token: str):
        """初始化Gitea客户端
        
        Args:
            server_url: Gitea服务器URL
            access_token: Gitea访问令牌
        """
        self.server_url = server_url.rstrip('/')
        self.access_token = access_token
        self.api_url = f"{self.server_url}/api/v1"
        self.headers = {
            "Authorization": f"token {self.access_token}",
            "Content-Type": "application/json"
        }
    
    def _request(self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Optional[Dict[str, Any]]:
        """发送API请求
        
        Args:
            method: HTTP方法（GET, POST, PUT, DELETE）
            endpoint: API端点
            data: 请求数据
            
        Returns:
            API响应数据
        """
        url = f"{self.api_url}/{endpoint}"
        
        try:
            if method == "GET":
                response = requests.get(url, headers=self.headers, params=data)
            elif method == "POST":
                response = requests.post(url, headers=self.headers, json=data)
            elif method == "PUT":
                response = requests.put(url, headers=self.headers, json=data)
            elif method == "PATCH":
                response = requests.patch(url, headers=self.headers, json=data)
            elif method == "DELETE":
                response = requests.delete(url, headers=self.headers)
            else:
                raise ValueError(f"Unsupported HTTP method: {method}")
            
            response.raise_for_status()
            
            if response.content:
                return response.json()
            return None
            
        except requests.RequestException as e:
            logger.error(f"Gitea API request failed: {str(e)}")
            return None
    
    def create_repository(self, owner: str, name: str, description: str = "", private: bool = False) -> Optional[Dict[str, Any]]:
        """创建仓库

        Args:
            owner: 所有者（用户或组织）
            name: 仓库名称
            description: 仓库描述
            private: 是否私有

        Returns:
            创建的仓库信息
        """
        data = {
            "name": name,
            "description": description,
            "private": private,
            "auto_init": True
        }

        # 优先尝试在指定的 owner（组织）下创建仓库
        logger.info(f"Attempting to create repository for owner: {owner}")
        owner_response = self._request("POST", f"org/{owner}/repos", data)
        if owner_response:
            logger.info(f"Repository created successfully under owner: {owner}")
            return owner_response

        # 如果组织创建失败，尝试在用户下创建仓库
        logger.info(f"Organization creation failed, trying user account")
        user_response = self._request("POST", f"user/repos", data)
        if user_response:
            logger.info(f"Repository created successfully in user account")
            return user_response

        logger.error(f"Failed to create repository for owner {owner}")
        return None
    
    def get_repository(self, owner: str, repo: str) -> Optional[Dict[str, Any]]:
        """获取仓库信息
        
        Args:
            owner: 所有者（用户或组织）
            repo: 仓库名称
            
        Returns:
            仓库信息
        """
        return self._request("GET", f"repos/{owner}/{repo}")
    
    def list_repositories(self, owner: str) -> Optional[List[Dict[str, Any]]]:
        """列出用户或组织的仓库
        
        Args:
            owner: 所有者（用户或组织）
            
        Returns:
            仓库列表
        """
        return self._request("GET", f"users/{owner}/repos")
    
    def delete_repository(self, owner: str, repo: str) -> bool:
        """删除仓库
        
        Args:
            owner: 所有者（用户或组织）
            repo: 仓库名称
            
        Returns:
            是否删除成功
        """
        response = self._request("DELETE", f"repos/{owner}/{repo}")
        return response is not None
    
    def create_file(self, owner: str, repo: str, path: str, content: str, message: str) -> Optional[Dict[str, Any]]:
        """创建或更新文件
        
        Args:
            owner: 所有者（用户或组织）
            repo: 仓库名称
            path: 文件路径
            content: 文件内容（base64编码）
            message: 提交消息
            
        Returns:
            操作结果
        """
        data = {
            "content": content,
            "message": message
        }
        
        return self._request("POST", f"repos/{owner}/{repo}/contents/{path}", data)
    
    def get_file(self, owner: str, repo: str, path: str) -> Optional[Dict[str, Any]]:
        """获取文件内容

        Args:
            owner: 所有者（用户或组织）
            repo: 仓库名称
            path: 文件路径

        Returns:
            文件信息和内容
        """
        return self._request("GET", f"repos/{owner}/{repo}/contents/{path}")

    def get_repository_files(self, owner: str, repo: str, path: str = "") -> Optional[List[Dict[str, Any]]]:
        """获取仓库文件列表

        Args:
            owner: 所有者（用户或组织）
            repo: 仓库名称
            path: 目录路径（默认为根目录）

        Returns:
            文件列表
        """
        result = self._request("GET", f"repos/{owner}/{repo}/contents/{path}")
        return result if isinstance(result, list) else None



    def check_connection(self) -> bool:
        """检查与Gitea服务器的连接
        
        Returns:
            是否连接成功
        """
        response = self._request("GET", "user")
        return response is not None
    
    def update_repository(self, owner: str, repo: str, name: Optional[str] = None, description: Optional[str] = None, private: Optional[bool] = None) -> Optional[Dict[str, Any]]:
        """更新仓库信息
        
        Args:
            owner: 所有者（用户或组织）
            repo: 仓库名称
            name: 新的仓库名称（可选）
            description: 新的仓库描述（可选）
            private: 是否私有（可选）
            
        Returns:
            更新后的仓库信息
        """
        data = {}
        if name is not None:
            data["name"] = name
        if description is not None:
            data["description"] = description
        if private is not None:
            data["private"] = private
        
        return self._request("PATCH", f"repos/{owner}/{repo}", data)