kairee
d971e718e9
Implement WBI video search with legacy fallback, classification scoring, candidate pool, permanent seen-aid dedupe, blacklist filtering, and batched Telegram notifications. Add write client for fav deal/sort, target-folder selection, insertion position calculation, read-back verification, fav_action locking, retry on failure, ignore and blacklist actions. Wire discovery, candidate cleanup, Telegram long polling, and new discover/candidates CLI commands into the scheduler. Add migration v3 for candidate duration and index stats.
140 lines
4.4 KiB
Python
140 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""WBI 签名与搜索解析测试（不联网）: uv run python tests/test_search.py"""
|
|
|
|
import os
|
|
import sys
|
|
import urllib.parse
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
from reaction_board.bilibili.client import BilibiliError, BilibiliProtocolError # noqa: E402
|
|
from reaction_board.bilibili.search import ( # noqa: E402
|
|
SearchClient,
|
|
key_from_url,
|
|
mixin_key,
|
|
parse_search_payload,
|
|
sign_params,
|
|
)
|
|
|
|
IMG_URL = "https://i0.hdslb.com/bfs/wbi/7cd084941338484aae1ad9425b84077c.png"
|
|
SUB_URL = "https://i0.hdslb.com/bfs/wbi/4932caff0ff746eab6f01bf08b70ac45.png"
|
|
|
|
NAV_PAYLOAD = {
|
|
"code": 0, "message": "0", "ttl": 1,
|
|
"data": {
|
|
"isLogin": True,
|
|
"wbi_img": {"img_url": IMG_URL, "sub_url": SUB_URL},
|
|
},
|
|
}
|
|
|
|
SEARCH_PAYLOAD = {
|
|
"code": 0, "message": "0", "ttl": 1,
|
|
"data": {
|
|
"result": [
|
|
{
|
|
"type": "video",
|
|
"id": 117530286889638,
|
|
"bvid": "BV1xvxi6LEZD",
|
|
"title": '<em class="keyword">咻咻满</em> reaction 初听',
|
|
"author": "测试UP",
|
|
"mid": 434442072,
|
|
"pubdate": 1793370433,
|
|
"duration": "105:5",
|
|
"pic": "//i0.hdslb.com/bfs/archive/example.jpg",
|
|
"play": 12345,
|
|
"like": 678,
|
|
"favorites": 90,
|
|
"danmaku": 12,
|
|
"review": 3,
|
|
},
|
|
{"type": "article", "id": 999, "title": "非视频"},
|
|
],
|
|
"numPages": 1,
|
|
"numResults": 1,
|
|
},
|
|
}
|
|
|
|
|
|
class FakeClient:
|
|
def __init__(self, responses):
|
|
self.responses = list(responses)
|
|
self.urls = []
|
|
|
|
def get_json(self, url, *, auth=False):
|
|
self.urls.append((url, auth))
|
|
item = self.responses.pop(0)
|
|
if isinstance(item, BaseException):
|
|
raise item
|
|
return item
|
|
|
|
|
|
def test_wbi_helpers():
|
|
assert key_from_url(IMG_URL) == "7cd084941338484aae1ad9425b84077c"
|
|
assert key_from_url(None) == ""
|
|
key = mixin_key(
|
|
"7cd084941338484aae1ad9425b84077c",
|
|
"4932caff0ff746eab6f01bf08b70ac45",
|
|
)
|
|
assert key == "ea1db124af3c7062474693fa704f4ff8"
|
|
|
|
signed = sign_params({"keyword": "咻咻满", "page": 1}, key, now=1_700_000_000)
|
|
assert signed["wts"] == 1_700_000_000
|
|
assert len(signed["w_rid"]) == 32
|
|
changed = sign_params({"keyword": "咻咻满", "page": 2}, key, now=1_700_000_000)
|
|
assert changed["w_rid"] != signed["w_rid"]
|
|
|
|
|
|
def test_parse_search_payload():
|
|
items = parse_search_payload(SEARCH_PAYLOAD)
|
|
assert len(items) == 1
|
|
item = items[0]
|
|
assert item.aid == 117530286889638
|
|
assert item.title == "咻咻满 reaction 初听"
|
|
assert item.duration_seconds == 105 * 60 + 5
|
|
assert item.cover.startswith("https://")
|
|
assert item.play == 12345 and item.like == 678
|
|
assert item.favorite == 90 and item.danmaku == 12 and item.review == 3
|
|
|
|
|
|
def test_wbi_search_and_fallback():
|
|
fake = FakeClient([NAV_PAYLOAD, SEARCH_PAYLOAD])
|
|
client = SearchClient(lambda: fake, clock=lambda: 1_700_000_000)
|
|
items = client.search("咻咻满 reaction", 1)
|
|
assert len(items) == 1
|
|
assert "/x/web-interface/nav" in fake.urls[0][0]
|
|
assert "/wbi/search/type" in fake.urls[1][0]
|
|
query = urllib.parse.parse_qs(urllib.parse.urlparse(fake.urls[1][0]).query)
|
|
assert query["w_rid"][0] and query["wts"][0] == "1700000000"
|
|
|
|
# nav 失败时降级旧接口
|
|
fake = FakeClient([BilibiliError("wbi unavailable"), SEARCH_PAYLOAD])
|
|
client = SearchClient(lambda: fake, clock=lambda: 1_700_000_000)
|
|
items = client.search("咻咻满 reaction", 2)
|
|
assert len(items) == 1
|
|
assert "/x/web-interface/search/type" in fake.urls[1][0]
|
|
|
|
|
|
def test_missing_result_is_protocol_error():
|
|
bad = {"code": 0, "message": "0", "data": {}}
|
|
fake = FakeClient([NAV_PAYLOAD, bad, bad])
|
|
client = SearchClient(lambda: fake, clock=lambda: 1_700_000_000)
|
|
try:
|
|
client.search("x", 1)
|
|
except BilibiliProtocolError as exc:
|
|
assert "result" in str(exc)
|
|
else:
|
|
raise AssertionError("缺少 result 应报协议错误")
|
|
|
|
|
|
def main():
|
|
test_wbi_helpers()
|
|
test_parse_search_payload()
|
|
test_wbi_search_and_fallback()
|
|
test_missing_result_is_protocol_error()
|
|
print("OK: WBI 签名、搜索解析与降级测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|