class AppServiceError(Exception): def __init__( self, message, *, category="internal", code="INTERNAL_ERROR", status=None, retryable=False, provider="system", detail=None, ): super().__init__(message) self.category = category self.code = code self.status = status self.retryable = retryable self.provider = provider self.detail = detail or {} def to_dict(self): return { "message": str(self), "category": self.category, "code": self.code, "status": self.status, "retryable": self.retryable, "provider": self.provider, "detail": self.detail, } def classify_http_error(status, code, retryable=False): code = str(code or "").upper() if status == 429 or code in {"RATE_LIMIT_EXCEEDED"}: return "rate_limit" if status == 401 or code in { "MISSING_API_KEY", "INVALID_API_KEY", "DISABLED_API_KEY", "EXPIRED_API_KEY", "INVALID_OPENAPI_USER_TOKEN", "OPENAPI_TOKEN_APP_MISMATCH", }: return "authentication" if status == 403 or code in {"OPENAPI_USER_REQUIRED", "SCOPE_NOT_ALLOWED", "USER_SCOPE_NOT_ALLOWED"}: return "authorization" if status == 404: return "not_found" if status == 400: return "validation" if status == 402 or code in {"INSUFFICIENT_POINTS", "VIP_REQUIRED"}: return "business_rule" if status and status >= 500: return "upstream" if retryable: return "upstream" return "unknown" def normalize_exception(error): if isinstance(error, AppServiceError): return error return AppServiceError(str(error), category="internal", code="INTERNAL_ERROR")