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

"""楽天ブックス（books.rakuten.co.jp）商品页解析

该站是传统服务端渲染页面，商品数据以 schema.org 微数据标注（Product / Offer /
AggregateRating），规格与简介在 `.sec-item` 分节里，分类路径则通过页面内联的
`var data_genres` 给出——其中的 rmsGenreId 就是市场侧的 genre_id。

站点不提供：SKU 组合（图书没有规格轴）、运费明细、店铺评分。
"""
from __future__ import annotations

import json
import re

from selectolax.parser import HTMLParser

from app.shared.errors import ScrapeParseError
from app.scraping.models.scrape import (
    Breadcrumb,
    ItemDetailData,
    ReviewSummary,
    ShippingInfo,
    ShopSummary,
    SkuAttribute,
    SkuInfo,
)
from app.scraping.parsers.subsites.base import SubsitePage, looks_sold_out, parse_price

HOST = "books.rakuten.co.jp"
SOURCE = "books"
SHOP_NAME = "楽天ブックス"

_GENRES_RE = re.compile(r"var\s+data_genres\s*=\s*")
# 折叠正文里由展开控件（checkbox/label）留下的连续空行
_BLANK_LINES_RE = re.compile(r"\n{2,}")
_IMAGE_RE = re.compile(r"//tshop\.r10s\.jp/book/cabinet/[^\"'\s?]+\.(?:jpg|jpeg|png)", re.I)
_CATEGORY_URL = "https://www.rakuten.co.jp/category/{}/"


def validate(html: str) -> str | None:
    """页面须带 Product 微数据，否则不是正常的商品页"""
    if 'itemprop="price"' in html and "schema.org/Product" in html:
        return None
    return f"books item page markup not found (body {len(html)} bytes)"


def _attr(tree: HTMLParser, selector: str, name: str) -> str:
    node = tree.css_first(selector)
    return (node.attributes.get(name) or "") if node else ""


def _text(tree: HTMLParser, selector: str) -> str:
    node = tree.css_first(selector)
    return node.text(strip=True) if node else ""


def _parse_spec(tree: HTMLParser) -> list[SkuAttribute]:
    """商品情報分节：每个 ul 内首个 li.product-title 是字段名，其后是取值"""
    attributes: list[SkuAttribute] = []
    for row in tree.css(".sec-item__identifier__list ul"):
        cells = row.css("li")
        if len(cells) < 2:
            continue
        title = cells[0]
        if "product-title" not in (title.attributes.get("class") or ""):
            continue
        value = " ".join(cell.text(strip=True) for cell in cells[1:] if cell.text(strip=True))
        if value:
            attributes.append(SkuAttribute(title=title.text(strip=True), value=value))
    return attributes


def _parse_description(tree: HTMLParser) -> str:
    """商品説明分节：把「内容紹介」「目次」等若干小节拼起来

    这些小节在 DOM 里是平铺的——标题 h3 与正文 div 互为兄弟节点而非父子，
    所以要从标题往后找同级的正文块，不能直接按容器取。
    """
    parts: list[str] = []
    for title in tree.css(".sec-item__extra__title"):
        node = title.next
        while node is not None:
            classes = node.attributes.get("class") or "" if node.tag != "-text" else ""
            if "sec-item__extra__content" in classes:
                body = _BLANK_LINES_RE.sub("\n", node.text(separator="\n", strip=True)).strip()
                if body:
                    parts.append(f"{title.text(strip=True)}\n{body}")
                break
            # 只在紧邻的兄弟里找，遇到下一个标题就说明这一节没有正文
            if node.tag != "-text" and "sec-item__extra__title" in classes:
                break
            node = node.next
    return "\n\n".join(parts)


def _parse_genres(html: str) -> tuple[str, list[Breadcrumb]]:
    """内联的 data_genres 给出分类路径，其中 rmsGenreId 对应市场侧 genre_id"""
    match = _GENRES_RE.search(html)
    if match is None:
        return "", []
    try:
        raw, _ = json.JSONDecoder().raw_decode(html, match.end())
    except ValueError:
        return "", []

    # 结构是 [[{...}, {...}]]，取第一条路径
    path = raw[0] if isinstance(raw, list) and raw and isinstance(raw[0], list) else raw
    crumbs: list[Breadcrumb] = []
    genre_id = ""
    for node in path if isinstance(path, list) else []:
        if not isinstance(node, dict):
            continue
        rms_id = str(node.get("rmsGenreId") or "")
        name = str(node.get("genreName") or "")
        if not name:
            continue
        crumbs.append(Breadcrumb(name=name, url=_CATEGORY_URL.format(rms_id) if rms_id else ""))
        if rms_id:
            genre_id = rms_id
    return genre_id, crumbs


def _extract_cart_form_item_id(tree: HTMLParser) -> str:
    """从购物车表单里抽 item_id

    楽天ブックス 的站内 item_id 与 URL 上的商品编号不是一回事（URL 17065211 →
    item_id 20600328）。下单要用表单里的 item_id，所以这里专门抽它。
    加购表单本身不再解析成 PurchaseInfo——加购由 trading 内部完成。
    """
    for form in tree.css("form"):
        action = form.attributes.get("action") or ""
        if "/bs/Cart" not in action:
            continue
        for inp in form.css("input"):
            if inp.attributes.get("name") == "item_id":
                return inp.attributes.get("value") or ""
    return ""


def parse(page: SubsitePage) -> ItemDetailData:
    """解析楽天ブックス商品页"""
    tree = HTMLParser(page.html)

    name = _text(tree, "#productTitle") or _text(tree, '[itemprop="name"]')
    if not name:
        raise ScrapeParseError("楽天ブックス页面未找到商品名")

    images = ["https:" + url if url.startswith("//") else url for url in _IMAGE_RE.findall(page.html)]
    # 同一张图可能带不同裁剪参数重复出现，去重但保留出现顺序
    images = list(dict.fromkeys(images))

    status = _text(tree, ".status")
    genre_id, breadcrumbs = _parse_genres(page.html)
    review_count = _text(tree, '[itemprop="reviewCount"]')
    # 站内 item_id 与 URL 编号不同，下单要用表单里的——保留抽取，但不构 PurchaseInfo
    item_id = _extract_cart_form_item_id(tree) or page.item_code

    return ItemDetailData(
        source=SOURCE,
        source_url=page.final_url,
        item_id=item_id,
        item_code=page.item_code,
        item_name=name,
        description=_parse_description(tree),
        item_url=page.requested_url,
        price=parse_price(_attr(tree, '[itemprop="price"]', "content")),
        purchase_condition=status,
        is_sold_out=looks_sold_out(status),
        images=images,
        shop=ShopSummary(shop_code=page.shop_code, shop_name=SHOP_NAME),
        review=ReviewSummary(
            score=float(_attr(tree, '[itemprop="ratingValue"]', "content") or 0) or 0.0,
            count=parse_price(review_count),
        ),
        genre_id=genre_id,
        breadcrumbs=breadcrumbs,
        # 图书统一由楽天ブックス发货，页面只给库存措辞，不给运费明细
        shipping=ShippingInfo(delivery_message=status),
        sku=SkuInfo(inventory_type="single", attributes=_parse_spec(tree), delivery_message=status),
    )