ax-backend-v1/app/core/exceptions.py

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class DocumentNotFoundError(Exception):
    def __init__(self, document_id: str):
        super().__init__(f"Document '{document_id}' not found")

class DocumentParseError(Exception):
    """Word 下载或解析失败。"""
    pass

class ExportError(Exception):
    pass

class RecordNotFoundError(Exception):
    def __init__(self, record_id: str):
        super().__init__(f"Export record '{record_id}' not found")

# 异常映射配置：异常类 -> (HTTP状态码, 业务错误码)
EXCEPTION_MAPPINGS: dict[type[Exception], tuple[int, int]] = {
    DocumentNotFoundError: (404, 404),
    DocumentParseError: (400, 400),
    ExportError: (500, 500),
    RecordNotFoundError: (404, 4041),
}

def register_exception_handlers(app: FastAPI) -> None:
    """注册全局异常处理器，所有自定义异常返回标准格式：{"code": int, "message": str, "data": None}"""
    async def unified_exception_handler(request: Request, exc: Exception) -> JSONResponse:
        """统一异常处理函数"""
        # 从映射中获取配置，如果不存在则使用默认值 (500, 500)
        http_status, error_code = EXCEPTION_MAPPINGS.get(type(exc), (500, 500))
        
        return JSONResponse(
            status_code=http_status,
            content={
                "code": error_code,
                "message": str(exc),
                "data": None,
            },
        )
    
    # 为所有配置的异常类型注册同一个处理函数
    for exc_class in EXCEPTION_MAPPINGS:
        app.add_exception_handler(exc_class, unified_exception_handler)