Spaces:
Paused
Paused
File size: 4,461 Bytes
1a9e2c2 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 | """异常处理器 - OpenAI兼容的错误响应"""
from fastapi import Request, status
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from starlette.exceptions import HTTPException as StarletteHTTPException
# HTTP错误映射
HTTP_ERROR_MAP = {
400: ("invalid_request_error", "请求格式错误或缺少必填参数"),
401: ("invalid_request_error", "令牌认证失败"),
403: ("permission_error", "没有权限访问此资源"),
404: ("invalid_request_error", "请求的资源不存在"),
429: ("rate_limit_error", "请求频率超出限制，请稍后再试"),
500: ("api_error", "内部服务器错误"),
503: ("api_error", "服务暂时不可用"),
}
# Grok错误码映射
GROK_STATUS_MAP = {
"NO_AUTH_TOKEN": status.HTTP_401_UNAUTHORIZED,
"INVALID_TOKEN": status.HTTP_401_UNAUTHORIZED,
"HTTP_ERROR": status.HTTP_502_BAD_GATEWAY,
"NETWORK_ERROR": status.HTTP_503_SERVICE_UNAVAILABLE,
"JSON_ERROR": status.HTTP_502_BAD_GATEWAY,
"API_ERROR": status.HTTP_502_BAD_GATEWAY,
"STREAM_ERROR": status.HTTP_502_BAD_GATEWAY,
"NO_RESPONSE": status.HTTP_502_BAD_GATEWAY,
"TOKEN_SAVE_ERROR": status.HTTP_500_INTERNAL_SERVER_ERROR,
"NO_AVAILABLE_TOKEN": status.HTTP_503_SERVICE_UNAVAILABLE,
}
GROK_TYPE_MAP = {
"NO_AUTH_TOKEN": "authentication_error",
"INVALID_TOKEN": "authentication_error",
"HTTP_ERROR": "api_error",
"NETWORK_ERROR": "api_error",
"JSON_ERROR": "api_error",
"API_ERROR": "api_error",
"STREAM_ERROR": "api_error",
"NO_RESPONSE": "api_error",
"TOKEN_SAVE_ERROR": "api_error",
"NO_AVAILABLE_TOKEN": "api_error",
}
class GrokApiException(Exception):
"""Grok API业务异常"""
def __init__(self, message: str, error_code: str = None, details: dict = None, context: dict = None, status_code: int = None):
self.message = message
self.error_code = error_code
self.details = details or {}
self.context = context or {}
self.status_code = status_code or GROK_STATUS_MAP.get(error_code)
super().__init__(self.message)
def build_error_response(message: str, error_type: str, code: str = None, param: str = None) -> dict:
"""构建OpenAI兼容的错误响应"""
error = {"message": message, "type": error_type}
if code:
error["code"] = code
if param:
error["param"] = param
return {"error": error}
async def http_exception_handler(_: Request, exc: StarletteHTTPException) -> JSONResponse:
"""处理HTTP异常"""
error_type, default_msg = HTTP_ERROR_MAP.get(exc.status_code, ("api_error", str(exc.detail)))
message = str(exc.detail) if exc.detail else default_msg
return JSONResponse(
status_code=exc.status_code,
content=build_error_response(message, error_type)
)
async def validation_exception_handler(_: Request, exc: RequestValidationError) -> JSONResponse:
"""处理验证错误"""
errors = exc.errors()
param = errors[0]["loc"][-1] if errors and errors[0].get("loc") else None
message = errors[0]["msg"] if errors and errors[0].get("msg") else "请求参数错误"
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content=build_error_response(message, "invalid_request_error", param=param)
)
async def grok_api_exception_handler(_: Request, exc: GrokApiException) -> JSONResponse:
"""处理Grok API异常"""
http_status = GROK_STATUS_MAP.get(exc.error_code, status.HTTP_500_INTERNAL_SERVER_ERROR)
error_type = GROK_TYPE_MAP.get(exc.error_code, "api_error")
return JSONResponse(
status_code=http_status,
content=build_error_response(exc.message, error_type, exc.error_code)
)
async def global_exception_handler(_: Request, exc: Exception) -> JSONResponse:
"""处理未捕获异常"""
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content=build_error_response("服务器遇到意外错误，请重试", "api_error")
)
def register_exception_handlers(app) -> None:
"""注册异常处理器"""
app.add_exception_handler(StarletteHTTPException, http_exception_handler)
app.add_exception_handler(RequestValidationError, validation_exception_handler)
app.add_exception_handler(GrokApiException, grok_api_exception_handler)
app.add_exception_handler(Exception, global_exception_handler)
|