rakuten-api/app/shared/errors.py

"""应用异常定义：所有业务异常均继承自 AppError

错误码规范：
- 1xxx: 请求/鉴权错误
- 2xxx: 抓取资源错误
- 3xxx: 反爬/上游阻断相关错误
- 4xxx: 页面解析错误
- 5xxx: 加购/下单错误（需要账号登录态的写操作）
"""


class AppError(Exception):
    """应用基础异常，携带错误码、HTTP 状态码和是否可重试信息"""

    def __init__(
        self,
        message: str,
        code: str,
        err_code: int,
        retryable: bool = False,
        status_code: int = 400,
        headers: dict[str, str] | None = None,
    ):
        super().__init__(message)
        self.message = message
        self.code = code
        self.err_code = err_code
        self.retryable = retryable
        self.status_code = status_code
        self.headers = headers


class AuthenticationError(AppError):
    """鉴权失败（Bearer Token 无效或缺失）"""

    def __init__(self, message: str = "Invalid credentials"):
        super().__init__(
            message=message,
            code="AUTH_INVALID",
            err_code=1001,
            retryable=False,
            status_code=401,
            headers={"WWW-Authenticate": "Bearer"},
        )


class InvalidRequestError(AppError):
    """请求参数不合法（如 URL 非乐天站点、缺少必填标识）"""

    def __init__(self, message: str = "Invalid request"):
        super().__init__(message=message, code="INVALID_REQUEST", err_code=1003, retryable=False)


class ResourceBusyError(AppError):
    """抓取资源繁忙（等待并发槽位超时）"""

    def __init__(self, message: str = "Timed out while waiting for a scrape slot"):
        super().__init__(message=message, code="RESOURCE_BUSY", err_code=2002, retryable=True)


class UpstreamRequestError(AppError):
    """上游请求失败（网络异常、超时、5xx）"""

    def __init__(self, message: str = "The upstream request failed"):
        super().__init__(message=message, code="UPSTREAM_ERROR", err_code=3001, retryable=True)


class UpstreamBlockedError(AppError):
    """上游请求被反爬阻断（Akamai 挑战页 / 403 / 页面缺少渲染数据）"""

    def __init__(self, message: str = "The upstream request was blocked"):
        super().__init__(message=message, code="UPSTREAM_BLOCKED", err_code=3002, retryable=True)


class ItemNotFoundError(AppError):
    """商品不存在或已下架（详情页 404）"""

    def __init__(self, message: str = "Item not found"):
        super().__init__(
            message=message,
            code="ITEM_NOT_FOUND",
            err_code=4004,
            retryable=False,
            status_code=404,
        )


class ScrapeParseError(AppError):
    """页面解析失败"""

    def __init__(self, message: str = "Failed to parse the target page"):
        super().__init__(message=message, code="PARSE_ERROR", err_code=4001, retryable=False)


class OffIchibaRedirectError(AppError):
    """商品页跳转到了市场之外的乐天官方子站

    楽天ブックス（book → books.rakuten.co.jp）、ビックカメラ
    （biccamera → biccamera.rakuten.co.jp）等官方旗舰店有各自独立的站点与
    页面结构，不返回市场统一模板，本服务的详情解析不适用。

    单独成一类错误是为了让上游能把这些商品路由到别处，而不是当成被反爬拦截去重试。
    """

    def __init__(self, requested_url: str, final_url: str):
        super().__init__(
            message=(
                f"商品页跳转至乐天市场以外的站点，当前不支持解析："
                f"{requested_url} -> {final_url}"
            ),
            code="OFF_ICHIBA_REDIRECT",
            err_code=4002,
            retryable=False,
        )
        self.requested_url = requested_url
        self.final_url = final_url


class NotLoggedInError(AppError):
    """账号登录态缺失或已失效

    加购与下单必须带已登录的账号会话。两站登录都要过 reCAPTCHA / 设备验证，
    无法自动恢复，因此这里明确标记 retryable=False，让上游停下来走一次
    `scripts/login.py` 重新人工登录，而不是原地重试。
    """

    def __init__(self, site: str, detail: str = ""):
        suffix = f"（{detail}）" if detail else ""
        super().__init__(
            message=(
                f"{site} 账号未登录或登录态已失效{suffix}，"
                f"请运行 scripts/login.py --site {site} 重新登录"
            ),
            code="NOT_LOGGED_IN",
            err_code=5001,
            retryable=False,
            status_code=401,
        )
        self.site = site
        self.detail = detail


class CartOperationError(AppError):
    """加购失败

    站点对加购请求几乎不返回结构化错误：缺必填选项、SKU 已售罄、商品下架
    都可能返回 200 并把用户导回商品页。因此判定依据是「加购后购物车里有没有
    这件商品」，而不是 HTTP 状态码。
    """

    def __init__(self, message: str = "加入购物车失败"):
        super().__init__(message=message, code="CART_FAILED", err_code=5002, retryable=False)


class OrderOperationError(AppError):
    """下单流程失败（确认页解析不出、金额校验不通过、提交被拒等）"""

    def __init__(self, message: str = "下单失败"):
        super().__init__(message=message, code="ORDER_FAILED", err_code=5003, retryable=False)


class OrderGuardError(AppError):
    """下单安全闸门未通过

    真实付款不可逆，因此把「调用方没有显式确认」「实际金额超出上限」这类拦截
    单独成一类错误，与站点侧失败区分开——前者是本服务主动拒绝，重试无意义，
    需要调用方修改入参后再来。
    """

    def __init__(self, message: str):
        super().__init__(message=message, code="ORDER_GUARD", err_code=5004, retryable=False)