211 lines
8.2 KiB
Python
211 lines
8.2 KiB
Python
"""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.core.errors import ScrapeParseError
|
|
from app.models.scrape import (
|
|
Breadcrumb,
|
|
ItemDetailData,
|
|
PurchaseInfo,
|
|
ShopSummary,
|
|
SkuAttribute,
|
|
SkuAxis,
|
|
SkuAxisValue,
|
|
SkuInfo,
|
|
SkuVariant,
|
|
)
|
|
from app.parsers.state import extract_initial_state
|
|
from app.parsers.subsites.base import SubsitePage, parse_price
|
|
from app.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 = "サイズ"
|
|
|
|
# cart_info.cart_url_type → 加购端点。站点前端用同名映射表（resolveCartUrl）解析；
|
|
# 若该字段本身已经是一个 URL，则直接使用。
|
|
_CART_URL_BY_TYPE = {
|
|
"1": "https://ts.basket.step.rakuten.co.jp/rms/mall/bs/cartadd/set",
|
|
"2": "https://ts.sp.basket.step.rakuten.co.jp/rms/mall/bss/cartadd/set",
|
|
"3": "https://t2.basket.step.rakuten.co.jp/rms/mall/bs/cartadd/set",
|
|
"4": "https://t2.sp.basket.step.rakuten.co.jp/rms/mall/bss/cartadd/set",
|
|
"5": "https://basket.step.rakuten.co.jp/rms/mall/bs/cartadd/set",
|
|
"6": "https://sp.basket.step.rakuten.co.jp/rms/mall/bss/cartadd/set",
|
|
}
|
|
_DEFAULT_PURCHASE_EVENT = "ES01_003_001"
|
|
|
|
|
|
def _resolve_cart_url(cart_url_type: str) -> str:
|
|
if cart_url_type.startswith("http"):
|
|
return cart_url_type
|
|
return _CART_URL_BY_TYPE.get(cart_url_type, "")
|
|
|
|
|
|
def _purchase_info(product: dict) -> PurchaseInfo:
|
|
"""加购字段与市场是同一套契约，差别只在端点由 cart_url_type 映射得到"""
|
|
cart_info = as_dict(as_dict(product.get("rms_info")).get("cart_info"))
|
|
if not cart_info:
|
|
return PurchaseInfo()
|
|
return PurchaseInfo(
|
|
cart_url=_resolve_cart_url(as_str(cart_info.get("cart_url_type"))),
|
|
form_fields={
|
|
"shop_bid": as_str(cart_info.get("shop_bid")),
|
|
"item_id": as_str(cart_info.get("item_id")),
|
|
"inventory_flag": as_str(cart_info.get("inventory_type")),
|
|
"__event": as_str(cart_info.get("event")) or _DEFAULT_PURCHASE_EVENT,
|
|
"encode": "utf8",
|
|
},
|
|
quantity_field="units",
|
|
variant_field="variant_id",
|
|
)
|
|
|
|
|
|
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),
|
|
purchase=_purchase_info(product),
|
|
)
|