core/errors.py

"""统一错误响应、上游报文脱敏与错误分类。

目的：杜绝把含 JWT / token / 上游原始响应体的字符串回吐给客户端，
同时为 OpenAI / Claude 两种方言提供规范错误体，并按类驱动 token 冷却。
"""
import re
from fastapi.responses import JSONResponse

# 匹配形如 "upstream API error 492" / "Tabbit API error 401" / "error 502"
_STATUS_RE = re.compile(r"(upstream API error \d{3}|Tabbit API error \d{3}|error \d{3})", re.I)


def safe_upstream_message(e: Exception) -> str:
    """只保留上游状态码语义，丢弃可能含密钥 / JWT / 响应体的原始字符串。"""
    msg = str(e)
    m = _STATUS_RE.search(msg)
    if m:
        return m.group(1)
    low = msg.lower()
    if "session" in low:
        return "failed to create upstream chat session"
    if "timeout" in low or "timed out" in low:
        return "upstream request timed out"
    if "connect" in low or "connection" in low:
        return "failed to connect to upstream"
    return "upstream request failed"


def classify_upstream_error(e: Exception) -> str:
    """错误分类：auth / rate_limit / signature / transient。"""
    msg = str(e)
    m = re.search(r"\b(\d{3})\b", msg)
    status = int(m.group(1)) if m else None
    if status in (401, 403):
        return "auth"
    if status == 429:
        return "rate_limit"
    if status in (492, 493):
        return "signature"
    return "transient"


def openai_error(status: int, message: str, etype: str = "upstream_error", code=None) -> JSONResponse:
    """OpenAI 风格错误体：{"error": {...}}。"""
    return JSONResponse(
        status_code=status,
        content={"error": {"message": message, "type": etype, "code": code, "param": None}},
    )


def claude_error(status: int, message: str, etype: str = "api_error") -> JSONResponse:
    """Anthropic 风格错误体：{"type":"error","error":{...}}。"""
    return JSONResponse(
        status_code=status,
        content={"type": "error", "error": {"type": etype, "message": message}},
    )