"""Rakuten Fashion / BRAND AVENUE(brandavenue.rakuten.co.jp)商品页解析 该站同样把整页数据内联在 `window.__INITIAL_STATE__`,但结构与市场页完全不同: 商品主体在 `itemDetail.data.product`,店铺信息在 `env`,市场侧的 genre_id 与 商品 ID 则藏在 `product.rms_info` 里。 SKU 有两个轴(颜色 / 尺码):product_sku 给出可售组合与售价,rms_info.inventory_list 给出各组合库存,两者按「尺码 + 颜色名」对齐。 站点不提供:商品评分(异步加载)、运费明细。面包屑用的是站内分类编码而非市场 genre_id,因此只给分类名不给链接。 """ from __future__ import annotations import re from app.shared.errors import ScrapeParseError from app.scraping.models.scrape import ( Breadcrumb, ItemDetailData, ShopSummary, SkuAttribute, SkuAxis, SkuAxisValue, SkuInfo, SkuVariant, ) from app.scraping.parsers.state import extract_initial_state from app.scraping.parsers.subsites.base import SubsitePage, parse_price from app.scraping.utils.coerce import as_dict, as_int, as_list, as_str HOST = "brandavenue.rakuten.co.jp" SOURCE = "brandavenue" SHOP_NAME = "Rakuten Fashion" _COLOR_AXIS = "カラー" _SIZE_AXIS = "サイズ" def validate(html: str) -> str | None: if "window.__INITIAL_STATE__" in html: return None return f"brandavenue state not found (body {len(html)} bytes)" def _images(html: str, image_folder: str, main_filename: str) -> list[str]: """从页面直出的图片地址中收集商品图 图片按商品编号做了目录分片,分片规则不对外暴露,所以直接取页面里已经渲染好的 地址,而不是自行拼接,避免规则变化导致图片全错。 """ if not image_folder: return [] pattern = re.compile( rf"https://[a-z0-9.\-]+/{re.escape(image_folder)}/[^\"'\s]+\.(?:jpg|jpeg|png)", re.I ) urls = list(dict.fromkeys(pattern.findall(html))) if not main_filename: return urls # 主图排在最前,便于调用方直接取 images[0] 当封面 main = main_filename.lower() urls.sort(key=lambda url: 0 if url.lower().endswith("/" + main) else 1) return urls def _breadcrumbs(product: dict) -> list[Breadcrumb]: """category_l_m_cd_name 是 [大类ID, 大类名, 中类ID, 中类名] 的扁平数组""" flat = [as_str(value) for value in as_list(product.get("category_l_m_cd_name"))] crumbs: list[Breadcrumb] = [] for index in range(0, len(flat) - 1, 2): name = flat[index + 1] if name: crumbs.append(Breadcrumb(name=name)) return crumbs def _sku(product: dict, *, include_variants: bool) -> SkuInfo: entries = [entry for entry in as_list(product.get("product_sku")) if isinstance(entry, dict)] inventory = { (as_str(row.get("size")), as_str(row.get("color_name"))): row for row in as_list(as_dict(product.get("rms_info")).get("inventory_list")) if isinstance(row, dict) } variants: list[SkuVariant] = [] colors: dict[str, bool] = {} sizes: dict[str, bool] = {} for entry in entries: color = as_str(entry.get("product_color_name")) size = as_str(entry.get("product_size_name")) in_stock = as_str(entry.get("inventory_exist_flg")) == "1" row = as_dict(inventory.get((size, color))) attributes = [ SkuAttribute(title=title, value=as_str(entry.get(key))) for title, key in (("素材", "material"), ("お手入れ", "cleaning"), ("お届け目安", "inventory_status_message")) if as_str(entry.get(key)) ] variants.append( SkuVariant( variant_id=as_str(row.get("variant_id")), selector_values=[color, size], price=parse_price(entry.get("selling_price")), quantity=as_int(row.get("stock")), is_sold_out=not in_stock, delivery_message=as_str(entry.get("inventory_status_message")), attributes=attributes, ) ) # 任一组合可售即认为该取值可选 colors[color] = colors.get(color, False) or in_stock sizes[size] = sizes.get(size, False) or in_stock axis = [ SkuAxis( key=key, label=key, values=[ SkuAxisValue(value=value, label=value, is_sold_out=not available) for value, available in mapping.items() if value ], ) for key, mapping in ((_COLOR_AXIS, colors), (_SIZE_AXIS, sizes)) if any(mapping) ] return SkuInfo( inventory_type="multiple" if len(variants) > 1 else "single", quantity=sum(variant.quantity for variant in variants), delivery_message=as_str(entries[0].get("inventory_status_message")) if entries else "", axis=axis, variants=variants if include_variants else [], variant_count=len(variants), ) def parse(page: SubsitePage) -> ItemDetailData: """解析 Rakuten Fashion 商品页""" state = extract_initial_state(page.html) product = as_dict(as_dict(as_dict(state.get("itemDetail")).get("data")).get("product")) if not product: raise ScrapeParseError("Rakuten Fashion 页面缺少 itemDetail.data.product") env = as_dict(state.get("env")) rms = as_dict(product.get("rms_info")) sold_out = as_int(product.get("soldout_flg")) == 1 return ItemDetailData( source=SOURCE, source_url=page.final_url, item_id=as_str(rms.get("rms_item_id")), item_code=page.item_code, item_name=as_str(product.get("product_name")), catch_copy=as_str(product.get("brand_name")), description=as_str(product.get("product_exp")), item_url=page.requested_url, price=parse_price(product.get("selling_price_no_format")), pre_tax_price=parse_price(product.get("fixed_price_no_format")), purchase_condition="soldOut" if sold_out else "enabled", is_sold_out=sold_out, images=_images(page.html, as_str(env.get("product_image_folder")), as_str(product.get("product_img_path"))), shop=ShopSummary( shop_id=as_int(env.get("shop_id")) or None, shop_code=as_str(env.get("shop_url")) or page.shop_code, shop_name=as_str(env.get("shop_name")) or SHOP_NAME, shop_url=f"https://www.rakuten.co.jp/{as_str(env.get('shop_url')) or page.shop_code}/", ), genre_id=as_str(rms.get("genre_id")), breadcrumbs=_breadcrumbs(product), sku=_sku(product, include_variants=page.include_sku_variants), )