diff --git a/reaction_board/bilibili/write_client.py b/reaction_board/bilibili/write_client.py
index 81a7e18..2dec0c6 100644
--- a/reaction_board/bilibili/write_client.py
+++ b/reaction_board/bilibili/write_client.py
@@ -1,5 +1,8 @@
# -*- coding: utf-8 -*-
-"""B 站收藏夹写客户端(``deal`` / ``sort``)。
+"""B 站收藏夹写客户端(``sort``)。
+
+收藏夹重排序是本项目**唯一**使用 ``bilibili.write`` 凭据的写操作,且频率极低。
+加入收藏夹(``deal``)等写操作已移除,收藏夹由人工维护。
只应使用 ``bilibili.write`` 凭据构造。
"""
@@ -39,22 +42,6 @@ class BilibiliWriteClient(BilibiliClient):
raise self._api_error(payload)
return payload
- def deal(self, aid: int, folder_id: int) -> dict[str, Any]:
- return self.post_form(
- f"{BASE}/x/v3/fav/resource/deal",
- {
- "rid": int(aid),
- "type": 2,
- "add_media_ids": int(folder_id),
- "del_media_ids": "",
- "csrf": self.csrf,
- },
- referer=(
- f"https://space.bilibili.com/{self.mid}/favlist"
- f"?fid={int(folder_id)}&ftype=create"
- ),
- )
-
def sort(self, media_id: int, sort_spec: str) -> dict[str, Any]:
return self.post_form(
f"{BASE}/x/v3/fav/resource/sort",
diff --git a/reaction_board/cli.py b/reaction_board/cli.py
index e87b882..ddd9269 100644
--- a/reaction_board/cli.py
+++ b/reaction_board/cli.py
@@ -2,7 +2,7 @@
"""看板命令行入口。
python app.py init # 建库 / 迁移 / 写入黑名单种子(离线)
- python app.py sync # 收藏夹只读同步(首次导入或增量)
+ python app.py ingest # 手动入库 B 站视频链接 / BV 号 / av 号(只写本地库)
python app.py snapshot # 采集当日七项快照与粉丝数
python app.py discover # 执行一次搜索发现并写入候选池
python app.py candidates # 查看候选池(离线)
@@ -10,6 +10,7 @@
python app.py run # 按配置周期常驻运行(--once 只跑一次)
python app.py status # 查看库内状态(离线)
+收藏夹重排序使用独立的 ``favsort.py``(唯一使用 ``bilibili.write`` 的功能)。
P2 起 ``python app.py`` 将在此基础上启动本地看板服务。
"""
@@ -27,7 +28,6 @@ from .bilibili.auth_flow import AuthService
from .bilibili.client import BilibiliClient, Pacer
from .bilibili.provider import ClientProvider
from .bilibili.search import SearchClient
-from .bilibili.write_client import BilibiliWriteClient
from .config import ConfigError, ConfigStore
from .database import Database
from .logging_json import JsonLogger
@@ -35,8 +35,7 @@ from .notify.telegram import TelegramClient
from .repository import Repository
from .scheduler import ScheduledTask, Scheduler
from .services.discovery import DiscoveryService
-from .services.favorites import FavoriteActionService
-from .services.folder_sync import FolderSyncService
+from .services.library import LibraryService
from .services.snapshot import SnapshotService
from .services.telegram_poller import TelegramPoller
from .services.up_profile import UpProfileService
@@ -66,20 +65,6 @@ def _build_parser() -> argparse.ArgumentParser:
sub = parser.add_subparsers(dest="command", required=False)
sub.add_parser("init", parents=[common], help="建库、迁移并写入黑名单种子")
- sync = sub.add_parser("sync", parents=[common], help="同步已配置的年度收藏夹")
- sync.add_argument(
- "--full", action="store_true", help="强制全量刷新元数据(忽略 6 小时周期)"
- )
- sync.add_argument(
- "--allow-empty-folder",
- action="store_true",
- help="允许把远端 0 条视为真实清空(默认按异常跳过差集)",
- )
- sync.add_argument(
- "--no-capture-followers",
- action="store_true",
- help="跳过同步末尾的 UP 粉丝采集(改用 backfill-followers 单独回填)",
- )
snapshot = sub.add_parser(
"snapshot", parents=[common], help="采集当日七项快照与粉丝数"
)
@@ -103,6 +88,14 @@ def _build_parser() -> argparse.ArgumentParser:
default=None,
help="只查看指定决策状态的候选",
)
+ ingest = sub.add_parser(
+ "ingest", parents=[common],
+ help="手动入库:B 站视频链接、BV 号或 av 号(只写本地库)",
+ )
+ ingest.add_argument(
+ "refs", nargs="+", metavar="REF",
+ help="视频链接、BV 号(BV…)、av 号(av12345)或含它们的字符串",
+ )
auth = sub.add_parser("auth", parents=[common], help="扫码授权并更新指定账号 Cookie")
auth.add_argument("--account", choices=("read", "write"), default="read",
help="要授权的账号角色(默认 read)")
@@ -112,15 +105,10 @@ def _build_parser() -> argparse.ArgumentParser:
auth_check = sub.add_parser("auth-check", parents=[common], help="检查账号登录态")
auth_check.add_argument("--account", choices=("read", "write", "all"), default="read",
help="要检查的账号角色(默认 read)")
- run = sub.add_parser("run", parents=[common], help="按配置周期常驻运行同步与快照")
+ run = sub.add_parser("run", parents=[common], help="按配置周期常驻运行发现与快照")
run.add_argument(
"--once", action="store_true", help="只执行一次到期任务后退出"
)
- run.add_argument(
- "--allow-empty-folder",
- action="store_true",
- help="允许把远端 0 条视为真实清空(默认按异常跳过差集)",
- )
serve = sub.add_parser("serve", parents=[common], help="启动只读本地看板(不运行调度)")
serve.add_argument(
"--preview", action="store_true",
@@ -209,32 +197,30 @@ def cmd_init(args, logger: JsonLogger) -> int:
return 0
-def cmd_sync(args, logger: JsonLogger) -> int:
- store = ConfigStore(args.config, require_auth=True)
+def cmd_ingest(args, logger: JsonLogger) -> int:
+ store = ConfigStore(args.config, require_auth=False)
cfg = store.get()
db = _open_db(cfg, logger)
try:
repo = Repository(db)
client = _client(cfg.bilibili.read, logger)
- service = FolderSyncService(
- store,
- repo,
- client,
- logger=logger,
- allow_empty_folders=args.allow_empty_folder,
- capture_followers=not getattr(args, "no_capture_followers", False),
+ service = LibraryService(
+ store, repo, lambda: client, logger=logger
)
- started = int(time.time())
- summary = service.run(full_refresh=args.full)
- payload = summary.as_dict()
+ text = " ".join(args.refs)
+ result = service.ingest_text(text)
+ payload = result.as_dict()
payload["database"] = str(cfg.storage.database_path)
- payload["counts"] = repo.counts()
_record_task(
- repo, "folder_sync", ok=summary.ok, started_at=started,
- finished_at=int(time.time()), payload=payload,
+ repo, "ingest", ok=bool(result.results) and not any(
+ item.state == "error" for item in result.results
+ ),
+ payload=payload,
)
_print(payload)
- return 0 if summary.ok else 1
+ return 0 if result.results and not any(
+ item.state == "error" for item in result.results
+ ) else 1
finally:
db.close()
@@ -403,7 +389,9 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
web_server = None
try:
repo = Repository(db)
- folder_provider = ClientProvider(
+ if cfg.up_blacklist_seed:
+ repo.seed_blacklist(cfg.up_blacklist_seed)
+ read_provider = ClientProvider(
store, "read", lambda credentials: _client(credentials, logger)
)
snapshot_provider = ClientProvider(
@@ -413,34 +401,22 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
credentials, logger, pacer=Pacer(1.0, 1.6), retries=1
),
)
- folder_service = FolderSyncService(
- store,
- repo,
- client_provider=folder_provider,
- logger=logger,
- allow_empty_folders=args.allow_empty_folder,
- )
snapshot_service = SnapshotService(
store, repo, client_provider=snapshot_provider, logger=logger
)
- write_provider = ClientProvider(
+ up_profile_service = UpProfileService(repo, read_provider.get, logger=logger)
+ library_service = LibraryService(
store,
- "write",
- lambda credentials: BilibiliWriteClient(
- cookie=credentials.cookie,
- csrf=credentials.csrf,
- mid=credentials.mid,
- logger=logger,
- ),
- )
- favorite_service = FavoriteActionService(
- store, repo, write_provider, logger=logger
+ repo,
+ read_provider,
+ up_profile_service=up_profile_service,
+ logger=logger,
)
notifier = _telegram(cfg, logger)
discovery_service = DiscoveryService(
store,
repo,
- SearchClient(folder_provider.get),
+ SearchClient(read_provider.get),
notifier=notifier,
logger=logger,
)
@@ -513,29 +489,6 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
logger.info("snapshot.summary", **payload)
return summary
- def run_sync():
- started = int(time.time())
- try:
- summary = folder_service.run()
- except Exception as exc: # noqa: BLE001
- _record_task(
- repo, "folder_sync", ok=False, started_at=started,
- finished_at=int(time.time()), error=str(exc),
- )
- raise
- payload = summary.as_dict()
- _record_task(
- repo, "folder_sync", ok=summary.ok, started_at=started,
- finished_at=int(time.time()), payload=payload,
- error=None if summary.ok else ";".join(summary.errors),
- )
- last["folder"] = summary
- logger.info("folder_sync.summary", **payload)
- # 新入库/恢复的视频立即补当天快照,而不是等到次日。
- if summary.added > 0:
- run_snapshot()
- return summary
-
def run_config_watch():
before = store.revision
store.get()
@@ -547,11 +500,6 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
snapshot_at = f"{hh:02d}:{mm:02d}"
scheduler = Scheduler(
[
- ScheduledTask(
- "folder_sync",
- float(cfg.schedule.folder_sync_minutes * 60),
- run_sync,
- ),
ScheduledTask(
"snapshot",
86400.0,
@@ -579,7 +527,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
daemon=True,
).start()
if notifier is not None and not args.once:
- poller = TelegramPoller(notifier, favorite_service, logger=logger)
+ poller = TelegramPoller(notifier, library_service, logger=logger)
threading.Thread(
target=poller.run_forever,
name="telegram-poller",
@@ -592,18 +540,15 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
else:
logger.info(
"scheduler.start",
- folder_sync_seconds=cfg.schedule.folder_sync_minutes * 60,
snapshot_at=snapshot_at,
)
scheduler.run_forever()
- folder_summary = last.get("folder")
snapshot_summary = last.get("snapshot")
discover_summary = last.get("discover")
- folder_ok = folder_summary is None or getattr(folder_summary, "ok", True)
snapshot_ok = snapshot_summary is None or getattr(snapshot_summary, "completed", True)
discover_ok = discover_summary is None or not getattr(discover_summary, "errors", [])
- return 0 if folder_ok and snapshot_ok and discover_ok else 1
+ return 0 if snapshot_ok and discover_ok else 1
finally:
if web_server is not None:
try:
@@ -703,16 +648,15 @@ def main(argv: Sequence[str] | None = None) -> int:
# 无子命令时启动完整服务:只读 Web + 调度 + Telegram。
args.command = "run"
args.once = False
- args.allow_empty_folder = False
return cmd_run(args, logger, serve_web=True)
if args.command == "serve":
return cmd_serve(args, logger)
if args.command == "init":
return cmd_init(args, logger)
- if args.command == "sync":
- return cmd_sync(args, logger)
if args.command == "snapshot":
return cmd_snapshot(args, logger)
+ if args.command == "ingest":
+ return cmd_ingest(args, logger)
if args.command == "backfill-followers":
return cmd_backfill_followers(args, logger)
if args.command == "discover":
diff --git a/reaction_board/config.py b/reaction_board/config.py
index 72362cc..eab2cf2 100644
--- a/reaction_board/config.py
+++ b/reaction_board/config.py
@@ -68,8 +68,6 @@ DEFAULT_CONFIG: dict[str, Any] = {
"up_blacklist_seed": [],
"schedule": {
"discover_minutes": 60,
- "folder_sync_minutes": 60,
- "full_folder_refresh_hours": 6,
"snapshot_at": "00:01",
"candidate_ttl_days": 7,
},
@@ -151,8 +149,6 @@ class FolderConfig:
@dataclass(frozen=True)
class ScheduleConfig:
discover_minutes: int
- folder_sync_minutes: int
- full_folder_refresh_hours: int
snapshot_at: str
candidate_ttl_days: int
@@ -370,16 +366,6 @@ def _build_config(data: Mapping[str, Any], path: Path) -> AppConfig:
discover_minutes=_as_int(
sched_raw.get("discover_minutes", 60), "schedule.discover_minutes", minimum=1
),
- folder_sync_minutes=_as_int(
- sched_raw.get("folder_sync_minutes", 60),
- "schedule.folder_sync_minutes",
- minimum=1,
- ),
- full_folder_refresh_hours=_as_int(
- sched_raw.get("full_folder_refresh_hours", 6),
- "schedule.full_folder_refresh_hours",
- minimum=1,
- ),
snapshot_at=snapshot_at,
candidate_ttl_days=_as_int(
sched_raw.get("candidate_ttl_days", 7),
diff --git a/reaction_board/fav_position.py b/reaction_board/fav_position.py
deleted file mode 100644
index 19d58f5..0000000
--- a/reaction_board/fav_position.py
+++ /dev/null
@@ -1,82 +0,0 @@
-# -*- coding: utf-8 -*-
-"""新视频在收藏夹中的插入位置计算(纯函数)。"""
-
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Mapping, Sequence
-
-
-@dataclass(frozen=True)
-class VideoMeta:
- up_mid: int | None
- pubdate: int
-
-
-@dataclass(frozen=True)
-class InsertionPlan:
- target_position: int
- moves: tuple[tuple[int, int], ...] # (pred_aid, moved_aid)
-
- @property
- def needs_head_move(self) -> bool:
- return len(self.moves) == 2
-
-
-def plan_insertion(
- current_ids: Sequence[int],
- new_aid: int,
- metadata: Mapping[int, VideoMeta],
- new_meta: VideoMeta,
-) -> InsertionPlan:
- """按“同 UP 连续、组间按最早投稿倒序”计算插入点。
-
- ``current_ids`` 可以包含 ``new_aid``(通常在末尾),函数会先排除它。
- """
- ordered = [int(aid) for aid in current_ids if int(aid) != int(new_aid)]
- if not ordered:
- return InsertionPlan(target_position=1, moves=())
-
- groups: list[dict] = []
- for index, aid in enumerate(ordered):
- meta = metadata.get(aid)
- up_mid = meta.up_mid if meta else None
- if up_mid is not None and groups and groups[-1]["up_mid"] == up_mid:
- groups[-1]["aids"].append(aid)
- else:
- groups.append({"up_mid": up_mid, "aids": [aid], "start": index})
-
- for group in groups:
- keys = [
- metadata[aid].pubdate
- for aid in group["aids"]
- if aid in metadata and metadata[aid].pubdate > 0
- ]
- group["key"] = min(keys) if keys else 0
-
- insert_index: int | None = None
- if new_meta.up_mid is not None:
- for group in groups:
- if group["up_mid"] == new_meta.up_mid:
- insert_index = group["start"]
- break
- if insert_index is None:
- for group in groups:
- if group["key"] < new_meta.pubdate:
- insert_index = group["start"]
- break
- if insert_index is None:
- insert_index = len(ordered)
-
- target_position = insert_index + 1
- if insert_index == 0:
- first = ordered[0]
- return InsertionPlan(
- target_position=1,
- moves=((first, int(new_aid)), (int(new_aid), first)),
- )
- predecessor = ordered[insert_index - 1]
- return InsertionPlan(
- target_position=target_position,
- moves=((predecessor, int(new_aid)),),
- )
diff --git a/reaction_board/migrations.py b/reaction_board/migrations.py
index a2c0a7d..4030adf 100644
--- a/reaction_board/migrations.py
+++ b/reaction_board/migrations.py
@@ -207,6 +207,15 @@ MIGRATIONS: tuple[Migration, ...] = (
""",
),
),
+ Migration(
+ version=5,
+ name="stop favorite-folder tracking",
+ statements=(
+ # 不再追踪收藏夹后,“取消收藏”状态失去语义;历史 unfaved 行归一到 active。
+ # 表结构与历史数据保留,仅停止写入。
+ "UPDATE video SET status = 'active' WHERE status = 'unfaved'",
+ ),
+ ),
)
diff --git a/reaction_board/services/favorites.py b/reaction_board/services/favorites.py
deleted file mode 100644
index a4a2d67..0000000
--- a/reaction_board/services/favorites.py
+++ /dev/null
@@ -1,237 +0,0 @@
-# -*- coding: utf-8 -*-
-"""候选收藏、忽略、拉黑动作。"""
-
-from __future__ import annotations
-
-import time
-from dataclasses import dataclass
-from typing import Any
-
-from ..bilibili.models import FavResource
-from ..config import ConfigStore
-from ..fav_position import VideoMeta, plan_insertion
-from ..repository import Repository
-from ..timeutil import shanghai_datetime
-
-
-@dataclass
-class ActionResult:
- ok: bool
- aid: int
- action: str
- state: str
- position: int | None = None
- already_in_folder: bool = False
- message: str = ""
-
- def as_dict(self) -> dict[str, Any]:
- return {
- "ok": self.ok,
- "aid": self.aid,
- "action": self.action,
- "state": self.state,
- "position": self.position,
- "already_in_folder": self.already_in_folder,
- "message": self.message,
- }
-
-
-class FavoriteActionService:
- def __init__(
- self,
- config_store: ConfigStore,
- repository: Repository,
- write_provider,
- *,
- clock=time.time,
- logger=None,
- ) -> None:
- self.config_store = config_store
- self.repo = repository
- self.write_provider = write_provider
- self._clock = clock
- self.logger = logger
-
- def _now(self) -> int:
- return int(self._clock())
-
- def _log(self, level: str, event: str, **fields) -> None:
- if self.logger is not None:
- getattr(self.logger, level)(event, **fields)
-
- def _target_folder(self, candidate: dict[str, Any]):
- cfg = self.config_store.get()
- up_mid = candidate.get("up_mid")
- if isinstance(up_mid, int) and not isinstance(up_mid, bool):
- folder_id = self.repo.folder_for_up(up_mid)
- if folder_id is not None:
- row = self.repo.get_folder(folder_id)
- if row is not None and row.enabled:
- return row
- year = shanghai_datetime(self._clock()).year
- folder = cfg.folder_for_year(year)
- if folder is not None:
- return folder
- latest = cfg.latest_folder()
- if latest is not None:
- return latest
- raise RuntimeError("没有可用的目标收藏夹,请检查 folders 配置")
-
- def favorite(self, aid: int) -> ActionResult:
- aid = int(aid)
- candidate = self.repo.get_candidate(aid)
- if candidate is None:
- return ActionResult(
- ok=False, aid=aid, action="favorite", state="missing",
- message="候选不存在或已处理",
- )
- try:
- folder = self._target_folder(candidate)
- except Exception as exc: # noqa: BLE001
- return ActionResult(
- ok=False, aid=aid, action="favorite", state="failed",
- message=str(exc),
- )
-
- token, state = self.repo.begin_fav_action(
- aid, folder.media_id, now=self._now()
- )
- if state == "done":
- action = self.repo.get_fav_action(token) or {}
- return ActionResult(
- ok=True, aid=aid, action="favorite", state="done",
- position=action.get("position"),
- message="已在收藏夹",
- )
- if state == "busy":
- return ActionResult(
- ok=False, aid=aid, action="favorite", state="busy",
- message="该候选正在处理中,请稍后重试",
- )
-
- already = False
- try:
- client = self.write_provider.get()
- ids = client.fav_ids(folder.media_id)
- if aid in {item.aid for item in ids}:
- already = True
- position = [item.aid for item in ids].index(aid) + 1
- else:
- client.deal(aid, folder.media_id)
- ids = client.fav_ids(folder.media_id)
- aid_order = [item.aid for item in ids]
- if aid not in aid_order:
- raise RuntimeError("deal 成功但回读收藏夹未找到该视频")
- metadata = {
- item_aid: VideoMeta(
- up_mid=(info.get("up_mid") if info else None),
- pubdate=int((info or {}).get("pubdate") or 0),
- )
- for item_aid, info in self.repo.video_metadata(aid_order).items()
- }
- plan = plan_insertion(
- aid_order,
- aid,
- metadata,
- VideoMeta(
- up_mid=candidate.get("up_mid"),
- pubdate=int(candidate.get("pubdate") or 0),
- ),
- )
- type_map = {item.aid: item.type for item in ids}
- for predecessor, moved in plan.moves:
- client.sort(
- folder.media_id,
- f"{predecessor}:{type_map.get(predecessor, 2)}:"
- f"{moved}:{type_map.get(moved, 2)}",
- )
- final_order = [item.aid for item in client.fav_ids(folder.media_id)]
- if aid not in final_order:
- raise RuntimeError("排序后回读收藏夹未找到该视频")
- position = final_order.index(aid) + 1
- if position != plan.target_position:
- raise RuntimeError(
- f"回读位置不符:期望 #{plan.target_position},实际 #{position}"
- )
-
- resource = FavResource(
- aid=aid,
- bvid=candidate.get("bvid"),
- title=str(candidate.get("title") or "(无标题)"),
- up_mid=candidate.get("up_mid"),
- up_name=candidate.get("up_name"),
- pubdate=int(candidate.get("pubdate") or 0),
- duration=None,
- cover=candidate.get("cover"),
- attr=0,
- fav_time=None,
- )
- self.repo.upsert_from_fav(
- resource, folder.media_id, was_added=True, now=self._now()
- )
- self.repo.delete_candidate(aid)
- self.repo.finish_fav_action(token, position=position, now=self._now())
- self._log(
- "info", "favorite.done",
- aid=aid, folder_id=folder.media_id, position=position,
- already=already,
- )
- self.repo.bump_cache_version("videos")
- return ActionResult(
- ok=True, aid=aid, action="favorite", state="done",
- position=position, already_in_folder=already,
- message="已在收藏夹" if already else "已加入收藏夹",
- )
- except Exception as exc: # noqa: BLE001 失败保留候选,允许重试
- self.repo.fail_fav_action(token, error=str(exc), now=self._now())
- self._log(
- "error", "favorite.failed",
- aid=aid, folder_id=folder.media_id, error=str(exc),
- )
- return ActionResult(
- ok=False, aid=aid, action="favorite", state="failed",
- message=str(exc),
- )
-
- def ignore(self, aid: int) -> ActionResult:
- aid = int(aid)
- candidate = self.repo.get_candidate(aid)
- if candidate is None:
- return ActionResult(
- ok=False, aid=aid, action="ignore", state="missing",
- message="候选不存在或已处理",
- )
- self.repo.mark_candidate_decision(aid, "rejected", now=self._now())
- self.repo.bump_cache_version("videos")
- return ActionResult(
- ok=True, aid=aid, action="ignore", state="rejected",
- message="已忽略",
- )
-
- def blacklist(self, aid: int) -> ActionResult:
- aid = int(aid)
- candidate = self.repo.get_candidate(aid)
- if candidate is None:
- return ActionResult(
- ok=False, aid=aid, action="blacklist", state="missing",
- message="候选不存在或已处理",
- )
- up_mid = candidate.get("up_mid")
- if not isinstance(up_mid, int) or isinstance(up_mid, bool):
- return ActionResult(
- ok=False, aid=aid, action="blacklist", state="failed",
- message="候选缺少 UP mid,无法拉黑",
- )
- self.repo.add_blacklist(
- up_mid,
- name=candidate.get("up_name"),
- reason="telegram",
- now=self._now(),
- )
- rejected = self.repo.reject_pending_for_up(up_mid, now=self._now())
- self.repo.mark_candidate_decision(aid, "rejected", now=self._now())
- self.repo.bump_cache_version("videos")
- return ActionResult(
- ok=True, aid=aid, action="blacklist", state="rejected",
- message=f"已拉黑 UP,并忽略该 UP 的 {rejected} 条待决策候选",
- )
diff --git a/reaction_board/services/folder_sync.py b/reaction_board/services/folder_sync.py
deleted file mode 100644
index f41c7db..0000000
--- a/reaction_board/services/folder_sync.py
+++ /dev/null
@@ -1,335 +0,0 @@
-# -*- coding: utf-8 -*-
-"""收藏夹同步(库的真相源)。
-
-每次运行:
-
-1. 读取每个配置年度夹的全部 aid;
-2. 与本地 ``folder_resource`` 做差集;
-3. 新增资源按 200 条分块拉取 ``infos`` 并入库;
-4. 从收藏夹移除的资源标记 ``unfaved``(不删除历史);
-5. 每 ``full_folder_refresh_hours`` 小时全量刷新一次元数据并识别失效资源;
-6. 容量接近上限时预警,但绝不自动换夹。
-
-黑名单只在发现流程生效,本服务不读取 ``up_blacklist``。
-"""
-
-from __future__ import annotations
-
-import time
-from dataclasses import dataclass, field
-from typing import Any
-
-from ..config import AppConfig, ConfigStore, FolderConfig
-from ..repository import Repository
-from .up_profile import UpProfileService
-
-CAPACITY_WARN = 950
-
-
-@dataclass
-class FolderSyncResult:
- media_id: int
- year: int
- title: str
- remote_count: int = 0
- added: int = 0
- removed: int = 0
- refreshed: int = 0
- missing: int = 0
- full_refresh: bool = False
- warning: str | None = None
- error: str | None = None
-
- def as_dict(self) -> dict[str, Any]:
- return {
- "media_id": self.media_id,
- "year": self.year,
- "title": self.title,
- "remote_count": self.remote_count,
- "added": self.added,
- "removed": self.removed,
- "refreshed": self.refreshed,
- "missing": self.missing,
- "full_refresh": self.full_refresh,
- "warning": self.warning,
- "error": self.error,
- }
-
-
-@dataclass
-class SyncSummary:
- started_at: int
- finished_at: int = 0
- folders: list[FolderSyncResult] = field(default_factory=list)
- blacklist_seeded: int = 0
- followers_updated: int = 0
- followers_failed: int = 0
-
- @property
- def added(self) -> int:
- return sum(f.added for f in self.folders)
-
- @property
- def removed(self) -> int:
- return sum(f.removed for f in self.folders)
-
- @property
- def refreshed(self) -> int:
- return sum(f.refreshed for f in self.folders)
-
- @property
- def missing(self) -> int:
- return sum(f.missing for f in self.folders)
-
- @property
- def errors(self) -> list[str]:
- return [f"folder {f.media_id}: {f.error}" for f in self.folders if f.error]
-
- @property
- def warnings(self) -> list[str]:
- return [f"folder {f.media_id}: {f.warning}" for f in self.folders if f.warning]
-
- @property
- def ok(self) -> bool:
- return not self.errors
-
- def as_dict(self) -> dict[str, Any]:
- return {
- "started_at": self.started_at,
- "finished_at": self.finished_at,
- "elapsed": max(0, self.finished_at - self.started_at),
- "added": self.added,
- "removed": self.removed,
- "refreshed": self.refreshed,
- "missing": self.missing,
- "blacklist_seeded": self.blacklist_seeded,
- "followers_updated": self.followers_updated,
- "followers_failed": self.followers_failed,
- "folders": [f.as_dict() for f in self.folders],
- "errors": self.errors,
- "warnings": self.warnings,
- "ok": self.ok,
- }
-
-
-class FolderSyncService:
- def __init__(
- self,
- config_store: ConfigStore,
- repository: Repository,
- client=None,
- *,
- client_provider=None,
- clock=time.time,
- logger=None,
- capacity_warn: int = CAPACITY_WARN,
- allow_empty_folders: bool = False,
- up_profile_service: UpProfileService | None = None,
- capture_followers: bool = True,
- ) -> None:
- if client is None and client_provider is None:
- raise ValueError("必须提供 client 或 client_provider")
- self.config_store = config_store
- self.repo = repository
- self.client = client
- self.client_provider = client_provider
- self._clock = clock
- self.logger = logger
- self.capacity_warn = capacity_warn
- self.allow_empty_folders = allow_empty_folders
- self.capture_followers = capture_followers
- self.up_profile = up_profile_service or UpProfileService(
- repository, self._client, clock=clock, logger=logger
- )
-
- def _client(self):
- if self.client_provider is not None:
- return self.client_provider.get()
- return self.client
-
- def _now(self) -> int:
- return int(self._clock())
-
- def _log(self, level: str, event: str, **fields) -> None:
- if self.logger is not None:
- getattr(self.logger, level)(event, **fields)
-
- def run(self, *, full_refresh: bool = False) -> SyncSummary:
- cfg = self.config_store.get()
- client = self._client()
- now = self._now()
- summary = SyncSummary(started_at=now)
- self._log(
- "info", "folder_sync.start", folders=len(cfg.folders), full=full_refresh
- )
-
- for folder in cfg.folders:
- self.repo.ensure_folder(folder, now=now)
- self.repo.disable_folders_except(f.media_id for f in cfg.folders)
- if cfg.up_blacklist_seed:
- summary.blacklist_seeded = self.repo.seed_blacklist(
- cfg.up_blacklist_seed, now=now
- )
-
- for folder in cfg.folders:
- try:
- result = self._sync_folder(
- client, cfg, folder, now=now, force_full=full_refresh
- )
- except Exception as exc: # noqa: BLE001 单个夹失败不阻塞其它夹
- result = FolderSyncResult(
- media_id=folder.media_id,
- year=folder.year,
- title=folder.title,
- error=str(exc),
- )
- self._log(
- "error",
- "folder_sync.folder_error",
- media_id=folder.media_id,
- error=str(exc),
- )
- else:
- self._log("info", "folder_sync.folder", **result.as_dict())
- if result.warning:
- self._log(
- "warning",
- "folder_sync.capacity",
- media_id=folder.media_id,
- warning=result.warning,
- )
- summary.folders.append(result)
-
- if self.capture_followers:
- try:
- missing = self.repo.up_mids_missing_follower()
- if missing:
- capture = self.up_profile.capture(
- [int(row["up_mid"]) for row in missing],
- names={
- int(row["up_mid"]): str(row.get("up_name") or "")
- for row in missing
- },
- live_since=summary.started_at - 3600,
- )
- summary.followers_updated = capture.updated_videos
- summary.followers_failed = capture.failed
- except Exception as exc: # noqa: BLE001 粉丝采集失败不影响收藏夹同步
- summary.followers_failed += 1
- self._log(
- "warning", "folder_sync.followers_failed", error=str(exc)
- )
-
- summary.finished_at = self._now()
- self._log(
- "info" if summary.ok else "error",
- "folder_sync.done",
- added=summary.added,
- removed=summary.removed,
- refreshed=summary.refreshed,
- missing=summary.missing,
- errors=len(summary.errors),
- elapsed=max(0, summary.finished_at - summary.started_at),
- )
- if summary.added or summary.removed or summary.refreshed or summary.followers_updated:
- self.repo.bump_cache_version("videos")
- return summary
-
- def _sync_folder(
- self,
- client,
- cfg: AppConfig,
- folder: FolderConfig,
- *,
- now: int,
- force_full: bool,
- ) -> FolderSyncResult:
- row = self.repo.get_folder(folder.media_id)
- full_due = (
- force_full
- or row is None
- or row.last_full_at is None
- or now - int(row.last_full_at)
- >= cfg.schedule.full_folder_refresh_hours * 3600
- )
-
- resources = client.fav_ids(folder.media_id)
- remote_map = {item.aid: item for item in resources if item.type == 2}
- remote = set(remote_map)
- local = self.repo.folder_active_aids(folder.media_id)
- added = remote - local
- removed = local - remote
-
- if not remote and local and not self.allow_empty_folders:
- raise RuntimeError(
- f"收藏夹 {folder.media_id} 远端返回 0 条,但本地仍有 {len(local)} 条;"
- f"疑似接口异常,已跳过差集。确认收藏夹确实被清空后,"
- f"可使用 --allow-empty-folder 重试"
- )
-
- result = FolderSyncResult(
- media_id=folder.media_id,
- year=folder.year,
- title=folder.title or (row.title if row else ""),
- remote_count=len(remote),
- added=len(added),
- removed=len(removed),
- full_refresh=full_due,
- )
-
- targets = set(added)
- if full_due:
- targets |= remote
-
- if targets:
- infos = client.fav_infos(
- folder.media_id,
- sorted(targets),
- folder_mid=cfg.bilibili.read.mid,
- )
- for aid, resource in infos.items.items():
- if aid not in remote:
- continue
- outcome = self.repo.upsert_from_fav(
- resource, folder.media_id, was_added=aid in added, now=now
- )
- if aid in added:
- # 人工已加入收藏夹:候选池不再保留同一条视频
- self.repo.delete_candidate(aid)
- if outcome.moved_folder:
- self._log(
- "warning",
- "folder_sync.multi_folder",
- aid=aid,
- previous_folder_id=outcome.previous_folder_id,
- folder_id=folder.media_id,
- )
- for aid in infos.missing:
- source = remote_map.get(aid)
- self.repo.mark_missing_resource(
- aid,
- folder_id=folder.media_id,
- bvid=source.bvid if source else None,
- now=now,
- )
- result.refreshed = len(infos.items)
- result.missing = len(infos.missing)
-
- for aid in removed:
- self.repo.mark_resource_removed(folder.media_id, aid, now=now)
-
- self.repo.touch_folder_resources(folder.media_id, now=now)
- self.repo.finish_folder_sync(
- folder,
- media_count=len(remote),
- full_refresh=full_due,
- now=now,
- )
-
- if len(remote) >= self.capacity_warn:
- result.warning = (
- f"收藏夹「{result.title or folder.media_id}」已有 {len(remote)} 条,"
- f"达到预警线 {self.capacity_warn};请人工准备下一年度收藏夹并更新配置,"
- f"系统不会自动换夹"
- )
- return result
diff --git a/reaction_board/web/queries.py b/reaction_board/web/queries.py
index 2141971..eb76f72 100644
--- a/reaction_board/web/queries.py
+++ b/reaction_board/web/queries.py
@@ -1611,12 +1611,10 @@ class DashboardQueries:
now = self._now()
next_runs = {
"discover": int(now + cfg.schedule.discover_minutes * 60),
- "folder_sync": int(now + cfg.schedule.folder_sync_minutes * 60),
"snapshot": int(next_daily_run(now, cfg.schedule.snapshot_at)),
}
for name, key, interval_minutes in (
("discover", "discovery", cfg.schedule.discover_minutes),
- ("folder_sync", "folder_sync", cfg.schedule.folder_sync_minutes),
):
record = tasks.get(key)
if isinstance(record, dict) and record.get("finished_at"):
@@ -1660,7 +1658,6 @@ class DashboardQueries:
"preview": self._kv(conn, "preview:seeded") == "1",
"schedule": {
"discover_minutes": cfg.schedule.discover_minutes,
- "folder_sync_minutes": cfg.schedule.folder_sync_minutes,
"snapshot_at": cfg.schedule.snapshot_at,
"candidate_ttl_days": cfg.schedule.candidate_ttl_days,
},
diff --git a/reaction_board/web/static/views/candidates.js b/reaction_board/web/static/views/candidates.js
index 20ed8ef..927de16 100644
--- a/reaction_board/web/static/views/candidates.js
+++ b/reaction_board/web/static/views/candidates.js
@@ -30,7 +30,7 @@ function candidateCard(item) {
${item.url ? `
打开 B 站` : ""}
-
请在 Telegram 中收藏 / 忽略 / 拉黑
+
请在 Telegram 中入库 / 忽略 / 拉黑
`;
@@ -51,7 +51,7 @@ export async function renderCandidates(root) {
${data.items.length ? `
候选处理
- ${data.telegram_enabled ? "请在 Telegram 消息中收藏、忽略或拉黑。" : "Telegram 未启用,请使用 CLI 或数据库处理候选。"}
+ ${data.telegram_enabled ? "请在 Telegram 消息中入库、忽略或拉黑。" : "Telegram 未启用,请使用 CLI 或数据库处理候选。"}
` : ""}
diff --git a/reaction_board/web/static/views/home.js b/reaction_board/web/static/views/home.js
index c0cf43b..e592b9b 100644
--- a/reaction_board/web/static/views/home.js
+++ b/reaction_board/web/static/views/home.js
@@ -30,7 +30,8 @@ function heatStrip(items) {
function capacity(folders) {
if (!folders.length) return "";
return `
- ${sectionHeader("年度收藏夹")}
+ ${sectionHeader("历史收藏夹")}
+ 系统已不再同步收藏夹;下列统计仅保留历史来源。
${folders.map((folder) => `
@@ -39,7 +40,6 @@ function capacity(folders) {
- ${folder.warning ? `
接近容量上限,请准备新建年度收藏夹。
` : ""}
`).join("")}
`;
}
@@ -95,7 +95,7 @@ export async function renderHome(root) {
${addedNote}
${added.items.length ? added.items.map((item) => videoCard(item)).join("")
- : emptyState("最近 7 天没有新入库视频", "收藏夹同步或搜索确认后会出现在这里。")}
+ : emptyState("最近 7 天没有新入库视频", "搜索发现确认或 Telegram 私聊手动入库后会出现在这里。")}
diff --git a/reaction_board/web/static/views/system.js b/reaction_board/web/static/views/system.js
index 481fcfd..858471d 100644
--- a/reaction_board/web/static/views/system.js
+++ b/reaction_board/web/static/views/system.js
@@ -47,7 +47,6 @@ export async function renderSystem(root) {
采集成功${formatCount(health.ok)}
采集失败${formatCount(health.failed)}
在库${formatCount(counts.active)}
- 已取消收藏${formatCount(counts.unfaved)}
已失效${formatCount(counts.removed)}
已移出库${formatCount(counts.dropped)}
@@ -60,16 +59,16 @@ export async function renderSystem(root) {
${sectionHeader("预计下次运行")}
-
收藏夹同步${escapeHtml(formatDateTime(next.folder_sync))}
搜索发现${escapeHtml(formatDateTime(next.discover))}
每日快照${escapeHtml(formatDateTime(next.snapshot))}
-
调度参数${system.schedule.discover_minutes} / ${system.schedule.folder_sync_minutes} 分钟
+
发现周期${system.schedule.discover_minutes} 分钟
预计时间按配置和最近任务完成时间推算;实际以常驻进程为准。
- ${sectionHeader("收藏夹容量")}
+ ${sectionHeader("历史收藏夹")}
+ 系统已不再同步收藏夹;下列统计仅保留历史来源。收藏夹由人工维护,重排序使用 favsort.py。
${(system.folders || []).map((folder) => `
@@ -78,8 +77,8 @@ export async function renderSystem(root) {
-
最近同步 ${escapeHtml(formatDateTime(folder.last_ids_at))}
-
`).join("") || emptyState("未配置收藏夹")}
+ 历史快照 ${escapeHtml(formatDateTime(folder.last_ids_at))}
+ `).join("") || emptyState("无历史收藏夹记录")}
diff --git a/tests/test_cli.py b/tests/test_cli.py
index a7e3c4b..372b796 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -1,352 +1,352 @@
-# -*- coding: utf-8 -*-
-"""CLI 入口测试(离线): uv run python tests/test_cli.py"""
-
-import contextlib
-import io
-import json
-import os
-import sys
-import tempfile
-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.models import LoginState # noqa: E402
-from reaction_board.cli import main # noqa: E402
-from reaction_board.database import Database # noqa: E402
-from reaction_board.repository import Repository # noqa: E402
-import reaction_board.cli as cli # noqa: E402
-
-
-def write_config(tmp, *, cookie="SESSDATA=x; buvid3=y", csrf="csrf"):
- path = Path(tmp) / "config.json"
- path.write_text(json.dumps({
- "bilibili": {
- "read": {"mid": 1, "csrf": csrf, "cookie": cookie},
- "write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"},
- },
- "storage": {
- "database_path": str(Path(tmp) / "app.db"),
- "backup_dir": str(Path(tmp) / "backup"),
- },
- "folders": [
- {"year": 2025, "media_id": 11, "title": "2025"},
- {"year": 2026, "media_id": 22, "title": "2026"},
- ],
- "up_blacklist_seed": [101, 202],
- }, ensure_ascii=False), encoding="utf-8")
- return path
-
-
-def run_cli(argv):
- out = io.StringIO()
- with contextlib.redirect_stdout(out):
- code = main(argv)
- return code, out.getvalue()
-
-
-def test_init_and_status():
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp)
- code, output = run_cli(["init", "--config", str(path)])
- assert code == 0
- assert '"schema_version": 4' in output
-
- db = Database(Path(tmp) / "app.db")
- db.initialize()
- try:
- repo = Repository(db)
- assert len(repo.list_folders()) == 2
- assert [r["mid"] for r in repo.blacklist()] == [101, 202]
- assert repo.counts()["video_total"] == 0
- finally:
- db.close()
-
- code, output = run_cli(["status", "--config", str(path)])
- assert code == 0
- assert '"configured_folders"' in output
- assert '"media_count": null' in output
- assert '"video_total": 0' in output
-
-
-def test_missing_config():
- with tempfile.TemporaryDirectory() as tmp:
- code, output = run_cli(["init", "--config", str(Path(tmp) / "nope.json")])
- assert code == 2
- payload = json.loads(output.strip().splitlines()[-1])
- assert payload["level"] == "error"
- assert "缺少配置文件" in payload["error"]
-
-
-def test_sync_requires_auth_without_network():
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp, cookie="", csrf="")
- code, output = run_cli(["sync", "--config", str(path)])
- assert code == 2
- payload = json.loads(output.strip().splitlines()[-1])
- assert "cookie" in payload["error"]
- assert not (Path(tmp) / "app.db").exists()
-
-
-def test_snapshot_offline_empty_db():
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp, cookie="", csrf="")
- data = json.loads(path.read_text(encoding="utf-8"))
- data["folders"] = []
- data["track_fans"] = []
- path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
- code, output = run_cli(["snapshot", "--config", str(path)])
- assert code == 0
- assert '"completed": true' in output
- assert '"total": 0' in output
- assert len(list((Path(tmp) / "backup").glob("app-*.db"))) == 1
-
-
-def test_auth_check_without_cookie():
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp, cookie="", csrf="")
- code, output = run_cli(["auth-check", "--config", str(path)])
- assert code == 1
- payload = json.loads(output)
- assert payload["is_login"] is False
- assert "未配置 cookie" in payload["message"]
-
-
-def test_auth_check_all_and_auth_dispatch():
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp, cookie="", csrf="")
- code, output = run_cli(["auth-check", "--account", "all", "--config", str(path)])
- assert code == 1
- payload = json.loads(output)
- assert payload["ok"] is False
- assert [item["account"] for item in payload["accounts"]] == ["read", "write"]
-
- original = cli.AuthService
- calls = []
-
- class FakeAuthService:
- def __init__(self, path, *, account, logger):
- calls.append((str(path), account))
-
- def run(self, *, timeout, no_browser):
- assert timeout == 9 and no_browser is True
- return 7
-
- cli.AuthService = FakeAuthService
- try:
- code, _ = run_cli([
- "auth", "--account", "write", "--timeout", "9", "--no-browser", "--config", str(path),
- ])
- finally:
- cli.AuthService = original
- assert code == 7
- assert calls == [(str(path), "write")]
-
-
-def test_run_once_offline():
- class FakeClient:
- def check_login(self):
- return LoginState(True, 1, "tester", 0, "OK")
-
- class FakeDiscovery:
- def __init__(self, *args, **kwargs):
- pass
-
- def run(self):
- from reaction_board.services.discovery import DiscoverySummary
- return DiscoverySummary(started_at=1, finished_at=1)
-
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp)
- data = json.loads(path.read_text(encoding="utf-8"))
- data["folders"] = []
- data["track_fans"] = []
- path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
-
- original = cli._client
- original_discovery = cli.DiscoveryService
- selected = []
-
- def fake_client(credentials, *args, **kwargs):
- selected.append(credentials)
- return FakeClient()
-
- cli._client = fake_client
- cli.DiscoveryService = FakeDiscovery
- try:
- code, output = run_cli(["run", "--once", "--config", str(path)])
- finally:
- cli._client = original
- cli.DiscoveryService = original_discovery
- assert code == 0
- assert '"name": "folder_sync"' in output
- assert '"name": "snapshot"' in output
- assert selected and all(credentials.mid == 1 for credentials in selected)
-
- db = Database(Path(tmp) / "app.db")
- db.initialize()
- try:
- repo = Repository(db)
- for name in ("folder_sync", "snapshot", "discovery", "candidate_cleanup"):
- assert repo.kv_get(f"task:last:{name}"), name
- finally:
- db.close()
-
-
-def test_serve_preview_guard():
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp)
- assert run_cli(["init", "--config", str(path)])[0] == 0
- db_path = Path(tmp) / "app.db"
- try:
- cli._assert_preview_database(db_path)
- except cli.ConfigError as exc:
- assert "preview:seeded" in str(exc)
- else:
- raise AssertionError("普通数据库不应通过 --preview 校验")
-
- db = Database(db_path)
- db.initialize()
- try:
- Repository(db).kv_set("preview:seeded", "1")
- finally:
- db.close()
- cli._assert_preview_database(db_path)
-
- try:
- cli._assert_preview_database(Path(tmp) / "missing.db")
- except cli.ConfigError as exc:
- assert "不存在" in str(exc)
- else:
- raise AssertionError("缺失数据库应报错")
-
-
-def test_default_command_starts_web_and_scheduler():
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp)
- calls = []
- original = cli.cmd_run
-
- def fake_run(args, logger, *, serve_web=False):
- calls.append((args.command, serve_web, args.once, args.allow_empty_folder))
- return 0
-
- cli.cmd_run = fake_run
- try:
- code = main(["--config", str(path)])
- finally:
- cli.cmd_run = original
- assert code == 0
- assert calls == [("run", True, False, False)]
-
-
-def test_snapshot_backup_failure_returns_error_and_retries():
- import reaction_board.services.snapshot as snapshot_module
-
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp, cookie="", csrf="")
- data = json.loads(path.read_text(encoding="utf-8"))
- data["folders"] = []
- data["track_fans"] = []
- path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
-
- original = snapshot_module.backup_daily
-
- def fail_backup(*args, **kwargs):
- raise OSError("disk full")
-
- snapshot_module.backup_daily = fail_backup
- try:
- code, output = run_cli(["snapshot", "--config", str(path)])
- finally:
- snapshot_module.backup_daily = original
- assert code == 1
- marker = output.find("{\n")
- assert marker >= 0, output
- payload = json.loads(output[marker:])
- assert payload["completed"] is True
- assert "disk full" in payload["backup_error"]
-
- code, output = run_cli(["snapshot", "--config", str(path)])
- assert code == 0
- marker = output.find("{\n")
- payload = json.loads(output[marker:])
- assert payload["backup_error"] is None
- assert Path(payload["backup_path"]).exists()
-
-
-def test_candidates_and_discover_commands():
- from reaction_board.services.discovery import DiscoverySummary
-
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp)
- assert run_cli(["init", "--config", str(path)])[0] == 0
- db = Database(Path(tmp) / "app.db")
- db.initialize()
- try:
- repo = Repository(db)
- repo.upsert_candidate(
- aid=123, bvid="BV123", title="咻咻满 reaction",
- up_mid=100, up_name="UP", pubdate=1,
- cover=None, score=3, channel="search:test#p1",
- )
- finally:
- db.close()
-
- code, output = run_cli(["candidates", "--config", str(path)])
- assert code == 0
- assert '"count": 1' in output
- assert '"aid": 123' in output
-
- class FakeDiscovery:
- def __init__(self, *args, **kwargs):
- pass
-
- def run(self):
- return DiscoverySummary(
- started_at=1, finished_at=2, candidates=3, notified=1
- )
-
- original = cli.DiscoveryService
- cli.DiscoveryService = FakeDiscovery
- try:
- code, output = run_cli(["discover", "--config", str(path)])
- finally:
- cli.DiscoveryService = original
- assert code == 0
- assert '"candidates": 3' in output
- assert '"notified": 1' in output
-
-
-def test_backfill_followers_dry_run():
- with tempfile.TemporaryDirectory() as tmp:
- path = write_config(tmp, cookie="", csrf="")
- code, output = run_cli([
- "backfill-followers", "--dry-run", "--config", str(path),
- ])
- assert code == 0
- assert '"requested": 0' in output
- assert '"failed": 0' in output
- assert '"dry_run": true' in output
- assert '"total": 0' in output
-
-
-def main_test():
- test_init_and_status()
- test_missing_config()
- test_sync_requires_auth_without_network()
- test_snapshot_offline_empty_db()
- test_auth_check_without_cookie()
- test_auth_check_all_and_auth_dispatch()
- test_run_once_offline()
- test_serve_preview_guard()
- test_default_command_starts_web_and_scheduler()
- test_snapshot_backup_failure_returns_error_and_retries()
- test_candidates_and_discover_commands()
- test_backfill_followers_dry_run()
- print("OK: CLI init/sync/snapshot/auth-check/run/status/backfill/错误路径测试全部通过")
-
-
-if __name__ == "__main__":
- main_test()
+# -*- coding: utf-8 -*-
+"""CLI 入口测试(离线): uv run python tests/test_cli.py"""
+
+import contextlib
+import io
+import json
+import os
+import sys
+import tempfile
+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.models import LoginState # noqa: E402
+from reaction_board.cli import main # noqa: E402
+from reaction_board.database import Database # noqa: E402
+from reaction_board.repository import Repository # noqa: E402
+import reaction_board.cli as cli # noqa: E402
+
+
+def write_config(tmp, *, cookie="SESSDATA=x; buvid3=y", csrf="csrf"):
+ path = Path(tmp) / "config.json"
+ path.write_text(json.dumps({
+ "bilibili": {
+ "read": {"mid": 1, "csrf": csrf, "cookie": cookie},
+ "write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"},
+ },
+ "storage": {
+ "database_path": str(Path(tmp) / "app.db"),
+ "backup_dir": str(Path(tmp) / "backup"),
+ },
+ "folders": [
+ {"year": 2025, "media_id": 11, "title": "2025"},
+ {"year": 2026, "media_id": 22, "title": "2026"},
+ ],
+ "up_blacklist_seed": [101, 202],
+ }, ensure_ascii=False), encoding="utf-8")
+ return path
+
+
+def run_cli(argv):
+ out = io.StringIO()
+ with contextlib.redirect_stdout(out):
+ code = main(argv)
+ return code, out.getvalue()
+
+
+def test_init_and_status():
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp)
+ code, output = run_cli(["init", "--config", str(path)])
+ assert code == 0
+ assert '"schema_version": 5' in output
+
+ db = Database(Path(tmp) / "app.db")
+ db.initialize()
+ try:
+ repo = Repository(db)
+ assert len(repo.list_folders()) == 2
+ assert [r["mid"] for r in repo.blacklist()] == [101, 202]
+ assert repo.counts()["video_total"] == 0
+ finally:
+ db.close()
+
+ code, output = run_cli(["status", "--config", str(path)])
+ assert code == 0
+ assert '"configured_folders"' in output
+ assert '"media_count": null' in output
+ assert '"video_total": 0' in output
+
+
+def test_missing_config():
+ with tempfile.TemporaryDirectory() as tmp:
+ code, output = run_cli(["init", "--config", str(Path(tmp) / "nope.json")])
+ assert code == 2
+ payload = json.loads(output.strip().splitlines()[-1])
+ assert payload["level"] == "error"
+ assert "缺少配置文件" in payload["error"]
+
+
+def test_ingest_without_refs_is_offline():
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp, cookie="", csrf="")
+ code, output = run_cli(["ingest", "--config", str(path), "没有视频"])
+ assert code == 1
+ data = json.loads(output[output.rindex("{\n"):])
+ assert data["refs"] == 0
+ assert data["results"] == []
+
+
+def test_snapshot_offline_empty_db():
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp, cookie="", csrf="")
+ data = json.loads(path.read_text(encoding="utf-8"))
+ data["folders"] = []
+ data["track_fans"] = []
+ path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+ code, output = run_cli(["snapshot", "--config", str(path)])
+ assert code == 0
+ assert '"completed": true' in output
+ assert '"total": 0' in output
+ assert len(list((Path(tmp) / "backup").glob("app-*.db"))) == 1
+
+
+def test_auth_check_without_cookie():
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp, cookie="", csrf="")
+ code, output = run_cli(["auth-check", "--config", str(path)])
+ assert code == 1
+ payload = json.loads(output)
+ assert payload["is_login"] is False
+ assert "未配置 cookie" in payload["message"]
+
+
+def test_auth_check_all_and_auth_dispatch():
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp, cookie="", csrf="")
+ code, output = run_cli(["auth-check", "--account", "all", "--config", str(path)])
+ assert code == 1
+ payload = json.loads(output)
+ assert payload["ok"] is False
+ assert [item["account"] for item in payload["accounts"]] == ["read", "write"]
+
+ original = cli.AuthService
+ calls = []
+
+ class FakeAuthService:
+ def __init__(self, path, *, account, logger):
+ calls.append((str(path), account))
+
+ def run(self, *, timeout, no_browser):
+ assert timeout == 9 and no_browser is True
+ return 7
+
+ cli.AuthService = FakeAuthService
+ try:
+ code, _ = run_cli([
+ "auth", "--account", "write", "--timeout", "9", "--no-browser", "--config", str(path),
+ ])
+ finally:
+ cli.AuthService = original
+ assert code == 7
+ assert calls == [(str(path), "write")]
+
+
+def test_run_once_offline():
+ class FakeClient:
+ def check_login(self):
+ return LoginState(True, 1, "tester", 0, "OK")
+
+ class FakeDiscovery:
+ def __init__(self, *args, **kwargs):
+ pass
+
+ def run(self):
+ from reaction_board.services.discovery import DiscoverySummary
+ return DiscoverySummary(started_at=1, finished_at=1)
+
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp)
+ data = json.loads(path.read_text(encoding="utf-8"))
+ data["folders"] = []
+ data["track_fans"] = []
+ path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+
+ original = cli._client
+ original_discovery = cli.DiscoveryService
+ selected = []
+
+ def fake_client(credentials, *args, **kwargs):
+ selected.append(credentials)
+ return FakeClient()
+
+ cli._client = fake_client
+ cli.DiscoveryService = FakeDiscovery
+ try:
+ code, output = run_cli(["run", "--once", "--config", str(path)])
+ finally:
+ cli._client = original
+ cli.DiscoveryService = original_discovery
+ assert code == 0
+ assert '"name": "snapshot"' in output
+ assert '"name": "folder_sync"' not in output
+ assert selected and all(credentials.mid == 1 for credentials in selected)
+
+ db = Database(Path(tmp) / "app.db")
+ db.initialize()
+ try:
+ repo = Repository(db)
+ for name in ("snapshot", "discovery", "candidate_cleanup"):
+ assert repo.kv_get(f"task:last:{name}"), name
+ finally:
+ db.close()
+
+
+def test_serve_preview_guard():
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp)
+ assert run_cli(["init", "--config", str(path)])[0] == 0
+ db_path = Path(tmp) / "app.db"
+ try:
+ cli._assert_preview_database(db_path)
+ except cli.ConfigError as exc:
+ assert "preview:seeded" in str(exc)
+ else:
+ raise AssertionError("普通数据库不应通过 --preview 校验")
+
+ db = Database(db_path)
+ db.initialize()
+ try:
+ Repository(db).kv_set("preview:seeded", "1")
+ finally:
+ db.close()
+ cli._assert_preview_database(db_path)
+
+ try:
+ cli._assert_preview_database(Path(tmp) / "missing.db")
+ except cli.ConfigError as exc:
+ assert "不存在" in str(exc)
+ else:
+ raise AssertionError("缺失数据库应报错")
+
+
+def test_default_command_starts_web_and_scheduler():
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp)
+ calls = []
+ original = cli.cmd_run
+
+ def fake_run(args, logger, *, serve_web=False):
+ calls.append((args.command, serve_web, args.once))
+ return 0
+
+ cli.cmd_run = fake_run
+ try:
+ code = main(["--config", str(path)])
+ finally:
+ cli.cmd_run = original
+ assert code == 0
+ assert calls == [("run", True, False)]
+
+
+def test_snapshot_backup_failure_returns_error_and_retries():
+ import reaction_board.services.snapshot as snapshot_module
+
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp, cookie="", csrf="")
+ data = json.loads(path.read_text(encoding="utf-8"))
+ data["folders"] = []
+ data["track_fans"] = []
+ path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
+
+ original = snapshot_module.backup_daily
+
+ def fail_backup(*args, **kwargs):
+ raise OSError("disk full")
+
+ snapshot_module.backup_daily = fail_backup
+ try:
+ code, output = run_cli(["snapshot", "--config", str(path)])
+ finally:
+ snapshot_module.backup_daily = original
+ assert code == 1
+ marker = output.find("{\n")
+ assert marker >= 0, output
+ payload = json.loads(output[marker:])
+ assert payload["completed"] is True
+ assert "disk full" in payload["backup_error"]
+
+ code, output = run_cli(["snapshot", "--config", str(path)])
+ assert code == 0
+ marker = output.find("{\n")
+ payload = json.loads(output[marker:])
+ assert payload["backup_error"] is None
+ assert Path(payload["backup_path"]).exists()
+
+
+def test_candidates_and_discover_commands():
+ from reaction_board.services.discovery import DiscoverySummary
+
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp)
+ assert run_cli(["init", "--config", str(path)])[0] == 0
+ db = Database(Path(tmp) / "app.db")
+ db.initialize()
+ try:
+ repo = Repository(db)
+ repo.upsert_candidate(
+ aid=123, bvid="BV123", title="咻咻满 reaction",
+ up_mid=100, up_name="UP", pubdate=1,
+ cover=None, score=3, channel="search:test#p1",
+ )
+ finally:
+ db.close()
+
+ code, output = run_cli(["candidates", "--config", str(path)])
+ assert code == 0
+ assert '"count": 1' in output
+ assert '"aid": 123' in output
+
+ class FakeDiscovery:
+ def __init__(self, *args, **kwargs):
+ pass
+
+ def run(self):
+ return DiscoverySummary(
+ started_at=1, finished_at=2, candidates=3, notified=1
+ )
+
+ original = cli.DiscoveryService
+ cli.DiscoveryService = FakeDiscovery
+ try:
+ code, output = run_cli(["discover", "--config", str(path)])
+ finally:
+ cli.DiscoveryService = original
+ assert code == 0
+ assert '"candidates": 3' in output
+ assert '"notified": 1' in output
+
+
+def test_backfill_followers_dry_run():
+ with tempfile.TemporaryDirectory() as tmp:
+ path = write_config(tmp, cookie="", csrf="")
+ code, output = run_cli([
+ "backfill-followers", "--dry-run", "--config", str(path),
+ ])
+ assert code == 0
+ assert '"requested": 0' in output
+ assert '"failed": 0' in output
+ assert '"dry_run": true' in output
+ assert '"total": 0' in output
+
+
+def main_test():
+ test_init_and_status()
+ test_missing_config()
+ test_ingest_without_refs_is_offline()
+ test_snapshot_offline_empty_db()
+ test_auth_check_without_cookie()
+ test_auth_check_all_and_auth_dispatch()
+ test_run_once_offline()
+ test_serve_preview_guard()
+ test_default_command_starts_web_and_scheduler()
+ test_snapshot_backup_failure_returns_error_and_retries()
+ test_candidates_and_discover_commands()
+ test_backfill_followers_dry_run()
+ print("OK: CLI init/sync/snapshot/auth-check/run/status/backfill/错误路径测试全部通过")
+
+
+if __name__ == "__main__":
+ main_test()
diff --git a/tests/test_database.py b/tests/test_database.py
index d9c949e..798ecf1 100644
--- a/tests/test_database.py
+++ b/tests/test_database.py
@@ -1,142 +1,142 @@
-# -*- coding: utf-8 -*-
-"""数据库单写队列测试(不联网): uv run python tests/test_database.py"""
-
-import os
-import sqlite3
-import sys
-import tempfile
-import threading
-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.database import Database, DatabaseClosed # noqa: E402
-
-
-def make_db(tmp):
- db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup")
- db.initialize()
- return db
-
-
-def test_write_read_transaction():
- with tempfile.TemporaryDirectory() as tmp:
- db = make_db(tmp)
- try:
- assert db.schema_version == 4
- db.write(lambda c: c.execute("INSERT INTO kv(k, v) VALUES ('a', '1')"))
- assert db.read(lambda c: c.execute("SELECT v FROM kv WHERE k='a'").fetchone()[0]) == "1"
-
- def boom(conn):
- conn.execute("UPDATE kv SET v='2' WHERE k='a'")
- raise RuntimeError("boom")
- try:
- db.transaction(boom)
- except RuntimeError as exc:
- assert str(exc) == "boom"
- else:
- raise AssertionError("事务应抛错")
- assert db.read(lambda c: c.execute("SELECT v FROM kv WHERE k='a'").fetchone()[0]) == "1"
- finally:
- db.close()
-
-
-def test_concurrent_writes_are_serialized():
- with tempfile.TemporaryDirectory() as tmp:
- db = make_db(tmp)
- try:
- db.write(lambda c: c.execute(
- "CREATE TABLE counter (id INTEGER PRIMARY KEY, n INTEGER NOT NULL)"
- ))
- db.write(lambda c: c.execute("INSERT INTO counter(id, n) VALUES (1, 0)"))
- errors = []
-
- def worker():
- try:
- for _ in range(10):
- db.transaction(
- lambda conn: conn.execute(
- "UPDATE counter SET n = n + 1 WHERE id = 1"
- )
- )
- except Exception as exc: # noqa: BLE001
- errors.append(exc)
-
- threads = [threading.Thread(target=worker) for _ in range(8)]
- for t in threads:
- t.start()
- for t in threads:
- t.join()
- assert not errors, errors
- assert db.read(lambda c: c.execute("SELECT n FROM counter").fetchone()[0]) == 80
- finally:
- db.close()
-
-
-def test_close_and_reinitialize():
- with tempfile.TemporaryDirectory() as tmp:
- db = make_db(tmp)
- db.close()
- db.close() # 幂等
- try:
- db.write(lambda c: None)
- except DatabaseClosed:
- pass
- else:
- raise AssertionError("关闭后写入应失败")
- try:
- db.read(lambda c: None)
- except DatabaseClosed:
- pass
- else:
- raise AssertionError("关闭后读取应失败")
-
- db2 = make_db(tmp)
- try:
- assert db2.schema_version == 4
- assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 4
- finally:
- db2.close()
-
-
-def test_wal_enabled():
- with tempfile.TemporaryDirectory() as tmp:
- db = make_db(tmp)
- try:
- mode = db.read(lambda c: c.execute("PRAGMA journal_mode").fetchone()[0])
- assert str(mode).lower() == "wal"
- assert db.read(lambda c: c.execute("PRAGMA foreign_keys").fetchone()[0]) == 1
- finally:
- db.close()
-
-
-def test_migration_backup_on_existing_db():
- """已有数据升级时由 Database 自动备份。"""
- with tempfile.TemporaryDirectory() as tmp:
- path = Path(tmp) / "app.db"
- # 先建一个旧库(只有 schema_version v0),Database.initialize 会升级并备份
- conn = sqlite3.connect(path)
- conn.execute("CREATE TABLE legacy (x INTEGER)")
- conn.commit()
- conn.close()
- db = Database(path, backup_dir=Path(tmp) / "backup")
- try:
- applied = db.initialize()
- assert applied == [1, 2, 3, 4]
- assert len(list((Path(tmp) / "backup").glob("app-pre-v0-*.db"))) == 1
- finally:
- db.close()
-
-
-def main():
- test_write_read_transaction()
- test_concurrent_writes_are_serialized()
- test_close_and_reinitialize()
- test_wal_enabled()
- test_migration_backup_on_existing_db()
- print("OK: 数据库单写队列与迁移入口测试全部通过")
-
-
-if __name__ == "__main__":
- main()
+# -*- coding: utf-8 -*-
+"""数据库单写队列测试(不联网): uv run python tests/test_database.py"""
+
+import os
+import sqlite3
+import sys
+import tempfile
+import threading
+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.database import Database, DatabaseClosed # noqa: E402
+
+
+def make_db(tmp):
+ db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup")
+ db.initialize()
+ return db
+
+
+def test_write_read_transaction():
+ with tempfile.TemporaryDirectory() as tmp:
+ db = make_db(tmp)
+ try:
+ assert db.schema_version == 5
+ db.write(lambda c: c.execute("INSERT INTO kv(k, v) VALUES ('a', '1')"))
+ assert db.read(lambda c: c.execute("SELECT v FROM kv WHERE k='a'").fetchone()[0]) == "1"
+
+ def boom(conn):
+ conn.execute("UPDATE kv SET v='2' WHERE k='a'")
+ raise RuntimeError("boom")
+ try:
+ db.transaction(boom)
+ except RuntimeError as exc:
+ assert str(exc) == "boom"
+ else:
+ raise AssertionError("事务应抛错")
+ assert db.read(lambda c: c.execute("SELECT v FROM kv WHERE k='a'").fetchone()[0]) == "1"
+ finally:
+ db.close()
+
+
+def test_concurrent_writes_are_serialized():
+ with tempfile.TemporaryDirectory() as tmp:
+ db = make_db(tmp)
+ try:
+ db.write(lambda c: c.execute(
+ "CREATE TABLE counter (id INTEGER PRIMARY KEY, n INTEGER NOT NULL)"
+ ))
+ db.write(lambda c: c.execute("INSERT INTO counter(id, n) VALUES (1, 0)"))
+ errors = []
+
+ def worker():
+ try:
+ for _ in range(10):
+ db.transaction(
+ lambda conn: conn.execute(
+ "UPDATE counter SET n = n + 1 WHERE id = 1"
+ )
+ )
+ except Exception as exc: # noqa: BLE001
+ errors.append(exc)
+
+ threads = [threading.Thread(target=worker) for _ in range(8)]
+ for t in threads:
+ t.start()
+ for t in threads:
+ t.join()
+ assert not errors, errors
+ assert db.read(lambda c: c.execute("SELECT n FROM counter").fetchone()[0]) == 80
+ finally:
+ db.close()
+
+
+def test_close_and_reinitialize():
+ with tempfile.TemporaryDirectory() as tmp:
+ db = make_db(tmp)
+ db.close()
+ db.close() # 幂等
+ try:
+ db.write(lambda c: None)
+ except DatabaseClosed:
+ pass
+ else:
+ raise AssertionError("关闭后写入应失败")
+ try:
+ db.read(lambda c: None)
+ except DatabaseClosed:
+ pass
+ else:
+ raise AssertionError("关闭后读取应失败")
+
+ db2 = make_db(tmp)
+ try:
+ assert db2.schema_version == 5
+ assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 5
+ finally:
+ db2.close()
+
+
+def test_wal_enabled():
+ with tempfile.TemporaryDirectory() as tmp:
+ db = make_db(tmp)
+ try:
+ mode = db.read(lambda c: c.execute("PRAGMA journal_mode").fetchone()[0])
+ assert str(mode).lower() == "wal"
+ assert db.read(lambda c: c.execute("PRAGMA foreign_keys").fetchone()[0]) == 1
+ finally:
+ db.close()
+
+
+def test_migration_backup_on_existing_db():
+ """已有数据升级时由 Database 自动备份。"""
+ with tempfile.TemporaryDirectory() as tmp:
+ path = Path(tmp) / "app.db"
+ # 先建一个旧库(只有 schema_version v0),Database.initialize 会升级并备份
+ conn = sqlite3.connect(path)
+ conn.execute("CREATE TABLE legacy (x INTEGER)")
+ conn.commit()
+ conn.close()
+ db = Database(path, backup_dir=Path(tmp) / "backup")
+ try:
+ applied = db.initialize()
+ assert applied == [1, 2, 3, 4, 5]
+ assert len(list((Path(tmp) / "backup").glob("app-pre-v0-*.db"))) == 1
+ finally:
+ db.close()
+
+
+def main():
+ test_write_read_transaction()
+ test_concurrent_writes_are_serialized()
+ test_close_and_reinitialize()
+ test_wal_enabled()
+ test_migration_backup_on_existing_db()
+ print("OK: 数据库单写队列与迁移入口测试全部通过")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/test_fav_position.py b/tests/test_fav_position.py
deleted file mode 100644
index a9405f8..0000000
--- a/tests/test_fav_position.py
+++ /dev/null
@@ -1,80 +0,0 @@
-# -*- coding: utf-8 -*-
-"""新视频插入位置纯函数测试: uv run python tests/test_fav_position.py"""
-
-import os
-import sys
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-sys.stdout.reconfigure(encoding="utf-8")
-
-from reaction_board.fav_position import VideoMeta, plan_insertion # noqa: E402
-
-
-def meta(up_mid, pubdate):
- return VideoMeta(up_mid=up_mid, pubdate=pubdate)
-
-
-def test_empty_and_single():
- assert plan_insertion([], 9, {}, meta(100, 200)).moves == ()
- assert plan_insertion([], 9, {}, meta(100, 200)).target_position == 1
- plan = plan_insertion([1], 9, {1: meta(200, 50)}, meta(100, 200))
- assert plan.target_position == 1
- assert plan.moves == ((1, 9), (9, 1))
-
-
-def test_same_up_group():
- current = [1, 2, 3, 4]
- metadata = {
- 1: meta(100, 100),
- 2: meta(100, 90),
- 3: meta(200, 80),
- 4: meta(200, 70),
- }
- # 同 UP 组在中间:插到组头
- plan = plan_insertion(current, 9, metadata, meta(200, 95))
- assert plan.target_position == 3
- assert plan.moves == ((2, 9),)
-
- # 同 UP 组在首位:两次移动
- plan = plan_insertion(current, 9, metadata, meta(100, 200))
- assert plan.target_position == 1
- assert plan.moves == ((1, 9), (9, 1))
-
-
-def test_new_group_by_earliest_pubdate():
- current = [1, 2, 3, 4]
- metadata = {
- 1: meta(100, 100),
- 2: meta(100, 90),
- 3: meta(200, 80),
- 4: meta(200, 70),
- }
- # 新 UP 的组键 85,应插到组键 70 的组之前
- plan = plan_insertion(current, 9, metadata, meta(300, 85))
- assert plan.target_position == 3
- assert plan.moves == ((2, 9),)
-
- # 比所有组都旧:追加到末尾
- plan = plan_insertion(current, 9, metadata, meta(300, 60))
- assert plan.target_position == 5
- assert plan.moves == ((4, 9),)
-
-
-def test_excludes_new_aid_if_present():
- current = [1, 9, 2]
- metadata = {1: meta(100, 100), 2: meta(100, 90)}
- plan = plan_insertion(current, 9, metadata, meta(100, 95))
- assert plan.target_position == 1
- assert plan.moves == ((1, 9), (9, 1))
-
-
-def main():
- test_empty_and_single()
- test_same_up_group()
- test_new_group_by_earliest_pubdate()
- test_excludes_new_aid_if_present()
- print("OK: 新视频插入位置计算测试全部通过")
-
-
-if __name__ == "__main__":
- main()
diff --git a/tests/test_favorites.py b/tests/test_favorites.py
deleted file mode 100644
index e08f5e5..0000000
--- a/tests/test_favorites.py
+++ /dev/null
@@ -1,211 +0,0 @@
-# -*- 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()
diff --git a/tests/test_folder_sync.py b/tests/test_folder_sync.py
deleted file mode 100644
index 17f3cc0..0000000
--- a/tests/test_folder_sync.py
+++ /dev/null
@@ -1,373 +0,0 @@
-# -*- coding: utf-8 -*-
-"""收藏夹同步集成测试(用真实接口形状的 transport Mock,不联网):
-uv run python tests/test_folder_sync.py
-"""
-
-import json
-import os
-import sys
-import tempfile
-import urllib.parse
-from pathlib import Path
-
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
-sys.stdout.reconfigure(encoding="utf-8")
-sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
-
-from fixtures import info_for # noqa: E402
-from reaction_board.bilibili.client import ( # noqa: E402
- BilibiliClient,
- HTTPResponse,
- Pacer,
- TransportError,
-)
-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.folder_sync import FolderSyncService # noqa: E402
-
-F1 = 4451053851
-F2 = 3700023351
-
-
-def http_json(payload):
- return HTTPResponse(
- 200, {"content-type": "application/json"},
- json.dumps(payload, ensure_ascii=False).encode("utf-8"),
- )
-
-
-class FakeFavTransport:
- def __init__(self, folders, infos, *, fail_media=None, omit_ids_data=None):
- self.folders = {k: list(v) for k, v in folders.items()}
- self.infos = dict(infos)
- self.fail_media = fail_media
- self.omit_ids_data = omit_ids_data
- self.ids_calls = []
- self.info_calls = []
- self.relation_calls = []
-
- def request(self, method, url, headers, data=None, timeout=30):
- query = urllib.parse.parse_qs(urllib.parse.urlparse(url).query)
- if "/x/relation/stat" in url:
- mid = int(query["vmid"][0])
- self.relation_calls.append(mid)
- return http_json({
- "code": 0, "message": "0", "ttl": 1,
- "data": {"mid": mid, "following": 5, "follower": 1000 + mid},
- })
- if "/x/v3/fav/resource/ids" in url:
- media_id = int(query["media_id"][0])
- self.ids_calls.append(media_id)
- if self.fail_media == media_id:
- raise TransportError("ids 网络错误")
- if self.omit_ids_data == media_id:
- return http_json({"code": 0, "message": "OK", "ttl": 1})
- data_rows = [
- {"id": aid, "type": 2, "bv_id": f"BV{aid}", "bvid": f"BV{aid}"}
- for aid in self.folders.get(media_id, [])
- ]
- return http_json({"code": 0, "message": "0", "ttl": 1, "data": data_rows})
- if "/x/v3/fav/resource/infos" in url:
- media_id = int(query["folder_id"][0])
- resources = query["resources"][0]
- aids = [int(part.split(":")[0]) for part in resources.split(",")]
- self.info_calls.append((media_id, aids))
- rows = [self.infos[aid] for aid in aids if aid in self.infos]
- return http_json({"code": 0, "message": "0", "ttl": 1, "data": rows})
- if "/x/v3/fav/folder/created/list-all" in url:
- return http_json({
- "code": 0, "message": "0", "ttl": 1,
- "data": {"count": 0, "list": []},
- })
- raise AssertionError(f"未预期的 URL: {url}")
-
-
-def write_config(tmp, folders):
- data = {
- "bilibili": {
- "read": {"mid": 28802351, "csrf": "csrf-secret", "cookie": "SESSDATA=secret; bili_jct=csrf-secret; buvid3=device"},
- "write": {"mid": 2, "csrf": "write-csrf", "cookie": "write-cookie"},
- },
- "storage": {
- "database_path": str(Path(tmp) / "app.db"),
- "backup_dir": str(Path(tmp) / "backup"),
- },
- "folders": folders,
- "up_blacklist_seed": [999],
- "schedule": {
- "discover_minutes": 60,
- "folder_sync_minutes": 60,
- "full_folder_refresh_hours": 6,
- "snapshot_at": "00:01",
- "candidate_ttl_days": 7,
- },
- }
- path = Path(tmp) / "config.json"
- path.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
- return path
-
-
-def make_service(tmp, transport, clock, *, capacity_warn=950):
- folders = [
- {"year": 2025, "media_id": F1, "title": "2025"},
- {"year": 2026, "media_id": F2, "title": "2026"},
- ]
- path = write_config(tmp, folders)
- store = ConfigStore(path, env={}, require_auth=True)
- db = Database(Path(tmp) / "app.db", backup_dir=Path(tmp) / "backup")
- db.initialize()
- repo = Repository(db, clock=lambda: clock[0])
- client = BilibiliClient(
- cookie="SESSDATA=secret; bvid3=device",
- csrf="csrf",
- mid=28802351,
- transport=transport,
- pacer=Pacer(0, 0),
- retries=1,
- )
- service = FolderSyncService(
- store, repo, client, clock=lambda: clock[0], capacity_warn=capacity_warn
- )
- return db, repo, service
-
-
-def valid(aid, *, up_mid=100, title=None):
- return info_for(aid, title=title or f"视频{aid}", up_mid=up_mid, up_name="UP")
-
-
-def test_initial_import_and_incremental_skip():
- with tempfile.TemporaryDirectory() as tmp:
- transport = FakeFavTransport(
- {F1: [1101, 1102, 1103, 1104], F2: []},
- {
- 1101: valid(1101),
- 1102: valid(1102, up_mid=999), # 黑名单 UP 也必须正常入库
- 1103: valid(1103, title="已失效", up_mid=100),
- },
- )
- # 1103 标记为 attr=9;1104 静默缺失
- transport.infos[1103]["attr"] = 9
- clock = [1_800_000_000]
- db, repo, service = make_service(tmp, transport, clock)
- try:
- repo.upsert_candidate(
- aid=1101, bvid="BV1101", title="咻咻满 reaction",
- up_mid=100, up_name="UP", pubdate=1_799_000_000,
- cover=None, score=3, channel="search:test#p1",
- now=1_799_999_000,
- )
- summary = service.run()
- assert summary.ok
- assert summary.added == 4
- assert summary.missing == 1
- assert summary.refreshed == 3
- # 人工加入收藏夹后,同 aid 候选自动移除
- assert repo.get_candidate(1101) is None
- assert repo.seed_blacklist([999], now=0) == 0 # 已由 service 写入
-
- assert repo.get_video(1101)["status"] == "active"
- assert repo.get_video(1102)["status"] == "active" # 黑名单不影响收藏夹导入
- assert repo.get_video(1103)["status"] == "removed"
- assert repo.get_video(1104)["status"] == "removed"
- assert repo.get_video(1104)["bvid"] == "BV1104"
- assert repo.folder_active_aids(F1) == {1101, 1102, 1103, 1104}
- assert repo.is_blacklisted(999)
- # 新入库视频记录入库时粉丝数;失效视频不再采集
- assert summary.followers_updated == 2
- assert summary.followers_failed == 0
- assert repo.get_video(1101)["up_follower"] == 1100
- assert repo.get_video(1101)["up_follower_backfilled"] == 0
- assert repo.get_video(1102)["up_follower"] == 1999
- assert sorted(transport.relation_calls) == [100, 999]
-
- folder = repo.get_folder(F1)
- assert folder.media_count == 4
- assert folder.last_full_at == 1_800_000_000
-
- # 6 小时内再次运行:没有新增,不请求 infos
- clock[0] += 60
- info_before = len(transport.info_calls)
- relation_before = len(transport.relation_calls)
- summary2 = service.run()
- assert summary2.added == 0
- assert summary2.refreshed == 0
- assert len(transport.info_calls) == info_before
- assert len(transport.relation_calls) == relation_before
- assert summary2.folders[0].full_refresh is False
- finally:
- db.close()
-
-
-def test_diff_add_and_remove():
- with tempfile.TemporaryDirectory() as tmp:
- transport = FakeFavTransport(
- {F1: [1101, 1102], F2: []},
- {1101: valid(1101), 1102: valid(1102)},
- )
- clock = [1_800_000_000]
- db, repo, service = make_service(tmp, transport, clock)
- try:
- service.run()
- assert repo.get_video(1101)["status"] == "active"
- assert repo.get_video(1102)["status"] == "active"
-
- transport.folders[F1] = [1102, 1103]
- transport.infos[1103] = valid(1103)
- clock[0] += 3600
- summary = service.run()
- assert summary.added == 1
- assert summary.removed == 1
- assert summary.refreshed == 1
- assert repo.get_video(1101)["status"] == "unfaved"
- assert repo.get_video(1102)["status"] == "active"
- assert repo.get_video(1103)["status"] == "active"
- assert repo.folder_active_aids(F1) == {1102, 1103}
- assert repo.get_folder(F1).media_count == 2
- finally:
- db.close()
-
-
-def test_full_refresh_marks_invalid_and_missing():
- with tempfile.TemporaryDirectory() as tmp:
- transport = FakeFavTransport(
- {F1: [1101, 1102], F2: []},
- {1101: valid(1101), 1102: valid(1102)},
- )
- clock = [1_800_000_000]
- db, repo, service = make_service(tmp, transport, clock)
- try:
- service.run()
- assert repo.get_video(1101)["status"] == "active"
- assert repo.get_video(1102)["status"] == "active"
-
- # 6 小时后:1101 失效,1102 静默缺失
- clock[0] += 6 * 3600 + 1
- transport.infos[1101]["attr"] = 9
- del transport.infos[1102]
- summary = service.run()
- assert summary.folders[0].full_refresh is True
- assert summary.missing == 1
- assert repo.get_video(1101)["status"] == "removed"
- assert repo.get_video(1102)["status"] == "removed"
- assert repo.get_folder(F1).last_full_at == clock[0]
- # 全量刷新请求了夹内全部 aid
- assert set(transport.info_calls[-1][1]) == {1101, 1102}
- finally:
- db.close()
-
-
-def test_capacity_warning_does_not_move_folder():
- with tempfile.TemporaryDirectory() as tmp:
- transport = FakeFavTransport(
- {F1: [1, 2, 3], F2: []},
- {i: valid(i) for i in (1, 2, 3)},
- )
- clock = [1_800_000_000]
- db, repo, service = make_service(tmp, transport, clock, capacity_warn=3)
- try:
- summary = service.run()
- assert summary.folders[0].warning
- assert "预警线 3" in summary.folders[0].warning
- assert "不会自动换夹" in summary.folders[0].warning
- # 视频仍在原夹,不被移动
- assert all(repo.get_video(i)["folder_id"] == F1 for i in (1, 2, 3))
- finally:
- db.close()
-
-
-def test_folder_error_does_not_block_other_folders():
- with tempfile.TemporaryDirectory() as tmp:
- transport = FakeFavTransport(
- {F1: [1101], F2: [2201]},
- {1101: valid(1101), 2201: valid(2201)},
- fail_media=F1,
- )
- clock = [1_800_000_000]
- db, repo, service = make_service(tmp, transport, clock)
- try:
- summary = service.run()
- assert summary.ok is False
- assert len(summary.errors) == 1 and str(F1) in summary.errors[0]
- assert summary.folders[0].error
- assert summary.folders[1].error is None
- assert repo.get_video(2201)["status"] == "active"
- finally:
- db.close()
-
-
-def test_blacklist_seed_only_once():
- with tempfile.TemporaryDirectory() as tmp:
- transport = FakeFavTransport({F1: [1101], F2: []}, {1101: valid(1101)})
- clock = [1_800_000_000]
- db, repo, service = make_service(tmp, transport, clock)
- try:
- first = service.run()
- assert first.blacklist_seeded == 1
- clock[0] += 60
- second = service.run()
- assert second.blacklist_seeded == 0
- assert len(repo.blacklist()) == 1
- finally:
- db.close()
-
-
-def test_empty_remote_requires_explicit_opt_in():
- with tempfile.TemporaryDirectory() as tmp:
- transport = FakeFavTransport(
- {F1: [1101, 1102], F2: []},
- {1101: valid(1101), 1102: valid(1102)},
- )
- clock = [1_800_000_000]
- db, repo, service = make_service(tmp, transport, clock)
- try:
- service.run()
- transport.folders[F1] = []
- clock[0] += 3600
- summary = service.run()
- assert summary.ok is False
- assert "疑似接口异常" in summary.folders[0].error
- assert repo.get_video(1101)["status"] == "active"
- assert repo.folder_active_aids(F1) == {1101, 1102}
-
- service.allow_empty_folders = True
- summary = service.run()
- assert summary.ok is True
- assert repo.get_video(1101)["status"] == "unfaved"
- assert repo.folder_active_aids(F1) == set()
- finally:
- db.close()
-
-
-def test_missing_ids_data_aborts_folder():
- with tempfile.TemporaryDirectory() as tmp:
- transport = FakeFavTransport(
- {F1: [1101], F2: []},
- {1101: valid(1101)},
- omit_ids_data=F1,
- )
- clock = [1_800_000_000]
- db, repo, service = make_service(tmp, transport, clock)
- try:
- summary = service.run()
- assert summary.ok is False
- assert "data" in summary.folders[0].error
- assert repo.get_video(1101) is None
- finally:
- db.close()
-
-
-def main():
- test_initial_import_and_incremental_skip()
- test_diff_add_and_remove()
- test_full_refresh_marks_invalid_and_missing()
- test_capacity_warning_does_not_move_folder()
- test_folder_error_does_not_block_other_folders()
- test_blacklist_seed_only_once()
- test_empty_remote_requires_explicit_opt_in()
- test_missing_ids_data_aborts_folder()
- print("OK: 收藏夹初始导入、差集同步、失效识别与容量预警测试全部通过")
-
-
-if __name__ == "__main__":
- main()
diff --git a/tests/test_migrations.py b/tests/test_migrations.py
index 311a93b..2d8186c 100644
--- a/tests/test_migrations.py
+++ b/tests/test_migrations.py
@@ -37,8 +37,8 @@ def test_fresh_and_idempotent():
conn = sqlite3.connect(Path(tmp) / "app.db")
try:
applied = run_migrations(conn)
- assert applied == [1, 2, 3, 4]
- assert current_version(conn) == 4
+ assert applied == [1, 2, 3, 4, 5]
+ assert current_version(conn) == 5
assert EXPECTED_TABLES <= table_names(conn)
assert "schema_version" in table_names(conn)
columns = {
@@ -66,7 +66,7 @@ def test_fresh_and_idempotent():
# 再跑一次不应重复应用
assert run_migrations(conn) == []
row = conn.execute("SELECT COUNT(*) FROM schema_version").fetchone()
- assert row[0] == 4
+ assert row[0] == 5
finally:
conn.close()
@@ -81,15 +81,15 @@ def test_backup_before_upgrade():
conn.execute("INSERT INTO kv(k, v) VALUES ('hello', 'world')")
conn.commit()
- v5 = Migration(5, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",))
+ v6 = Migration(6, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",))
applied = run_migrations(
conn,
backup_dir=backup_dir,
now=1_800_000_000,
- migrations=MIGRATIONS + (v5,),
+ migrations=MIGRATIONS + (v6,),
)
- assert applied == [5]
- backups = list(backup_dir.glob("app-pre-v4-*.db"))
+ assert applied == [6]
+ backups = list(backup_dir.glob("app-pre-v5-*.db"))
assert len(backups) == 1
check = sqlite3.connect(backups[0])
try:
@@ -159,12 +159,35 @@ def test_validate_migration_order():
conn.close()
+def test_v5_normalizes_unfaved():
+ with tempfile.TemporaryDirectory() as tmp:
+ conn = sqlite3.connect(Path(tmp) / "app.db")
+ try:
+ run_migrations(conn, migrations=MIGRATIONS[:4])
+ conn.execute(
+ "INSERT INTO video(aid, title, pubdate, status, first_seen) "
+ "VALUES (1, 't', 1, 'unfaved', 1)"
+ )
+ conn.execute(
+ "INSERT INTO video(aid, title, pubdate, status, first_seen) "
+ "VALUES (2, 't', 1, 'dropped', 1)"
+ )
+ conn.commit()
+ applied = run_migrations(conn)
+ assert applied == [5]
+ rows = dict(conn.execute("SELECT aid, status FROM video").fetchall())
+ assert rows == {1: "active", 2: "dropped"}
+ finally:
+ conn.close()
+
+
def main():
test_fresh_and_idempotent()
test_backup_before_upgrade()
test_failed_migration_rolls_back()
test_prune_backups()
test_validate_migration_order()
+ test_v5_normalizes_unfaved()
print("OK: SQLite 迁移、备份与回滚测试全部通过")
diff --git a/tests/test_web_queries.py b/tests/test_web_queries.py
index f4b2f7c..2eb8839 100644
--- a/tests/test_web_queries.py
+++ b/tests/test_web_queries.py
@@ -1,492 +1,491 @@
-# -*- coding: utf-8 -*-
-"""只读查询层测试(离线): uv run python tests/test_web_queries.py"""
-
-import os
-import sys
-import tempfile
-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.config import ConfigStore # noqa: E402
-from reaction_board.database import Database # noqa: E402
-from reaction_board.repository import Repository # noqa: E402
-from reaction_board.timeutil import day_bounds # noqa: E402
-from reaction_board.web.cache import DASHBOARD_CACHE # noqa: E402
-from reaction_board.web.queries import DashboardQueries # noqa: E402
-from tests.web_fixtures import DATES, build_db, write_config # noqa: E402
-
-FIXED_NOW = 1_793_459_000 # 2026-10-31 前后
-
-
-def setup(tmp: str):
- build_db(tmp)
- config = write_config(tmp)
- db = Database(Path(tmp) / "app.db")
- db.initialize()
- store = ConfigStore(config)
- queries = DashboardQueries(db, store, clock=lambda: FIXED_NOW)
- return db, queries
-
-
-def test_overview():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- overview = q.overview()
- assert overview["data_date"] == DATES[-1]
- assert overview["snapshot_days"] == 3
- assert overview["is_first_day"] is False
- assert overview["initial_import_count"] == 2
- assert overview["counts"]["active"] == 2
- assert overview["counts"]["unfaved"] == 1
- assert overview["counts"]["removed"] == 1
- assert overview["counts"]["active_ups"] == 2
- assert overview["counts"]["candidates_pending"] == 1
- assert overview["totals"]["view"] == 1200 + 5200 + 120
- assert overview["interactions"] > 0
- assert overview["view_delta"] == (100 + 100 + 0) # 101/102 的日增量
- assert overview["snapshot_health"]["total"] == 3
- assert overview["folders"][0]["media_count"] == 3
- assert overview["folders"][0]["ratio"] < 1
- assert overview["fans"]["follower"] == 1020
- assert overview["fans"]["delta_1d"] == 10
- assert overview["preview"] is False
- assert overview["years"] == [2026]
- assert "snapshot" in overview["tasks"]
- finally:
- db.close()
-
-
-def test_video_listing_filters_and_sorting():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- heat = q.list_videos(sort="heat")
- aids = [item["aid"] for item in heat["items"]]
- assert 104 not in aids, "removed 视频不应进入榜单"
- # 综合热度榜应用最小播放门槛 2000:101(1200)与 103(120)被排除
- assert aids == [102], aids
- assert heat["items"][0]["heat_percentile"] == 100.0
-
- views = q.list_videos(sort="view")
- assert [item["aid"] for item in views["items"]] == [102, 101, 103]
-
- interactions = q.list_videos(sort="interaction")
- assert interactions["items"][0]["aid"] == 101
-
- added = q.list_videos(sort="added", within_days=1)
- assert [item["aid"] for item in added["items"]] == [103]
-
- published = q.list_videos(sort="published", within_days=2)
- assert {item["aid"] for item in published["items"]} == {101, 103}
-
- searched = q.list_videos(sort="view", q="UP B")
- assert [item["aid"] for item in searched["items"]] == [103]
- searched = q.list_videos(sort="view", q="视频 102")
- assert [item["aid"] for item in searched["items"]] == [102]
-
- folder = q.list_videos(sort="view", folder_id=3700023351)
- assert folder["total"] == 3
- folder = q.list_videos(sort="view", folder_id=999)
- assert folder["total"] == 0
-
- page = q.list_videos(sort="view", page=2, page_size=1)
- assert page["total"] == 3
- assert page["pages"] == 3
- assert [item["aid"] for item in page["items"]] == [101]
- finally:
- db.close()
-
-
-def test_growth_and_cold_start():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- growth = q.list_videos(sort="growth", growth_days=2)
- assert growth["available"] is True
- assert {item["aid"] for item in growth["items"]} == {101, 102}
- first = next(item for item in growth["items"] if item["aid"] == 101)
- assert first["growth"]["available"] is True
- assert first["growth"]["delta"] == 200
- assert first["growth"]["gap_days"] == 2
-
- rate = q.list_videos(sort="growth_rate", growth_days=2)
- assert {item["aid"] for item in rate["items"]} == {101, 102}
-
- # 只有一个快照日时增长不可用,不能返回假 0
- for day in DATES[:-1]:
- db.write(lambda conn, day=day: conn.execute(
- "DELETE FROM snapshot WHERE stat_date = ?", (day,)
- ))
- DASHBOARD_CACHE.clear()
- cold = q.list_videos(sort="growth", growth_days=2)
- assert cold["available"] is False
- assert cold["items"] == []
- assert "两天" in cold["reason"]
- finally:
- db.close()
-
-
-def test_video_detail_and_trend():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- detail = q.video_detail(101)
- assert detail["aid"] == 101
- assert detail["metrics"]["view"] == 1200
- assert detail["growth"]["delta"] == 100
- assert detail["snapshot_count"] == 3
- assert detail["folder_title"] == "2026"
- assert detail["url"].endswith("BV101")
- assert q.video_detail(999999) is None
-
- trend = q.video_trend(101, days=3)
- assert trend["dates"] == list(DATES)
- assert trend["series"]["view"] == [1000, 1100, 1200]
- assert trend["available"] is True
- assert len(trend["points"]) == 3
-
- # 请求区间早于首次快照时,从首次快照开始,不制造空白历史
- wide = q.video_trend(101, days=30)
- assert wide["dates"] == list(DATES)
-
- # 删除中间一天,折线在缺失日期应为 None,而不是插值
- db.write(lambda conn: conn.execute(
- "DELETE FROM snapshot WHERE aid=101 AND stat_date=?",
- ("2026-10-30",),
- ))
- trend = q.video_trend(101, days=3)
- assert trend["series"]["view"] == [1000, None, 1200]
-
- # 失败快照也视为缺失
- db.write(lambda conn: conn.execute(
- "UPDATE snapshot SET ok=0, view=NULL WHERE aid=101 AND stat_date=?",
- (DATES[-1],),
- ))
- trend = q.video_trend(101, days=3)
- assert trend["series"]["view"][-1] is None
- finally:
- db.close()
-
-
-def test_up_queries():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- ups = q.list_ups()
- assert ups["sort"] == "avg_views"
- assert ups["total"] == 2
- assert ups["items"][0]["up_mid"] == 1
- assert ups["items"][0]["video_count"] == 2
- assert ups["items"][0]["total_views"] == 6400
- assert ups["items"][0]["avg_views"] == 3200.0
-
- one = q.list_ups(min_videos=2)
- assert one["total"] == 1
-
- detail = q.up_detail(1)
- assert detail["video_count"] == 2
- assert detail["total_views"] == 6400
- assert detail["blacklisted"] is False
- assert detail["series"]["dates"] == list(DATES)
- assert detail["series"]["views"] == [6000, 6200, 6400]
- assert detail["space_url"].endswith("/1")
-
- blacklisted = q.up_detail(2)
- assert blacklisted["blacklisted"] is True
- assert blacklisted["video_count"] == 1
- assert q.up_detail(999999) is None
- finally:
- db.close()
-
-
-def test_query_cache_invalidation():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- DASHBOARD_CACHE.clear()
- first = q.overview()
- db.write(lambda conn: conn.execute(
- "UPDATE snapshot SET \"view\" = 999999 WHERE aid=101 AND stat_date=?",
- (DATES[-1],),
- ))
- cached = q.overview()
- assert cached["totals"]["view"] == first["totals"]["view"]
-
- repo = Repository(db)
- repo.bump_cache_version("videos")
- refreshed = q.overview()
- assert refreshed["totals"]["view"] > first["totals"]["view"]
-
- fans_first = q.fans(days=3)
- db.write(lambda conn: conn.execute(
- "UPDATE up_stat SET follower=999999 WHERE mid=37754047 AND stat_date=?",
- (DATES[-1],),
- ))
- fans_cached = q.fans(days=3)
- assert fans_cached["follower"] == fans_first["follower"]
-
- repo.bump_cache_version("fans")
- fans_refreshed = q.fans(days=3)
- assert fans_refreshed["follower"] == 999999
- finally:
- db.close()
-
-
-def test_fans_custom_range_modes():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- daily = q.fans(start="2026-10-29", end="2026-10-31")
- assert daily["mode"] == "daily"
- assert daily["range_start"] == "2026-10-29"
- assert daily["range_end"] == "2026-10-31"
- assert daily["dates"] == list(DATES)
- assert daily["series"] == [1000, 1010, 1020]
-
- # 超过一年触发按月峰值模式
- db.write(lambda conn: conn.execute(
- "INSERT OR REPLACE INTO up_stat"
- "(stat_date, mid, follower, following, captured_at, ok) "
- "VALUES ('2025-01-01', 37754047, 500, NULL, 1, 1)"
- ))
- DASHBOARD_CACHE.clear()
- monthly = q.fans(start="2025-01-01", end="2026-10-31")
- assert monthly["mode"] == "monthly"
- assert monthly["span_days"] == 669
- assert monthly["dates"][0] == "2025-01-01"
- assert monthly["series"][0] == 500
- assert monthly["dates"][-1] == "2026-10-31"
- assert monthly["series"][-1] == 1020
-
- # 同月多条时取峰值日期,并优先较晚日期
- db.write(lambda conn: conn.execute(
- "INSERT OR REPLACE INTO up_stat"
- "(stat_date, mid, follower, following, captured_at, ok) "
- "VALUES ('2026-10-15', 37754047, 5000, NULL, 1, 1), "
- "('2026-10-16', 37754047, 5000, NULL, 1, 1)"
- ))
- DASHBOARD_CACHE.clear()
- monthly = q.fans(start="2025-01-01", end="2026-10-31")
- assert monthly["dates"][-1] == "2026-10-16"
- assert monthly["series"][-1] == 5000
-
- try:
- q.fans(start="2026-11-01", end="2026-10-01")
- except ValueError:
- pass
- else:
- raise AssertionError("start 晚于 end 应报错")
- finally:
- db.close()
-
-
-def test_fans_candidates_system_and_export():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- fans = q.fans(days=3)
- assert fans["mid"] == 37754047
- assert fans["name"] == "咻咻满"
- assert fans["dates"] == list(DATES)
- assert fans["series"] == [1000, 1010, 1020]
- assert fans["delta_1d"] == 10
- assert fans["delta_7d"] is None
- assert fans["delta_cum"] == 20
- assert fans["options"][0]["mid"] == 37754047
- wide = q.fans(days=30)
- assert wide["dates"] == list(DATES)
-
- # “最近记录”只返回最近 14 天,早于窗口的记录不进入表格
- db.write(lambda conn: conn.execute(
- "INSERT OR REPLACE INTO up_stat"
- "(stat_date, mid, follower, following, captured_at, ok) "
- "VALUES (?, ?, ?, NULL, 1, 1)",
- ("2026-10-01", 37754047, 900),
- ))
- DASHBOARD_CACHE.clear()
- recent = q.fans(days=90)
- assert recent["recent_days"] == 14
- assert [point["stat_date"] for point in recent["recent_points"]] == list(DATES)
- deltas = {point["stat_date"]: point["delta"] for point in recent["recent_points"]}
- assert deltas["2026-10-29"] == 100 # 对比 2026-10-01 的 900
- assert deltas["2026-10-30"] == 10
- assert deltas["2026-10-31"] == 10
- one_day = q.fans(days=90, recent_days=1)
- assert [point["stat_date"] for point in one_day["recent_points"]] == [DATES[-1]]
-
- # 缺失前一天时不能把跨多天的变化标成“日增”
- db.write(lambda conn: conn.execute(
- "DELETE FROM up_stat WHERE mid=37754047 AND stat_date=?",
- ("2026-10-30",),
- ))
- DASHBOARD_CACHE.clear()
- gapped = q.fans(days=3)
- assert gapped["delta_1d"] is None
- assert gapped["delta_7d"] is None
-
- candidates = q.candidates()
- assert candidates["total"] == 2
- assert candidates["pending"] == 1
- assert candidates["items"][0]["aid"] == 201
- assert 1 <= candidates["items"][0]["days_left"] <= 7
- assert candidates["items"][1]["decision"] == "rejected"
-
- system = q.system()
- assert system["data_date"] == DATES[-1]
- assert system["snapshot_days"] == 3
- assert system["blacklist"][0]["video_count"] == 1
- assert system["tasks"]["snapshot"]["ok"] is True
- assert system["errors"] == []
- assert system["preview"] is False
- assert system["next_runs"]["snapshot"] > 0
- assert system["next_runs"]["folder_sync"] > FIXED_NOW
- assert system["next_runs"]["discover"] > FIXED_NOW
- finally:
- db.close()
-
-
-def test_duration_efficiency_queries():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- db.write(lambda conn: conn.executemany(
- "UPDATE video SET duration = ?, pubdate = ? WHERE aid = ?",
- [
- (9000, 1_756_684_800, 101), # 2025-09-01 前后,2.5 小时
- (7200, 1_756_771_200, 102), # 2 小时
- (300, 1_756_857_600, 103), # 5 分钟,应被排除
- ],
- ))
- DASHBOARD_CACHE.clear()
-
- efficiency = q.list_videos(sort="efficiency")
- assert [item["aid"] for item in efficiency["items"]] == [102]
- first = efficiency["items"][0]
- assert first["duration"] == 7200
- assert first["content_hours"] == 2.0
- assert first["efficiency"] == 2600.0 # 原始播放/小时,详情页口径
- assert abs(first["efficiency_score"] - 2207.5) < 0.1
- assert efficiency["duration_filter"] == {
- "min_duration_seconds": 600, "maturity_days": 2,
- "min_view": 2000, "shrink_minutes": 60,
- }
-
- # 播放不足 2000 的视频不进入效率榜(101 只有 1200)
- assert all(item["aid"] != 101 for item in efficiency["items"])
-
- interaction = q.list_videos(sort="interaction_efficiency")
- assert [item["aid"] for item in interaction["items"]] == [102]
- assert abs(interaction["items"][0]["interaction_efficiency_score"] - 159.36) < 0.1
-
- # 综合热度榜也应用最小播放门槛
- heat = q.list_videos(sort="heat")
- assert [item["aid"] for item in heat["items"]] == [102]
-
- # 时长缺失不计入时长效率榜
- ups = q.list_ups(sort="efficiency")
- assert ups["sort"] == "efficiency"
- up_a = next(item for item in ups["items"] if item["up_mid"] == 1)
- assert up_a["total_duration"] == int(9000 + 7200)
- assert up_a["content_hours"] == 4.5
- assert up_a["views_per_hour"] == round(6400 / 4.5, 1)
- assert up_a["efficiency"] is not None
- assert up_a["interaction_efficiency"] is not None
-
- by_duration = q.list_ups(sort="duration")
- assert [item["up_mid"] for item in by_duration["items"]][0] == 1
-
- db.write(lambda conn: conn.execute(
- "UPDATE video SET duration = NULL WHERE aid = 101"
- ))
- DASHBOARD_CACHE.clear()
- missing = q.list_videos(sort="efficiency")
- assert [item["aid"] for item in missing["items"]] == [102]
-
- # 发布未满 2 天的视频不进入效率榜
- db.write(lambda conn: conn.execute(
- "UPDATE video SET duration = 7200, pubdate = ? WHERE aid = 102",
- (int(day_bounds("2026-10-30")[0]),),
- ))
- DASHBOARD_CACHE.clear()
- fresh = q.list_videos(sort="efficiency")
- assert fresh["items"] == []
- assert fresh["available"] is False
- assert "发布满 2 天" in fresh["reason"]
- assert "播放 ≥ 2000" in fresh["reason"]
- finally:
- db.close()
-
-
-def test_year_filter_and_published_sort():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- by_year = q.list_videos(sort="view", year=2026)
- assert [item["aid"] for item in by_year["items"]] == [102, 101, 103]
- assert by_year["year"] == 2026
-
- none = q.list_videos(sort="view", year=2025)
- assert none["items"] == []
- assert none["total"] == 0
-
- published = q.list_videos(sort="published")
- assert [item["aid"] for item in published["items"]] == [103, 101, 102]
-
- added = q.list_videos(sort="added")
- assert added["total"] == 3
-
- ups = q.list_ups(sort="efficiency", year=2026)
- assert ups["year"] == 2026 and ups["total"] == 2
- ups_none = q.list_ups(sort="views", year=2025)
- assert ups_none["total"] == 0
- finally:
- db.close()
-
-
-def test_about_payload():
- with tempfile.TemporaryDirectory() as tmp:
- db, q = setup(tmp)
- try:
- about = q.about()
- assert about["data_date"] == DATES[-1]
- assert about["snapshot_days"] == 3
- assert about["snapshot_first_date"] == DATES[0]
- assert about["fans_start_date"] == DATES[0]
- assert about["schema_version"] == 4
- assert about["analytics"] == {
- "min_duration_seconds": 600,
- "maturity_days": 2,
- "min_view_for_ranking": 2000,
- "video_shrink_minutes": 60,
- "shrink_minutes": None,
- }
- assert about["heat_weights"]["view"] == 0.30
- assert about["followers"]["videos_total"] == 3
- assert about["followers"]["videos_covered"] == 0
- assert about["followers"]["videos_missing"] == 3
- assert about["folders"][0]["media_id"] == 3700023351
- assert about["preview"] is False
- finally:
- db.close()
-
-
-def main_test():
- test_overview()
- test_video_listing_filters_and_sorting()
- test_growth_and_cold_start()
- test_video_detail_and_trend()
- test_up_queries()
- test_query_cache_invalidation()
- test_duration_efficiency_queries()
- test_year_filter_and_published_sort()
- test_about_payload()
- test_fans_custom_range_modes()
- test_fans_candidates_system_and_export()
- print("OK: Web 查询层总览/榜单/详情/UP/时长效率/粉丝/候选/系统测试全部通过")
-
-
-if __name__ == "__main__":
- main_test()
+# -*- coding: utf-8 -*-
+"""只读查询层测试(离线): uv run python tests/test_web_queries.py"""
+
+import os
+import sys
+import tempfile
+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.config import ConfigStore # noqa: E402
+from reaction_board.database import Database # noqa: E402
+from reaction_board.repository import Repository # noqa: E402
+from reaction_board.timeutil import day_bounds # noqa: E402
+from reaction_board.web.cache import DASHBOARD_CACHE # noqa: E402
+from reaction_board.web.queries import DashboardQueries # noqa: E402
+from tests.web_fixtures import DATES, build_db, write_config # noqa: E402
+
+FIXED_NOW = 1_793_459_000 # 2026-10-31 前后
+
+
+def setup(tmp: str):
+ build_db(tmp)
+ config = write_config(tmp)
+ db = Database(Path(tmp) / "app.db")
+ db.initialize()
+ store = ConfigStore(config)
+ queries = DashboardQueries(db, store, clock=lambda: FIXED_NOW)
+ return db, queries
+
+
+def test_overview():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ overview = q.overview()
+ assert overview["data_date"] == DATES[-1]
+ assert overview["snapshot_days"] == 3
+ assert overview["is_first_day"] is False
+ assert overview["initial_import_count"] == 2
+ assert overview["counts"]["active"] == 2
+ assert overview["counts"]["unfaved"] == 1
+ assert overview["counts"]["removed"] == 1
+ assert overview["counts"]["active_ups"] == 2
+ assert overview["counts"]["candidates_pending"] == 1
+ assert overview["totals"]["view"] == 1200 + 5200 + 120
+ assert overview["interactions"] > 0
+ assert overview["view_delta"] == (100 + 100 + 0) # 101/102 的日增量
+ assert overview["snapshot_health"]["total"] == 3
+ assert overview["folders"][0]["media_count"] == 3
+ assert overview["folders"][0]["ratio"] < 1
+ assert overview["fans"]["follower"] == 1020
+ assert overview["fans"]["delta_1d"] == 10
+ assert overview["preview"] is False
+ assert overview["years"] == [2026]
+ assert "snapshot" in overview["tasks"]
+ finally:
+ db.close()
+
+
+def test_video_listing_filters_and_sorting():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ heat = q.list_videos(sort="heat")
+ aids = [item["aid"] for item in heat["items"]]
+ assert 104 not in aids, "removed 视频不应进入榜单"
+ # 综合热度榜应用最小播放门槛 2000:101(1200)与 103(120)被排除
+ assert aids == [102], aids
+ assert heat["items"][0]["heat_percentile"] == 100.0
+
+ views = q.list_videos(sort="view")
+ assert [item["aid"] for item in views["items"]] == [102, 101, 103]
+
+ interactions = q.list_videos(sort="interaction")
+ assert interactions["items"][0]["aid"] == 101
+
+ added = q.list_videos(sort="added", within_days=1)
+ assert [item["aid"] for item in added["items"]] == [103]
+
+ published = q.list_videos(sort="published", within_days=2)
+ assert {item["aid"] for item in published["items"]} == {101, 103}
+
+ searched = q.list_videos(sort="view", q="UP B")
+ assert [item["aid"] for item in searched["items"]] == [103]
+ searched = q.list_videos(sort="view", q="视频 102")
+ assert [item["aid"] for item in searched["items"]] == [102]
+
+ folder = q.list_videos(sort="view", folder_id=3700023351)
+ assert folder["total"] == 3
+ folder = q.list_videos(sort="view", folder_id=999)
+ assert folder["total"] == 0
+
+ page = q.list_videos(sort="view", page=2, page_size=1)
+ assert page["total"] == 3
+ assert page["pages"] == 3
+ assert [item["aid"] for item in page["items"]] == [101]
+ finally:
+ db.close()
+
+
+def test_growth_and_cold_start():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ growth = q.list_videos(sort="growth", growth_days=2)
+ assert growth["available"] is True
+ assert {item["aid"] for item in growth["items"]} == {101, 102}
+ first = next(item for item in growth["items"] if item["aid"] == 101)
+ assert first["growth"]["available"] is True
+ assert first["growth"]["delta"] == 200
+ assert first["growth"]["gap_days"] == 2
+
+ rate = q.list_videos(sort="growth_rate", growth_days=2)
+ assert {item["aid"] for item in rate["items"]} == {101, 102}
+
+ # 只有一个快照日时增长不可用,不能返回假 0
+ for day in DATES[:-1]:
+ db.write(lambda conn, day=day: conn.execute(
+ "DELETE FROM snapshot WHERE stat_date = ?", (day,)
+ ))
+ DASHBOARD_CACHE.clear()
+ cold = q.list_videos(sort="growth", growth_days=2)
+ assert cold["available"] is False
+ assert cold["items"] == []
+ assert "两天" in cold["reason"]
+ finally:
+ db.close()
+
+
+def test_video_detail_and_trend():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ detail = q.video_detail(101)
+ assert detail["aid"] == 101
+ assert detail["metrics"]["view"] == 1200
+ assert detail["growth"]["delta"] == 100
+ assert detail["snapshot_count"] == 3
+ assert detail["folder_title"] == "2026"
+ assert detail["url"].endswith("BV101")
+ assert q.video_detail(999999) is None
+
+ trend = q.video_trend(101, days=3)
+ assert trend["dates"] == list(DATES)
+ assert trend["series"]["view"] == [1000, 1100, 1200]
+ assert trend["available"] is True
+ assert len(trend["points"]) == 3
+
+ # 请求区间早于首次快照时,从首次快照开始,不制造空白历史
+ wide = q.video_trend(101, days=30)
+ assert wide["dates"] == list(DATES)
+
+ # 删除中间一天,折线在缺失日期应为 None,而不是插值
+ db.write(lambda conn: conn.execute(
+ "DELETE FROM snapshot WHERE aid=101 AND stat_date=?",
+ ("2026-10-30",),
+ ))
+ trend = q.video_trend(101, days=3)
+ assert trend["series"]["view"] == [1000, None, 1200]
+
+ # 失败快照也视为缺失
+ db.write(lambda conn: conn.execute(
+ "UPDATE snapshot SET ok=0, view=NULL WHERE aid=101 AND stat_date=?",
+ (DATES[-1],),
+ ))
+ trend = q.video_trend(101, days=3)
+ assert trend["series"]["view"][-1] is None
+ finally:
+ db.close()
+
+
+def test_up_queries():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ ups = q.list_ups()
+ assert ups["sort"] == "avg_views"
+ assert ups["total"] == 2
+ assert ups["items"][0]["up_mid"] == 1
+ assert ups["items"][0]["video_count"] == 2
+ assert ups["items"][0]["total_views"] == 6400
+ assert ups["items"][0]["avg_views"] == 3200.0
+
+ one = q.list_ups(min_videos=2)
+ assert one["total"] == 1
+
+ detail = q.up_detail(1)
+ assert detail["video_count"] == 2
+ assert detail["total_views"] == 6400
+ assert detail["blacklisted"] is False
+ assert detail["series"]["dates"] == list(DATES)
+ assert detail["series"]["views"] == [6000, 6200, 6400]
+ assert detail["space_url"].endswith("/1")
+
+ blacklisted = q.up_detail(2)
+ assert blacklisted["blacklisted"] is True
+ assert blacklisted["video_count"] == 1
+ assert q.up_detail(999999) is None
+ finally:
+ db.close()
+
+
+def test_query_cache_invalidation():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ DASHBOARD_CACHE.clear()
+ first = q.overview()
+ db.write(lambda conn: conn.execute(
+ "UPDATE snapshot SET \"view\" = 999999 WHERE aid=101 AND stat_date=?",
+ (DATES[-1],),
+ ))
+ cached = q.overview()
+ assert cached["totals"]["view"] == first["totals"]["view"]
+
+ repo = Repository(db)
+ repo.bump_cache_version("videos")
+ refreshed = q.overview()
+ assert refreshed["totals"]["view"] > first["totals"]["view"]
+
+ fans_first = q.fans(days=3)
+ db.write(lambda conn: conn.execute(
+ "UPDATE up_stat SET follower=999999 WHERE mid=37754047 AND stat_date=?",
+ (DATES[-1],),
+ ))
+ fans_cached = q.fans(days=3)
+ assert fans_cached["follower"] == fans_first["follower"]
+
+ repo.bump_cache_version("fans")
+ fans_refreshed = q.fans(days=3)
+ assert fans_refreshed["follower"] == 999999
+ finally:
+ db.close()
+
+
+def test_fans_custom_range_modes():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ daily = q.fans(start="2026-10-29", end="2026-10-31")
+ assert daily["mode"] == "daily"
+ assert daily["range_start"] == "2026-10-29"
+ assert daily["range_end"] == "2026-10-31"
+ assert daily["dates"] == list(DATES)
+ assert daily["series"] == [1000, 1010, 1020]
+
+ # 超过一年触发按月峰值模式
+ db.write(lambda conn: conn.execute(
+ "INSERT OR REPLACE INTO up_stat"
+ "(stat_date, mid, follower, following, captured_at, ok) "
+ "VALUES ('2025-01-01', 37754047, 500, NULL, 1, 1)"
+ ))
+ DASHBOARD_CACHE.clear()
+ monthly = q.fans(start="2025-01-01", end="2026-10-31")
+ assert monthly["mode"] == "monthly"
+ assert monthly["span_days"] == 669
+ assert monthly["dates"][0] == "2025-01-01"
+ assert monthly["series"][0] == 500
+ assert monthly["dates"][-1] == "2026-10-31"
+ assert monthly["series"][-1] == 1020
+
+ # 同月多条时取峰值日期,并优先较晚日期
+ db.write(lambda conn: conn.execute(
+ "INSERT OR REPLACE INTO up_stat"
+ "(stat_date, mid, follower, following, captured_at, ok) "
+ "VALUES ('2026-10-15', 37754047, 5000, NULL, 1, 1), "
+ "('2026-10-16', 37754047, 5000, NULL, 1, 1)"
+ ))
+ DASHBOARD_CACHE.clear()
+ monthly = q.fans(start="2025-01-01", end="2026-10-31")
+ assert monthly["dates"][-1] == "2026-10-16"
+ assert monthly["series"][-1] == 5000
+
+ try:
+ q.fans(start="2026-11-01", end="2026-10-01")
+ except ValueError:
+ pass
+ else:
+ raise AssertionError("start 晚于 end 应报错")
+ finally:
+ db.close()
+
+
+def test_fans_candidates_system_and_export():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ fans = q.fans(days=3)
+ assert fans["mid"] == 37754047
+ assert fans["name"] == "咻咻满"
+ assert fans["dates"] == list(DATES)
+ assert fans["series"] == [1000, 1010, 1020]
+ assert fans["delta_1d"] == 10
+ assert fans["delta_7d"] is None
+ assert fans["delta_cum"] == 20
+ assert fans["options"][0]["mid"] == 37754047
+ wide = q.fans(days=30)
+ assert wide["dates"] == list(DATES)
+
+ # “最近记录”只返回最近 14 天,早于窗口的记录不进入表格
+ db.write(lambda conn: conn.execute(
+ "INSERT OR REPLACE INTO up_stat"
+ "(stat_date, mid, follower, following, captured_at, ok) "
+ "VALUES (?, ?, ?, NULL, 1, 1)",
+ ("2026-10-01", 37754047, 900),
+ ))
+ DASHBOARD_CACHE.clear()
+ recent = q.fans(days=90)
+ assert recent["recent_days"] == 14
+ assert [point["stat_date"] for point in recent["recent_points"]] == list(DATES)
+ deltas = {point["stat_date"]: point["delta"] for point in recent["recent_points"]}
+ assert deltas["2026-10-29"] == 100 # 对比 2026-10-01 的 900
+ assert deltas["2026-10-30"] == 10
+ assert deltas["2026-10-31"] == 10
+ one_day = q.fans(days=90, recent_days=1)
+ assert [point["stat_date"] for point in one_day["recent_points"]] == [DATES[-1]]
+
+ # 缺失前一天时不能把跨多天的变化标成“日增”
+ db.write(lambda conn: conn.execute(
+ "DELETE FROM up_stat WHERE mid=37754047 AND stat_date=?",
+ ("2026-10-30",),
+ ))
+ DASHBOARD_CACHE.clear()
+ gapped = q.fans(days=3)
+ assert gapped["delta_1d"] is None
+ assert gapped["delta_7d"] is None
+
+ candidates = q.candidates()
+ assert candidates["total"] == 2
+ assert candidates["pending"] == 1
+ assert candidates["items"][0]["aid"] == 201
+ assert 1 <= candidates["items"][0]["days_left"] <= 7
+ assert candidates["items"][1]["decision"] == "rejected"
+
+ system = q.system()
+ assert system["data_date"] == DATES[-1]
+ assert system["snapshot_days"] == 3
+ assert system["blacklist"][0]["video_count"] == 1
+ assert system["tasks"]["snapshot"]["ok"] is True
+ assert system["errors"] == []
+ assert system["preview"] is False
+ assert system["next_runs"]["snapshot"] > 0
+ assert system["next_runs"]["discover"] > FIXED_NOW
+ finally:
+ db.close()
+
+
+def test_duration_efficiency_queries():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ db.write(lambda conn: conn.executemany(
+ "UPDATE video SET duration = ?, pubdate = ? WHERE aid = ?",
+ [
+ (9000, 1_756_684_800, 101), # 2025-09-01 前后,2.5 小时
+ (7200, 1_756_771_200, 102), # 2 小时
+ (300, 1_756_857_600, 103), # 5 分钟,应被排除
+ ],
+ ))
+ DASHBOARD_CACHE.clear()
+
+ efficiency = q.list_videos(sort="efficiency")
+ assert [item["aid"] for item in efficiency["items"]] == [102]
+ first = efficiency["items"][0]
+ assert first["duration"] == 7200
+ assert first["content_hours"] == 2.0
+ assert first["efficiency"] == 2600.0 # 原始播放/小时,详情页口径
+ assert abs(first["efficiency_score"] - 2207.5) < 0.1
+ assert efficiency["duration_filter"] == {
+ "min_duration_seconds": 600, "maturity_days": 2,
+ "min_view": 2000, "shrink_minutes": 60,
+ }
+
+ # 播放不足 2000 的视频不进入效率榜(101 只有 1200)
+ assert all(item["aid"] != 101 for item in efficiency["items"])
+
+ interaction = q.list_videos(sort="interaction_efficiency")
+ assert [item["aid"] for item in interaction["items"]] == [102]
+ assert abs(interaction["items"][0]["interaction_efficiency_score"] - 159.36) < 0.1
+
+ # 综合热度榜也应用最小播放门槛
+ heat = q.list_videos(sort="heat")
+ assert [item["aid"] for item in heat["items"]] == [102]
+
+ # 时长缺失不计入时长效率榜
+ ups = q.list_ups(sort="efficiency")
+ assert ups["sort"] == "efficiency"
+ up_a = next(item for item in ups["items"] if item["up_mid"] == 1)
+ assert up_a["total_duration"] == int(9000 + 7200)
+ assert up_a["content_hours"] == 4.5
+ assert up_a["views_per_hour"] == round(6400 / 4.5, 1)
+ assert up_a["efficiency"] is not None
+ assert up_a["interaction_efficiency"] is not None
+
+ by_duration = q.list_ups(sort="duration")
+ assert [item["up_mid"] for item in by_duration["items"]][0] == 1
+
+ db.write(lambda conn: conn.execute(
+ "UPDATE video SET duration = NULL WHERE aid = 101"
+ ))
+ DASHBOARD_CACHE.clear()
+ missing = q.list_videos(sort="efficiency")
+ assert [item["aid"] for item in missing["items"]] == [102]
+
+ # 发布未满 2 天的视频不进入效率榜
+ db.write(lambda conn: conn.execute(
+ "UPDATE video SET duration = 7200, pubdate = ? WHERE aid = 102",
+ (int(day_bounds("2026-10-30")[0]),),
+ ))
+ DASHBOARD_CACHE.clear()
+ fresh = q.list_videos(sort="efficiency")
+ assert fresh["items"] == []
+ assert fresh["available"] is False
+ assert "发布满 2 天" in fresh["reason"]
+ assert "播放 ≥ 2000" in fresh["reason"]
+ finally:
+ db.close()
+
+
+def test_year_filter_and_published_sort():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ by_year = q.list_videos(sort="view", year=2026)
+ assert [item["aid"] for item in by_year["items"]] == [102, 101, 103]
+ assert by_year["year"] == 2026
+
+ none = q.list_videos(sort="view", year=2025)
+ assert none["items"] == []
+ assert none["total"] == 0
+
+ published = q.list_videos(sort="published")
+ assert [item["aid"] for item in published["items"]] == [103, 101, 102]
+
+ added = q.list_videos(sort="added")
+ assert added["total"] == 3
+
+ ups = q.list_ups(sort="efficiency", year=2026)
+ assert ups["year"] == 2026 and ups["total"] == 2
+ ups_none = q.list_ups(sort="views", year=2025)
+ assert ups_none["total"] == 0
+ finally:
+ db.close()
+
+
+def test_about_payload():
+ with tempfile.TemporaryDirectory() as tmp:
+ db, q = setup(tmp)
+ try:
+ about = q.about()
+ assert about["data_date"] == DATES[-1]
+ assert about["snapshot_days"] == 3
+ assert about["snapshot_first_date"] == DATES[0]
+ assert about["fans_start_date"] == DATES[0]
+ assert about["schema_version"] == 5
+ assert about["analytics"] == {
+ "min_duration_seconds": 600,
+ "maturity_days": 2,
+ "min_view_for_ranking": 2000,
+ "video_shrink_minutes": 60,
+ "shrink_minutes": None,
+ }
+ assert about["heat_weights"]["view"] == 0.30
+ assert about["followers"]["videos_total"] == 3
+ assert about["followers"]["videos_covered"] == 0
+ assert about["followers"]["videos_missing"] == 3
+ assert about["folders"][0]["media_id"] == 3700023351
+ assert about["preview"] is False
+ finally:
+ db.close()
+
+
+def main_test():
+ test_overview()
+ test_video_listing_filters_and_sorting()
+ test_growth_and_cold_start()
+ test_video_detail_and_trend()
+ test_up_queries()
+ test_query_cache_invalidation()
+ test_duration_efficiency_queries()
+ test_year_filter_and_published_sort()
+ test_about_payload()
+ test_fans_custom_range_modes()
+ test_fans_candidates_system_and_export()
+ print("OK: Web 查询层总览/榜单/详情/UP/时长效率/粉丝/候选/系统测试全部通过")
+
+
+if __name__ == "__main__":
+ main_test()
diff --git a/tests/test_web_server.py b/tests/test_web_server.py
index c56b4cc..ea677ba 100644
--- a/tests/test_web_server.py
+++ b/tests/test_web_server.py
@@ -1,308 +1,308 @@
-# -*- coding: utf-8 -*-
-"""只读 Web 服务测试(离线): uv run python tests/test_web_server.py"""
-
-import http.client
-import json
-import os
-import sys
-import tempfile
-import threading
-import urllib.parse
-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.config import ConfigStore # noqa: E402
-from reaction_board.database import Database # noqa: E402
-from reaction_board.web.server import create_server # noqa: E402
-from tests.web_fixtures import build_db, write_config # noqa: E402
-
-
-class RunningServer:
- def __init__(self, tmp: str, *, token: str = "", preview: bool = False):
- self.config = write_config(tmp, token=token)
- build_db(tmp)
- self.db = Database(Path(tmp) / "app.db")
- self.db.initialize()
- if preview:
- self.db.write(lambda conn: conn.execute(
- "INSERT OR REPLACE INTO kv(k, v) VALUES ('preview:seeded', '1')"
- ))
- self.store = ConfigStore(self.config)
- self.server = create_server(
- self.store, self.db, port=0, preview=preview,
- clock=lambda: 1_793_459_000,
- )
- self.port = self.server.server_address[1]
- self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
- self.thread.start()
-
- def request(self, path: str, *, method: str = "GET", headers=None, body=None):
- conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
- conn.request(method, path, body=body, headers=headers or {})
- response = conn.getresponse()
- payload = response.read()
- result = response.status, dict(response.getheaders()), payload
- conn.close()
- return result
-
- def json(self, path: str, **kwargs):
- status, headers, payload = self.request(path, **kwargs)
- return status, headers, json.loads(payload.decode("utf-8"))
-
- def close(self):
- self.server.shutdown()
- self.server.server_close()
- self.db.close()
-
-
-def test_readonly_routes_and_static():
- with tempfile.TemporaryDirectory() as tmp:
- running = RunningServer(tmp)
- try:
- status, _, health = running.json("/api/health")
- assert status == 200 and health["ok"] is True
- assert health["data"]["preview"] is False
-
- status, _, payload = running.json("/api/session")
- assert status == 200
- assert payload["data"]["auth_enabled"] is False
- assert payload["data"]["authenticated"] is True
- status, _, payload = running.json("/api/candidates")
- assert status == 200 and payload["ok"] is True
- status, _, payload = running.json("/api/system")
- assert status == 200 and payload["ok"] is True
- status, _, payload = running.json("/api/about")
- assert status == 200
- assert payload["data"]["schema_version"] == 4
- assert payload["data"]["analytics"]["min_duration_seconds"] == 600
-
- status, _, overview = running.json("/api/overview")
- assert status == 200 and overview["ok"] is True
- assert overview["data"]["data_date"] == "2026-10-31"
-
- status, headers, payload = running.request("/")
- assert status == 200
- assert "text/html" in headers["Content-Type"]
- assert b"= "2026-10-18"
- for point in fans["recent_points"]
- )
- assert all("delta" in point for point in fans["recent_points"])
-
- status, _, payload = running.request(
- "/api/fans?start=2026-10-29&end=2026-10-31"
- )
- assert status == 200
- data = json.loads(payload)["data"]
- assert data["mode"] == "daily"
- assert data["dates"] == ["2026-10-29", "2026-10-30", "2026-10-31"]
-
- running.db.write(lambda conn: conn.execute(
- "INSERT OR REPLACE INTO up_stat"
- "(stat_date, mid, follower, following, captured_at, ok) "
- "VALUES ('2025-01-01', 37754047, 500, NULL, 1, 1)"
- ))
- status, _, payload = running.request(
- "/api/fans?start=2025-01-01&end=2026-10-31"
- )
- assert status == 200
- assert json.loads(payload)["data"]["mode"] == "monthly"
-
- status, headers, payload = running.request("/api/videos", method="HEAD")
- assert status == 200 and payload == b""
- assert "Content-Length" in headers
-
- # 只读:所有业务写方法都被拒绝,且数据库不变
- before = running.db.read(lambda conn: conn.execute(
- "SELECT COUNT(*) FROM candidate"
- ).fetchone()[0])
- for method in ("POST", "PUT", "PATCH", "DELETE"):
- status, _, _ = running.request("/api/videos/101", method=method)
- assert status == 405, method
- after = running.db.read(lambda conn: conn.execute(
- "SELECT COUNT(*) FROM candidate"
- ).fetchone()[0])
- assert after == before
-
- status, _, error = running.json("/api/videos?sort=bad")
- assert status == 400 and error["ok"] is False
- status, _, error = running.json("/api/videos?page=0")
- assert status == 400
- status, _, error = running.json("/api/videos/abc")
- assert status == 400
- status, _, error = running.json("/api/nope")
- assert status == 404
-
- status, _, payload = running.request("/api/export.csv?sort=view")
- assert status == 404
-
- status, headers, _ = running.request("/api/overview")
- assert headers["X-Content-Type-Options"] == "nosniff"
- assert "default-src 'self'" in headers["Content-Security-Policy"]
- assert headers["X-Frame-Options"] == "DENY"
- finally:
- running.close()
-
-
-def test_token_auth_and_login():
- with tempfile.TemporaryDirectory() as tmp:
- running = RunningServer(tmp, token="test-token")
- try:
- # 公共页面与公共 API 访客可见
- status, _, body = running.json("/api/overview")
- assert status == 200 and body["ok"] is True
- status, _, body = running.json("/api/videos")
- assert status == 200 and body["ok"] is True
- status, _, body = running.json("/api/fans?days=7")
- assert status == 200 and body["ok"] is True
- status, _, payload = running.request("/")
- assert status == 200 and b"= "2026-10-18"
+ for point in fans["recent_points"]
+ )
+ assert all("delta" in point for point in fans["recent_points"])
+
+ status, _, payload = running.request(
+ "/api/fans?start=2026-10-29&end=2026-10-31"
+ )
+ assert status == 200
+ data = json.loads(payload)["data"]
+ assert data["mode"] == "daily"
+ assert data["dates"] == ["2026-10-29", "2026-10-30", "2026-10-31"]
+
+ running.db.write(lambda conn: conn.execute(
+ "INSERT OR REPLACE INTO up_stat"
+ "(stat_date, mid, follower, following, captured_at, ok) "
+ "VALUES ('2025-01-01', 37754047, 500, NULL, 1, 1)"
+ ))
+ status, _, payload = running.request(
+ "/api/fans?start=2025-01-01&end=2026-10-31"
+ )
+ assert status == 200
+ assert json.loads(payload)["data"]["mode"] == "monthly"
+
+ status, headers, payload = running.request("/api/videos", method="HEAD")
+ assert status == 200 and payload == b""
+ assert "Content-Length" in headers
+
+ # 只读:所有业务写方法都被拒绝,且数据库不变
+ before = running.db.read(lambda conn: conn.execute(
+ "SELECT COUNT(*) FROM candidate"
+ ).fetchone()[0])
+ for method in ("POST", "PUT", "PATCH", "DELETE"):
+ status, _, _ = running.request("/api/videos/101", method=method)
+ assert status == 405, method
+ after = running.db.read(lambda conn: conn.execute(
+ "SELECT COUNT(*) FROM candidate"
+ ).fetchone()[0])
+ assert after == before
+
+ status, _, error = running.json("/api/videos?sort=bad")
+ assert status == 400 and error["ok"] is False
+ status, _, error = running.json("/api/videos?page=0")
+ assert status == 400
+ status, _, error = running.json("/api/videos/abc")
+ assert status == 400
+ status, _, error = running.json("/api/nope")
+ assert status == 404
+
+ status, _, payload = running.request("/api/export.csv?sort=view")
+ assert status == 404
+
+ status, headers, _ = running.request("/api/overview")
+ assert headers["X-Content-Type-Options"] == "nosniff"
+ assert "default-src 'self'" in headers["Content-Security-Policy"]
+ assert headers["X-Frame-Options"] == "DENY"
+ finally:
+ running.close()
+
+
+def test_token_auth_and_login():
+ with tempfile.TemporaryDirectory() as tmp:
+ running = RunningServer(tmp, token="test-token")
+ try:
+ # 公共页面与公共 API 访客可见
+ status, _, body = running.json("/api/overview")
+ assert status == 200 and body["ok"] is True
+ status, _, body = running.json("/api/videos")
+ assert status == 200 and body["ok"] is True
+ status, _, body = running.json("/api/fans?days=7")
+ assert status == 200 and body["ok"] is True
+ status, _, payload = running.request("/")
+ assert status == 200 and b"