rakuten-api/app/scraping/parsers/subsites/biccamera.py

"""ビックカメラ楽天市場店（biccamera.rakuten.co.jp）商品页解析

Nuxt 应用，整页数据内联在 `window.__NUXT__`（纯 JSON 对象，可直接增量解析）。
商品主体在 `state.item`，字段命名已经很接近市场侧语义，且直接给出市场的
shop_id / genre_id / 分类路径与库存数。

站点不提供：SKU 组合（该店商品都是单一规格）、商品评分（异步加载）。
"""
from __future__ import annotations

import json
import re

from app.shared.errors import ScrapeParseError
from app.scraping.models.scrape import (
    Breadcrumb,
    ItemDetailData,
    ShippingInfo,
    ShopSummary,
    SkuInfo,
)
from app.scraping.parsers.subsites.base import SubsitePage
from app.scraping.utils.coerce import as_dict, as_int, as_list, as_str

HOST = "biccamera.rakuten.co.jp"
SOURCE = "biccamera"
SHOP_NAME = "ビックカメラ楽天市場店"

_NUXT_RE = re.compile(r"window\.__NUXT__\s*=\s*")
_CATEGORY_URL = "https://www.rakuten.co.jp/category/{}/"


def validate(html: str) -> str | None:
    if _NUXT_RE.search(html):
        return None
    return f"biccamera nuxt state not found (body {len(html)} bytes)"


def _extract_nuxt(html: str) -> dict:
    match = _NUXT_RE.search(html)
    if match is None:
        raise ScrapeParseError("ビックカメラ页面未找到 window.__NUXT__")
    try:
        state, _ = json.JSONDecoder().raw_decode(html, match.end())
    except ValueError as exc:
        raise ScrapeParseError(f"window.__NUXT__ 解析失败：{exc}") from exc
    if not isinstance(state, dict):
        raise ScrapeParseError("window.__NUXT__ 不是 JSON 对象")
    return state


def parse(page: SubsitePage) -> ItemDetailData:
    """解析ビックカメラ商品页"""
    nuxt = _extract_nuxt(page.html)
    item = as_dict(as_dict(nuxt.get("state")).get("item"))
    if not item or not as_str(item.get("item_name")):
        raise ScrapeParseError("ビックカメラ页面缺少 state.item")

    sold_out = bool(item.get("sold_out_flag"))
    inventory = as_int(item.get("inventory"))
    delivery = as_str(item.get("delivery_schedule_text"))

    breadcrumbs = [
        Breadcrumb(
            name=as_str(genre.get("genre_name")),
            url=_CATEGORY_URL.format(as_str(genre.get("genre_id"))),
        )
        for genre in as_list(item.get("genres"))
        if isinstance(genre, dict) and as_str(genre.get("genre_name"))
    ]

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

    shop_code = as_str(item.get("shop_url")) or page.shop_code
    return ItemDetailData(
        source=SOURCE,
        source_url=page.final_url,
        item_id=str(as_int(item.get("item_id"))) if item.get("item_id") is not None else "",
        item_code=as_str(item.get("item_number")) or page.item_code,
        item_name=as_str(item.get("item_name")),
        catch_copy=as_str(item.get("catch_copy")),
        description=as_str(item.get("caption")),
        item_url=page.requested_url,
        price=as_int(item.get("price_with_tax")),
        pre_tax_price=as_int(item.get("original_price")),
        tax_flag=bool(item.get("included_tax_flag")),
        purchase_condition="soldOut" if sold_out else "enabled",
        is_sold_out=sold_out,
        purchase_unit=as_int(item.get("units")),
        images=images,
        shop=ShopSummary(
            shop_id=as_int(item.get("shop_id")) or None,
            shop_code=shop_code,
            shop_name=SHOP_NAME,
            shop_url=f"https://www.rakuten.co.jp/{shop_code}/",
        ),
        genre_id=str(as_int(item.get("genre_id"))) if item.get("genre_id") is not None else "",
        breadcrumbs=breadcrumbs,
        shipping=ShippingInfo(
            # 该店商品价格含运费时站点会置位此标记，不再单独给运费金额
            is_shipping_free=bool(item.get("included_shipping_fee_flag")),
            delivery_message=delivery,
        ),
        sku=SkuInfo(
            inventory_type="single",
            quantity=inventory,
            show_inventory=inventory > 0,
            delivery_message=delivery,
        ),
    )