rakuten-api/app/scraping/parsers/search.py

"""搜索页 __INITIAL_STATE__ → SearchResultData

数据位于 state.data.ichibaSearch，含 pagination 与 items 两部分。

需要注意的两个站点行为：
- items 里会混入 CPC 广告位，其 url 指向 grp07.ias.rakuten.co.jp 跳转域，
  真实商品地址在 originalItemUrl；判定依据是 itemOptions.cpc 非空。
- pagination.numFound 是命中总数，但实际只能翻到 pagination.subset 条为止。
"""
from __future__ import annotations

from typing import Any

from app.scraping.core import site
from app.shared.errors import ScrapeParseError
from app.scraping.models.scrape import ReviewSummary, SearchItem, SearchResultData, ShopSummary
from app.scraping.utils.coerce import as_dict, as_float, as_int, as_list, as_str
from app.scraping.utils.urls import item_url_parts


def parse_search_item(raw: dict[str, Any]) -> SearchItem:
    """解析单个搜索结果条目"""
    item_options = as_dict(raw.get("itemOptions"))
    is_ad = bool(item_options.get("cpc"))

    # 广告位的 url 是跳转链接，真实商品地址只在 originalItemUrl 里
    original_url = as_str(raw.get("originalItemUrl"))
    fallback_url = as_str(raw.get("url"))
    item_url = original_url or fallback_url
    shop_code, item_code = item_url_parts(item_url)
    if not item_code and fallback_url and fallback_url != item_url:
        shop_code, item_code = item_url_parts(fallback_url)

    images = [
        as_str(image.get("url"))
        for image in as_list(raw.get("images"))
        if isinstance(image, dict) and as_str(image.get("url"))
    ]

    raw_shop = as_dict(raw.get("shop"))
    shop_review = as_dict(raw_shop.get("review"))
    shop = ShopSummary(
        shop_id=as_int(raw_shop.get("id")) or None,
        shop_code=as_str(raw_shop.get("urlCode")) or shop_code,
        shop_name=as_str(raw_shop.get("name")),
        shop_url=as_str(raw_shop.get("url")),
        review_score=as_float(shop_review.get("score")),
        review_count=as_int(shop_review.get("count")),
    )

    raw_review = as_dict(raw.get("review"))
    review = ReviewSummary(
        score=as_float(raw_review.get("score")),
        count=as_int(raw_review.get("numReviews")),
        url=as_str(raw_review.get("url")),
    )

    genre_path = as_str(raw.get("genreIdList"))
    genre_names = [
        as_str(genre.get("name"))
        for genre in as_list(raw.get("genres"))
        if isinstance(genre, dict) and as_str(genre.get("name"))
    ]
    # genreIdList 形如 /0/101205/565950/566404，末段为最具体的分类；根节点 0 不算
    genre_id = next(
        (segment for segment in reversed(genre_path.split("/")) if segment and segment != "0"),
        "",
    )

    raw_shipping = as_dict(raw.get("shipping"))
    shipping_price = raw_shipping.get("price")
    sku_info = as_dict(raw.get("skuInfo"))

    return SearchItem(
        item_id=as_str(raw.get("code")),
        item_code=item_code,
        item_name=as_str(raw.get("name")),
        item_url=item_url,
        catch_copy=as_str(raw.get("subtitle")),
        price=as_int(raw.get("price")),
        price_range=as_str(sku_info.get("priceRange")),
        has_price_range=bool(raw.get("hasPriceRange")),
        image_url=images[0] if images else "",
        image_urls=images,
        shop=shop,
        review=review,
        genre_id=genre_id,
        genre_path=genre_path,
        genre_names=genre_names,
        shipping_fee=as_int(shipping_price) if shipping_price is not None else None,
        delivery_message=as_str(raw_shipping.get("estimateDeliveryDay")),
        point_count=as_int(as_dict(raw.get("point")).get("count")),
        is_sold_out=bool(raw.get("isSoldOut")),
        is_ad=is_ad,
        has_multi_sku=bool(sku_info.get("hasMultiSku")),
        variant_id=as_str(raw.get("variantId")),
        item_options=item_options,
    )


def parse_search(
    state: dict[str, Any],
    *,
    request_url: str,
    page: int,
    exclude_ads: bool,
) -> SearchResultData:
    """把搜索页状态解析为搜索结果

    Raises:
        ScrapeParseError: 状态中不存在 ichibaSearch 节点，或站点返回了错误
    """
    data = as_dict(as_dict(state.get("state")).get("data"))
    search = data.get("ichibaSearch")
    if not isinstance(search, dict):
        raise ScrapeParseError("搜索结果缺少 ichibaSearch 节点")

    error = search.get("error")
    if error:
        raise ScrapeParseError(f"站点返回搜索错误：{error}")

    pagination = as_dict(search.get("pagination"))
    ui_question = as_dict(data.get("effectiveUiQuestion"))

    # 站点声明的每页条数；items 实际长度会因为混入广告位而略大于它
    page_size = as_int(pagination.get("pageSize"))
    total_count = as_int(pagination.get("numFound"))
    subset = as_int(pagination.get("subset")) or site.DEFAULT_SUBSET_LIMIT
    reachable_count = min(total_count, subset) if total_count else 0
    start = as_int(pagination.get("start"))

    # 请求页超出可达窗口时，站点会静默回绕到第 1 页并把 start/page 重置为 0/1。
    # 拿站点回报的页码与请求页码对账即可识别，比自行按 subset 推算更可靠。
    site_page = as_int(ui_question.get("page"), page)
    out_of_range = page > 1 and site_page != page

    raw_items = [item for item in as_list(search.get("items")) if isinstance(item, dict)]
    items = [parse_search_item(item) for item in raw_items]
    ad_count = sum(1 for item in items if item.is_ad)
    if exclude_ads:
        items = [item for item in items if not item.is_ad]
    if out_of_range:
        # 这一页的内容是第 1 页的副本，交给上游只会造成重复入库
        items = []
        ad_count = 0

    has_more = (
        not out_of_range
        and bool(raw_items)
        and (start + (page_size or len(raw_items))) < reachable_count
    )

    return SearchResultData(
        keyword=as_str(ui_question.get("keywords")),
        page=page,
        page_size=page_size,
        total_count=total_count,
        reachable_count=reachable_count,
        has_more=has_more,
        out_of_range=out_of_range,
        ad_count=ad_count,
        request_url=request_url,
        items=items,
    )