File size: 4,408 Bytes
85270f9 | 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 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | from __future__ import annotations
from typing import Any
from fastapi.responses import JSONResponse
from utils.sensitive_data import sanitize_sensitive_text
def _translate(message: str) -> str:
"""先做敏感信息脱敏，再交给管理员配置的报错文案映射。
映射只影响前台展示；调用方（日志）拿到的仍是脱敏后的原始报错。
"""
cleaned = sanitize_sensitive_text(message)
if not cleaned:
return cleaned
from services.error_message_service import error_message_service
return error_message_service.translate(cleaned, channel="api")
def _message_from_value(value: object) -> str:
if isinstance(value, str):
return value
if not isinstance(value, dict):
return ""
message = value.get("message")
if isinstance(message, str) and message:
return message
return _message_from_value(value.get("error"))
def error_message_from_detail(detail: object) -> str:
if isinstance(detail, list):
messages = []
for item in detail:
if not isinstance(item, dict):
continue
location = ".".join(str(part) for part in item.get("loc", []) if part != "body")
message = str(item.get("msg") or "").strip()
if location and message:
messages.append(f"{location}: {message}")
elif message:
messages.append(message)
return _translate("; ".join(messages))
if isinstance(detail, dict):
message = _message_from_value(detail.get("error")) or _message_from_value(detail)
if message:
return _translate(message)
return _translate(str(detail or "").strip())
def _default_error_type(status_code: int) -> str:
if status_code == 401:
return "authentication_error"
if status_code == 403:
return "permission_error"
if status_code == 429:
return "rate_limit_error"
if 400 <= status_code < 500:
return "invalid_request_error"
return "server_error"
def _default_error_code(status_code: int) -> str:
if status_code == 401:
return "invalid_api_key"
if status_code == 403:
return "permission_denied"
if status_code == 429:
return "rate_limit_exceeded"
if 400 <= status_code < 500:
return "bad_request"
return "upstream_error"
def openai_error_payload(
detail: object,
status_code: int,
*,
error_type: str | None = None,
code: object | None = None,
param: object | None = None,
) -> dict[str, Any]:
error_detail = detail.get("error") if isinstance(detail, dict) else None
if isinstance(error_detail, dict):
return {
"error": {
"message": error_message_from_detail(error_detail) or "request failed",
"type": str(error_detail.get("type") or error_type or _default_error_type(status_code)),
"param": error_detail.get("param", param),
"code": error_detail.get("code", code if code is not None else _default_error_code(status_code)),
}
}
return {
"error": {
"message": error_message_from_detail(detail) or "request failed",
"type": error_type or _default_error_type(status_code),
"param": param,
"code": code if code is not None else _default_error_code(status_code),
}
}
def openai_error_response(
detail: object,
status_code: int,
*,
headers: dict[str, str] | None = None,
error_type: str | None = None,
code: object | None = None,
param: object | None = None,
) -> JSONResponse:
return JSONResponse(
status_code=status_code,
content=openai_error_payload(detail, status_code, error_type=error_type, code=code, param=param),
headers=headers,
)
def anthropic_error_response(
detail: object,
status_code: int,
*,
headers: dict[str, str] | None = None,
) -> JSONResponse:
error_type = "api_error" if status_code >= 500 else _default_error_type(status_code)
return JSONResponse(
status_code=status_code,
content={
"type": "error",
"error": {
"type": error_type,
"message": error_message_from_detail(detail) or "request failed",
},
},
headers=headers,
)
|