"""基础模型:错误码 / BridgeError 异常 / ErrorResponse。""" from __future__ import annotations from typing import Any, Optional from pydantic import BaseModel, Field # --------------------------------------------------------------------------- # 错误码定义(与 spec 一致) # --------------------------------------------------------------------------- # 每个错误码对应一个 HTTP 状态码 ERROR_CODES: dict[str, int] = { "WECHAT_NOT_RUNNING": 503, "WECHAT_NOT_LOGGED_IN": 401, "WINDOW_NOT_FOUND": 503, "CONTACT_NOT_FOUND": 404, "SEND_FAILED": 500, "DB_LOCKED": 503, "DB_NOT_FOUND": 500, "INVALID_PARAMS": 400, "BRIDGE_INTERNAL_ERROR": 500, "LOGIN_TIMEOUT": 408, "MEDIA_NOT_FOUND": 404, "RATE_LIMITED": 429, "DB_ENCRYPTED": 503, "DB_NEED_INIT": 503, "DB_INIT_IN_PROGRESS": 503, "DB_KEY_INVALID": 503, "LOGOUT_FAILED": 500, "RESTART_TIMEOUT": 408, "REVOKE_WINDOW_EXPIRED": 409, # UI 自动化 Flow 错误码(P3) "SEND_TIMEOUT": 503, "STATE_DIRTY": 503, "X11_TEMP_UNAVAILABLE": 503, "X11_UNAVAILABLE": 503, "ELEMENT_NOT_FOUND": 503, "DB_VERIFY_FAILED": 500, "AMBIGUOUS_CONTACT": 400, "BRIDGE_CIRCUITED": 503, "WECHAT_NOT_READY": 503, } class BridgeError(Exception): """bridge 统一业务异常。 路由层直接 raise BridgeError(code="...", message="...", details=...), 由全局异常处理器捕获后转换为统一 ErrorResponse 并设置对应 HTTP 状态码。 """ def __init__( self, code: str = "BRIDGE_INTERNAL_ERROR", message: str = "bridge internal error", http_status: Optional[int] = None, details: Optional[Any] = None, ) -> None: self.code = code self.message = message # 若未显式指定 http_status,则查表取默认值;查不到则回落到 500 self.http_status = http_status if http_status is not None else ERROR_CODES.get(code, 500) self.details = details super().__init__(f"[{code}] {message}") # --------------------------------------------------------------------------- # 通用错误响应 # --------------------------------------------------------------------------- class ErrorResponse(BaseModel): """统一错误响应结构。""" success: bool = Field(default=False, description="固定为 false") error: dict[str, Any] = Field( description="错误详情,含 code/message/details 三个字段" )