rakuten-api/app/scraping/utils/urls.py

"""乐天页面 URL 的构建与解析

搜索页 URL 形如：
    https://search.rakuten.co.jp/search/mall/{关键词}/{分类ID}/?p=2&s=2&min=1000&f=2&f=101
其中关键词段在只按分类检索时用 `-` 占位；`f` 可重复出现，每个值代表一项筛选。

商品页 URL 形如：
    https://item.rakuten.co.jp/{店铺代码}/{商品编号}/
"""
from __future__ import annotations

from urllib.parse import quote, urlencode, urlparse, urlsplit, urlunsplit, parse_qsl

from app.scraping.core import site
from app.shared.errors import InvalidRequestError
from app.scraping.models.scrape import SearchRequest

# 只按分类检索（无关键词）时，关键词段的占位符
_KEYWORD_PLACEHOLDER = "-"

# www.rakuten.co.jp 下已被站点自身占用、不可能是店铺代码的一级路径
_RESERVED_WWW_PATHS = frozenset({"category", "event", "search", "rms", "info", "help"})


def build_search_url(payload: SearchRequest) -> str:
    """根据搜索请求参数构建乐天搜索页 URL"""
    keyword = payload.keyword.strip()
    path = quote(keyword, safe="") if keyword else _KEYWORD_PLACEHOLDER
    url = f"{site.SEARCH_BASE_URL}{path}/"
    if payload.genre_id:
        url = f"{url}{quote(str(payload.genre_id), safe='')}/"

    params: list[tuple[str, str]] = []
    if payload.page > 1:
        params.append(("p", str(payload.page)))

    sort_code = site.SORT_CODES.get(payload.sort.value)
    if sort_code:
        params.append(("s", sort_code))

    if payload.min_price is not None:
        params.append(("min", str(payload.min_price)))
    if payload.max_price is not None:
        params.append(("max", str(payload.max_price)))
    if payload.shop_id is not None:
        params.append(("sid", str(payload.shop_id)))
    if payload.exclude_keyword:
        params.append(("nitem", payload.exclude_keyword))
    if payload.title_only:
        params.append(("sf", "1"))
    if payload.or_query:
        params.append(("st", "O"))
    if payload.min_review_score is not None:
        params.append(("review", str(payload.min_review_score)))
    if payload.tags:
        params.append(("tg", ",".join(str(tag) for tag in payload.tags)))

    # 成色与各布尔筛选共用可重复的 `f` 参数
    if payload.condition is not None:
        params.append(("f", site.CONDITION_CODES[payload.condition.value]))
    for field_name, code in site.BOOL_FILTER_CODES.items():
        if getattr(payload, field_name):
            params.append(("f", code))

    if not params:
        return url
    return f"{url}?{urlencode(params)}"


def normalize_search_url(raw_url: str, page: int) -> str:
    """校验透传的搜索 URL，并在需要时覆盖其中的页码

    page 大于 1 时以入参为准覆盖 URL 中的 `p`，这样上游可以复用同一条 URL 翻页。
    """
    parsed = urlsplit(raw_url)
    if parsed.hostname != site.SEARCH_HOST:
        raise InvalidRequestError(f"search_url 必须是 {site.SEARCH_HOST} 下的搜索页地址")

    if page <= 1:
        return raw_url

    query = [(key, value) for key, value in parse_qsl(parsed.query, keep_blank_values=True) if key != "p"]
    query.append(("p", str(page)))
    return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, urlencode(query), parsed.fragment))


def build_genre_url(genre_id: str | None) -> str:
    """构建取分类树用的页面地址

    指定分类时用分类页——它比搜索页多给 genreInfo（完整分类名与描述）。
    不指定时退回搜索页：分类分面与查询内容无关，用哨兵关键词能拿到干净的顶层列表。
    """
    if genre_id:
        return f"{site.CATEGORY_BASE_URL}{quote(str(genre_id).strip().strip('/'), safe='')}/"
    return f"{site.SEARCH_BASE_URL}{site.GENRE_FACET_PROBE_KEYWORD}/"


def build_item_url(shop_code: str, item_code: str) -> str:
    """由店铺代码与商品编号拼出商品详情页 URL"""
    shop = quote(shop_code.strip().strip("/"), safe="")
    item = quote(item_code.strip().strip("/"), safe="")
    if not shop or not item:
        raise InvalidRequestError("shop_code 与 item_code 均不能为空")
    return f"{site.ITEM_BASE_URL}{shop}/{item}/"


def split_item_url(raw_url: str) -> tuple[str, str]:
    """从商品页 URL 中解析出 (店铺代码, 商品编号)

    Raises:
        InvalidRequestError: 不是 item.rakuten.co.jp 下的商品页地址
    """
    parsed = urlsplit(raw_url)
    if parsed.hostname != site.ITEM_HOST:
        raise InvalidRequestError(f"item_url 必须是 {site.ITEM_HOST} 下的商品页地址")

    segments = [segment for segment in parsed.path.split("/") if segment]
    if len(segments) < 2:
        raise InvalidRequestError(f"无法从 item_url 中解析店铺代码与商品编号：{raw_url}")
    return segments[0], segments[1]


def build_shop_url(shop_code: str) -> str:
    """由店铺代码拼出店铺首页 URL"""
    shop = quote(shop_code.strip().strip("/"), safe="")
    if not shop:
        raise InvalidRequestError("shop_code 不能为空")
    return f"{site.WWW_BASE_URL}{shop}/"


def split_shop_url(raw_url: str) -> str:
    """从店铺页 URL 中解析出店铺代码

    Raises:
        InvalidRequestError: 不是 www.rakuten.co.jp 下的店铺页地址
    """
    parsed = urlsplit(raw_url)
    if parsed.hostname != site.WWW_HOST:
        raise InvalidRequestError(f"shop_url 必须是 {site.WWW_HOST} 下的店铺页地址")

    segments = [segment for segment in parsed.path.split("/") if segment]
    if not segments:
        raise InvalidRequestError(f"无法从 shop_url 中解析店铺代码：{raw_url}")
    # 排除分类页等非店铺路径，避免把 /category/101205/ 当成店铺代码 category
    if segments[0] in _RESERVED_WWW_PATHS:
        raise InvalidRequestError(f"shop_url 不是店铺首页地址：{raw_url}")
    return segments[0]


def item_url_parts(url: str) -> tuple[str, str]:
    """尽力从任意商品链接中解析 (店铺代码, 商品编号)，失败时返回空串

    用于搜索结果——广告位链接可能指向跳转域名，解析不出来不应中断整页解析。
    """
    try:
        parsed = urlparse(url)
    except ValueError:
        return "", ""
    if parsed.hostname != site.ITEM_HOST:
        return "", ""
    segments = [segment for segment in parsed.path.split("/") if segment]
    if len(segments) < 2:
        return "", ""
    return segments[0], segments[1]