Files
012-kaopeilian/backend/app/services/ai/coze/exceptions.py
111 998211c483 feat: 初始化考培练系统项目
- 从服务器拉取完整代码
- 按框架规范整理项目结构
- 配置 Drone CI 测试环境部署
- 包含后端(FastAPI)、前端(Vue3)、管理端

技术栈: Vue3 + TypeScript + FastAPI + MySQL
2026-01-24 19:33:28 +08:00

102 lines
2.2 KiB
Python

"""
Coze 服务异常定义
"""
from typing import Optional, Dict, Any
class CozeException(Exception):
"""Coze 服务基础异常"""
def __init__(
self,
message: str,
code: Optional[str] = None,
status_code: Optional[int] = None,
details: Optional[Dict[str, Any]] = None,
):
super().__init__(message)
self.message = message
self.code = code
self.status_code = status_code
self.details = details or {}
class CozeAuthError(CozeException):
"""认证异常"""
pass
class CozeAPIError(CozeException):
"""API 调用异常"""
pass
class CozeRateLimitError(CozeException):
"""速率限制异常"""
pass
class CozeTimeoutError(CozeException):
"""超时异常"""
pass
class CozeStreamError(CozeException):
"""流式响应异常"""
pass
def map_coze_error_to_exception(error: Exception) -> CozeException:
"""
将 Coze SDK 错误映射为统一异常
Args:
error: 原始异常
Returns:
CozeException: 映射后的异常
"""
error_message = str(error)
# 根据错误消息判断错误类型
if (
"authentication" in error_message.lower()
or "unauthorized" in error_message.lower()
):
return CozeAuthError(
message="Coze 认证失败",
code="COZE_AUTH_ERROR",
status_code=401,
details={"original_error": error_message},
)
if "rate limit" in error_message.lower():
return CozeRateLimitError(
message="Coze API 速率限制",
code="COZE_RATE_LIMIT",
status_code=429,
details={"original_error": error_message},
)
if "timeout" in error_message.lower():
return CozeTimeoutError(
message="Coze API 调用超时",
code="COZE_TIMEOUT",
status_code=504,
details={"original_error": error_message},
)
# 默认映射为 API 错误
return CozeAPIError(
message="Coze API 调用失败",
code="COZE_API_ERROR",
status_code=500,
details={"original_error": error_message},
)