"""ビックカメラ楽天市場店(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, ), )