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

"""ラクマ（fril.jp）页面 URL 的构建与解析

搜索页 URL 形如：
    https://fril.jp/s?query=switch&category_id=788&statuses=5,4&min=1000&transaction=selling

参数名与取值取自站点前端 bundle 中 SearchPanel 组件的 `_url()` 方法，
不是猜测。注意站点对无法识别的参数值不会报错，而是**静默返回首页**
（如 statuses=99 返回 75KB 的首页 HTML），因此所有枚举值都必须来自码表。

商品页 URL 形如：https://item.fril.jp/{商品hash}
店铺页 URL 形如：https://fril.jp/shop/{店铺hash}
"""
from __future__ import annotations

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

from app.scraping.core import rakuma_site as site
from app.shared.errors import InvalidRequestError
from app.scraping.models.scrape import RakumaSearchRequest


def build_search_url(payload: RakumaSearchRequest) -> str:
    """根据搜索请求参数构建 ラクマ 搜索页 URL"""
    params: list[tuple[str, str]] = []

    keyword = payload.keyword.strip()
    if keyword:
        params.append(("query", keyword))
    if payload.exclude_keyword:
        params.append(("excluded_query", payload.exclude_keyword))
    if payload.category_id:
        params.append(("category_id", str(payload.category_id).strip()))
    # 站点前端把这两项做成互斥分支：勾选「排除无品牌」时不下发 brand_id
    if payload.except_for_no_brand:
        params.append(("except_for_no_brand", "true"))
    elif payload.brand_id:
        params.append(("brand_id", str(payload.brand_id).strip()))

    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.authenticity_types:
        codes = [site.AUTHENTICITY_CODES[value.value] for value in payload.authenticity_types]
        params.append(("authenticity_types", ",".join(codes)))
    if payload.conditions:
        codes = [site.CONDITION_CODES[value.value] for value in payload.conditions]
        params.append(("statuses", ",".join(codes)))
    if payload.free_shipping:
        params.append(("carriage", site.CARRIAGE_INCLUDED))
    if payload.transaction is not None:
        params.append(("transaction", site.TRANSACTION_CODES[payload.transaction.value]))
    if payload.anonymous_shipping:
        params.append(("anonymous_shipping", "true"))

    sort_code, order_code = site.SORT_CODES[payload.sort.value]
    params.append(("sort", sort_code))
    params.append(("order", order_code))

    if payload.page > 1:
        params.append(("page", str(payload.page)))

    return f"{site.SEARCH_BASE_URL}?{urlencode(params)}"


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

    page 大于 1 时以入参为准覆盖 URL 中的 `page`，这样上游可以复用同一条 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 != "page"
    ]
    query.append(("page", str(page)))
    return urlunsplit((parsed.scheme, parsed.netloc, parsed.path, urlencode(query), parsed.fragment))


def build_category_url(category_id: str) -> str:
    """由分类 ID 拼出分类商品列表页 URL（`/category/{id}`）

    这是对外返回的分类页地址；抓取分类树用的是 `/category` 一览页
    （见 rakuma_site.CATEGORY_LIST_URL），两者不是同一个页面。
    """
    category = quote(str(category_id).strip().strip("/"), safe="")
    return f"{site.CATEGORY_BASE_URL}{category}" if category else ""


def build_item_url(item_id: str) -> str:
    """由商品 hash 拼出商品详情页 URL"""
    item = quote(item_id.strip().strip("/"), safe="")
    if not item:
        raise InvalidRequestError("item_id 不能为空")
    return f"{site.ITEM_BASE_URL}{item}"


def split_item_url(raw_url: str) -> str:
    """从商品页 URL 中解析出商品 hash

    Raises:
        InvalidRequestError: 不是 item.fril.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 not segments:
        raise InvalidRequestError(f"无法从 item_url 中解析商品编号：{raw_url}")
    return segments[0]


def build_shop_url(shop_id: str, *, page: int = 1, review: bool = False) -> str:
    """由店铺 hash 拼出店铺页 URL

    Args:
        page: 商品列表页码，1 时不带参数
        review: 取评价页（/review）而非商品列表页
    """
    shop = quote(shop_id.strip().strip("/"), safe="")
    if not shop:
        raise InvalidRequestError("shop_id 不能为空")

    url = f"{site.SHOP_BASE_URL}{shop}"
    if review:
        return f"{url}/review"
    return f"{url}?page={page}" if page > 1 else url


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

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

    segments = [segment for segment in parsed.path.split("/") if segment]
    if len(segments) < 2 or segments[0] != "shop":
        raise InvalidRequestError(f"无法从 shop_url 中解析店铺编号：{raw_url}")
    return segments[1]


def item_id_from_url(url: str) -> str:
    """尽力从任意商品链接中解析商品 hash，失败时返回空串

    用于列表页解析——单条链接解析不出来不应中断整页解析。
    """
    try:
        parsed = urlsplit(url)
    except ValueError:
        return ""
    if parsed.hostname != site.ITEM_HOST:
        return ""
    segments = [segment for segment in parsed.path.split("/") if segment]
    return segments[0] if segments else ""