"""乐天页面 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]