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.
212 lines
6.7 KiB
Python
212 lines
6.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""收藏/忽略/拉黑动作测试（假写客户端，不联网）: uv run python tests/test_favorites.py"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
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 BilibiliAPIError # noqa: E402
|
|
from reaction_board.bilibili.models import FavResource # noqa: E402
|
|
from reaction_board.config import ConfigStore # noqa: E402
|
|
from reaction_board.database import Database # noqa: E402
|
|
from reaction_board.repository import Repository # noqa: E402
|
|
from reaction_board.services.favorites import FavoriteActionService # noqa: E402
|
|
|
|
NOW = 1_793_444_393
|
|
FOLDER_ID = 3700023351
|
|
|
|
|
|
@dataclass
|
|
class FakeFavId:
|
|
aid: int
|
|
type: int = 2
|
|
|
|
|
|
class FakeWriteClient:
|
|
def __init__(self, order, *, fail_deal=False):
|
|
self.order = list(order)
|
|
self.fail_deal = fail_deal
|
|
self.dealt = []
|
|
self.sorts = []
|
|
|
|
def fav_ids(self, media_id):
|
|
return [FakeFavId(aid) for aid in self.order]
|
|
|
|
def deal(self, aid, folder_id):
|
|
if self.fail_deal:
|
|
raise BilibiliAPIError(-1, "收藏夹已满")
|
|
self.dealt.append((aid, folder_id))
|
|
self.order.append(aid)
|
|
|
|
def sort(self, media_id, spec):
|
|
predecessor, _, moved, _ = (int(part) for part in spec.split(":"))
|
|
self.sorts.append(spec)
|
|
self.order.remove(moved)
|
|
index = self.order.index(predecessor)
|
|
self.order.insert(index + 1, moved)
|
|
|
|
|
|
class FakeProvider:
|
|
def __init__(self, client):
|
|
self.client = client
|
|
|
|
def get(self):
|
|
return self.client
|
|
|
|
|
|
def write_config(tmp):
|
|
data = {
|
|
"bilibili": {
|
|
"read": {"mid": 1, "csrf": "r", "cookie": "read-cookie"},
|
|
"write": {"mid": 2, "csrf": "w", "cookie": "write-cookie"},
|
|
},
|
|
"storage": {
|
|
"database_path": str(Path(tmp) / "app.db"),
|
|
"backup_dir": str(Path(tmp) / "backup"),
|
|
},
|
|
"folders": [{"year": 2026, "media_id": FOLDER_ID, "title": "2026"}],
|
|
"discover": {"keywords": ["x"], "pages": 1, "window_days": 14},
|
|
}
|
|
path = Path(tmp) / "config.json"
|
|
path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def add_video(repo, aid, up_mid, pubdate):
|
|
repo.upsert_from_fav(
|
|
FavResource(
|
|
aid=aid, bvid=f"BV{aid}", title=f"视频{aid}", up_mid=up_mid,
|
|
up_name=f"UP{up_mid}", pubdate=pubdate, duration=600,
|
|
cover=None, attr=0, fav_time=None,
|
|
),
|
|
FOLDER_ID,
|
|
was_added=True,
|
|
)
|
|
|
|
|
|
def add_candidate(repo, aid, up_mid=100, pubdate=NOW - 3600):
|
|
return repo.upsert_candidate(
|
|
aid=aid,
|
|
bvid=f"BV{aid}",
|
|
title=f"咻咻满 reaction {aid}",
|
|
up_mid=up_mid,
|
|
up_name=f"UP{up_mid}",
|
|
pubdate=pubdate,
|
|
cover="https://example/cover.jpg",
|
|
score=3,
|
|
channel="search:test#p1",
|
|
now=NOW,
|
|
)
|
|
|
|
|
|
def setup(tmp, order, *, fail_deal=False):
|
|
path = write_config(tmp)
|
|
store = ConfigStore(path, env={})
|
|
db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup")
|
|
db.initialize()
|
|
repo = Repository(db, clock=lambda: NOW)
|
|
add_video(repo, 1, 100, 1_792_000_000)
|
|
add_video(repo, 2, 200, 1_791_000_000)
|
|
add_candidate(repo, 9, up_mid=100, pubdate=1_793_000_000)
|
|
client = FakeWriteClient(order, fail_deal=fail_deal)
|
|
service = FavoriteActionService(
|
|
store, repo, FakeProvider(client), clock=lambda: NOW
|
|
)
|
|
return db, repo, client, service
|
|
|
|
|
|
def test_favorite_inserts_and_positions():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db, repo, client, service = setup(tmp, [1, 2])
|
|
try:
|
|
result = service.favorite(9)
|
|
assert result.ok is True
|
|
assert result.position == 1
|
|
assert result.already_in_folder is False
|
|
assert client.dealt == [(9, FOLDER_ID)]
|
|
assert client.sorts == ["1:2:9:2", "9:2:1:2"]
|
|
assert client.order == [9, 1, 2]
|
|
assert repo.get_candidate(9) is None
|
|
video = repo.get_video(9)
|
|
assert video["status"] == "active"
|
|
assert video["folder_id"] == FOLDER_ID
|
|
action = repo.get_fav_action("fav:9")
|
|
assert action["state"] == "done"
|
|
assert action["position"] == 1
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_favorite_already_in_folder_skips_writes():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db, repo, client, service = setup(tmp, [1, 9, 2])
|
|
try:
|
|
result = service.favorite(9)
|
|
assert result.ok is True
|
|
assert result.already_in_folder is True
|
|
assert result.position == 2
|
|
assert client.dealt == []
|
|
assert client.sorts == []
|
|
assert repo.get_candidate(9) is None
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_favorite_failure_keeps_candidate_and_allows_retry():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db, repo, client, service = setup(tmp, [1, 2], fail_deal=True)
|
|
try:
|
|
result = service.favorite(9)
|
|
assert result.ok is False
|
|
assert "收藏夹已满" in result.message
|
|
assert repo.get_candidate(9) is not None
|
|
action = repo.get_fav_action("fav:9")
|
|
assert action["state"] == "failed"
|
|
assert "收藏夹已满" in action["err"]
|
|
|
|
client.fail_deal = False
|
|
retry = service.favorite(9)
|
|
assert retry.ok is True
|
|
assert retry.position == 1
|
|
assert repo.get_candidate(9) is None
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_ignore_and_blacklist():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db, repo, client, service = setup(tmp, [1, 2])
|
|
try:
|
|
add_candidate(repo, 11, up_mid=300)
|
|
add_candidate(repo, 12, up_mid=300)
|
|
ignored = service.ignore(11)
|
|
assert ignored.ok is True
|
|
assert repo.get_candidate(11)["decision"] == "rejected"
|
|
|
|
result = service.blacklist(12)
|
|
assert result.ok is True
|
|
assert repo.is_blacklisted(300)
|
|
assert repo.get_candidate(12)["decision"] == "rejected"
|
|
# 已有视频不受拉黑影响
|
|
assert repo.get_video(1)["status"] == "active"
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def main():
|
|
test_favorite_inserts_and_positions()
|
|
test_favorite_already_in_folder_skips_writes()
|
|
test_favorite_failure_keeps_candidate_and_allows_retry()
|
|
test_ignore_and_blacklist()
|
|
print("OK: 收藏、忽略、拉黑与失败重试测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|