165 lines
6.3 KiB
Python
165 lines
6.3 KiB
Python
"""乐天页面 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.core import site
|
|
from app.core.errors import InvalidRequestError
|
|
from app.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]
|