gemini-balance/app/core/uploader.py

import requests
from app.schemas.image_models import ImageMetadata, ImageUploader, UploadResponse
from enum import Enum
from typing import Optional, Any

class UploadErrorType(Enum):
    """上传错误类型枚举"""
    NETWORK_ERROR = "network_error"  # 网络请求错误
    AUTH_ERROR = "auth_error"        # 认证错误
    INVALID_FILE = "invalid_file"    # 无效文件
    SERVER_ERROR = "server_error"    # 服务器错误
    PARSE_ERROR = "parse_error"      # 响应解析错误
    UNKNOWN = "unknown"              # 未知错误


class UploadError(Exception):
    """图片上传错误异常类"""
    
    def __init__(
        self,
        message: str,
        error_type: UploadErrorType = UploadErrorType.UNKNOWN,
        status_code: Optional[int] = None,
        details: Optional[dict] = None,
        original_error: Optional[Exception] = None
    ):
        """
        初始化上传错误异常
        
        Args:
            message: 错误消息
            error_type: 错误类型
            status_code: HTTP状态码
            details: 详细错误信息
            original_error: 原始异常
        """
        self.message = message
        self.error_type = error_type
        self.status_code = status_code
        self.details = details or {}
        self.original_error = original_error
        
        # 构建完整错误信息
        full_message = f"[{error_type.value}] {message}"
        if status_code:
            full_message = f"{full_message} (Status: {status_code})"
        if details:
            full_message = f"{full_message} - Details: {details}"
            
        super().__init__(full_message)
    
    @classmethod
    def from_response(cls, response: Any, message: Optional[str] = None) -> "UploadError":
        """
        从HTTP响应创建错误实例
        
        Args:
            response: HTTP响应对象
            message: 自定义错误消息
        """
        try:
            error_data = response.json()
            details = error_data.get("data", {})
            return cls(
                message=message or error_data.get("message", "Unknown error"),
                error_type=UploadErrorType.SERVER_ERROR,
                status_code=response.status_code,
                details=details
            )
        except Exception:
            return cls(
                message=message or "Failed to parse error response",
                error_type=UploadErrorType.PARSE_ERROR,
                status_code=response.status_code
            )


class SmMsUploader(ImageUploader):
    API_URL = "https://sm.ms/api/v2/upload"
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        
    def upload(self, file: bytes, filename: str) -> UploadResponse:
        try:
            # 准备请求头
            headers = {
                "Authorization": f"Basic {self.api_key}"
            }
            
            # 准备文件数据
            files = {
                "smfile": (filename, file, "image/png")
            }
            
            # 发送请求
            response = requests.post(
                self.API_URL,
                headers=headers,
                files=files
            )
            
            # 检查响应状态
            response.raise_for_status()
            
            # 解析响应
            result = response.json()
            
            # 验证上传是否成功
            if not result.get("success"):
                raise UploadError(result.get("message", "Upload failed"))
                
            # 转换为统一格式
            data = result["data"]
            image_metadata = ImageMetadata(
                width=data["width"],
                height=data["height"],
                filename=data["filename"],
                size=data["size"],
                url=data["url"],
                delete_url=data["delete"]
            )
            
            return UploadResponse(
                success=True,
                code="success",
                message="Upload success",
                data=image_metadata
            )
            
        except requests.RequestException as e:
            # 处理网络请求相关错误
            raise UploadError(f"Upload request failed: {str(e)}")
        except (KeyError, ValueError) as e:
            # 处理响应解析错误
            raise UploadError(f"Invalid response format: {str(e)}")
        except Exception as e:
            # 处理其他未预期的错误
            raise UploadError(f"Upload failed: {str(e)}")
    
    
class QiniuUploader(ImageUploader):
    def __init__(self, access_key: str, secret_key: str):
        self.access_key = access_key
        self.secret_key = secret_key
        
    def upload(self, file: bytes, filename: str) -> UploadResponse:
        # 实现七牛云的具体上传逻辑
        pass
    
    
class PicGoUploader(ImageUploader):
    """Chevereto API 图片上传器"""
    
    def __init__(self, api_key: str, api_url: str = "https://www.picgo.net/api/1/upload"):
        """
        初始化 Chevereto 上传器
        
        Args:
            api_key: Chevereto API 密钥
            api_url: Chevereto API 上传地址
        """
        self.api_key = api_key
        self.api_url = api_url
        
    def upload(self, file: bytes, filename: str) -> UploadResponse:
        """
        上传图片到 Chevereto 服务
        
        Args:
            file: 图片文件二进制数据
            filename: 文件名
            
        Returns:
            UploadResponse: 上传响应对象
            
        Raises:
            UploadError: 上传失败时抛出异常
        """
        try:
            # 准备请求头
            headers = {
                "X-API-Key": self.api_key
            }
            
            # 准备文件数据
            files = {
                "source": (filename, file)
            }
            
            # 发送请求
            response = requests.post(
                self.api_url,
                headers=headers,
                files=files
            )
            
            # 检查响应状态
            response.raise_for_status()
            
            # 解析响应
            result = response.json()
            
            # 验证上传是否成功
            if result.get("status_code") != 200:
                error_message = "Upload failed"
                if "error" in result:
                    error_message = result["error"].get("message", error_message)
                raise UploadError(
                    message=error_message,
                    error_type=UploadErrorType.SERVER_ERROR,
                    status_code=result.get("status_code"),
                    details=result.get("error")
                )
                
            # 从响应中提取图片信息
            image_data = result.get("image", {})
            
            # 构建图片元数据
            image_metadata = ImageMetadata(
                width=image_data.get("width", 0),
                height=image_data.get("height", 0),
                filename=image_data.get("filename", filename),
                size=image_data.get("size", 0),
                url=image_data.get("url", ""),
                delete_url=image_data.get("delete_url", None)
            )
            
            return UploadResponse(
                success=True,
                code="success",
                message=result.get("success", {}).get("message", "Upload success"),
                data=image_metadata
            )
            
        except requests.RequestException as e:
            # 处理网络请求相关错误
            raise UploadError(
                message=f"Upload request failed: {str(e)}",
                error_type=UploadErrorType.NETWORK_ERROR,
                original_error=e
            )
        except (KeyError, ValueError, TypeError) as e:
            # 处理响应解析错误
            raise UploadError(
                message=f"Invalid response format: {str(e)}",
                error_type=UploadErrorType.PARSE_ERROR,
                original_error=e
            )
        except UploadError:
            # 重新抛出已经是 UploadError 类型的异常
            raise
        except Exception as e:
            # 处理其他未预期的错误
            raise UploadError(
                message=f"Upload failed: {str(e)}",
                error_type=UploadErrorType.UNKNOWN,
                original_error=e
            )

    
class ImageUploaderFactory:
    @staticmethod
    def create(provider: str, **credentials) -> ImageUploader:
        if provider == "smms":
            return SmMsUploader(credentials["api_key"])
        elif provider == "qiniu":
            return QiniuUploader(
                credentials["access_key"], 
                credentials["secret_key"]
            )
        elif provider == "picgo":
            api_url = credentials.get("api_url", "https://www.picgo.net/api/1/upload")
            return PicGoUploader(credentials["api_key"], api_url)
        raise ValueError(f"Unknown provider: {provider}")