# -*- coding: utf-8 -*- """B 站视频搜索（WBI 签名 + 旧版降级）。""" from __future__ import annotations import hashlib import re import time import urllib.parse from dataclasses import dataclass from typing import Any, Callable, Mapping from ..classify import parse_duration from .client import BASE, BilibiliClient, BilibiliError, BilibiliProtocolError # B 站前端固定的 mixinKey 重排表。 MIXIN_KEY_ENC_TAB = ( 46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35, 27, 43, 5, 49, 33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13, 37, 48, 7, 16, 24, 55, 40, 61, 26, 17, 0, 1, 60, 51, 30, 4, 22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11, 36, 20, 34, 44, 52, ) WBI_KEY_TTL = 6 * 3600 _TAG_RE = re.compile(r"<[^>]+>") @dataclass(frozen=True) class SearchItem: aid: int bvid: str | None title: str up_mid: int | None up_name: str | None pubdate: int duration_seconds: int cover: str | None play: int = 0 like: int = 0 favorite: int = 0 danmaku: int = 0 review: int = 0 def key_from_url(url: str | None) -> str: """从 ``.../bfs/wbi/xxx.png`` 提取 ``xxx``。""" if not isinstance(url, str) or not url: return "" path = urllib.parse.urlparse(url).path name = path.rsplit("/", 1)[-1] return name.split(".", 1)[0] def mixin_key(img_key: str, sub_key: str) -> str: raw = f"{img_key}{sub_key}" if len(raw) < 64: raise ValueError("WBI key 长度不足") return "".join(raw[index] for index in MIXIN_KEY_ENC_TAB)[:32] def sign_params( params: Mapping[str, Any], key: str, *, now: float | None = None ) -> dict[str, Any]: signed = {str(k): v for k, v in params.items()} signed["wts"] = int(time.time() if now is None else now) query = urllib.parse.urlencode(sorted(signed.items())) signed["w_rid"] = hashlib.md5(f"{query}{key}".encode("utf-8")).hexdigest() return signed def _as_int(value: Any) -> int: if isinstance(value, bool) or not isinstance(value, int): return 0 return value def _clean_title(value: Any) -> str: text = value if isinstance(value, str) else "" return _TAG_RE.sub("", text).strip() def _cover(value: Any) -> str | None: if not isinstance(value, str) or not value: return None if value.startswith("//"): return f"https:{value}" return value def _parse_duration(value: Any) -> int: return parse_duration(value) def parse_search_payload(payload: Mapping[str, Any]) -> list[SearchItem]: data = payload.get("data") if not isinstance(data, Mapping): raise BilibiliProtocolError("search data 不是对象") if "result" not in data: raise BilibiliProtocolError("search data 缺少 result 字段") rows = data["result"] if rows is None: return [] if not isinstance(rows, list): raise BilibiliProtocolError("search data.result 不是数组") items: list[SearchItem] = [] for row in rows: if not isinstance(row, Mapping): raise BilibiliProtocolError("search result item 不是对象") raw_type = row.get("type") if isinstance(raw_type, str) and raw_type and raw_type != "video": continue aid = row.get("id") if isinstance(aid, bool) or not isinstance(aid, int): continue up_mid = row.get("mid") if isinstance(up_mid, bool) or not isinstance(up_mid, int): up_mid = None bvid = row.get("bvid") if not isinstance(bvid, str) or not bvid: bvid = None up_name = row.get("author") if not isinstance(up_name, str) or not up_name: up_name = None pubdate = row.get("pubdate") if isinstance(pubdate, bool) or not isinstance(pubdate, int): pubdate = 0 items.append(SearchItem( aid=aid, bvid=bvid, title=_clean_title(row.get("title")), up_mid=up_mid, up_name=up_name, pubdate=pubdate, duration_seconds=_parse_duration(row.get("duration")), cover=_cover(row.get("pic")), play=_as_int(row.get("play")), like=_as_int(row.get("like")), favorite=_as_int(row.get("favorites")), danmaku=_as_int(row.get("danmaku")), review=_as_int(row.get("review")), )) return items class SearchClient: """带 WBI 密钥缓存与旧接口降级的搜索客户端。""" def __init__( self, client_getter: Callable[[], BilibiliClient], *, clock=time.time, ) -> None: self._client_getter = client_getter self._clock = clock self._keys: tuple[str, str] | None = None self._keys_at = 0.0 def _wbi_keys(self) -> tuple[str, str]: now = self._clock() if self._keys is not None and now - self._keys_at < WBI_KEY_TTL: return self._keys payload = self._client_getter().get_json( f"{BASE}/x/web-interface/nav", auth=True ) data = payload.get("data") if not isinstance(data, Mapping): raise BilibiliProtocolError("nav data 不是对象") wbi = data.get("wbi_img") if not isinstance(wbi, Mapping): raise BilibiliProtocolError("nav data.wbi_img 缺失") img_key = key_from_url(wbi.get("img_url")) sub_key = key_from_url(wbi.get("sub_url")) if not img_key or not sub_key: raise BilibiliProtocolError("WBI key 提取失败") self._keys = (img_key, sub_key) self._keys_at = now return self._keys def search( self, keyword: str, page: int, *, page_size: int = 42 ) -> list[SearchItem]: params = { "search_type": "video", "keyword": keyword, "order": "pubdate", "page": int(page), "page_size": int(page_size), } try: img_key, sub_key = self._wbi_keys() signed = sign_params( params, mixin_key(img_key, sub_key), now=self._clock() ) url = ( f"{BASE}/x/web-interface/wbi/search/type?" f"{urllib.parse.urlencode(signed)}" ) return parse_search_payload(self._client_getter().get_json(url)) except BilibiliError: # WBI 失败时降级到旧接口；旧接口也失败则抛出。 url = ( f"{BASE}/x/web-interface/search/type?" f"{urllib.parse.urlencode(params)}" ) return parse_search_payload(self._client_getter().get_json(url))