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.scraping.models.scrape import (
    Breadcrumb,
    ItemDetailData,
    PurchaseInfo,
    PurchaseOption,
    PurchaseOptionValue,
    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"

# 普通购买的事件标识，站点前端构造加购表单时固定带上
_NORMAL_PURCHASE_EVENT = "ES01_003_001"

# 库存类型 → 加购表单里的 inventory_flag
_INVENTORY_FLAG = {"multiple": 2}
_DEFAULT_INVENTORY_FLAG = 1


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 parse_purchase_options(raw: Any) -> list[PurchaseOption]:
    """解析商品选项（選択肢）

    结构为 {id, name, type: select|check|text, isRequired, values:[{id, name}]}；
    type=text 的选项没有候选值，由买家自由填写（如刻字内容）。
    """
    options: list[PurchaseOption] = []
    for option in as_list(raw):
        if not isinstance(option, dict):
            continue
        options.append(
            PurchaseOption(
                option_id=str(as_int(option.get("id"))) if option.get("id") is not None else "",
                name=as_str(option.get("name")),
                type=as_str(option.get("type")),
                is_required=bool(option.get("isRequired")),
                values=[
                    PurchaseOptionValue(
                        value_id=str(as_int(value.get("id"))) if value.get("id") is not None else "",
                        name=as_str(value.get("name")),
                    )
                    for value in as_list(option.get("values"))
                    if isinstance(value, dict)
                ],
            )
        )
    return options


def _purchase_info(
    *,
    sell_type: dict[str, Any],
    raw_sku: dict[str, Any],
    information: dict[str, Any],
    shop_id: int,
    item_id: str,
    item_variant_id: str,
) -> PurchaseInfo:
    """组装加购所需的端点与字段

    字段名与取值来自站点前端构造加购表单的逻辑（getPurchaseFormData）。
    basketDomain 逐商品不同（不同店铺落在不同的 basket 集群），不能写死。
    """
    inventory_flag = _INVENTORY_FLAG.get(as_str(raw_sku.get("inventoryType")), _DEFAULT_INVENTORY_FLAG)
    form_fields = {
        "shop_bid": str(shop_id),
        "item_id": item_id,
        "inventory_flag": str(inventory_flag),
        "__event": _NORMAL_PURCHASE_EVENT,
    }
    # 单一库存商品的规格是固定的，直接填好，调用方无需再选
    if inventory_flag == _DEFAULT_INVENTORY_FLAG and item_variant_id:
        form_fields["variant_id"] = item_variant_id

    options = parse_purchase_options(information.get("options"))
    return PurchaseInfo(
        cart_url=as_str(sell_type.get("basketDomain")),
        form_fields=form_fields,
        quantity_field="units",
        variant_field="variant_id",
        options_field="choice" if options else "",
        options=options,
        has_required_options=any(option.is_required for option in options),
    )


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"))

    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(as_dict(purchase.get("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,
        purchase=_purchase_info(
            sell_type=sell_type,
            raw_sku=raw_sku,
            information=as_dict(purchase.get("information")),
            shop_id=as_int(shop_information.get("shopId")),
            item_id=item_id,
            item_variant_id=as_str(item.get("variantId")),
        ),
    )