rakuten-api/app/scraping/services/rakuten_client.py

"""乐天抓取客户端：把请求参数翻译成站点 URL，抓取后解析为结构化数据

两个接口分别走不同的指纹通道：
- 搜索：PC 通道（search.rakuten.co.jp 的搜索页）
- 详情：手机通道（item.rakuten.co.jp 只有手机 UA 才返回统一模板）
"""
from __future__ import annotations

import logging

from app.scraping.core import site
from app.shared.config import Settings
from app.shared.errors import ScrapeParseError
from app.scraping.models.scrape import (
    GenreData,
    GenreRequest,
    ItemDetailData,
    ItemDetailRequest,
    SearchRequest,
    SearchResultData,
    ShopDetailData,
    ShopDetailRequest,
    ShopItemsRequest,
)
from app.scraping.parsers.genre import parse_genres
from app.scraping.parsers.item import parse_item_detail
from app.scraping.parsers.search import parse_search
from app.scraping.parsers.shop import parse_shop_detail
from app.scraping.parsers.state import extract_initial_state
from app.scraping.parsers.subsites import (
    SubsitePage,
    build_item_page_validator,
    host_of,
    parse_subsite_item,
)
from app.scraping.services.site_session import SiteSession
from app.scraping.utils.urls import (
    build_genre_url,
    build_item_url,
    build_search_url,
    build_shop_url,
    normalize_search_url,
    split_item_url,
    split_shop_url,
)

logger = logging.getLogger(__name__)


class RakutenClient:
    """乐天市场抓取客户端"""

    def __init__(self, settings: Settings, session: SiteSession):
        self._settings = settings
        self._session = session

    async def search(self, payload: SearchRequest) -> SearchResultData:
        """抓取搜索结果列表"""
        if payload.search_url is not None:
            url = normalize_search_url(str(payload.search_url), payload.page)
        else:
            url = build_search_url(payload)

        logger.info("抓取搜索页：url=%s", url)
        html = await self._session.fetch_html(url, mobile=False)
        state = extract_initial_state(html)
        result = parse_search(
            state,
            request_url=url,
            page=payload.page,
            exclude_ads=payload.exclude_ads,
        )
        logger.info(
            "搜索完成：url=%s items=%s ads=%s total=%s",
            url, len(result.items), result.ad_count, result.total_count,
        )
        return result

    async def genres(self, payload: GenreRequest) -> GenreData:
        """抓取分类树：顶层分类列表，或指定分类的信息与直接子分类"""
        genre_id = (payload.genre_id or "").strip() or None
        url = build_genre_url(genre_id)

        logger.info("抓取分类树：url=%s genre_id=%s", url, genre_id)
        html = await self._session.fetch_html(url, mobile=False)
        state = extract_initial_state(html)
        result = parse_genres(state, genre_id=genre_id)
        logger.info(
            "分类树完成：genre_id=%s name=%s children=%s",
            result.genre_id or "root", result.name, len(result.children),
        )
        return result

    async def shop_detail(self, payload: ShopDetailRequest) -> ShopDetailData:
        """抓取商家详情"""
        if payload.shop_url is not None:
            shop_code = split_shop_url(str(payload.shop_url))
        else:
            shop_code = (payload.shop_code or "").strip()
        url = build_shop_url(shop_code)

        logger.info("抓取店铺页：url=%s", url)
        html = await self._session.fetch_html(url, mobile=False)
        state = extract_initial_state(html)
        result = parse_shop_detail(state, shop_code=shop_code, html=html)
        logger.info(
            "店铺详情完成：shop_code=%s shop_id=%s name=%s reviews=%s",
            result.shop_code, result.shop_id, result.shop_name, result.review_count,
        )
        return result

    async def shop_items(self, payload: ShopItemsRequest) -> SearchResultData:
        """抓取商家名下的商品列表

        站点没有可分页抓取的「店铺内商品」页，店铺商品实际就是搜索结果按
        `sid=` 限定店铺后的产物，因此这里转成一次搜索。只给 shop_code 时
        需要先取一次店铺详情换出 shop_id。
        """
        shop_id = payload.shop_id
        if shop_id is None:
            detail = await self.shop_detail(ShopDetailRequest(shop_code=payload.shop_code))
            if detail.shop_id is None:
                raise ScrapeParseError(f"未能取得店铺 {payload.shop_code} 的 shop_id")
            shop_id = detail.shop_id

        return await self.search(payload.to_search_request(shop_id))

    async def item_detail(self, payload: ItemDetailRequest) -> ItemDetailData:
        """抓取商品详情"""
        if payload.item_url is not None:
            shop_code, item_code = split_item_url(str(payload.item_url))
        else:
            # 模型校验已保证此分支下两者都非空
            shop_code, item_code = payload.shop_code or "", payload.item_code or ""

        # 统一收敛到规范地址，去掉 variantId 等查询参数——所有 SKU 组合都会随详情返回
        url = build_item_url(shop_code, item_code)

        logger.info("抓取商品详情：url=%s", url)
        # 部分官方旗舰店的商品页会跳出市场域名，落到各自的独立站点；
        # 校验与解析都按最终落地域名分派，落到未登记站点时由校验器抛错。
        page = await self._session.fetch(
            url, mobile=True, validator=build_item_page_validator(url)
        )

        if host_of(page.url) == site.ITEM_HOST:
            detail = parse_item_detail(
                extract_initial_state(page.html),
                item_url=url,
                shop_code=shop_code,
                include_sku_variants=payload.include_sku_variants,
            )
        else:
            detail = parse_subsite_item(
                SubsitePage(
                    html=page.html,
                    requested_url=url,
                    final_url=page.url,
                    shop_code=shop_code,
                    item_code=item_code,
                    include_sku_variants=payload.include_sku_variants,
                )
            )

        logger.info(
            "详情完成：url=%s source=%s name=%s price=%s skus=%s",
            url, detail.source, detail.item_name[:40], detail.price, detail.sku.variant_count,
        )
        return detail