rakuten-api/app/scraping/parsers/rakuma/item.py

"""ラクマ 商品详情页 HTML → RakumaItemDetailData

页面数据分三处，各取所长：
- `<script type="application/ld+json">` 的 Product 微数据：名称、价格、描述、图片
- `.item__details` 规格表：成色、尺码、配送方式与地区（每行 th 上的
  `item-status-{key}` class 是稳定键，比日文标签文案可靠）
- 埋点属性 `data-rat-cp-*` 与 dataLayer JSON：数值 ID、分类 ID、品牌 ID

售罄判定用页面上的 SOLD OUT 标记，而不是 ld+json 的 availability——
实测已售出商品的 ld+json 仍写 InStock，不可信。
"""
from __future__ import annotations

import json
import re
from typing import Any

from selectolax.parser import HTMLParser

from app.scraping.core import rakuma_site as site
from app.shared.errors import ScrapeParseError
from app.scraping.models.scrape import Breadcrumb, RakumaItemDetailData, RakumaSeller
from app.scraping.parsers.rakuma.base import (
    attr,
    find_item_payload,
    image_url,
    node_text,
    parse_float,
    parse_int,
    rat_params,
)
from app.scraping.utils.rakuma_urls import split_shop_url

_LD_JSON_RE = re.compile(
    r'<script[^>]*type="application/ld\+json"[^>]*>(.*?)</script>', re.S
)
# 商品主图：站点用 slider 展示，主图挂在 .sp-image 上（推荐位的图不在其中）
_MAIN_IMAGE_SELECTOR = ".sp-slide img.sp-image, .item-photos img, .slider img.sp-image"

# 规格表里 th 图标 class 上的稳定键 → 模型字段
_SPEC_KEYS = {
    "item-status-status": "condition",
    "item-status-size": "size",
    "item-status-carriage": "shipping_payer",
    "item-status-delivery_method": "shipping_method",
    "item-status-delivery_date": "shipping_date_estimate",
    "item-status-delivery_area": "shipping_from",
}

# 站点上表示「没有填写该项」的占位文案
_SPEC_EMPTY_VALUES = ("なし", "未定", "指定なし", "-", "―")

# 规格表缺失时的兜底：页面级埋点属性 data-rat-cp-{key} → 模型字段。
# 注意这组值的措辞与规格表不完全一致（如运费负担规格表写「送料込」，
# 埋点写「出品者」），原样透出，不做归一。
_RAT_SPEC_KEYS = {
    "condition": "item_condition",
    "shipping_payer": "shipping_cost_payer",
    "shipping_date_estimate": "shipping_date_estimate",
    "shipping_from": "shipping_from",
}


def _parse_ld_product(html: str) -> dict[str, Any]:
    """取出 ld+json 里的 Product 节点"""
    for match in _LD_JSON_RE.finditer(html):
        try:
            data = json.loads(match.group(1))
        except ValueError:
            continue
        if isinstance(data, dict) and data.get("@type") == "Product":
            return data
    return {}


def _parse_specs(tree: HTMLParser, rat: dict[str, str]) -> dict[str, str]:
    """解析商品情報规格表

    每行的 th 里有个 `<i class="icon-status ... item-status-{key}">`，
    这个 key 比日文标签稳定，用它做映射。

    已售出商品的页面会换成另一套布局、规格表整体消失，此时退回页面级埋点
    属性——它给的项少一些（没有配送方法与尺码），但成色、运费负担与发货地
    仍在，好过整片留空。
    """
    specs: dict[str, str] = {}
    for row in tree.css("table.item__details tr"):
        icon = row.css_first("th i")
        value_node = row.css_first("td")
        if icon is None or value_node is None:
            continue
        classes = attr(icon, "class").split()
        field = next((_SPEC_KEYS[name] for name in classes if name in _SPEC_KEYS), None)
        if field is None:
            continue
        value = node_text(value_node)
        specs[field] = "" if value in _SPEC_EMPTY_VALUES else value

    for field, key in _RAT_SPEC_KEYS.items():
        if not specs.get(field) and rat.get(key):
            specs[field] = rat[key]
    return specs


def _parse_breadcrumbs(tree: HTMLParser) -> tuple[list[Breadcrumb], str]:
    """解析分类面包屑，并返回最具体的一级分类 ID

    取规格表里的分类行而非页头面包屑：页头那条会把品牌也混进来，
    规格表里的是纯分类链。
    """
    crumbs: list[Breadcrumb] = []
    category_id = ""
    for row in tree.css("table.item__details tr"):
        icon = row.css_first("th i")
        if icon is None or "item-status-category" not in attr(icon, "class"):
            continue
        for link in row.css("td a"):
            url = attr(link, "href")
            crumbs.append(Breadcrumb(name=node_text(link), url=url))
            segments = [segment for segment in url.split("/") if segment]
            if segments:
                category_id = segments[-1]
        break
    return crumbs, category_id


def _parse_seller(tree: HTMLParser, payload: dict[str, Any]) -> RakumaSeller:
    """解析出品者信息块"""
    link = tree.css_first("a.shop_link, a[href*='/shop/']")
    shop_url = attr(link, "href")
    shop_id = ""
    if shop_url:
        try:
            shop_id = split_shop_url(shop_url)
        except Exception:
            shop_id = ""

    seller_user_id = payload.get("seller_user_id")
    return RakumaSeller(
        shop_id=shop_id,
        user_id=str(seller_user_id) if seller_user_id is not None else "",
        shop_name=node_text(tree.css_first(".header-shopinfo__shop-name")),
        user_name=node_text(tree.css_first(".header-shopinfo__user-name")),
        shop_url=shop_url,
        icon_url=image_url(tree.css_first(".header-shopinfo__user-icon img")),
        seller_type=str(payload.get("seller_user_type") or ""),
        review_score=parse_float(node_text(tree.css_first(".shop_score__score"))),
        # 商品页只给评分不给评价数，需要评价数请调 /api/rakuma/shop_detail
        is_verified=tree.css_first(".header-shopinfo__verified-badge-item") is not None,
    )


def parse_item_detail(html: str, *, item_id: str, item_url: str) -> RakumaItemDetailData:
    """把商品详情页 HTML 解析为商品详情

    Raises:
        ScrapeParseError: 页面不是商品详情页
    """
    tree = HTMLParser(html)
    info = tree.css_first(f".{site.ITEM_PAGE_MARKER}")
    if info is None:
        raise ScrapeParseError("页面不是商品详情页（缺少商品信息区块）")

    product = _parse_ld_product(html)
    # 这段埋点挂在哪个元素上因页面状态而异，按属性全页扫描
    payload = find_item_payload(tree)
    rat = rat_params(html)
    specs = _parse_specs(tree, rat)
    breadcrumbs, category_id = _parse_breadcrumbs(tree)

    images = [
        url
        for url in dict.fromkeys(image_url(node) for node in tree.css(_MAIN_IMAGE_SELECTOR))
        if url and "img.fril.jp" in url
    ]
    if not images and isinstance(product.get("image"), str):
        images = [product["image"]]

    # ld+json 的 availability 对已售商品仍写 InStock，只能按页面标记判断
    page_text = info.text()
    is_sold_out = any(marker in page_text for marker in site.SOLD_OUT_MARKERS)

    brand = product.get("brand") if isinstance(product.get("brand"), dict) else {}
    offers = product.get("offers") if isinstance(product.get("offers"), dict) else {}

    return RakumaItemDetailData(
        item_id=item_id,
        item_number=str(payload.get("item_id") or ""),
        item_name=str(product.get("name") or "") or node_text(tree.css_first("h1.item__name")),
        description=str(product.get("description") or "")
        or node_text(tree.css_first(".item__description__line-limited")),
        item_url=item_url,
        price=parse_int(offers.get("price")) or parse_int(node_text(tree.css_first(".item__price"))),
        is_sold_out=is_sold_out,
        images=images,
        condition=specs.get("condition", ""),
        size=specs.get("size", ""),
        brand_id=str(payload.get("brand_id") or "") or rat.get("brand_id", ""),
        brand_name=str(brand.get("name") or "") or str(payload.get("brand_name") or ""),
        category_id=category_id or str(payload.get("category_id") or ""),
        breadcrumbs=breadcrumbs,
        shipping_payer=specs.get("shipping_payer", ""),
        shipping_method=specs.get("shipping_method", ""),
        shipping_date_estimate=specs.get("shipping_date_estimate", ""),
        shipping_from=specs.get("shipping_from", ""),
        is_anonymous_shipping=tree.css_first(".item__icon.anonymous") is not None,
        like_count=parse_int(node_text(tree.css_first(".like_button_set"))),
        comment_count=parse_int(node_text(tree.css_first(".go-to-comment-button"))),
        posted_at=node_text(tree.css_first(".time_ago")),
        seller=_parse_seller(tree, payload),
    )