rakuten-api/app/scraping/parsers/item.py

"""商品详情页 __INITIAL_STATE__ → ItemDetailData

与搜索页不同，详情页的状态没有 `state` 包裹层，各业务块直接挂在顶层：
item / purchase / shop / review / shipping / breadcrumbs。

注意：只有手机 UA 才会拿到这套统一模板；PC UA 返回的是各店铺自定义的 EUC-JP
老页面，里面没有 __INITIAL_STATE__。UA 的选择由 RakutenClient 负责。
"""
from __future__ import annotations

from typing import Any

from app.shared.errors import ScrapeParseError
from app.shared.purchase_contract import ItemOption as SharedItemOption
from app.shared.purchase_contract import auto_choice_for, parse_options
from app.scraping.models.scrape import (
    Breadcrumb,
    ItemDetailData,
    ItemOption,
    ItemOptionValue,
    ReviewSummary,
    ShippingInfo,
    ShopSummary,
    SkuAttribute,
    SkuAxis,
    SkuAxisValue,
    SkuInfo,
    SkuVariant,
)
from app.scraping.utils.coerce import as_dict, as_float, as_int, as_list, as_str

# purchase.sellType 下表示「可正常购买」的状态值
_PURCHASABLE_CONDITION = "enabled"

# 库存类型 → 加购表单里的 inventory_flag（常量与基础字段构造在 app.shared.purchase_contract）


def _to_item_options(options: list[SharedItemOption]) -> list[ItemOption]:
    """把共用契约的 ItemOption dataclass 搬成对外的 pydantic 模型

    解析与占位项判定都在 `app.shared.purchase_contract.parse_options`——trading
    下单时用同一份逻辑挑 choice 取值，两侧对「哪个取值是合法的」必须完全一致，
    否则本接口告诉上游能选的值、下单时却填了别的。本函数只做搬运，不加判断。
    """
    return [
        ItemOption(
            option_id=option.option_id,
            name=option.name,
            type=option.type,
            is_required=option.is_required,
            values=[
                ItemOptionValue(
                    value_id=value.value_id,
                    name=value.name,
                    is_placeholder=value.is_placeholder,
                )
                for value in option.values
            ],
            selectable_value_count=len(option.selectable_values),
        )
        for option in options
    ]


def _parse_attributes(raw: Any) -> list[SkuAttribute]:
    return [
        SkuAttribute(title=as_str(attr.get("title")), value=as_str(attr.get("value")))
        for attr in as_list(raw)
        if isinstance(attr, dict)
    ]


def _parse_axis(raw: Any) -> list[SkuAxis]:
    axes: list[SkuAxis] = []
    for axis in as_list(raw):
        if not isinstance(axis, dict):
            continue
        axes.append(
            SkuAxis(
                key=as_str(axis.get("key")),
                label=as_str(axis.get("label")),
                values=[
                    SkuAxisValue(
                        value=as_str(value.get("value")),
                        label=as_str(value.get("label")),
                        is_sold_out=bool(value.get("isSoldOut")),
                    )
                    for value in as_list(axis.get("values"))
                    if isinstance(value, dict)
                ],
            )
        )
    return axes


def _parse_variants(raw: Any) -> list[SkuVariant]:
    variants: list[SkuVariant] = []
    for variant in as_list(raw):
        if not isinstance(variant, dict):
            continue
        quantity = as_int(variant.get("quantity"))
        variants.append(
            SkuVariant(
                variant_id=as_str(variant.get("variantId")),
                selector_values=[as_str(value) for value in as_list(variant.get("selectorValues"))],
                price=as_int(variant.get("price")),
                quantity=quantity,
                # 站点未给 SKU 级的售罄标记，库存为 0 即视为该组合不可购买
                is_sold_out=quantity <= 0,
                delivery_message=as_str(variant.get("deliveryMessageRMS")),
                attributes=_parse_attributes(variant.get("attributes")),
            )
        )
    return variants


def _pick_sell_type(sell_type: dict[str, Any]) -> dict[str, Any]:
    """取售卖方式信息，优先普通购买，其次任意一种带价格的方式（如定期购）"""
    normal = as_dict(sell_type.get("normalPurchase"))
    if normal:
        return normal
    for value in sell_type.values():
        candidate = as_dict(value)
        if "minPrice" in candidate:
            return candidate
    return {}


def parse_item_detail(
    state: dict[str, Any],
    *,
    item_url: str,
    shop_code: str,
    include_sku_variants: bool,
) -> ItemDetailData:
    """把商品详情页状态解析为商品详情

    Raises:
        ScrapeParseError: 状态中不存在 item 节点
    """
    item = state.get("item")
    if not isinstance(item, dict) or not item:
        raise ScrapeParseError("商品详情缺少 item 节点")

    purchase = as_dict(state.get("purchase"))
    sell_type = _pick_sell_type(as_dict(purchase.get("sellType")))
    purchase_condition = as_str(sell_type.get("purchaseCondition"))

    purchase_information = as_dict(purchase.get("information"))
    # 只解析一次，两个派生结果都从这份结果来
    shared_options = parse_options(purchase_information)
    # 无法自动选值的必填项：与 trading 自动填 choice 时的判定同源，上游据此知道
    # 「哪些项必须自己给值」，而不是等下单时才被站点拒绝
    _, unfillable_required = auto_choice_for(shared_options)

    raw_sku = as_dict(purchase.get("sku"))
    variants = _parse_variants(raw_sku.get("variants"))
    sku = SkuInfo(
        inventory_type=as_str(raw_sku.get("inventoryType")),
        quantity=as_int(raw_sku.get("quantity")),
        show_inventory=bool(raw_sku.get("showInventory")),
        delivery_message=as_str(raw_sku.get("deliveryMessageRMS")),
        attributes=_parse_attributes(raw_sku.get("attributes")),
        axis=_parse_axis(raw_sku.get("axis")),
        variants=variants if include_sku_variants else [],
        variant_count=len(variants),
    )

    shop_information = as_dict(as_dict(state.get("shop")).get("information"))
    shop_review = as_dict(shop_information.get("shopReview"))
    resolved_shop_code = as_str(shop_information.get("shopUrl")) or shop_code
    shop = ShopSummary(
        shop_id=as_int(shop_information.get("shopId")) or None,
        shop_code=resolved_shop_code,
        shop_name=as_str(shop_information.get("shopName")),
        shop_url=f"https://www.rakuten.co.jp/{resolved_shop_code}/" if resolved_shop_code else "",
        review_score=as_float(shop_review.get("rating")),
        review_count=as_int(shop_review.get("total")),
    )

    item_review = as_dict(as_dict(state.get("review")).get("item"))
    review = ReviewSummary(
        score=as_float(item_review.get("totalRating")),
        count=as_int(item_review.get("count")),
    )

    shipping_raw = as_dict(as_dict(state.get("shipping")).get("informationFromServer"))
    shipping_fee = shipping_raw.get("shippingFee")
    threshold = shipping_raw.get("freeShippingThreshold")
    shipping = ShippingInfo(
        shipping_fee=as_int(shipping_fee) if shipping_fee is not None else None,
        is_shipping_free=bool(shipping_raw.get("isShippingFree")),
        is_asuraku=bool(shipping_raw.get("isAsuraku")),
        is_next_day_delivery=bool(shipping_raw.get("isNextDayDelivery")),
        free_shipping_threshold=as_int(threshold) if threshold is not None else None,
        prefecture_id=as_int(shipping_raw.get("prefectureId")) or None,
        delivery_message=as_str(raw_sku.get("deliveryMessageRMS")),
    )

    breadcrumbs = [
        Breadcrumb(name=as_str(crumb.get("name")), url=as_str(crumb.get("url")))
        for crumb in as_list(as_dict(state.get("breadcrumbs")).get("genreBreadcrumbs"))
        if isinstance(crumb, dict)
    ]

    images = [
        as_str(image.get("imageUrl"))
        for image in as_list(as_dict(item.get("media")).get("images"))
        if isinstance(image, dict) and as_str(image.get("imageUrl"))
    ]

    item_id = str(as_int(item.get("itemId"))) if item.get("itemId") is not None else ""
    return ItemDetailData(
        source="ichiba",
        source_url=item_url,  # 市场页无跳转，解析地址即商品地址
        item_id=item_id,
        item_code=as_str(item.get("itemNumber")),
        item_name=as_str(item.get("itemName")),
        catch_copy=as_str(item.get("catchCopy")),
        description=as_str(item.get("description")),
        item_url=item_url,
        price=as_int(sell_type.get("minPrice")),
        pre_tax_price=as_int(sell_type.get("preTaxPrice")),
        tax_flag=bool(item.get("taxFlag")),
        tax_rate=as_float(shop_information.get("taxRate")),
        purchase_condition=purchase_condition,
        # purchaseCondition 是站点判定能否下单的直接依据；缺失时不臆断为售罄
        is_sold_out=bool(purchase_condition) and purchase_condition != _PURCHASABLE_CONDITION,
        purchase_unit=as_int(purchase_information.get("unit")),
        images=images,
        shop=shop,
        review=review,
        genre_id=str(as_int(item.get("genreId"))) if item.get("genreId") is not None else "",
        breadcrumbs=breadcrumbs,
        shipping=shipping,
        sku=sku,
        options=_to_item_options(shared_options),
        has_required_options=any(option.is_required for option in shared_options),
        unfillable_required_options=unfillable_required,
    )