diff --git a/reaction_board/cli.py b/reaction_board/cli.py
index 4ec1240..904346e 100644
--- a/reaction_board/cli.py
+++ b/reaction_board/cli.py
@@ -19,6 +19,7 @@ import argparse
import json
import sys
import threading
+import time
from pathlib import Path
from typing import Sequence
@@ -48,7 +49,7 @@ def _build_parser() -> argparse.ArgumentParser:
common = argparse.ArgumentParser(add_help=False)
common.add_argument(
"--config",
- default=str(DEFAULT_CONFIG),
+ default=argparse.SUPPRESS,
help="配置文件路径（默认项目根目录 config.json）",
)
@@ -56,7 +57,12 @@ def _build_parser() -> argparse.ArgumentParser:
prog="app.py",
description="咻咻满 Reaction 数据看板 —— 数据底座、同步与每日快照",
)
- sub = parser.add_subparsers(dest="command", required=True)
+ parser.add_argument(
+ "--config",
+ default=str(DEFAULT_CONFIG),
+ help="配置文件路径（默认项目根目录 config.json）",
+ )
+ sub = parser.add_subparsers(dest="command", required=False)
sub.add_parser("init", parents=[common], help="建库、迁移并写入黑名单种子")
sync = sub.add_parser("sync", parents=[common], help="同步已配置的年度收藏夹")
@@ -109,6 +115,11 @@ def _build_parser() -> argparse.ArgumentParser:
action="store_true",
help="允许把远端 0 条视为真实清空（默认按异常跳过差集）",
)
+ serve = sub.add_parser("serve", parents=[common], help="启动只读本地看板（不运行调度）")
+ serve.add_argument(
+ "--preview", action="store_true",
+ help="以预览模式启动（要求数据库含 preview:seeded=1 标记）",
+ )
sub.add_parser("status", parents=[common], help="查看数据库状态（离线）")
return parser
@@ -128,6 +139,32 @@ def _print(payload) -> None:
print(json.dumps(payload, ensure_ascii=False, indent=2), flush=True)
+def _record_task(
+ repo: Repository,
+ name: str,
+ *,
+ ok: bool,
+ started_at: int | None = None,
+ finished_at: int | None = None,
+ payload=None,
+ error: str | None = None,
+) -> None:
+ """把最近一次任务结果写入 kv，供只读系统页展示。"""
+ now = int(time.time())
+ record = {
+ "name": name,
+ "ok": bool(ok),
+ "started_at": int(started_at if started_at is not None else now),
+ "finished_at": int(finished_at if finished_at is not None else now),
+ "error": error,
+ "summary": payload,
+ }
+ try:
+ repo.kv_set(f"task:last:{name}", json.dumps(record, ensure_ascii=False))
+ except Exception: # noqa: BLE001 任务状态记录失败不影响主流程
+ pass
+
+
def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient:
return BilibiliClient(
cookie=credentials.cookie,
@@ -174,10 +211,15 @@ def cmd_sync(args, logger: JsonLogger) -> int:
logger=logger,
allow_empty_folders=args.allow_empty_folder,
)
+ started = int(time.time())
summary = service.run(full_refresh=args.full)
payload = summary.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,
+ )
_print(payload)
return 0 if summary.ok else 1
finally:
@@ -199,8 +241,15 @@ def cmd_snapshot(args, logger: JsonLogger) -> int:
payload["database"] = str(cfg.storage.database_path)
payload["snapshot_counts"] = repo.snapshot_counts(summary.stat_date)
payload["counts"] = repo.counts()
+ ok = bool(summary.completed and not summary.backup_error)
+ _record_task(
+ repo, "snapshot", ok=ok,
+ started_at=int(summary.started_at or time.time()),
+ finished_at=int(time.time()), payload=payload,
+ error=summary.backup_error,
+ )
_print(payload)
- return 0 if summary.completed and not summary.backup_error else 1
+ return 0 if ok else 1
finally:
db.close()
@@ -239,6 +288,13 @@ def cmd_discover(args, logger: JsonLogger) -> int:
payload = summary.as_dict()
payload["candidate_total"] = len(repo.list_candidates())
payload["database"] = str(cfg.storage.database_path)
+ _record_task(
+ repo, "discovery", ok=not summary.errors,
+ started_at=int(summary.started_at),
+ finished_at=int(summary.finished_at or time.time()),
+ payload=payload,
+ error="；".join(summary.errors) if summary.errors else None,
+ )
_print(payload)
return 0 if not summary.errors else 1
finally:
@@ -299,11 +355,12 @@ def cmd_auth_check(args, logger: JsonLogger) -> int:
return 0 if all(result["ok"] for result in results) else 1
-def cmd_run(args, logger: JsonLogger) -> int:
+def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
store = ConfigStore(args.config, require_auth=True)
cfg = store.get()
db = _open_db(cfg, logger)
last: dict[str, object] = {}
+ web_server = None
try:
repo = Repository(db)
folder_provider = ClientProvider(
@@ -349,12 +406,29 @@ def cmd_run(args, logger: JsonLogger) -> int:
)
def run_discover():
- summary = discovery_service.run()
+ started = int(time.time())
+ try:
+ summary = discovery_service.run()
+ except Exception as exc: # noqa: BLE001 记录后交给调度器隔离
+ _record_task(
+ repo, "discovery", ok=False, started_at=started,
+ finished_at=int(time.time()), error=str(exc),
+ )
+ raise
+ payload = summary.as_dict()
+ _record_task(
+ repo, "discovery", ok=not summary.errors,
+ started_at=summary.started_at,
+ finished_at=summary.finished_at or int(time.time()),
+ payload=payload,
+ error="；".join(summary.errors) if summary.errors else None,
+ )
last["discover"] = summary
- logger.info("discovery.summary", **summary.as_dict())
+ logger.info("discovery.summary", **payload)
return summary
def run_cleanup():
+ started = int(time.time())
removed = repo.delete_expired_candidates(
ttl_days=cfg.schedule.candidate_ttl_days
)
@@ -363,21 +437,58 @@ def cmd_run(args, logger: JsonLogger) -> int:
"cleanup.done",
candidates_removed=removed, fav_actions_expired=expired,
)
+ _record_task(
+ repo, "candidate_cleanup", ok=True, started_at=started,
+ finished_at=int(time.time()),
+ payload={
+ "candidates_removed": removed,
+ "fav_actions_expired": expired,
+ },
+ )
return {
"candidates_removed": removed,
"fav_actions_expired": expired,
}
def run_snapshot():
- summary = snapshot_service.run_if_needed()
+ started = int(time.time())
+ try:
+ summary = snapshot_service.run_if_needed()
+ except Exception as exc: # noqa: BLE001
+ _record_task(
+ repo, "snapshot", ok=False, started_at=started,
+ finished_at=int(time.time()), error=str(exc),
+ )
+ raise
+ payload = summary.as_dict()
+ ok = bool(summary.completed and not summary.backup_error)
+ _record_task(
+ repo, "snapshot", ok=ok, started_at=summary.started_at,
+ finished_at=summary.finished_at or int(time.time()),
+ payload=payload, error=summary.backup_error,
+ )
last["snapshot"] = summary
- logger.info("snapshot.summary", **summary.as_dict())
+ logger.info("snapshot.summary", **payload)
return summary
def run_sync():
- summary = folder_service.run()
+ 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", **summary.as_dict())
+ logger.info("folder_sync.summary", **payload)
# 新入库/恢复的视频立即补当天快照，而不是等到次日。
if summary.added > 0:
run_snapshot()
@@ -416,6 +527,15 @@ def cmd_run(args, logger: JsonLogger) -> int:
],
logger=logger,
)
+ if serve_web and not args.once:
+ from .web.server import create_server, serve_forever
+
+ web_server = create_server(store, db, logger=logger)
+ threading.Thread(
+ target=lambda: serve_forever(web_server, logger=logger),
+ name="web-server",
+ daemon=True,
+ ).start()
if notifier is not None and not args.once:
poller = TelegramPoller(notifier, favorite_service, logger=logger)
threading.Thread(
@@ -442,6 +562,52 @@ def cmd_run(args, logger: JsonLogger) -> int:
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
+ finally:
+ if web_server is not None:
+ try:
+ web_server.shutdown()
+ finally:
+ web_server.server_close()
+ db.close()
+
+
+def _assert_preview_database(path: Path) -> None:
+ import sqlite3
+
+ if not path.exists():
+ raise ConfigError(f"预览数据库不存在: {path}")
+ uri = f"file:{path.resolve().as_posix()}?mode=ro"
+ try:
+ conn = sqlite3.connect(uri, uri=True)
+ except sqlite3.Error as exc:
+ raise ConfigError(f"无法读取预览数据库: {exc}") from exc
+ try:
+ row = conn.execute(
+ "SELECT v FROM kv WHERE k = 'preview:seeded'"
+ ).fetchone()
+ except sqlite3.Error as exc:
+ raise ConfigError(f"目标数据库不是预览库: {exc}") from exc
+ finally:
+ conn.close()
+ if row is None or str(row[0]) != "1":
+ raise ConfigError(
+ "目标数据库缺少 preview:seeded=1 标记；请先运行 "
+ "tools/seed_preview_db.py，或去掉 --preview"
+ )
+
+
+def cmd_serve(args, logger: JsonLogger) -> int:
+ from .web.server import create_server, serve_forever
+
+ store = ConfigStore(args.config, require_auth=False)
+ cfg = store.get()
+ if args.preview:
+ _assert_preview_database(cfg.storage.database_path)
+ db = _open_db(cfg, logger)
+ try:
+ server = create_server(store, db, preview=args.preview, logger=logger)
+ serve_forever(server, logger=logger)
+ return 0
finally:
db.close()
@@ -491,6 +657,14 @@ def main(argv: Sequence[str] | None = None) -> int:
args = _build_parser().parse_args(list(argv) if argv is not None else None)
logger = JsonLogger()
try:
+ if args.command is None:
+ # 无子命令时启动完整服务：只读 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":
diff --git a/reaction_board/web/__init__.py b/reaction_board/web/__init__.py
new file mode 100644
index 0000000..4ee913c
--- /dev/null
+++ b/reaction_board/web/__init__.py
@@ -0,0 +1,2 @@
+# -*- coding: utf-8 -*-
+"""只读 Web 看板。"""
diff --git a/reaction_board/web/queries.py b/reaction_board/web/queries.py
new file mode 100644
index 0000000..bea885a
--- /dev/null
+++ b/reaction_board/web/queries.py
@@ -0,0 +1,1090 @@
+# -*- coding: utf-8 -*-
+"""只读看板查询层。
+
+所有 SQL 都在这里集中，Web 层只负责参数解析与 JSON 序列化。查询不会写入
+数据库，也不会访问 B 站。
+"""
+
+from __future__ import annotations
+
+import json
+import math
+import sqlite3
+import time
+from datetime import date, timedelta
+from typing import Any, Iterable, Mapping, Sequence
+
+from ..analytics import (
+ DEFAULT_HEAT_WEIGHTS,
+ METRICS,
+ MIN_VIEW_FOR_GROWTH_RATE,
+ compute_growth,
+ heat_scores,
+ interaction_rate,
+ normalize_cover,
+ score_percentiles,
+)
+from ..config import ConfigStore
+from ..database import Database
+from ..timeutil import day_bounds, next_daily_run, shanghai_date, shanghai_datetime
+
+TRACKED_STATUSES = ("active", "unfaved")
+VIDEO_SORTS = {
+ "heat", "view", "like", "coin", "favorite", "share", "reply", "danmaku",
+ "interaction", "growth", "growth_rate", "added", "published",
+}
+UP_SORTS = {"views", "count", "avg_views", "interaction", "avg_interaction", "latest"}
+CAPACITY = 1000
+CAPACITY_WARNING = 950
+TRACK_FANS_NAMES = {37754047: "咻咻满"}
+
+
+def _chunks(values: Sequence[int], size: int = 400) -> Iterable[Sequence[int]]:
+ for index in range(0, len(values), size):
+ yield values[index:index + size]
+
+
+def _date_minus(day: str, days: int) -> str:
+ return (date.fromisoformat(day) - timedelta(days=int(days))).isoformat()
+
+
+def _date_range(start: str, end: str) -> list[str]:
+ first = date.fromisoformat(start)
+ last = date.fromisoformat(end)
+ if last < first:
+ return []
+ return [
+ (first + timedelta(days=offset)).isoformat()
+ for offset in range((last - first).days + 1)
+ ]
+
+
+def _ts_date(timestamp: int | float | None) -> str | None:
+ if timestamp is None:
+ return None
+ return shanghai_date(float(timestamp))
+
+
+def _like_pattern(value: str) -> str:
+ escaped = value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
+ return f"%{escaped}%"
+
+
+def _metric_map(row: Mapping[str, Any] | None) -> dict[str, int] | None:
+ if row is None:
+ return None
+ result: dict[str, int] = {}
+ for metric in METRICS:
+ value = row.get(metric)
+ result[metric] = 0 if value is None else int(value)
+ return result
+
+
+def _safe_float(value: Any) -> float | None:
+ if value is None or isinstance(value, bool):
+ return None
+ try:
+ return float(value)
+ except (TypeError, ValueError):
+ return None
+
+
+class DashboardQueries:
+ def __init__(
+ self,
+ db: Database,
+ config_store: ConfigStore,
+ *,
+ clock=time.time,
+ ) -> None:
+ self.db = db
+ self.config_store = config_store
+ self._clock = clock
+
+ # ------------------------------------------------------------ helpers
+
+ def _now(self) -> float:
+ return float(self._clock())
+
+ def _today(self) -> str:
+ return shanghai_date(self._now())
+
+ @staticmethod
+ def _latest_date(conn: sqlite3.Connection) -> str | None:
+ row = conn.execute(
+ "SELECT MAX(stat_date) AS d FROM snapshot WHERE ok = 1"
+ ).fetchone()
+ return None if row is None or row["d"] is None else str(row["d"])
+
+ @staticmethod
+ def _snapshot_days(conn: sqlite3.Connection) -> int:
+ row = conn.execute(
+ "SELECT COUNT(DISTINCT stat_date) AS n FROM snapshot WHERE ok = 1"
+ ).fetchone()
+ return int(row["n"] or 0)
+
+ @staticmethod
+ def _snapshot_map(
+ conn: sqlite3.Connection,
+ aids: Sequence[int],
+ *,
+ before: str | None = None,
+ before_inclusive: bool = False,
+ ) -> dict[int, dict[str, Any]]:
+ if not aids:
+ return {}
+ result: dict[int, dict[str, Any]] = {}
+ for chunk in _chunks(list(aids)):
+ placeholders = ",".join("?" for _ in chunk)
+ if before is None:
+ sql = f"""
+ SELECT s.* FROM snapshot s
+ JOIN (
+ SELECT aid, MAX(stat_date) AS stat_date
+ FROM snapshot WHERE ok = 1 AND aid IN ({placeholders})
+ GROUP BY aid
+ ) m ON m.aid = s.aid AND m.stat_date = s.stat_date
+ """
+ params: list[Any] = list(chunk)
+ else:
+ operator = "<=" if before_inclusive else "<"
+ sql = f"""
+ SELECT s.* FROM snapshot s
+ JOIN (
+ SELECT aid, MAX(stat_date) AS stat_date
+ FROM snapshot
+ WHERE ok = 1 AND aid IN ({placeholders}) AND stat_date {operator} ?
+ GROUP BY aid
+ ) m ON m.aid = s.aid AND m.stat_date = s.stat_date
+ """
+ params = [*chunk, before]
+ for row in conn.execute(sql, params).fetchall():
+ data = dict(row)
+ result[int(data["aid"])] = data
+ return result
+
+ @staticmethod
+ def _videos(
+ conn: sqlite3.Connection,
+ *,
+ statuses: Sequence[str] = TRACKED_STATUSES,
+ folder_id: int | None = None,
+ q: str | None = None,
+ within_field: str | None = None,
+ within_start: str | None = None,
+ up_mid: int | None = None,
+ ) -> list[dict[str, Any]]:
+ clauses = [f"v.status IN ({','.join('?' for _ in statuses)})"]
+ params: list[Any] = list(statuses)
+ if folder_id is not None:
+ clauses.append("v.folder_id = ?")
+ params.append(int(folder_id))
+ if up_mid is not None:
+ clauses.append("v.up_mid = ?")
+ params.append(int(up_mid))
+ if q:
+ pattern = _like_pattern(q)
+ clauses.append(
+ "(v.title LIKE ? ESCAPE '\\' OR v.up_name LIKE ? ESCAPE '\\')"
+ )
+ params.extend([pattern, pattern])
+ if within_field and within_start:
+ if within_field not in {"pubdate", "first_seen"}:
+ raise ValueError("within_field 必须是 pubdate 或 first_seen")
+ start_ts, _ = day_bounds(within_start)
+ clauses.append(f"v.{within_field} >= ?")
+ params.append(start_ts)
+ sql = (
+ "SELECT v.*, f.title AS folder_title, f.year AS folder_year "
+ "FROM video v LEFT JOIN folder f ON f.media_id = v.folder_id "
+ f"WHERE {' AND '.join(clauses)} ORDER BY v.aid"
+ )
+ return [dict(row) for row in conn.execute(sql, params).fetchall()]
+
+ def _global_heat(
+ self, conn: sqlite3.Connection
+ ) -> tuple[list[dict[str, Any]], dict[int, dict[str, Any]], dict[int, float], dict[int, float]]:
+ videos = self._videos(conn)
+ snapshots = self._snapshot_map(conn, [int(v["aid"]) for v in videos])
+ rows = []
+ for video in videos:
+ snapshot = snapshots.get(int(video["aid"]))
+ if snapshot is None:
+ continue
+ item = {"aid": int(video["aid"])}
+ item.update(_metric_map(snapshot) or {})
+ rows.append(item)
+ weights = self.config_store.get().heat_weights
+ scores = heat_scores(rows, weights or DEFAULT_HEAT_WEIGHTS)
+ percentiles = score_percentiles(scores)
+ return videos, snapshots, scores, percentiles
+
+ @staticmethod
+ def _growth_for(
+ snapshot: Mapping[str, Any] | None,
+ previous: Mapping[str, Any] | None,
+ metric: str = "view",
+ ):
+ if snapshot is None:
+ return compute_growth(None, None, None)
+ current = snapshot.get(metric)
+ before = None if previous is None else previous.get(metric)
+ gap = None
+ if previous is not None and snapshot.get("stat_date") and previous.get("stat_date"):
+ try:
+ gap = (
+ date.fromisoformat(str(snapshot["stat_date"]))
+ - date.fromisoformat(str(previous["stat_date"]))
+ ).days
+ except ValueError:
+ gap = None
+ return compute_growth(current, before, gap)
+
+ def _item(
+ self,
+ video: Mapping[str, Any],
+ snapshot: Mapping[str, Any] | None,
+ *,
+ heat: float | None = None,
+ percentile: float | None = None,
+ previous: Mapping[str, Any] | None = None,
+ include_growth: bool = False,
+ ) -> dict[str, Any]:
+ metrics = _metric_map(snapshot)
+ item: dict[str, Any] = {
+ "aid": int(video["aid"]),
+ "bvid": video.get("bvid"),
+ "title": video.get("title") or "(无标题)",
+ "up_mid": video.get("up_mid"),
+ "up_name": video.get("up_name"),
+ "pubdate": int(video.get("pubdate") or 0),
+ "duration": video.get("duration"),
+ "cover": normalize_cover(video.get("cover")),
+ "folder_id": video.get("folder_id"),
+ "folder_title": video.get("folder_title"),
+ "folder_year": video.get("folder_year"),
+ "status": video.get("status"),
+ "first_seen": int(video.get("first_seen") or 0),
+ "attr": video.get("attr"),
+ "metrics": metrics,
+ "snapshot_date": None if snapshot is None else snapshot.get("stat_date"),
+ "heat": None if heat is None else round(float(heat), 3),
+ "heat_percentile": percentile,
+ "interaction_rate": None if metrics is None else interaction_rate(metrics),
+ }
+ if include_growth:
+ item["growth"] = self._growth_for(snapshot, previous).as_dict()
+ return item
+
+ # ------------------------------------------------------------ overview
+
+ def overview(self) -> dict[str, Any]:
+ def run(conn: sqlite3.Connection) -> dict[str, Any]:
+ latest = self._latest_date(conn)
+ snapshot_days = self._snapshot_days(conn)
+ counts_rows = conn.execute(
+ "SELECT status, COUNT(*) AS n FROM video GROUP BY status"
+ ).fetchall()
+ by_status = {str(row["status"]): int(row["n"]) for row in counts_rows}
+ videos = self._videos(conn)
+ aids = [int(video["aid"]) for video in videos]
+ snapshots = self._snapshot_map(conn, aids)
+ previous = self._snapshot_map(conn, aids, before=latest or self._today())
+ totals = {metric: 0 for metric in METRICS}
+ for video in videos:
+ metrics = _metric_map(snapshots.get(int(video["aid"])))
+ if metrics:
+ for metric in METRICS:
+ totals[metric] += metrics[metric]
+
+ view_delta: int | None = None
+ available_deltas = 0
+ view_delta_sum = 0
+ for video in videos:
+ aid = int(video["aid"])
+ snapshot = snapshots.get(aid)
+ prior = previous.get(aid)
+ if snapshot is None or prior is None:
+ continue
+ growth = self._growth_for(snapshot, prior)
+ if growth.available and growth.delta is not None:
+ view_delta_sum += growth.delta
+ available_deltas += 1
+ if available_deltas:
+ view_delta = view_delta_sum
+
+ first_seen_row = conn.execute(
+ "SELECT MIN(first_seen) AS ts FROM video"
+ ).fetchone()
+ first_seen_ts = first_seen_row["ts"] if first_seen_row else None
+ first_seen_date = _ts_date(first_seen_ts)
+ initial_import_count = 0
+ if first_seen_ts is not None:
+ initial_import_count = int(conn.execute(
+ "SELECT COUNT(*) FROM video WHERE first_seen = ?",
+ (int(first_seen_ts),),
+ ).fetchone()[0])
+
+ start_ts, end_ts = day_bounds(self._today())
+ added_today = int(conn.execute(
+ "SELECT COUNT(*) FROM video WHERE first_seen >= ? AND first_seen < ?",
+ (start_ts, end_ts),
+ ).fetchone()[0])
+ is_first_day = snapshot_days <= 1
+ if is_first_day:
+ added_today = 0
+
+ snapshot_health = {"total": 0, "ok": 0, "failed": 0}
+ if latest:
+ rows = conn.execute(
+ "SELECT ok, COUNT(*) AS n FROM snapshot WHERE stat_date = ? GROUP BY ok",
+ (latest,),
+ ).fetchall()
+ by_ok = {int(row["ok"]): int(row["n"]) for row in rows}
+ snapshot_health = {
+ "total": sum(by_ok.values()),
+ "ok": by_ok.get(1, 0),
+ "failed": by_ok.get(0, 0),
+ }
+
+ candidate_pending = int(conn.execute(
+ "SELECT COUNT(*) FROM candidate WHERE decision IS NULL"
+ ).fetchone()[0])
+ candidate_total = int(conn.execute(
+ "SELECT COUNT(*) FROM candidate"
+ ).fetchone()[0])
+
+ cfg = self.config_store.get()
+ fans = self._fans_overview(conn, cfg.track_fans, latest, snapshot_days)
+ folders = self._folders_overview(conn)
+ tasks = self._task_records(conn)
+ auth = {
+ "ok": self._kv(conn, "auth:ok"),
+ "message": self._kv(conn, "auth:message"),
+ "last_check": self._kv(conn, "auth:last_check"),
+ "mid": self._kv(conn, "auth:mid"),
+ }
+ preview = self._kv(conn, "preview:seeded") == "1"
+ snapshot_completed_today = self._kv(
+ conn, f"snapshot:completed:{self._today()}"
+ ) == "1"
+ return {
+ "generated_at": int(self._now()),
+ "today": self._today(),
+ "data_date": latest,
+ "snapshot_days": snapshot_days,
+ "first_seen_date": first_seen_date,
+ "is_first_day": is_first_day,
+ "initial_import_count": initial_import_count,
+ "added_today": added_today,
+ "counts": {
+ **by_status,
+ "active_ups": int(conn.execute(
+ "SELECT COUNT(DISTINCT up_mid) FROM video "
+ "WHERE status IN ('active','unfaved') AND up_mid IS NOT NULL"
+ ).fetchone()[0]),
+ "candidates_pending": candidate_pending,
+ "candidates_total": candidate_total,
+ },
+ "totals": totals,
+ "interactions": totals["like"] + totals["coin"] + totals["favorite"],
+ "view_delta": view_delta,
+ "view_delta_available": view_delta is not None,
+ "snapshot_health": snapshot_health,
+ "snapshot_completed_today": snapshot_completed_today,
+ "fans": fans,
+ "folders": folders,
+ "tasks": tasks,
+ "auth": auth,
+ "preview": preview,
+ "capacity": CAPACITY,
+ "capacity_warning": CAPACITY_WARNING,
+ }
+ return self.db.read(run)
+
+ @staticmethod
+ def _kv(conn: sqlite3.Connection, key: str) -> str | None:
+ row = conn.execute("SELECT v FROM kv WHERE k = ?", (key,)).fetchone()
+ return None if row is None else str(row["v"])
+
+ def _task_records(self, conn: sqlite3.Connection) -> dict[str, Any]:
+ rows = conn.execute(
+ "SELECT k, v FROM kv WHERE k LIKE 'task:last:%' ORDER BY k"
+ ).fetchall()
+ result: dict[str, Any] = {}
+ for row in rows:
+ name = str(row["k"]).removeprefix("task:last:")
+ try:
+ value = json.loads(str(row["v"]))
+ except (TypeError, json.JSONDecodeError):
+ value = {"raw": row["v"]}
+ result[name] = value
+ return result
+
+ def _folders_overview(self, conn: sqlite3.Connection) -> list[dict[str, Any]]:
+ rows = conn.execute(
+ """
+ SELECT f.*,
+ (SELECT COUNT(*) FROM folder_resource fr
+ WHERE fr.media_id = f.media_id AND fr.active = 1) AS local_count
+ FROM folder f ORDER BY f.year DESC, f.media_id DESC
+ """
+ ).fetchall()
+ result = []
+ for row in rows:
+ media_count = row["media_count"]
+ count = int(media_count) if media_count is not None else int(row["local_count"] or 0)
+ result.append({
+ "media_id": int(row["media_id"]),
+ "year": int(row["year"]),
+ "title": row["title"] or "",
+ "media_count": count,
+ "local_count": int(row["local_count"] or 0),
+ "last_ids_at": row["last_ids_at"],
+ "last_full_at": row["last_full_at"],
+ "enabled": bool(row["enabled"]),
+ "capacity": CAPACITY,
+ "ratio": min(1.0, count / CAPACITY),
+ "warning": count >= CAPACITY_WARNING,
+ })
+ return result
+
+ def _fans_overview(
+ self,
+ conn: sqlite3.Connection,
+ mid_list: Sequence[int],
+ latest: str | None,
+ snapshot_days: int,
+ ) -> dict[str, Any] | None:
+ mids = list(mid_list) or [
+ int(row["mid"])
+ for row in conn.execute("SELECT DISTINCT mid FROM up_stat ORDER BY mid").fetchall()
+ ]
+ if not mids:
+ return None
+ mid = int(mids[0])
+ rows = conn.execute(
+ "SELECT * FROM up_stat WHERE mid = ? ORDER BY stat_date", (mid,)
+ ).fetchall()
+ if not rows:
+ return None
+ ok_rows = [dict(row) for row in rows if int(row["ok"]) == 1]
+ if not ok_rows:
+ return None
+ current = ok_rows[-1]
+ previous = ok_rows[-2] if len(ok_rows) >= 2 else None
+ baseline = None
+ if latest:
+ cutoff = _date_minus(latest, 7)
+ candidates = [row for row in ok_rows if str(row["stat_date"]) <= cutoff]
+ baseline = candidates[-1] if candidates else None
+ first = ok_rows[0]
+ return {
+ "mid": mid,
+ "name": TRACK_FANS_NAMES.get(mid, f"UP {mid}"),
+ "stat_date": current["stat_date"],
+ "follower": current["follower"],
+ "following": current["following"],
+ "delta_1d": None if previous is None else int(current["follower"] or 0) - int(previous["follower"] or 0),
+ "delta_7d": None if baseline is None else int(current["follower"] or 0) - int(baseline["follower"] or 0),
+ "delta_cum": int(current["follower"] or 0) - int(first["follower"] or 0),
+ "first_date": first["stat_date"],
+ "snapshot_days": snapshot_days,
+ "points": [
+ {
+ "date": row["stat_date"],
+ "follower": row["follower"] if int(row["ok"]) == 1 else None,
+ "ok": bool(row["ok"]),
+ }
+ for row in rows[-30:]
+ ],
+ }
+
+ # ------------------------------------------------------------ videos
+
+ def list_videos(
+ self,
+ *,
+ sort: str = "heat",
+ within_days: int | None = None,
+ growth_days: int = 7,
+ folder_id: int | None = None,
+ q: str | None = None,
+ page: int = 1,
+ page_size: int = 20,
+ export: bool = False,
+ ) -> dict[str, Any]:
+ if sort not in VIDEO_SORTS:
+ raise ValueError(f"不支持的排序: {sort}")
+ if growth_days < 1 or growth_days > 365:
+ raise ValueError("growth_days 必须在 1~365 之间")
+ page = max(1, int(page))
+ page_size = max(1, min(int(page_size), 200 if not export else 10_000))
+
+ def run(conn: sqlite3.Connection) -> dict[str, Any]:
+ latest = self._latest_date(conn) or self._today()
+ within_field = None
+ if within_days is not None:
+ if within_days < 1:
+ raise ValueError("within_days 必须大于 0")
+ within_field = "first_seen" if sort == "added" else "pubdate"
+ videos = self._videos(
+ conn,
+ folder_id=folder_id,
+ q=q,
+ within_field=within_field,
+ within_start=(
+ _date_minus(latest, int(within_days) - 1)
+ if within_days is not None else None
+ ),
+ )
+ aids = [int(video["aid"]) for video in videos]
+ snapshots = self._snapshot_map(conn, aids)
+ _, _, heat, percentiles = self._global_heat(conn)
+
+ previous: dict[int, dict[str, Any]] = {}
+ if sort in {"growth", "growth_rate", "interaction"}:
+ baseline_date = _date_minus(latest, growth_days)
+ previous = self._snapshot_map(conn, aids, before=baseline_date, before_inclusive=True)
+ elif sort in {"heat", "view", "like", "coin", "favorite", "share", "reply", "danmaku"}:
+ previous = {}
+
+ items = [
+ self._item(
+ video,
+ snapshots.get(int(video["aid"])),
+ heat=heat.get(int(video["aid"])),
+ percentile=percentiles.get(int(video["aid"])),
+ previous=previous.get(int(video["aid"])),
+ include_growth=sort in {"growth", "growth_rate"},
+ )
+ for video in videos
+ ]
+
+ if sort == "interaction":
+ items = [item for item in items if item["interaction_rate"] is not None]
+ if sort in {"growth", "growth_rate"}:
+ items = [
+ item for item in items
+ if item.get("growth", {}).get("available")
+ ]
+ if sort == "growth_rate":
+ items = [
+ item for item in items
+ if (item["metrics"] or {}).get("view", 0) >= MIN_VIEW_FOR_GROWTH_RATE
+ ]
+
+ def sort_key(item: Mapping[str, Any]):
+ if sort == "added":
+ return (int(item["first_seen"]), int(item["pubdate"]))
+ if sort == "published":
+ return (int(item["pubdate"]), int(item["first_seen"]))
+ if sort == "heat":
+ return item["heat"] if item["heat"] is not None else float("-inf")
+ if sort == "interaction":
+ return item["interaction_rate"] if item["interaction_rate"] is not None else float("-inf")
+ if sort in {"growth", "growth_rate"}:
+ field = "delta" if sort == "growth" else "rate"
+ value = item.get("growth", {}).get(field)
+ return value if value is not None else float("-inf")
+ value = (item["metrics"] or {}).get(sort)
+ return value if value is not None else -1
+
+ items.sort(key=lambda item: item["aid"])
+ items.sort(key=sort_key, reverse=True)
+ total = len(items)
+ if not export:
+ start = (page - 1) * page_size
+ items = items[start:start + page_size]
+ available = True
+ reason = None
+ if sort in {"growth", "growth_rate"} and not any(
+ item.get("growth", {}).get("available") for item in items
+ ):
+ available = False
+ reason = "需要至少两天的快照才能计算增长"
+ return {
+ "sort": sort,
+ "within_days": within_days,
+ "growth_days": growth_days,
+ "page": page,
+ "page_size": page_size,
+ "total": total,
+ "pages": max(1, math.ceil(total / page_size)) if total else 0,
+ "items": items,
+ "available": available,
+ "reason": reason,
+ }
+ return self.db.read(run)
+
+ def video_detail(self, aid: int) -> dict[str, Any] | None:
+ aid = int(aid)
+
+ def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
+ video_rows = self._videos(conn)
+ video = next((item for item in video_rows if int(item["aid"]) == aid), None)
+ if video is None:
+ row = conn.execute(
+ "SELECT v.*, f.title AS folder_title, f.year AS folder_year "
+ "FROM video v LEFT JOIN folder f ON f.media_id = v.folder_id "
+ "WHERE v.aid = ?",
+ (aid,),
+ ).fetchone()
+ if row is None:
+ return None
+ video = dict(row)
+ snapshots = self._snapshot_map(conn, [aid])
+ snapshot = snapshots.get(aid)
+ previous = self._snapshot_map(conn, [aid], before=snapshot["stat_date"] if snapshot else None).get(aid)
+ _, _, heat, percentiles = self._global_heat(conn)
+ growth = self._growth_for(snapshot, previous)
+ item = self._item(
+ video,
+ snapshot,
+ heat=heat.get(aid),
+ percentile=percentiles.get(aid),
+ )
+ item["growth"] = growth.as_dict()
+ item["snapshot_count"] = int(conn.execute(
+ "SELECT COUNT(*) FROM snapshot WHERE aid = ?", (aid,)
+ ).fetchone()[0])
+ item["blacklisted"] = bool(conn.execute(
+ "SELECT 1 FROM up_blacklist WHERE mid = ?", (video.get("up_mid"),)
+ ).fetchone()) if video.get("up_mid") is not None else False
+ item["url"] = (
+ f"https://www.bilibili.com/video/{video.get('bvid')}"
+ if video.get("bvid") else None
+ )
+ return item
+ return self.db.read(run)
+
+ def video_trend(self, aid: int, *, days: int | None = 30) -> dict[str, Any]:
+ aid = int(aid)
+ if days is not None and days < 1:
+ raise ValueError("days 必须大于 0")
+
+ def run(conn: sqlite3.Connection) -> dict[str, Any]:
+ latest = self._latest_date(conn)
+ rows = conn.execute(
+ "SELECT * FROM snapshot WHERE aid = ? ORDER BY stat_date", (aid,)
+ ).fetchall()
+ points = [dict(row) for row in rows]
+ if latest is None or not points:
+ return {
+ "aid": aid,
+ "days": days,
+ "dates": [],
+ "series": {metric: [] for metric in METRICS},
+ "points": [],
+ "available": False,
+ }
+ start = _date_minus(latest, days - 1) if days is not None else str(points[0]["stat_date"])
+ dates = _date_range(start, latest)
+ by_date = {str(point["stat_date"]): point for point in points}
+ series: dict[str, list[int | None]] = {}
+ for metric in METRICS:
+ values: list[int | None] = []
+ for day in dates:
+ point = by_date.get(day)
+ if point is None or int(point["ok"]) != 1 or point.get(metric) is None:
+ values.append(None)
+ else:
+ values.append(int(point[metric]))
+ series[metric] = values
+ filtered_points = [
+ point for point in points if str(point["stat_date"]) in set(dates)
+ ]
+ return {
+ "aid": aid,
+ "days": days,
+ "dates": dates,
+ "series": series,
+ "points": filtered_points,
+ "available": any(point["ok"] == 1 for point in filtered_points),
+ }
+ return self.db.read(run)
+
+ # ------------------------------------------------------------ ups
+
+ def list_ups(
+ self,
+ *,
+ sort: str = "views",
+ within_days: int | None = None,
+ q: str | None = None,
+ min_videos: int = 1,
+ page: int = 1,
+ page_size: int = 20,
+ export: bool = False,
+ ) -> dict[str, Any]:
+ if sort not in UP_SORTS:
+ raise ValueError(f"不支持的 UP 排序: {sort}")
+ page = max(1, int(page))
+ page_size = max(1, min(int(page_size), 200 if not export else 10_000))
+ min_videos = max(1, int(min_videos))
+
+ def run(conn: sqlite3.Connection) -> dict[str, Any]:
+ latest = self._latest_date(conn) or self._today()
+ videos = self._videos(
+ conn,
+ q=q,
+ within_field="pubdate" if within_days is not None else None,
+ within_start=(
+ _date_minus(latest, int(within_days) - 1)
+ if within_days is not None else None
+ ),
+ )
+ snapshots = self._snapshot_map(conn, [int(video["aid"]) for video in videos])
+ groups: dict[int, dict[str, Any]] = {}
+ for video in videos:
+ mid = video.get("up_mid")
+ if mid is None:
+ continue
+ mid = int(mid)
+ group = groups.setdefault(mid, {
+ "up_mid": mid,
+ "up_name": video.get("up_name") or f"UP {mid}",
+ "videos": [],
+ })
+ metrics = _metric_map(snapshots.get(int(video["aid"])))
+ group["videos"].append((video, metrics))
+ items = []
+ for group in groups.values():
+ group_videos = group["videos"]
+ metrics_list = [metrics for _, metrics in group_videos if metrics]
+ total_views = sum(metrics["view"] for metrics in metrics_list)
+ total_interactions = sum(
+ metrics["like"] + metrics["coin"] + metrics["favorite"]
+ for metrics in metrics_list
+ )
+ rates = [
+ interaction_rate(metrics)
+ for metrics in metrics_list
+ if interaction_rate(metrics) is not None
+ ]
+ best = max(
+ group_videos,
+ key=lambda pair: (pair[1] or {}).get("view", 0),
+ default=None,
+ )
+ items.append({
+ "up_mid": group["up_mid"],
+ "up_name": group["up_name"],
+ "video_count": len(group_videos),
+ "total_views": total_views,
+ "avg_views": round(total_views / len(metrics_list), 1) if metrics_list else None,
+ "total_interactions": total_interactions,
+ "avg_interaction": round(sum(rates) / len(rates), 6) if rates else None,
+ "latest_pubdate": max(int(video["pubdate"]) for video, _ in group_videos),
+ "first_pubdate": min(int(video["pubdate"]) for video, _ in group_videos),
+ "best_video": None if best is None else {
+ "aid": int(best[0]["aid"]),
+ "title": best[0].get("title"),
+ "cover": normalize_cover(best[0].get("cover")),
+ "view": (best[1] or {}).get("view"),
+ },
+ })
+ items = [item for item in items if item["video_count"] >= min_videos]
+ sort_map = {
+ "views": lambda item: item["total_views"],
+ "count": lambda item: item["video_count"],
+ "avg_views": lambda item: item["avg_views"] or 0,
+ "interaction": lambda item: item["total_interactions"],
+ "avg_interaction": lambda item: item["avg_interaction"] or 0,
+ "latest": lambda item: item["latest_pubdate"],
+ }
+ key = sort_map[sort]
+ items.sort(key=lambda item: (key(item), item["up_mid"]), reverse=True)
+ total = len(items)
+ if not export:
+ start = (page - 1) * page_size
+ items = items[start:start + page_size]
+ return {
+ "sort": sort,
+ "within_days": within_days,
+ "min_videos": min_videos,
+ "page": page,
+ "page_size": page_size,
+ "total": total,
+ "pages": max(1, math.ceil(total / page_size)) if total else 0,
+ "items": items,
+ }
+ return self.db.read(run)
+
+ def up_detail(self, mid: int) -> dict[str, Any] | None:
+ mid = int(mid)
+
+ def run(conn: sqlite3.Connection) -> dict[str, Any] | None:
+ latest = self._latest_date(conn) or self._today()
+ videos = self._videos(conn, up_mid=mid)
+ if not videos:
+ row = conn.execute(
+ "SELECT 1 FROM up_blacklist WHERE mid = ?", (mid,)
+ ).fetchone()
+ if row is None:
+ return None
+ return {
+ "up_mid": mid,
+ "up_name": f"UP {mid}",
+ "videos": [],
+ "video_count": 0,
+ "blacklisted": True,
+ "series": {"dates": [], "views": []},
+ }
+ snapshots = self._snapshot_map(conn, [int(video["aid"]) for video in videos])
+ items = [
+ self._item(video, snapshots.get(int(video["aid"])))
+ for video in videos
+ ]
+ metrics_list = [item["metrics"] for item in items if item["metrics"]]
+ rates = [item["interaction_rate"] for item in items if item["interaction_rate"] is not None]
+ total_views = sum(metrics["view"] for metrics in metrics_list)
+ total_interactions = sum(
+ metrics["like"] + metrics["coin"] + metrics["favorite"]
+ for metrics in metrics_list
+ )
+ aids = [int(video["aid"]) for video in videos]
+ rows = conn.execute(
+ f"SELECT stat_date, SUM(view) AS view FROM snapshot "
+ f"WHERE ok = 1 AND aid IN ({','.join('?' for _ in aids)}) "
+ "GROUP BY stat_date ORDER BY stat_date",
+ aids,
+ ).fetchall()
+ view_by_date = {str(row["stat_date"]): int(row["view"] or 0) for row in rows}
+ dates = _date_range(min(view_by_date), latest) if view_by_date else []
+ items.sort(key=lambda item: ((item["metrics"] or {}).get("view", 0), item["pubdate"]), reverse=True)
+ folder_titles = sorted({
+ str(item["folder_title"]) for item in items if item.get("folder_title")
+ })
+ return {
+ "up_mid": mid,
+ "up_name": items[0].get("up_name") or f"UP {mid}",
+ "video_count": len(items),
+ "total_views": total_views,
+ "avg_views": round(total_views / len(metrics_list), 1) if metrics_list else None,
+ "total_interactions": total_interactions,
+ "avg_interaction": round(sum(rates) / len(rates), 6) if rates else None,
+ "first_pubdate": min(item["pubdate"] for item in items),
+ "latest_pubdate": max(item["pubdate"] for item in items),
+ "folders": folder_titles,
+ "blacklisted": bool(conn.execute(
+ "SELECT 1 FROM up_blacklist WHERE mid = ?", (mid,)
+ ).fetchone()),
+ "space_url": f"https://space.bilibili.com/{mid}",
+ "videos": items,
+ "series": {
+ "dates": dates,
+ "views": [view_by_date.get(day) for day in dates],
+ },
+ }
+ return self.db.read(run)
+
+ # ------------------------------------------------------------ fans
+
+ def fans(self, mid: int | None = None, *, days: int = 90) -> dict[str, Any]:
+ if days < 1:
+ raise ValueError("days 必须大于 0")
+
+ def run(conn: sqlite3.Connection) -> dict[str, Any]:
+ cfg = self.config_store.get()
+ mids = list(cfg.track_fans)
+ if not mids:
+ mids = [
+ int(row["mid"]) for row in conn.execute(
+ "SELECT DISTINCT mid FROM up_stat ORDER BY mid"
+ ).fetchall()
+ ]
+ selected = int(mid) if mid is not None else (mids[0] if mids else None)
+ options = [
+ {"mid": value, "name": TRACK_FANS_NAMES.get(value, f"UP {value}")}
+ for value in mids
+ ]
+ if selected is None:
+ return {
+ "mid": None, "name": None, "available": False,
+ "options": options, "dates": [], "series": [],
+ "points": [], "days": days,
+ }
+ rows = [
+ dict(row) for row in conn.execute(
+ "SELECT * FROM up_stat WHERE mid = ? ORDER BY stat_date",
+ (selected,),
+ ).fetchall()
+ ]
+ if not rows:
+ return {
+ "mid": selected,
+ "name": TRACK_FANS_NAMES.get(selected, f"UP {selected}"),
+ "available": False,
+ "options": options,
+ "dates": [],
+ "series": [],
+ "points": [],
+ "days": days,
+ }
+ latest = max(str(row["stat_date"]) for row in rows if int(row["ok"]) == 1) if any(int(row["ok"]) == 1 for row in rows) else str(rows[-1]["stat_date"])
+ start = _date_minus(latest, days - 1)
+ dates = _date_range(start, latest)
+ by_date = {str(row["stat_date"]): row for row in rows}
+ series = []
+ for day in dates:
+ row = by_date.get(day)
+ if row is None or int(row["ok"]) != 1:
+ series.append(None)
+ else:
+ series.append(row["follower"])
+ ok_rows = [row for row in rows if int(row["ok"]) == 1]
+ current = ok_rows[-1] if ok_rows else rows[-1]
+ previous = ok_rows[-2] if len(ok_rows) >= 2 else None
+ cutoff = _date_minus(latest, 7)
+ baselines = [row for row in ok_rows if str(row["stat_date"]) <= cutoff]
+ baseline = baselines[-1] if baselines else None
+ first = ok_rows[0] if ok_rows else rows[0]
+ return {
+ "mid": selected,
+ "name": TRACK_FANS_NAMES.get(selected, f"UP {selected}"),
+ "available": bool(ok_rows),
+ "options": options,
+ "days": days,
+ "dates": dates,
+ "series": series,
+ "points": rows,
+ "stat_date": current["stat_date"],
+ "follower": current.get("follower"),
+ "following": current.get("following"),
+ "delta_1d": None if previous is None else int(current.get("follower") or 0) - int(previous.get("follower") or 0),
+ "delta_7d": None if baseline is None else int(current.get("follower") or 0) - int(baseline.get("follower") or 0),
+ "delta_cum": int(current.get("follower") or 0) - int(first.get("follower") or 0),
+ "first_date": first["stat_date"],
+ }
+ return self.db.read(run)
+
+ # ------------------------------------------------------------ candidates
+
+ def candidates(self) -> dict[str, Any]:
+ def run(conn: sqlite3.Connection) -> dict[str, Any]:
+ cfg = self.config_store.get()
+ now = int(self._now())
+ ttl = int(cfg.schedule.candidate_ttl_days)
+ rows = conn.execute(
+ "SELECT * FROM candidate ORDER BY "
+ "CASE WHEN decision IS NULL THEN 0 ELSE 1 END, "
+ "COALESCE(score, 0) DESC, first_seen DESC, aid"
+ ).fetchall()
+ items = []
+ for row in rows:
+ data = dict(row)
+ first_seen = int(data.get("first_seen") or 0)
+ expires_at = first_seen + ttl * 86400
+ items.append({
+ **data,
+ "cover": normalize_cover(data.get("cover")),
+ "expires_at": expires_at,
+ "days_left": max(0, math.ceil((expires_at - now) / 86400)),
+ "url": (
+ f"https://www.bilibili.com/video/{data.get('bvid')}"
+ if data.get("bvid") else None
+ ),
+ })
+ pending = sum(1 for item in items if item.get("decision") is None)
+ return {
+ "items": items,
+ "total": len(items),
+ "pending": pending,
+ "ttl_days": ttl,
+ "telegram_enabled": bool(cfg.telegram.enabled),
+ "now": now,
+ }
+ return self.db.read(run)
+
+ # ------------------------------------------------------------ system
+
+ def system(self) -> dict[str, Any]:
+ def run(conn: sqlite3.Connection) -> dict[str, Any]:
+ cfg = self.config_store.get()
+ latest = self._latest_date(conn)
+ snapshot_days = self._snapshot_days(conn)
+ health = {"total": 0, "ok": 0, "failed": 0}
+ if latest:
+ rows = conn.execute(
+ "SELECT ok, COUNT(*) AS n FROM snapshot WHERE stat_date = ? GROUP BY ok",
+ (latest,),
+ ).fetchall()
+ by_ok = {int(row["ok"]): int(row["n"]) for row in rows}
+ health = {
+ "total": sum(by_ok.values()),
+ "ok": by_ok.get(1, 0),
+ "failed": by_ok.get(0, 0),
+ }
+ folders = self._folders_overview(conn)
+ blacklist = [
+ dict(row) for row in conn.execute(
+ """
+ SELECT b.*,
+ (SELECT COUNT(*) FROM video v WHERE v.up_mid = b.mid
+ AND v.status IN ('active','unfaved')) AS video_count
+ FROM up_blacklist b ORDER BY b.created_at, b.mid
+ """
+ ).fetchall()
+ ]
+ tasks = self._task_records(conn)
+ 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 in (("discover", "discover"), ("folder_sync", "folder_sync")):
+ record = tasks.get(key)
+ if isinstance(record, dict) and record.get("finished_at"):
+ next_runs[name] = int(record["finished_at"]) + (
+ cfg.schedule.discover_minutes if name == "discover"
+ else cfg.schedule.folder_sync_minutes
+ ) * 60
+ errors = [
+ {"task": name, **record}
+ for name, record in tasks.items()
+ if isinstance(record, dict) and record.get("ok") is False
+ ]
+ counts = {str(row["status"]): int(row["n"]) for row in conn.execute(
+ "SELECT status, COUNT(*) AS n FROM video GROUP BY status"
+ ).fetchall()}
+ return {
+ "generated_at": int(now),
+ "today": self._today(),
+ "data_date": latest,
+ "snapshot_days": snapshot_days,
+ "snapshot_health": health,
+ "counts": counts,
+ "folders": folders,
+ "blacklist": blacklist,
+ "tasks": tasks,
+ "errors": errors,
+ "next_runs": next_runs,
+ "auth": {
+ "ok": self._kv(conn, "auth:ok"),
+ "message": self._kv(conn, "auth:message"),
+ "last_check": self._kv(conn, "auth:last_check"),
+ "mid": self._kv(conn, "auth:mid"),
+ },
+ "backup": {
+ "last_completed": self._kv(conn, "snapshot:last_completed"),
+ "last_attempt": self._kv(conn, "snapshot:last_attempt"),
+ "ok": self._kv(conn, f"snapshot:backup:{latest}" if latest else "snapshot:backup"),
+ "error": self._kv(conn, f"snapshot:backup_error:{latest}" if latest else "snapshot:backup_error"),
+ },
+ "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,
+ },
+ "telegram_enabled": bool(cfg.telegram.enabled),
+ }
+ return self.db.read(run)
+
+ def export_videos(self, **filters) -> dict[str, Any]:
+ filters.pop("page", None)
+ filters.pop("page_size", None)
+ return self.list_videos(**filters, export=True)
diff --git a/reaction_board/web/server.py b/reaction_board/web/server.py
new file mode 100644
index 0000000..11b9a07
--- /dev/null
+++ b/reaction_board/web/server.py
@@ -0,0 +1,649 @@
+# -*- coding: utf-8 -*-
+"""只读看板 HTTP 服务。
+
+只注册 GET（以及登录所需的 POST /login）。数据管理仍由 Telegram、CLI 或
+数据库完成，Web 不提供任何业务写接口。
+"""
+
+from __future__ import annotations
+
+import csv
+import hashlib
+import hmac
+import io
+import json
+import mimetypes
+import os
+import time
+from dataclasses import dataclass
+from datetime import date
+from http import HTTPStatus
+from http.cookies import SimpleCookie
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any, Callable, Mapping
+from urllib.parse import parse_qs, unquote, urlsplit
+
+from ..analytics import is_loopback
+from ..config import ConfigStore
+from ..database import Database
+from ..timeutil import shanghai_date
+from .queries import DashboardQueries
+
+STATIC_ROOT = Path(__file__).resolve().parent / "static"
+COOKIE_NAME = "rb_session"
+SESSION_TTL = 7 * 86400
+
+
+class ApiError(Exception):
+ def __init__(self, status: int, code: str, message: str) -> None:
+ super().__init__(message)
+ self.status = status
+ self.code = code
+ self.message = message
+
+
+@dataclass
+class WebContext:
+ queries: DashboardQueries
+ config_store: ConfigStore
+ preview: bool = False
+ logger: Any = None
+ clock: Callable[[], float] = time.time
+ db_path: str | None = None
+ schema_version: int | None = None
+
+ @property
+ def auth_token(self) -> str:
+ return self.config_store.get().web.auth_token
+
+
+class DashboardHTTPServer(ThreadingHTTPServer):
+ daemon_threads = True
+ allow_reuse_address = True
+
+ def __init__(self, address, context: WebContext) -> None:
+ self.context = context
+ super().__init__(address, DashboardHandler)
+
+
+class DashboardHandler(BaseHTTPRequestHandler):
+ protocol_version = "HTTP/1.1"
+ server_version = "ReactionBoard/0.1"
+
+ # ------------------------------------------------------------ logging
+
+ def log_message(self, fmt: str, *args: Any) -> None:
+ # 请求级日志容易刷屏，只保留错误在 _send_error 里记录。
+ return
+
+ @property
+ def context(self) -> WebContext:
+ return self.server.context # type: ignore[attr-defined]
+
+ # ------------------------------------------------------------ HTTP
+
+ def do_GET(self) -> None:
+ self._dispatch("GET")
+
+ def do_HEAD(self) -> None:
+ self._dispatch("HEAD")
+
+ def do_POST(self) -> None:
+ self._dispatch("POST")
+
+ def do_PUT(self) -> None:
+ self._method_not_allowed()
+
+ def do_PATCH(self) -> None:
+ self._method_not_allowed()
+
+ def do_DELETE(self) -> None:
+ self._method_not_allowed()
+
+ def do_OPTIONS(self) -> None:
+ self._method_not_allowed()
+
+ def _method_not_allowed(self) -> None:
+ self._send_error(HTTPStatus.METHOD_NOT_ALLOWED, "method_not_allowed", "Web 看板只读")
+
+ # ------------------------------------------------------------ dispatch
+
+ def _dispatch(self, method: str) -> None:
+ parsed = urlsplit(self.path)
+ path = unquote(parsed.path)
+ query = parse_qs(parsed.query, keep_blank_values=True)
+ try:
+ if path == "/api/health":
+ self._send_json(HTTPStatus.OK, {
+ "ok": True,
+ "preview": self.context.preview,
+ "time": int(self.context.clock()),
+ })
+ return
+
+ if path == "/login":
+ if method == "POST":
+ self._login_post()
+ else:
+ self._login_page()
+ return
+ if path == "/logout":
+ self._logout()
+ return
+
+ if path.startswith("/api/"):
+ if not self._authorized():
+ self._send_error(HTTPStatus.UNAUTHORIZED, "unauthorized", "需要登录")
+ return
+ if method == "POST":
+ self._send_error(
+ HTTPStatus.METHOD_NOT_ALLOWED,
+ "method_not_allowed",
+ "Web 看板只读",
+ )
+ return
+ self._api(path, query, method)
+ return
+
+ if not self._authorized():
+ self._redirect("/login")
+ return
+ self._static(path, method)
+ except ApiError as exc:
+ self._send_error(exc.status, exc.code, exc.message)
+ except ValueError as exc:
+ self._send_error(HTTPStatus.BAD_REQUEST, "bad_request", str(exc))
+ except BrokenPipeError:
+ return
+ except Exception as exc: # noqa: BLE001
+ if self.context.logger is not None:
+ self.context.logger.error("web.error", path=path, error=str(exc))
+ self._send_error(
+ HTTPStatus.INTERNAL_SERVER_ERROR, "internal_error", "服务端错误"
+ )
+
+ # ------------------------------------------------------------ auth
+
+ def _session_value(self, expires_at: int) -> str:
+ token = self.context.auth_token
+ payload = f"rb:{int(expires_at)}"
+ signature = hmac.new(
+ token.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256
+ ).hexdigest()
+ return f"{int(expires_at)}.{signature}"
+
+ def _valid_session(self, value: str) -> bool:
+ token = self.context.auth_token
+ if not token or "." not in value:
+ return False
+ raw_exp, signature = value.split(".", 1)
+ try:
+ expires_at = int(raw_exp)
+ except ValueError:
+ return False
+ if expires_at < int(self.context.clock()):
+ return False
+ payload = f"rb:{expires_at}"
+ expected = hmac.new(
+ token.encode("utf-8"), payload.encode("utf-8"), hashlib.sha256
+ ).hexdigest()
+ return hmac.compare_digest(signature, expected)
+
+ def _authorized(self) -> bool:
+ token = self.context.auth_token
+ if not token:
+ return True
+ header = self.headers.get("Authorization", "")
+ if header.startswith("Bearer "):
+ return hmac.compare_digest(header[7:].strip(), token)
+ cookie_header = self.headers.get("Cookie")
+ if not cookie_header:
+ return False
+ cookie = SimpleCookie()
+ try:
+ cookie.load(cookie_header)
+ except Exception: # noqa: BLE001
+ return False
+ morsel = cookie.get(COOKIE_NAME)
+ return morsel is not None and self._valid_session(morsel.value)
+
+ def _login_page(self) -> None:
+ if not self.context.auth_token:
+ self._redirect("/")
+ return
+ html = """
+
+
+登录 · Reaction 看板
+
+Reaction 看板
输入配置中的 web.auth_token 以继续。
+"""
+ self._send_html(HTTPStatus.OK, html)
+
+ def _login_post(self) -> None:
+ if not self.context.auth_token:
+ self._redirect("/")
+ return
+ try:
+ length = int(self.headers.get("Content-Length", "0"))
+ except ValueError:
+ length = 0
+ if length <= 0 or length > 4096:
+ self._send_error(HTTPStatus.BAD_REQUEST, "bad_request", "请求体过大")
+ return
+ body = self.rfile.read(length).decode("utf-8", errors="replace")
+ token = (parse_qs(body).get("token") or [""])[0]
+ if not hmac.compare_digest(token, self.context.auth_token):
+ self._send_error(HTTPStatus.UNAUTHORIZED, "unauthorized", "令牌不正确")
+ return
+ expires_at = int(self.context.clock()) + SESSION_TTL
+ value = self._session_value(expires_at)
+ self.send_response(HTTPStatus.SEE_OTHER)
+ self.send_header("Location", "/")
+ self.send_header("Set-Cookie", (
+ f"{COOKIE_NAME}={value}; Path=/; HttpOnly; SameSite=Lax; "
+ f"Max-Age={SESSION_TTL}"
+ ))
+ self.send_header("Content-Length", "0")
+ self._security_headers()
+ self.end_headers()
+
+ def _logout(self) -> None:
+ self.send_response(HTTPStatus.SEE_OTHER)
+ self.send_header("Location", "/login" if self.context.auth_token else "/")
+ self.send_header("Set-Cookie", f"{COOKIE_NAME}=; Path=/; Max-Age=0; HttpOnly; SameSite=Lax")
+ self.send_header("Content-Length", "0")
+ self._security_headers()
+ self.end_headers()
+
+ # ------------------------------------------------------------ API
+
+ def _api(self, path: str, query: Mapping[str, list[str]], method: str) -> None:
+ queries = self.context.queries
+ if path == "/api/overview":
+ self._send_json(HTTPStatus.OK, queries.overview())
+ return
+ if path == "/api/videos":
+ self._send_json(HTTPStatus.OK, queries.list_videos(
+ sort=_param(query, "sort", "heat"),
+ within_days=_optional_int(query, "within_days"),
+ growth_days=_int_param(query, "growth_days", 7, 1, 365),
+ folder_id=_optional_int(query, "folder"),
+ q=_optional_str(query, "q"),
+ page=_int_param(query, "page", 1, 1, 1_000_000),
+ page_size=_int_param(
+ query, "page_size",
+ _optional_int(query, "limit") or 20,
+ 1, 200,
+ ),
+ ))
+ return
+ if path.startswith("/api/videos/"):
+ parts = [part for part in path.split("/") if part]
+ if len(parts) == 3:
+ aid = _path_int(parts[2], "aid")
+ data = queries.video_detail(aid)
+ if data is None:
+ raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "视频不存在")
+ self._send_json(HTTPStatus.OK, data)
+ return
+ if len(parts) == 4 and parts[3] == "trend":
+ aid = _path_int(parts[2], "aid")
+ days = _range_days(query)
+ self._send_json(HTTPStatus.OK, queries.video_trend(aid, days=days))
+ return
+ raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "接口不存在")
+ if path == "/api/rankings":
+ scope = _param(query, "scope", "video")
+ if scope == "up":
+ self._send_json(HTTPStatus.OK, queries.list_ups(
+ sort=_param(query, "sort", "views"),
+ within_days=_optional_int(query, "within_days"),
+ q=_optional_str(query, "q"),
+ min_videos=_int_param(query, "min_videos", 1, 1, 10000),
+ page=_int_param(query, "page", 1, 1, 1_000_000),
+ page_size=_int_param(
+ query, "page_size",
+ _optional_int(query, "limit") or 20,
+ 1, 200,
+ ),
+ ))
+ return
+ if scope != "video":
+ raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", "scope 必须是 video 或 up")
+ self._send_json(HTTPStatus.OK, queries.list_videos(
+ sort=_param(query, "sort", "heat"),
+ within_days=_optional_int(query, "within_days"),
+ growth_days=_int_param(query, "growth_days", 7, 1, 365),
+ folder_id=_optional_int(query, "folder"),
+ q=_optional_str(query, "q"),
+ page=_int_param(query, "page", 1, 1, 1_000_000),
+ page_size=_int_param(
+ query, "page_size",
+ _optional_int(query, "limit") or 20,
+ 1, 200,
+ ),
+ ))
+ return
+ if path == "/api/ups":
+ self._send_json(HTTPStatus.OK, queries.list_ups(
+ sort=_param(query, "sort", "views"),
+ within_days=_optional_int(query, "within_days"),
+ q=_optional_str(query, "q"),
+ min_videos=_int_param(query, "min_videos", 1, 1, 10000),
+ page=_int_param(query, "page", 1, 1, 1_000_000),
+ page_size=_int_param(
+ query, "page_size",
+ _optional_int(query, "limit") or 20,
+ 1, 200,
+ ),
+ ))
+ return
+ if path.startswith("/api/ups/"):
+ parts = [part for part in path.split("/") if part]
+ if len(parts) != 3:
+ raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "接口不存在")
+ mid = _path_int(parts[2], "mid")
+ data = queries.up_detail(mid)
+ if data is None:
+ raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "UP 不存在")
+ self._send_json(HTTPStatus.OK, data)
+ return
+ if path == "/api/fans":
+ self._send_json(HTTPStatus.OK, queries.fans(
+ _optional_int(query, "mid"),
+ days=_int_param(query, "days", 90, 1, 3650),
+ ))
+ return
+ if path == "/api/candidates":
+ self._send_json(HTTPStatus.OK, queries.candidates())
+ return
+ if path == "/api/system":
+ self._send_json(HTTPStatus.OK, queries.system())
+ return
+ if path == "/api/export.csv":
+ self._export_csv(query)
+ return
+ raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "接口不存在")
+
+ def _export_csv(self, query: Mapping[str, list[str]]) -> None:
+ data = self.context.queries.export_videos(
+ sort=_param(query, "sort", "heat"),
+ within_days=_optional_int(query, "within_days"),
+ growth_days=_int_param(query, "growth_days", 7, 1, 365),
+ folder_id=_optional_int(query, "folder"),
+ q=_optional_str(query, "q"),
+ page=1,
+ page_size=10_000,
+ )
+ output = io.StringIO()
+ writer = csv.writer(output, lineterminator="\n")
+ writer.writerow([
+ "排名", "aid", "bvid", "标题", "UP", "发布时间", "时长（秒）",
+ "播放", "点赞", "投币", "收藏", "转发", "评论", "弹幕",
+ "互动率", "综合热度", "状态", "收藏夹", "快照日期",
+ ])
+ for index, item in enumerate(data["items"], start=1):
+ metrics = item.get("metrics") or {}
+ writer.writerow([
+ index,
+ item["aid"],
+ item.get("bvid") or "",
+ item.get("title") or "",
+ item.get("up_name") or "",
+ _format_date(item.get("pubdate")),
+ item.get("duration") or "",
+ metrics.get("view", ""),
+ metrics.get("like", ""),
+ metrics.get("coin", ""),
+ metrics.get("favorite", ""),
+ metrics.get("share", ""),
+ metrics.get("reply", ""),
+ metrics.get("danmaku", ""),
+ "" if item.get("interaction_rate") is None else f"{item['interaction_rate']:.6f}",
+ "" if item.get("heat") is None else item["heat"],
+ item.get("status") or "",
+ item.get("folder_title") or "",
+ item.get("snapshot_date") or "",
+ ])
+ payload = ("\ufeff" + output.getvalue()).encode("utf-8")
+ filename = f"reaction-board-{shanghai_date(self.context.clock())}.csv"
+ self.send_response(HTTPStatus.OK)
+ self.send_header("Content-Type", "text/csv; charset=utf-8")
+ self.send_header("Content-Disposition", f'attachment; filename="{filename}"')
+ self.send_header("Content-Length", str(len(payload)))
+ self.send_header("Cache-Control", "no-store")
+ self._security_headers()
+ self.end_headers()
+ if self.command != "HEAD":
+ self.wfile.write(payload)
+
+ # ------------------------------------------------------------ static
+
+ def _static(self, path: str, method: str) -> None:
+ if path in {"/", "/index.html"}:
+ target = STATIC_ROOT / "index.html"
+ elif path.startswith("/static/"):
+ relative = path[len("/static/"):]
+ target = (STATIC_ROOT / relative).resolve()
+ try:
+ target.relative_to(STATIC_ROOT.resolve())
+ except ValueError:
+ raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "文件不存在")
+ else:
+ raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "页面不存在")
+ if not target.is_file():
+ raise ApiError(HTTPStatus.NOT_FOUND, "not_found", "页面不存在")
+ content_type = mimetypes.guess_type(str(target))[0] or "application/octet-stream"
+ if content_type.startswith("text/") or content_type in {
+ "application/javascript", "application/json", "image/svg+xml",
+ }:
+ content_type += "; charset=utf-8"
+ payload = target.read_bytes()
+ self.send_response(HTTPStatus.OK)
+ self.send_header("Content-Type", content_type)
+ self.send_header("Content-Length", str(len(payload)))
+ self.send_header(
+ "Cache-Control",
+ "no-cache" if target.name == "index.html" else "public, max-age=3600",
+ )
+ self._security_headers()
+ self.end_headers()
+ if method != "HEAD":
+ self.wfile.write(payload)
+
+ # ------------------------------------------------------------ responses
+
+ def _security_headers(self) -> None:
+ self.send_header(
+ "Content-Security-Policy",
+ "default-src 'self'; img-src 'self' https: data:; "
+ "style-src 'self' 'unsafe-inline'; script-src 'self'; "
+ "connect-src 'self'; base-uri 'none'; frame-ancestors 'none'",
+ )
+ self.send_header("X-Content-Type-Options", "nosniff")
+ self.send_header("Referrer-Policy", "no-referrer")
+ self.send_header("X-Frame-Options", "DENY")
+
+ def _send_json(self, status: int, data: Any) -> None:
+ payload = json.dumps({"ok": True, "data": data}, ensure_ascii=False).encode("utf-8")
+ self.send_response(int(status))
+ self.send_header("Content-Type", "application/json; charset=utf-8")
+ self.send_header("Content-Length", str(len(payload)))
+ self.send_header("Cache-Control", "no-store")
+ self._security_headers()
+ self.end_headers()
+ if self.command != "HEAD":
+ self.wfile.write(payload)
+
+ def _send_error(self, status: int, code: str, message: str) -> None:
+ payload = json.dumps(
+ {"ok": False, "error": {"code": code, "message": message}},
+ ensure_ascii=False,
+ ).encode("utf-8")
+ self.send_response(int(status))
+ self.send_header("Content-Type", "application/json; charset=utf-8")
+ self.send_header("Content-Length", str(len(payload)))
+ self.send_header("Cache-Control", "no-store")
+ self._security_headers()
+ self.end_headers()
+ if self.command != "HEAD":
+ self.wfile.write(payload)
+
+ def _send_html(self, status: int, html: str) -> None:
+ payload = html.encode("utf-8")
+ self.send_response(int(status))
+ self.send_header("Content-Type", "text/html; charset=utf-8")
+ self.send_header("Content-Length", str(len(payload)))
+ self.send_header("Cache-Control", "no-store")
+ self._security_headers()
+ self.end_headers()
+ if self.command != "HEAD":
+ self.wfile.write(payload)
+
+ def _redirect(self, location: str) -> None:
+ self.send_response(HTTPStatus.SEE_OTHER)
+ self.send_header("Location", location)
+ self.send_header("Content-Length", "0")
+ self._security_headers()
+ self.end_headers()
+
+
+# ------------------------------------------------------------ helpers
+
+
+def _param(query: Mapping[str, list[str]], name: str, default: str) -> str:
+ values = query.get(name)
+ return default if not values else str(values[0])
+
+
+def _optional_str(query: Mapping[str, list[str]], name: str) -> str | None:
+ value = _param(query, name, "").strip()
+ return value or None
+
+
+def _optional_int(query: Mapping[str, list[str]], name: str) -> int | None:
+ value = _optional_str(query, name)
+ if value is None:
+ return None
+ try:
+ return int(value)
+ except ValueError as exc:
+ raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", f"{name} 必须是整数") from exc
+
+
+def _int_param(
+ query: Mapping[str, list[str]],
+ name: str,
+ default: int,
+ minimum: int,
+ maximum: int,
+) -> int:
+ value = _optional_int(query, name)
+ if value is None:
+ return int(default)
+ if value < minimum or value > maximum:
+ raise ApiError(
+ HTTPStatus.BAD_REQUEST, "bad_request",
+ f"{name} 必须在 {minimum}~{maximum} 之间",
+ )
+ return value
+
+
+def _path_int(value: str, name: str) -> int:
+ try:
+ return int(value)
+ except ValueError as exc:
+ raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", f"{name} 必须是整数") from exc
+
+
+def _range_days(query: Mapping[str, list[str]]) -> int | None:
+ raw = _optional_str(query, "range")
+ if raw in {None, "all", "0"}:
+ return None
+ if raw in {"7", "30", "90"}:
+ return int(raw)
+ days = _optional_int(query, "days")
+ if days is None:
+ return 30
+ if days < 1 or days > 3650:
+ raise ApiError(HTTPStatus.BAD_REQUEST, "bad_request", "days 超出范围")
+ return days
+
+
+def _format_date(timestamp: int | float | None) -> str:
+ if not timestamp:
+ return ""
+ try:
+ return shanghai_date(float(timestamp))
+ except (OverflowError, OSError, ValueError):
+ return ""
+
+
+def _preview_marker(db: Database) -> bool:
+ def run(conn):
+ row = conn.execute("SELECT v FROM kv WHERE k = 'preview:seeded'").fetchone()
+ return row is not None and str(row["v"]) == "1"
+ return bool(db.read(run))
+
+
+def create_server(
+ config_store: ConfigStore,
+ db: Database,
+ *,
+ host: str | None = None,
+ port: int | None = None,
+ preview: bool = False,
+ logger: Any = None,
+ clock: Callable[[], float] = time.time,
+) -> DashboardHTTPServer:
+ cfg = config_store.get()
+ bind_host = cfg.web.host if host is None else host
+ bind_port = cfg.web.port if port is None else int(port)
+ if not cfg.web.auth_token and not is_loopback(bind_host):
+ raise ValueError(
+ "非回环地址必须配置 web.auth_token，或只绑定 127.0.0.1"
+ )
+ if preview and not _preview_marker(db):
+ raise ValueError("目标数据库不是预览库（缺少 preview:seeded=1 标记）")
+ context = WebContext(
+ queries=DashboardQueries(db, config_store, clock=clock),
+ config_store=config_store,
+ preview=preview,
+ logger=logger,
+ clock=clock,
+ db_path=str(db.path),
+ schema_version=db.schema_version,
+ )
+ return DashboardHTTPServer((bind_host, bind_port), context)
+
+
+def serve_forever(server: DashboardHTTPServer, *, logger: Any = None) -> None:
+ host, port = server.server_address[:2]
+ if logger is not None:
+ logger.info(
+ "web.start",
+ url=f"http://{host}:{port}",
+ preview=server.context.preview,
+ auth=bool(server.context.auth_token),
+ )
+ try:
+ server.serve_forever()
+ finally:
+ server.server_close()
diff --git a/reaction_board/web/static/index.html b/reaction_board/web/static/index.html
new file mode 100644
index 0000000..01042ea
--- /dev/null
+++ b/reaction_board/web/static/index.html
@@ -0,0 +1,13 @@
+
+
+
+
+
+ 咻咻满 Reaction 看板
+
+
+
+ 正在加载…
+
+
+
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..b9e3e81
--- /dev/null
+++ b/tests/__init__.py
@@ -0,0 +1,2 @@
+# -*- coding: utf-8 -*-
+"""离线测试与共享夹具。"""
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 407fef6..fd5805b 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -184,6 +184,34 @@ def test_run_once_offline():
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_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
@@ -271,6 +299,7 @@ def main_test():
test_auth_check_without_cookie()
test_auth_check_all_and_auth_dispatch()
test_run_once_offline()
+ test_default_command_starts_web_and_scheduler()
test_snapshot_backup_failure_returns_error_and_retries()
test_candidates_and_discover_commands()
print("OK: CLI init/sync/snapshot/auth-check/run/status/错误路径测试全部通过")
diff --git a/tests/test_web_queries.py b/tests/test_web_queries.py
new file mode 100644
index 0000000..77afa66
--- /dev/null
+++ b/tests/test_web_queries.py
@@ -0,0 +1,242 @@
+# -*- 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.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["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 "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 视频不应进入榜单"
+ assert aids[0] == 102, aids
+ assert heat["items"][0]["heat"] > heat["items"][-1]["heat"]
+ 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,)
+ ))
+ 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
+
+ # 删除中间一天，折线在缺失日期应为 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["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_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
+
+ 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
+
+ exported = q.export_videos(sort="view")
+ assert exported["total"] == 3
+ assert all("metrics" in item for item in exported["items"])
+ 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_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
new file mode 100644
index 0000000..f5ed821
--- /dev/null
+++ b/tests/test_web_server.py
@@ -0,0 +1,215 @@
+# -*- 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, _, 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" Path:
+ root = Path(tmp)
+ path = root / "config.json"
+ path.write_text(json.dumps({
+ "bilibili": {
+ "read": {"mid": 1, "csrf": "read-csrf", "cookie": "SESSDATA=fake-read"},
+ "write": {"mid": 2, "csrf": "write-csrf", "cookie": "SESSDATA=fake-write"},
+ },
+ "telegram": {"enabled": False, "bot_token": "", "chat_id": 0},
+ "web": {
+ "host": host, "port": port, "auth_token": token,
+ "public_base_url": "",
+ },
+ "storage": {
+ "database_path": str(root / "app.db"),
+ "backup_dir": str(root / "backup"),
+ "backup_keep_days": 7,
+ },
+ "folders": [
+ {"year": 2026, "media_id": 3700023351, "title": "2026"},
+ ],
+ "track_fans": [37754047],
+ "up_blacklist_seed": [],
+ "schedule": {
+ "discover_minutes": 60,
+ "folder_sync_minutes": 60,
+ "full_folder_refresh_hours": 6,
+ "snapshot_at": "00:01",
+ "candidate_ttl_days": 7,
+ },
+ }, ensure_ascii=False, indent=2), encoding="utf-8")
+ return path
+
+
+def resource(
+ aid: int,
+ *,
+ up_mid: int,
+ up_name: str,
+ pubdate: int,
+ attr: int = 0,
+ title: str | None = None,
+) -> FavResource:
+ return FavResource(
+ aid=aid,
+ bvid=f"BV{aid}",
+ title=title or f"咻咻满 reaction 视频 {aid}",
+ up_mid=up_mid,
+ up_name=up_name,
+ pubdate=pubdate,
+ duration=7200,
+ cover=f"//i0.hdslb.com/{aid}.jpg",
+ attr=attr,
+ fav_time=None,
+ )
+
+
+def build_db(tmp: str | Path) -> Path:
+ root = Path(tmp)
+ path = root / "app.db"
+ db = Database(path)
+ db.initialize()
+ try:
+ repo = Repository(db)
+ repo.ensure_folder(FOLDER)
+ repo.seed_blacklist([2])
+
+ def ts(day: str) -> int:
+ return int(day_bounds(day)[0])
+
+ videos = {
+ 101: (1, "UP A", int(day_bounds("2026-10-30")[0]), 0, "2026-10-29"),
+ 102: (1, "UP A", int(day_bounds("2026-10-20")[0]), 0, "2026-10-29"),
+ 103: (2, "UP B", int(day_bounds("2026-10-31")[0]), 0, "2026-10-31"),
+ 104: (3, "UP C", int(day_bounds("2026-09-01")[0]), 9, "2026-10-29"),
+ }
+ for aid, (up_mid, up_name, pubdate, attr, first_day) in videos.items():
+ repo.upsert_from_fav(
+ resource(aid, up_mid=up_mid, up_name=up_name, pubdate=pubdate, attr=attr),
+ FOLDER.media_id,
+ was_added=True,
+ now=ts(first_day),
+ )
+ repo.set_video_status(103, "unfaved")
+
+ metrics = {
+ 101: {"view": 1000, "like": 100, "coin": 20, "favorite": 30, "share": 2, "reply": 10, "danmaku": 12},
+ 102: {"view": 5000, "like": 200, "coin": 50, "favorite": 100, "share": 10, "reply": 30, "danmaku": 40},
+ 104: {"view": 50, "like": 1, "coin": 0, "favorite": 0, "share": 0, "reply": 0, "danmaku": 0},
+ }
+ for index, day in enumerate(DATES):
+ for aid in (101, 102):
+ base = metrics[aid]
+ values = {}
+ for key, value in base.items():
+ values[key] = value + (index * 100 if key == "view" else index * 2)
+ repo.upsert_snapshot(
+ day, aid, metrics=values, ok=True, err=None,
+ captured_at=ts(day) + 60,
+ )
+ repo.upsert_up_stat(
+ day, 37754047, follower=1000 + index * 10, following=5,
+ ok=True, captured_at=ts(day) + 60,
+ )
+ # 103 只有最后一天快照，用于验证样本不足；104 另加一条失败快照
+ repo.upsert_snapshot(
+ DATES[-1], 103,
+ metrics={"view": 120, "like": 5, "coin": 1, "favorite": 2, "share": 0, "reply": 1, "danmaku": 1},
+ ok=True, err=None, captured_at=ts(DATES[-1]) + 60,
+ )
+ repo.upsert_snapshot(
+ "2026-10-30", 104,
+ metrics=None, ok=False, err="simulated failure",
+ captured_at=ts("2026-10-30") + 60,
+ )
+
+ repo.upsert_candidate(
+ aid=201, bvid="BV201", title="咻咻满 reaction 候选",
+ up_mid=9, up_name="候选UP", pubdate=1_790_000_000,
+ cover=None, score=6, channel="search:test#p1",
+ now=ts("2026-10-30"), play=1000, like=50,
+ )
+ repo.mark_notified(
+ 201, channel="telegram", batch_id="b1", ok=True,
+ now=ts("2026-10-30") + 10,
+ )
+ repo.upsert_candidate(
+ aid=202, bvid="BV202", title="咻咻满 切片候选",
+ up_mid=10, up_name="切片UP", pubdate=1_790_000_000,
+ cover=None, score=1, channel="search:test#p1",
+ now=ts("2026-10-29"),
+ )
+ repo.mark_candidate_decision(202, "rejected", now=ts("2026-10-30"))
+ repo.finish_folder_sync(FOLDER, media_count=3, full_refresh=True)
+ repo.kv_set("snapshot:last_completed", DATES[-1])
+ repo.kv_set("auth:ok", "1")
+ repo.kv_set("auth:message", "OK")
+ repo.kv_set(
+ "task:last:snapshot",
+ json.dumps({
+ "name": "snapshot", "ok": True,
+ "started_at": ts(DATES[-1]), "finished_at": ts(DATES[-1]) + 60,
+ "error": None, "summary": {"completed": True},
+ }),
+ )
+ finally:
+ db.close()
+ return path
+
+
+def day_date(value: str) -> date:
+ return date.fromisoformat(value)