core/exceptions.py

"""
統一異常處理
定義自訂異常類別和錯誤響應格式
"""

from typing import Optional, Dict, Any
from fastapi import HTTPException
from fastapi.responses import JSONResponse


class BloomWareException(Exception):
    """Bloom Ware 基礎異常類別"""

    def __init__(
        self,
        message: str,
        code: str = "UNKNOWN_ERROR",
        status_code: int = 500,
        details: Optional[Dict[str, Any]] = None,
    ):
        self.message = message
        self.code = code
        self.status_code = status_code
        self.details = details or {}
        super().__init__(message)

    def to_dict(self) -> Dict[str, Any]:
        """轉換為字典格式"""
        return {
            "success": False,
            "error": {
                "code": self.code,
                "message": self.message,
                "details": self.details,
            }
        }

    def to_response(self) -> JSONResponse:
        """轉換為 JSON 響應"""
        return JSONResponse(
            status_code=self.status_code,
            content=self.to_dict(),
        )


# ==================== 認證相關異常 ====================

class AuthenticationError(BloomWareException):
    """認證錯誤"""

    def __init__(self, message: str = "認證失敗", details: Optional[Dict[str, Any]] = None):
        super().__init__(
            message=message,
            code="AUTHENTICATION_ERROR",
            status_code=401,
            details=details,
        )


class TokenExpiredError(AuthenticationError):
    """Token 過期"""

    def __init__(self):
        super().__init__(
            message="Token 已過期，請重新登入",
            details={"reason": "token_expired"},
        )


class InvalidTokenError(AuthenticationError):
    """無效的 Token"""

    def __init__(self):
        super().__init__(
            message="無效的認證令牌",
            details={"reason": "invalid_token"},
        )


class PermissionDeniedError(BloomWareException):
    """權限不足"""

    def __init__(self, message: str = "權限不足"):
        super().__init__(
            message=message,
            code="PERMISSION_DENIED",
            status_code=403,
        )


# ==================== 資源相關異常 ====================

class ResourceNotFoundError(BloomWareException):
    """資源不存在"""

    def __init__(self, resource_type: str, resource_id: str):
        super().__init__(
            message=f"{resource_type} 不存在",
            code="RESOURCE_NOT_FOUND",
            status_code=404,
            details={
                "resource_type": resource_type,
                "resource_id": resource_id,
            },
        )


class ChatNotFoundError(ResourceNotFoundError):
    """對話不存在"""

    def __init__(self, chat_id: str):
        super().__init__("對話", chat_id)


class UserNotFoundError(ResourceNotFoundError):
    """用戶不存在"""

    def __init__(self, user_id: str):
        super().__init__("用戶", user_id)


# ==================== 驗證相關異常 ====================

class ValidationError(BloomWareException):
    """驗證錯誤"""

    def __init__(self, field: str, message: str):
        super().__init__(
            message=f"參數 '{field}' 驗證失敗: {message}",
            code="VALIDATION_ERROR",
            status_code=400,
            details={"field": field},
        )


class RateLimitExceededError(BloomWareException):
    """請求頻率超限"""

    def __init__(self, retry_after: int = 60):
        super().__init__(
            message="請求頻率超過限制，請稍後再試",
            code="RATE_LIMIT_EXCEEDED",
            status_code=429,
            details={"retry_after": retry_after},
        )


# ==================== 服務相關異常 ====================

class ServiceUnavailableError(BloomWareException):
    """服務不可用"""

    def __init__(self, service_name: str):
        super().__init__(
            message=f"{service_name} 服務暫時不可用",
            code="SERVICE_UNAVAILABLE",
            status_code=503,
            details={"service": service_name},
        )


class DatabaseError(BloomWareException):
    """數據庫錯誤"""

    def __init__(self, message: str = "數據庫操作失敗"):
        super().__init__(
            message=message,
            code="DATABASE_ERROR",
            status_code=500,
        )


class AIServiceError(BloomWareException):
    """AI 服務錯誤"""

    def __init__(self, message: str = "AI 服務暫時不可用"):
        super().__init__(
            message=message,
            code="AI_SERVICE_ERROR",
            status_code=503,
        )


class ExternalAPIError(BloomWareException):
    """外部 API 錯誤"""

    def __init__(self, api_name: str, message: str):
        super().__init__(
            message=f"{api_name} API 錯誤: {message}",
            code="EXTERNAL_API_ERROR",
            status_code=502,
            details={"api": api_name},
        )


# ==================== 語音相關異常 ====================

class VoiceAuthError(BloomWareException):
    """語音認證錯誤"""

    def __init__(self, message: str, reason: str):
        super().__init__(
            message=message,
            code="VOICE_AUTH_ERROR",
            status_code=400,
            details={"reason": reason},
        )


class SpeakerLabelTakenError(VoiceAuthError):
    """語音標籤已被使用"""

    def __init__(self):
        super().__init__(
            message="此語音標籤已被其他用戶綁定",
            reason="speaker_label_taken",
        )


# ==================== 異常處理器 ====================

def create_error_response(
    code: str,
    message: str,
    status_code: int = 500,
    details: Optional[Dict[str, Any]] = None,
) -> JSONResponse:
    """創建標準錯誤響應"""
    return JSONResponse(
        status_code=status_code,
        content={
            "success": False,
            "error": {
                "code": code,
                "message": message,
                "details": details or {},
            }
        }
    )


def handle_exception(exc: Exception) -> JSONResponse:
    """
    統一異常處理
    
    將各種異常轉換為標準 JSON 響應
    """
    if isinstance(exc, BloomWareException):
        return exc.to_response()
    
    if isinstance(exc, HTTPException):
        return create_error_response(
            code="HTTP_ERROR",
            message=exc.detail,
            status_code=exc.status_code,
        )
    
    # 未知異常
    import logging
    logger = logging.getLogger("core.exceptions")
    logger.exception(f"未處理的異常: {exc}")
    
    return create_error_response(
        code="INTERNAL_ERROR",
        message="內部伺服器錯誤",
        status_code=500,
    )