app/errors.py

"""统一错误模型。

完全兼容原 Netlify 版返回格式：
    { "ok": false, "error": { "code", "message", "status", "retryable", "hint", "upstream?", "details?" } }

错误码集合：bad_request/unauthorized/forbidden/not_found/request_timeout/
quota_exceeded/internal_error/service_unavailable/upstream_timeout/
model_disabled/server_misconfigured
"""
from __future__ import annotations

from typing import Any, Optional

from fastapi import Request
from fastapi.responses import JSONResponse


# 状态码 -> 错误码
_STATUS_TO_CODE = {
    400: "bad_request",
    401: "unauthorized",
    402: "payment_required",
    403: "forbidden",
    404: "not_found",
    405: "method_not_allowed",
    408: "request_timeout",
    409: "conflict",
    422: "unprocessable_entity",
    429: "quota_exceeded",
    500: "internal_error",
    502: "bad_gateway",
    503: "service_unavailable",
    504: "upstream_timeout",
}

# 可重试状态码
_RETRYABLE_STATUS = {408, 409, 429, 500, 502, 503, 504}


class HttpError(Exception):
    """统一 HTTP 异常。"""

    def __init__(
        self,
        message: str,
        status: int = 500,
        code: Optional[str] = None,
        details: Any = None,
        upstream: Any = None,
    ) -> None:
        super().__init__(message)
        self.message = message
        self.status = status
        self.code = code or _STATUS_TO_CODE.get(status, "internal_error")
        self.details = details
        self.upstream = upstream


def error_code_for_status(status: int, fallback: str = "upstream_error") -> str:
    return _STATUS_TO_CODE.get(status, fallback)


def is_retryable_status(status: int) -> bool:
    return status in _RETRYABLE_STATUS


def get_error_hint(*, code: str, status: int, message: str) -> Optional[str]:
    """生成给前端展示的提示文案。"""
    text = (message or "").lower()
    if code == "unauthorized" or status == 401:
        return "请检查 access_key/access_token 是否正确，或先重新生成临时令牌"
    if code == "forbidden" or status == 403:
        return "当前账号或策略不允许此操作，请检查后台厂商与模型策略"
    if code == "not_found" or status == 404:
        return "资源不存在，请检查接口路径、provider 与 model 参数"
    if code == "quota_exceeded" or status == 429:
        return "请求过快或额度不足，建议稍后重试或切换厂商/模型"
    if code == "upstream_timeout" or status == 504 or "timeout" in text:
        return "上游超时，建议降低 max_tokens 或稍后重试"
    if code == "bad_request" or status in (400, 422):
        return "请求参数格式可能有误，请检查 messages/images/model 字段"
    if status >= 500:
        return "服务暂时不可用，请稍后重试"
    return None


def make_error_response(
    *,
    message: str,
    status: int = 500,
    code: Optional[str] = None,
    details: Any = None,
    upstream: Any = None,
) -> JSONResponse:
    err_code = code or _STATUS_TO_CODE.get(status, "internal_error")
    hint = get_error_hint(code=err_code, status=status, message=message)
    payload = {
        "ok": False,
        "error": {
            "code": err_code,
            "message": message,
            "status": status,
            "retryable": is_retryable_status(status),
            "hint": hint,
        },
    }
    if details is not None:
        payload["error"]["details"] = details
    if upstream is not None:
        payload["error"]["upstream"] = upstream
    return JSONResponse(status_code=status, content=payload, headers=_CORS_HEADERS)


def make_upstream_error_response(
    *,
    status: int,
    text: str,
    json_body: Any = None,
    fallback_code: str = "upstream_error",
) -> JSONResponse:
    """从上游响应构造统一错误。"""
    upstream_err = None
    if isinstance(json_body, dict):
        upstream_err = json_body.get("error") if isinstance(json_body.get("error"), dict) else json_body
    code = (upstream_err or {}).get("code") if isinstance(upstream_err, dict) else None
    code = code or error_code_for_status(status, fallback_code)
    message = (upstream_err or {}).get("message") if isinstance(upstream_err, dict) else None
    message = message or text or f"Upstream error ({status})"
    hint = get_error_hint(code=code, status=status, message=message)
    payload = {
        "ok": False,
        "error": {
            "code": code,
            "message": message,
            "status": status,
            "retryable": is_retryable_status(status),
            "hint": hint,
            "upstream": upstream_err,
        },
    }
    return JSONResponse(status_code=status, content=payload, headers=_CORS_HEADERS)


_CORS_HEADERS = {
    "Access-Control-Allow-Origin": "*",
    "Access-Control-Allow-Methods": "*",
    "Access-Control-Allow-Headers": "*",
}


async def http_error_handler(request: Request, exc: HttpError) -> JSONResponse:
    return make_error_response(
        message=exc.message,
        status=exc.status,
        code=exc.code,
        details=exc.details,
        upstream=exc.upstream,
    )


async def unhandled_exception_handler(request: Request, exc: Exception) -> JSONResponse:
    import traceback
    traceback.print_exc()
    return make_error_response(
        message=str(exc) or "Unexpected error",
        status=500,
        code="internal_error",
    )