Compare commits

...
Author SHA1 Message Date
bot-ci 51ffa44577 fix(scheduler): do not force full snapshot on post-seed restart
Schedule snapshot_seed/snapshot/fans to the next wall-clock seed_at /
snapshot_at on process start (optionally from last task record), instead
of next_run=0. Force-seal of seed shells only after today's snapshot_at
(or CLI --force), so a daytime restart cannot scrape the whole site.
2026-11-03 01:09:50 +08:00
bot-ci 1697573d95 fix(scheduler): address !7 REQUEST_CHANGES
- seed: check completed_key before no_source so same-day last_completed
  is already_completed (not 10m retry)
- scheduler: resolve TaskControl ok/error before logging; ok=False logs
  task_fail, not task_ok
- candidate_cleanup: catch failures, record task, retry ~5m like discover
2026-11-03 01:08:34 +08:00
bot-ci bb18d232f6 feat(scheduler): add snapshot-seed CLI and 10m seed deferral
Add `python app.py snapshot-seed [--date]` (offline clone, Debian /
docker compose exec friendly). When seed fires while full-site snapshot
is running, or last_completed is missing (no_source), record skip as
non-success and retry in 10 minutes until a real seed succeeds; fans
and quiet-window rules unchanged.
2026-11-03 00:22:11 +08:00
bot-ci 9969e65202 fix(scheduler): drop cross-task mutex, quiet window + retry
Incident: evening full snapshot could be recorded as ok with
already_completed/elapsed=0 after mutual skip with hourly sync, then
next_run jumped to the next day.

- Full-site snapshot never skips because another task is running
- 23:00-01:00 Asia/Shanghai: defer discovery / snapshot_recent /
  candidate_cleanup to after 01:00; fans never defers
- Skip records ok=false + skipped=true (not success / not completed)
- Skip/fail reschedule soon (~5m); daily success still next snapshot_at
- TaskControl return value drives next_run; system UI shows skip fields
2026-11-03 00:07:19 +08:00
kairee 39378c65b2 Merge pull request 'feat(growth): same-day 1-day baselines for new videos'
Reviewed-on: #6
2026-11-02 22:12:18 +08:00
kairee 00572a3429 fix(growth): require a 1000-view baseline on the rate board
Keep small-base videos on the delta board only, and raise the smoothing prior to 1000 so videos that just clear the floor do not dominate the ranking.
2026-11-02 22:11:30 +08:00
kairee 39a6537f67 fix(growth): rank the rate board by a smoothed score
Show the real percentage on the card, but sort with delta / (baseline + prior) so a small baseline cannot dominate. Same-day publishes stay off the rate board.
2026-11-02 22:06:41 +08:00
kairee 0f2ac612f5 fix(ui): widen mobile video-card cover so duration stays on the thumbnail 2026-11-02 21:39:06 +08:00
kairee 9412475473 Merge branch 'main' into cursor/same-day-growth-baseline 2026-11-02 20:45:30 +08:00
kairee c5b7185675 feat(growth): same-day 1-day baselines for new videos
Let 1-day view growth use a zero baseline for videos published today, and the first successful sample for older videos ingested today, without mixing those into 7/30-day rankings.
2026-11-02 20:44:17 +08:00
34 changed files with 1865 additions and 180 deletions

No files matched your search

+14 -2
View File
@@ -456,6 +456,9 @@ P1 实现说明:
- 时区使用固定 UTC+8(中国无夏令时),避免 Windows 缺少 `tzdata`;
- 午夜 `seed_at`(默认 `00:00`)把 `snapshot:last_completed` 的 ok=1 行克隆为当日壳,写入 `snapshot:seeded:<date>=1` / `fans:seeded:<date>=1`,**不**推进完成日;
- 也可手动:`python app.py snapshot-seed [--date YYYY-MM-DD]`(Debian / `docker compose exec re-xxm python app.py snapshot-seed`);
- 目标日已有 `snapshot:completed:<date>` 时 seed 记 `already_completed` 成功(即使 `last_completed` 已等于当日),**不得**误判 `no_source`;
- 若全站 `snapshot` 正在跑,或目标日未完成且 `last_completed` 暂不可用(`no_source`),seed **不记成功**,约 10 分钟后重试直到真正克隆成功;
- 晚间 `snapshot_at`(默认 `23:15`)对 seeded 日 force 重采;成功后写 `snapshot:completed:<date>=1`、推进 `last_completed` 并清除 seeded;
- 当日全部视频与粉丝采集成功后在 `kv` 写入 `snapshot:completed:<date>=1`;启动时未完成会补跑(纯 seed 日不误标完成);
- 完成标记不是唯一门槛:每次检查 `pending_snapshot_aids(date)`,新入库或从 `dropped` 恢复的视频会立即补采当天快照;
@@ -466,6 +469,15 @@ P1 实现说明:
- Cookie 登录态通过 `x/web-interface/nav` 检查,失效只告警,不阻塞匿名 `view`;
- Telegram 摘要推迟到 P3,P1 先写结构化 JSON 日志和 `kv` 状态。
调度避让与补跑:
- 常驻启动时日更任务(`snapshot_seed` / `snapshot` / `fans`)按墙钟下一档 `seed_at` / `snapshot_at` 排期(可参考上次成功/失败记录),**不得** `next_run=0` 重启即跑;
- 仅到达当日 `snapshot_at` 后(或 CLI `--force`)才对 seed 壳 `force` 打穿;白天重启不因 `snapshot:seeded` 全量强采;
- 全站 `snapshot` **不因**其它任务 running 而 skip;到期/强制必须执行;
- 上海时区 **23:00–01:00** 为全站快照窗:`discovery` / `snapshot_recent` / `candidate_cleanup` 不执行,延后到 01:00 后;`fans` 短耗时,不避开任何任务;
- 任务 skip(含静默窗延后)写入 `ok=false` + `skipped=true`,**不得**记成成功或推进「当日已完成」;
- skip / fail 后短间隔补跑(约 5 分钟;`snapshot_seed` 避让/无源为约 10 分钟);日更成功才推到次日 `snapshot_at` / `seed_at`。
---
## 8. Telegram 与入库操作
@@ -608,7 +620,7 @@ interaction_rate = (like + coin + favorite) / max(view, 1)
- 查询前一有效快照而不是填补缺失日期;
- 缺失日期在图表中断开;
- 增速榜要求播放基数至少 200;
- 增速榜要求播放基数至少 1000;
- 互动率榜要求播放至少 100;
- 样本不足两天的视频明确标注,不伪造增长数据。
@@ -756,7 +768,7 @@ uv run python tests/test_moves.py
- 粉丝采集对每条 `relation/stat` 加 45s 硬超时看门狗(防 DNS/传输层挂起),每 25 条输出 `up_profile.progress` 进度;单条失败不阻塞整批;
- `app.py backfill-followers`:一次性回填历史视频粉丝数,支持 `--limit` / `--dry-run`;
- 新增内容效率/互动效率榜与 UP 内容效率(Gamma–Poisson EB 收缩),过滤 < 600 秒与未满 2 天的视频;
- `analytics` 配置段:`min_duration_seconds`、`maturity_days`、`min_view_for_ranking`、`video_shrink_minutes`、`shrink_minutes`;
- `analytics` 配置段:`min_duration_seconds`、`maturity_days`、`min_view_for_ranking`、`video_shrink_minutes`、`shrink_minutes`、`growth_rate_prior`;
- 主导航拆分:`/videos`(视频)与 `/ups`(UP主)独立;旧 `/rank` 保留兼容;
- 新增 `/about` 数据说明页与 `/api/about`;
- 卡片精简:列表只保留主指标与互动率,明细移到详情页;
+1
View File
@@ -108,5 +108,6 @@
"video_shrink_minutes": 60,
"shrink_minutes": null,
"home_growth_days": "auto",
"growth_rate_prior": 1000,
},
}
+2
View File
@@ -123,5 +123,7 @@
"shrink_minutes": null,
// "auto" = 7 日有数据用 7,否则 1
"home_growth_days": "auto",
// 增速榜排序:Δ / (基线 + prior)。卡片上的百分比仍是 Δ / 基线。
"growth_rate_prior": 1000,
},
}
+10 -5
View File
@@ -380,7 +380,7 @@ for folder in 配置的年度夹(2025、2026、…):
- 两个夹共 831 条,一次性导入 `video` 表;
- **历史快照无法回填**:B 站不提供任意日期历史数据,只能从部署当天开始积累;
- **开发联调阶段产生的数据一律保留**(不清库、不重建),正式上线直接续用同一个 `app.db`;
- 新发现/新入库的视频从"入库当天"开始积累快照,第一天没有增量是正常的,看板对"仅 1 个数据点"的视频标注"样本不足"。
- 新发现/新入库的视频从"入库当天"开始积累快照。近 1 日:当日发布按 0 起算(一次成功采样即可);更早发布的新入库按入库时播放量起算,第二次采样走晚间全量,不进小时快照。近 7/30 日仍要跨日快照。
---
@@ -473,6 +473,8 @@ CREATE TABLE snapshot (
ok INTEGER NOT NULL DEFAULT 1,
err TEXT,
src TEXT DEFAULT 'view', -- view | infos(降级时标记)
view_open INTEGER, -- 同日第一次成功采样的播放量
open_captured_at INTEGER, -- 开盘采样时刻
PRIMARY KEY (stat_date, aid)
);
CREATE INDEX idx_snap_aid ON snapshot(aid, stat_date);
@@ -672,23 +674,26 @@ w = {view:0.30, like:0.20, coin:0.20, favorite:0.15,
### 8.2 增长榜
> 看板 UI 中“增长”标注为**播放增量**(区间内绝对增量),“增速”标注为**播放增速**(增量 / 基线播放)。两者基线都取不早于 `最新快照 - N 天` 的最近一条成功快照。
> 看板 UI 中“增长”标注为**播放增量**(区间内绝对增量),“增速”标注为**播放增速**(增量 / 基线播放)。近 7/30 日基线取不早于 `最新快照 - N 天` 的最近一条成功快照。近 1 日:有更早日期快照则用前一日;否则当日发布基准为 0,更早发布的新入库用当天第一次成功采样(`view_open`)。
```
d7 = (view_today - view_{today-7}) / 实际间隔天数
r7 = (view_today - view_{today-7}) / max(view_{today-7}, 1)
d1_published_today = view_today - 0 # 只进增量榜,不计算增速
d1_imported_today = view_today - view_open # 需同日第二次采样
sort_rate = Δ / (基线 + analytics.growth_rate_prior) # 默认 prior=1000,只用于增速榜排序
```
| 榜 | 口径 | 基数门槛 |
|---|---|---|
| 累计热度 | §8.1 | — |
| 近 7 天播放增量 | `d7(view)` | — |
| 近 7 天增速 | `r7(view)` | `view >= 200` |
| 近 7 天增速 | `r7(view)` | 基线播放 `>= 1000` |
| 互动率 | `(like+coin+fav)/view` | `view >= 100` |
| 单项榜 | 7 项直接排序 | — |
| UP 榜 | 按 `up_mid` 聚合 Σ/AVG | — |
**冷启动**:不足 2 天数据的视频用"发布后 24h 增量",并标注"样本不足";缺失日不插值。
**冷启动**:近 1 日不要求跨日快照。当日发布相对 0,只进播放增量、不进播放增速;更早发布且当天入库的片子相对 `view_open`(第二次采样走晚间全量,不进小时快照)。近 7/30 日仍要跨日快照。缺失日不插值。
### 8.3 趋势
@@ -795,7 +800,7 @@ secrets.example.json 凭据样例(auth 回写目标)
**进程结构**
```
scheduler 线程:discovery(每 60min) / snapshot_seed(每日 00:00) / snapshot+fans(每日 23:15) / snapshot_recent(小时) / candidate_cleanup(每日)
scheduler 线程:discovery(每 discover_minutes,避开 23:00–01:00) / snapshot_seed(每日 00:00;snapshot 占用或 no_source 则 10 分钟重试,CLI: snapshot-seed) / snapshot(每日 23:15,不因其它任务 skip) / fans(每日 23:15,不避开) / snapshot_recent(小时,避开 23:00–01:00) / candidate_cleanup(每日,避开 23:00–01:00)
web 线程:ThreadingHTTPServer :8080
telegram:getUpdates long polling(回调 + 私聊文本入库)
storage:SQLite WAL,单写线程
+32 -4
View File
@@ -34,7 +34,7 @@ DEFAULT_HEAT_WEIGHTS: dict[str, float] = {
}
MIN_VIEW_FOR_INTERACTION = 100
MIN_VIEW_FOR_GROWTH_RATE = 200
MIN_VIEW_FOR_GROWTH_RATE = 1000
MIN_EFFICIENCY_DURATION = 600
DEFAULT_MATURITY_DAYS = 2
DEFAULT_SHRINK_MINUTES = 120.0
@@ -313,15 +313,20 @@ def compute_growth(
current: int | float | None,
previous: int | float | None,
gap_days: int | float | None,
*,
allow_zero_gap: bool = False,
) -> Growth:
"""计算累计差值、日均差值和相对增速。
缺失基线或间隔不足 1 天时返回不可用,不把 0 当作真实增长。
缺失基线或间隔为负时返回不可用。默认间隔不足 1 天也不可用,不把 0 当成真实增长。
``allow_zero_gap=True`` 时允许同一天的两次采样(或相对 0)计算增量,``per_day`` 为 None。
"""
if current is None or previous is None or gap_days is None:
return Growth(available=False)
gap = int(gap_days)
if gap <= 0:
if gap < 0:
return Growth(available=False)
if gap == 0 and not allow_zero_gap:
return Growth(available=False)
current_value = max(0, int(current))
previous_value = max(0, int(previous))
@@ -329,12 +334,35 @@ def compute_growth(
return Growth(
available=True,
delta=delta,
per_day=delta / gap,
per_day=None if gap == 0 else delta / gap,
rate=delta / max(previous_value, 1),
gap_days=gap,
)
def smoothed_growth_rate(
delta: int | float | None,
previous: int | float | None,
prior: int | float,
) -> float | None:
"""增速榜排序分:``Δ / (基线 + prior)``。
``rate`` 仍是真实相对增速。``prior`` 只削弱小基数在排序里的优势。
"""
if delta is None or previous is None or isinstance(delta, bool) or isinstance(previous, bool):
return None
if isinstance(prior, bool):
return None
try:
strength = float(prior)
base = max(0, int(previous)) + strength
except (TypeError, ValueError):
return None
if strength < 0 or base <= 0:
return None
return int(delta) / base
def normalize_cover(url: object) -> str | None:
"""把 B 站封面统一为 https;空值返回 None。"""
if not isinstance(url, str):
+384 -76
View File
@@ -4,6 +4,7 @@
python app.py init # 建库 / 迁移 / 写入黑名单种子(离线)
python app.py ingest # 手动入库 B 站视频链接 / BV 号 / av 号(只写本地库)
python app.py snapshot # 采集当日七项快照与粉丝数
python app.py snapshot-seed # 用上一完成日克隆当日壳(离线,可 --date)
python app.py discover # 执行一次搜索发现并写入候选池
python app.py candidates # 查看候选池(离线)
python app.py auth-check # 检查 Cookie 登录态
@@ -33,13 +34,22 @@ from .database import Database
from .logging_json import JsonLogger
from .notify.telegram import TelegramClient
from .repository import Repository
from .scheduler import ScheduledTask, Scheduler
from .scheduler import ScheduledTask, Scheduler, TaskControl
from .services.discovery import DiscoveryService
from .services.library import LibraryService
from .services.snapshot import SnapshotService
from .services.telegram_poller import TelegramPoller
from .services.up_profile import UpProfileService
from .timeutil import next_daily_run, parse_hhmm, shanghai_date
from .timeutil import (
in_snapshot_quiet_window,
interval_next_run,
is_date,
next_after_snapshot_quiet_window,
next_daily_run,
parse_hhmm,
shanghai_date,
shanghai_datetime,
)
ROOT = Path(__file__).resolve().parent.parent
DEFAULT_CONFIG = default_config_path(ROOT)
@@ -71,6 +81,17 @@ def _build_parser() -> argparse.ArgumentParser:
snapshot.add_argument(
"--force", action="store_true", help="忽略当日完成标记,重新采集"
)
snapshot_seed = sub.add_parser(
"snapshot-seed",
parents=[common],
help="用上一完成日克隆当日快照壳(离线,不调 B 站)",
)
snapshot_seed.add_argument(
"--date",
default=None,
metavar="YYYY-MM-DD",
help="目标统计日(默认上海时区今天)",
)
discover = sub.add_parser(
"discover", parents=[common], help="执行一次搜索发现并写入候选池"
)
@@ -164,6 +185,7 @@ def _clear_task_running(repo: Repository, name: str) -> None:
def _is_task_running(repo: Repository, name: str) -> bool:
"""供 seed 等任务检测全站快照是否在跑(避让,而非互斥跳过成功)。"""
try:
return bool(repo.kv_get(_task_running_key(name)))
except Exception: # noqa: BLE001
@@ -179,12 +201,19 @@ def _record_task(
finished_at: int | None = None,
payload=None,
error: str | None = None,
skipped: bool = False,
) -> None:
"""把最近一次任务结果写入 kv,并清除 running 标记。"""
"""把最近一次任务结果写入 kv,并清除 running 标记。
``skipped=True`` 表示主动延后/避让,不得记成成功,也不推进「当日已完成」。
"""
now = int(time.time())
if skipped:
ok = False
record = {
"name": name,
"ok": bool(ok),
"skipped": bool(skipped),
"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,
@@ -197,6 +226,125 @@ def _record_task(
_clear_task_running(repo, name)
# 全站快照 / 粉丝失败后的补跑间隔(秒)
_TASK_RETRY_SECONDS = 300.0
# snapshot_seed:全站快照占用 / 尚无 last_completed 时的补跑间隔(秒)
_SEED_RETRY_SECONDS = 600.0
_QUIET_SKIP_REASON = "snapshot_quiet_window"
_SEED_SKIP_SNAPSHOT_RUNNING = "snapshot_running"
_SEED_SKIP_NO_SOURCE = "no_source"
def _quiet_window_defer(
repo: Repository,
logger: JsonLogger,
name: str,
*,
started_at: int,
) -> TaskControl:
"""静默窗内不执行:记 skip(非成功),并排到 01:00 之后。"""
finished = int(time.time())
logger.info(f"{name}.skip", reason=_QUIET_SKIP_REASON)
_record_task(
repo,
name,
ok=False,
skipped=True,
started_at=started_at,
finished_at=finished,
payload={"skipped": True, "reason": _QUIET_SKIP_REASON},
error=_QUIET_SKIP_REASON,
)
return TaskControl(
ok=False,
next_run=next_after_snapshot_quiet_window(float(finished)),
error=_QUIET_SKIP_REASON,
)
def _retry_soon(now: float | None = None) -> float:
return (time.time() if now is None else now) + _TASK_RETRY_SECONDS
def _seed_retry_soon(now: float | None = None) -> float:
return (time.time() if now is None else now) + _SEED_RETRY_SECONDS
def _defer_snapshot_seed(
repo: Repository,
logger: JsonLogger,
*,
started_at: int,
reason: str,
finished_at: int | None = None,
) -> TaskControl:
"""seed 因 snapshot 占用 / no_source 延后:记 skip(非成功),10 分钟后再试。"""
finished = int(finished_at if finished_at is not None else time.time())
logger.info("snapshot_seed.skip", reason=reason)
_record_task(
repo,
"snapshot_seed",
ok=False,
skipped=True,
started_at=started_at,
finished_at=finished,
payload={"skipped": True, "reason": reason},
error=reason,
)
return TaskControl(
ok=False,
next_run=_seed_retry_soon(float(finished)),
error=reason,
)
def _initial_daily_next_run(
repo: Repository,
name: str,
at: str,
*,
now: float | None = None,
retry_seconds: float | None = None,
) -> float:
"""进程启动时日更任务的下次运行:墙钟下一档,或与系统页一致的上次记录推导。
不得返回 ``0`` / 过去时间来“立刻跑一遍”——午夜 seed 后重启否则会白天全量强采。
"""
current = time.time() if now is None else float(now)
wall = next_daily_run(current, at)
raw = repo.kv_get(f"task:last:{name}")
if not raw:
return wall
try:
record = json.loads(raw)
except (TypeError, ValueError, json.JSONDecodeError):
return wall
if not isinstance(record, dict) or record.get("finished_at") is None:
return wall
finished_at = float(record["finished_at"])
summary = record.get("summary") if isinstance(record.get("summary"), dict) else {}
skipped = bool(record.get("skipped") or summary.get("skipped"))
failed = record.get("ok") is False
if skipped or failed:
gap = (
_SEED_RETRY_SECONDS if name == "snapshot_seed"
else float(_TASK_RETRY_SECONDS if retry_seconds is None else retry_seconds)
)
retry_at = finished_at + gap
return retry_at if retry_at > current else wall
scheduled = next_daily_run(finished_at, at)
# 成功记录推到的下一档若已过,退回墙钟下一档(严格 > now)
return scheduled if scheduled > current else wall
def _reached_daily_slot(at: str, now: float | None = None) -> bool:
"""当前上海时间是否已到达(含)今日 ``at`` 墙钟点。"""
current = shanghai_datetime(now)
hour, minute = parse_hhmm(at)
slot = current.replace(hour=hour, minute=minute, second=0, microsecond=0)
return current >= slot
def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient:
return BilibiliClient(
cookie=credentials.cookie,
@@ -354,6 +502,82 @@ def cmd_snapshot(args, logger: JsonLogger) -> int:
db.close()
def cmd_snapshot_seed(args, logger: JsonLogger) -> int:
"""离线克隆:``python app.py snapshot-seed [--date YYYY-MM-DD]``。
与调度器同一套规则:全站 ``snapshot`` 运行中或 ``no_source`` 时不记成功,
返回码非 0,便于 ``docker compose exec re-xxm python app.py snapshot-seed`` 排查。
"""
store = ConfigStore(args.config, require_auth=False)
cfg = store.get()
db = _open_db(cfg, logger)
try:
repo = Repository(db)
target = args.date
if target is not None and not is_date(target):
raise ConfigError(f"日期格式必须是 YYYY-MM-DD: {target!r}")
started = int(time.time())
if _is_task_running(repo, "snapshot"):
payload = {
"ok": False,
"skipped": True,
"reason": _SEED_SKIP_SNAPSHOT_RUNNING,
"stat_date": target or shanghai_date(),
"database": str(cfg.storage.database_path),
}
_record_task(
repo,
"snapshot_seed",
ok=False,
skipped=True,
started_at=started,
finished_at=int(time.time()),
payload=payload,
error=_SEED_SKIP_SNAPSHOT_RUNNING,
)
_print(payload)
return 1
# seed 只读本地库,不调 B 站;占位 client 满足构造约束。
service = SnapshotService(store, repo, object(), logger=logger)
_mark_task_running(repo, "snapshot_seed", started_at=started)
try:
summary = service.run_seed_from_last_completed(date=target)
except Exception as exc: # noqa: BLE001
_record_task(
repo, "snapshot_seed", ok=False, started_at=started,
finished_at=int(time.time()), error=str(exc),
)
raise
payload = summary.as_dict()
payload["database"] = str(cfg.storage.database_path)
finished = int(summary.finished_at or time.time())
if summary.reason == _SEED_SKIP_NO_SOURCE:
_record_task(
repo,
"snapshot_seed",
ok=False,
skipped=True,
started_at=int(summary.started_at or started),
finished_at=finished,
payload={**payload, "skipped": True, "reason": _SEED_SKIP_NO_SOURCE},
error=_SEED_SKIP_NO_SOURCE,
)
_print(payload)
return 1
_record_task(
repo,
"snapshot_seed",
ok=True,
started_at=int(summary.started_at or started),
finished_at=finished,
payload=payload,
)
_print(payload)
return 0
finally:
db.close()
def _telegram(cfg, logger: JsonLogger):
if not cfg.telegram.enabled:
return None
@@ -504,173 +728,248 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
logger=logger,
)
hh, mm = parse_hhmm(cfg.schedule.snapshot_at)
snapshot_at = f"{hh:02d}:{mm:02d}"
sh, sm = parse_hhmm(cfg.schedule.seed_at)
seed_at = f"{sh:02d}:{sm:02d}"
recent_minutes = int(store.get().schedule.recent_snapshot_minutes)
discover_minutes = int(store.get().schedule.discover_minutes)
def run_discover():
started = int(time.time())
if in_snapshot_quiet_window(float(started)):
return _quiet_window_defer(
repo, logger, "discovery", started_at=started,
)
_mark_task_running(repo, "discovery", started_at=started)
interval = float(store.get().schedule.discover_minutes * 60)
try:
summary = discovery_service.run()
except Exception as exc: # noqa: BLE001 记录后交给调度器隔离
except Exception as exc: # noqa: BLE001 记录后补跑,不抛给次日
finished = int(time.time())
_record_task(
repo, "discovery", ok=False, started_at=started,
finished_at=int(time.time()), error=str(exc),
finished_at=finished, error=str(exc),
)
return TaskControl(
ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
)
raise
payload = summary.as_dict()
ok = not summary.errors
finished = int(summary.finished_at or time.time())
_record_task(
repo, "discovery", ok=not summary.errors,
repo, "discovery", ok=ok,
started_at=summary.started_at,
finished_at=summary.finished_at or int(time.time()),
finished_at=finished,
payload=payload,
error=";".join(summary.errors) if summary.errors else None,
)
last["discover"] = summary
logger.info("discovery.summary", **payload)
return summary
next_run = (
interval_next_run(float(finished), interval) if ok
else _retry_soon(float(finished))
)
return TaskControl(ok=ok, next_run=next_run)
def run_cleanup():
started = int(time.time())
removed = repo.delete_expired_candidates(
ttl_days=cfg.schedule.candidate_ttl_days
)
expired = repo.expire_fav_actions()
if in_snapshot_quiet_window(float(started)):
return _quiet_window_defer(
repo, logger, "candidate_cleanup", started_at=started,
)
try:
removed = repo.delete_expired_candidates(
ttl_days=cfg.schedule.candidate_ttl_days
)
expired = repo.expire_fav_actions()
except Exception as exc: # noqa: BLE001 记录后约 5 分钟补跑
finished = int(time.time())
_record_task(
repo, "candidate_cleanup", ok=False, started_at=started,
finished_at=finished, error=str(exc),
)
return TaskControl(
ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
)
if removed or expired:
repo.bump_cache_version("videos")
finished = int(time.time())
logger.info(
"cleanup.done",
candidates_removed=removed, fav_actions_expired=expired,
)
_record_task(
repo, "candidate_cleanup", ok=True, started_at=started,
finished_at=int(time.time()),
finished_at=finished,
payload={
"candidates_removed": removed,
"fav_actions_expired": expired,
},
)
return {
"candidates_removed": removed,
"fav_actions_expired": expired,
}
return TaskControl(
ok=True,
next_run=interval_next_run(float(finished), 86400.0),
)
def run_snapshot():
"""全站快照:不因其它任务 running 而 skip;失败/未完成则短间隔补跑。"""
started = int(time.time())
if _is_task_running(repo, "snapshot_recent"):
logger.info(
"snapshot.skip", reason="snapshot_recent_running",
)
_record_task(
repo, "snapshot", ok=True, started_at=started,
finished_at=int(time.time()),
payload={"skipped": True, "reason": "snapshot_recent_running"},
)
return None
_mark_task_running(repo, "snapshot", started_at=started)
try:
# 晚间全量:若当日是午夜 seed 壳,必须 force 打穿 ok=1 克隆行。
# 仅晚间 snapshot_at 及之后:seed 壳才 force 打穿 ok=1 克隆行。
# 白天/重启不得因 seeded 标记立刻全量强采。
today = shanghai_date()
force_seed = repo.kv_get(
SnapshotService.seeded_key(today)
) == "1"
force_seed = (
repo.kv_get(SnapshotService.seeded_key(today)) == "1"
and _reached_daily_slot(snapshot_at, float(started))
)
summary = snapshot_service.run_if_needed(force=force_seed)
except Exception as exc: # noqa: BLE001
finished = int(time.time())
_record_task(
repo, "snapshot", ok=False, started_at=started,
finished_at=int(time.time()), error=str(exc),
finished_at=finished, error=str(exc),
)
return TaskControl(
ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
)
raise
payload = summary.as_dict()
ok = bool(summary.completed and not summary.backup_error)
finished = int(summary.finished_at or time.time())
_record_task(
repo, "snapshot", ok=ok, started_at=summary.started_at,
finished_at=summary.finished_at or int(time.time()),
finished_at=finished,
payload=payload, error=summary.backup_error,
)
last["snapshot"] = summary
logger.info("snapshot.summary", **payload)
return summary
next_run = (
next_daily_run(float(finished), snapshot_at) if ok
else _retry_soon(float(finished))
)
return TaskControl(ok=ok, next_run=next_run)
def run_snapshot_recent():
started = int(time.time())
if _is_task_running(repo, "snapshot"):
logger.info(
"snapshot.recent_skip", reason="snapshot_running",
if in_snapshot_quiet_window(float(started)):
return _quiet_window_defer(
repo, logger, "snapshot_recent", started_at=started,
)
_record_task(
repo, "snapshot_recent", ok=True, started_at=started,
finished_at=int(time.time()),
payload={"skipped": True, "reason": "snapshot_running"},
)
return None
_mark_task_running(repo, "snapshot_recent", started_at=started)
interval = float(store.get().schedule.recent_snapshot_minutes * 60)
try:
summary = snapshot_service.run_recent_videos()
except Exception as exc: # noqa: BLE001
finished = int(time.time())
_record_task(
repo, "snapshot_recent", ok=False, started_at=started,
finished_at=int(time.time()), error=str(exc),
finished_at=finished, error=str(exc),
)
return TaskControl(
ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
)
raise
payload = summary.as_dict()
ok = summary.failed == 0
error = ";".join(summary.errors) if summary.errors else None
finished = int(summary.finished_at or time.time())
_record_task(
repo, "snapshot_recent", ok=ok, started_at=summary.started_at,
finished_at=summary.finished_at or int(time.time()),
finished_at=finished,
payload=payload, error=error,
)
last["snapshot_recent"] = summary
logger.info("snapshot.recent_summary", **payload)
return summary
next_run = (
interval_next_run(float(finished), interval) if ok
else _retry_soon(float(finished))
)
return TaskControl(ok=ok, next_run=next_run)
def run_fans():
"""粉丝更新短耗时:不避开任何任务,可与全量并行。"""
started = int(time.time())
_mark_task_running(repo, "fans", started_at=started)
try:
today = shanghai_date()
force_seed = repo.kv_get(
SnapshotService.fans_seeded_key(today)
) == "1"
force_seed = (
repo.kv_get(SnapshotService.fans_seeded_key(today)) == "1"
and _reached_daily_slot(snapshot_at, float(started))
)
summary = snapshot_service.run_fans_if_needed(force=force_seed)
except Exception as exc: # noqa: BLE001
finished = int(time.time())
_record_task(
repo, "fans", ok=False, started_at=started,
finished_at=int(time.time()), error=str(exc),
finished_at=finished, error=str(exc),
)
return TaskControl(
ok=False, next_run=_retry_soon(float(finished)), error=str(exc),
)
raise
payload = summary.as_dict()
ok = summary.fans_failed == 0
error = ";".join(
err for err in summary.errors if str(err).startswith("fans ")
) or None
finished = int(summary.finished_at or time.time())
_record_task(
repo, "fans", ok=ok, started_at=summary.started_at,
finished_at=summary.finished_at or int(time.time()),
finished_at=finished,
payload=payload, error=error,
)
last["fans"] = summary
logger.info("fans.summary", **payload)
return summary
next_run = (
next_daily_run(float(finished), snapshot_at) if ok
else _retry_soon(float(finished))
)
return TaskControl(ok=ok, next_run=next_run)
def run_snapshot_seed():
started = int(time.time())
# 全站快照占用时不执行、不记成功;10 分钟后重试直到真正 seed 成功。
if _is_task_running(repo, "snapshot"):
return _defer_snapshot_seed(
repo,
logger,
started_at=started,
reason=_SEED_SKIP_SNAPSHOT_RUNNING,
)
_mark_task_running(repo, "snapshot_seed", started_at=started)
try:
summary = snapshot_service.run_seed_from_last_completed()
except Exception as exc: # noqa: BLE001
finished = int(time.time())
_record_task(
repo, "snapshot_seed", ok=False, started_at=started,
finished_at=int(time.time()), error=str(exc),
finished_at=finished, error=str(exc),
)
return TaskControl(
ok=False,
next_run=_seed_retry_soon(float(finished)),
error=str(exc),
)
raise
payload = summary.as_dict()
_record_task(
repo, "snapshot_seed", ok=True, started_at=summary.started_at,
finished_at=summary.finished_at or int(time.time()),
payload=payload,
)
finished = int(summary.finished_at or time.time())
last["snapshot_seed"] = summary
logger.info("snapshot.seed_summary", **payload)
return summary
if summary.reason == _SEED_SKIP_NO_SOURCE:
return _defer_snapshot_seed(
repo,
logger,
started_at=int(summary.started_at or started),
reason=_SEED_SKIP_NO_SOURCE,
finished_at=finished,
)
# already_completed / already_seeded / seeded → 成功,推到次日 seed_at
_record_task(
repo, "snapshot_seed", ok=True, started_at=summary.started_at,
finished_at=finished,
payload=payload,
)
return TaskControl(
ok=True, next_run=next_daily_run(float(finished), seed_at),
)
def run_config_watch():
before = store.revision
@@ -679,33 +978,39 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
logger.info("config.reloaded", revision=store.revision)
return {"revision": store.revision}
hh, mm = parse_hhmm(cfg.schedule.snapshot_at)
snapshot_at = f"{hh:02d}:{mm:02d}"
sh, sm = parse_hhmm(cfg.schedule.seed_at)
seed_at = f"{sh:02d}:{sm:02d}"
recent_minutes = int(store.get().schedule.recent_snapshot_minutes)
# --once:保持 next_run=0 立刻跑到期逻辑;常驻:日更对齐墙钟,避免重启即强采。
now_ts = time.time()
if args.once:
seed_next = snapshot_next = fans_next = 0.0
else:
seed_next = _initial_daily_next_run(
repo, "snapshot_seed", seed_at, now=now_ts,
)
snapshot_next = _initial_daily_next_run(
repo, "snapshot", snapshot_at, now=now_ts,
)
fans_next = _initial_daily_next_run(
repo, "fans", snapshot_at, now=now_ts,
)
scheduler = Scheduler(
[
ScheduledTask(
"snapshot_seed",
86400.0,
run_snapshot_seed,
next_run=0.0,
next_run_fn=lambda started: next_daily_run(started, seed_at),
next_run=seed_next,
),
ScheduledTask(
"snapshot",
86400.0,
run_snapshot,
next_run=0.0,
next_run_fn=lambda started: next_daily_run(started, snapshot_at),
next_run=snapshot_next,
),
ScheduledTask(
"fans",
86400.0,
run_fans,
next_run=0.0,
next_run_fn=lambda started: next_daily_run(started, snapshot_at),
next_run=fans_next,
),
ScheduledTask(
"snapshot_recent",
@@ -714,7 +1019,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
),
ScheduledTask(
"discovery",
float(cfg.schedule.discover_minutes * 60),
float(discover_minutes * 60),
run_discover,
),
ScheduledTask("candidate_cleanup", 86400.0, run_cleanup),
@@ -722,6 +1027,7 @@ def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int:
],
logger=logger,
)
if serve_web and not args.once:
from .web.server import create_server, serve_forever
@@ -863,6 +1169,8 @@ def main(argv: Sequence[str] | None = None) -> int:
return cmd_init(args, logger)
if args.command == "snapshot":
return cmd_snapshot(args, logger)
if args.command == "snapshot-seed":
return cmd_snapshot_seed(args, logger)
if args.command == "ingest":
return cmd_ingest(args, logger)
if args.command == "backfill-followers":
+7
View File
@@ -52,6 +52,7 @@ DEFAULT_ANALYTICS: dict[str, Any] = {
"video_shrink_minutes": 60,
"shrink_minutes": None,
"home_growth_days": "auto",
"growth_rate_prior": 1000,
}
DEFAULT_CONFIG: dict[str, Any] = {
@@ -210,6 +211,7 @@ class AnalyticsConfig:
video_shrink_minutes: int
shrink_minutes: int | None
home_growth_days: str | int
growth_rate_prior: int
@dataclass(frozen=True)
@@ -592,6 +594,11 @@ def _build_config(data: Mapping[str, Any], path: Path, *, secrets_path: Path) ->
)
),
home_growth_days=home_growth_days,
growth_rate_prior=_as_int(
analytics_raw.get("growth_rate_prior", 1000),
"analytics.growth_rate_prior",
minimum=0,
),
)
return AppConfig(
+12
View File
@@ -244,6 +244,18 @@ MIGRATIONS: tuple[Migration, ...] = (
"CREATE INDEX idx_visit_date ON visit(stat_date)",
),
),
Migration(
version=8,
name="snapshot open view",
statements=(
# 同日第一次成功采样的播放量,供近 1 日「入库后增量」做基准。
# 已有成功行无法找回被覆盖的入库值,只能把当前 view 当作开盘。
'ALTER TABLE snapshot ADD COLUMN view_open INTEGER',
"ALTER TABLE snapshot ADD COLUMN open_captured_at INTEGER",
'UPDATE snapshot SET view_open = "view", open_captured_at = captured_at '
"WHERE ok = 1 AND view_open IS NULL",
),
),
)
+30 -8
View File
@@ -690,12 +690,17 @@ class Repository:
src: str = "view",
) -> None:
values = metrics or {}
ok_flag = 1 if ok else 0
view_value = values.get("view")
open_view = view_value if ok else None
open_at = int(captured_at) if ok else None
def run(conn: sqlite3.Connection) -> None:
conn.execute(
"""
INSERT INTO snapshot(stat_date, aid, "view", "like", favorite, coin,
share, reply, danmaku, captured_at, ok, err, src)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
share, reply, danmaku, captured_at, ok, err, src,
view_open, open_captured_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(stat_date, aid) DO UPDATE SET
"view" = excluded."view",
"like" = excluded."like",
@@ -707,12 +712,22 @@ class Repository:
captured_at = excluded.captured_at,
ok = excluded.ok,
err = excluded.err,
src = excluded.src
src = excluded.src,
view_open = CASE
WHEN excluded.ok = 1 AND snapshot.view_open IS NULL
THEN excluded."view"
ELSE snapshot.view_open
END,
open_captured_at = CASE
WHEN excluded.ok = 1 AND snapshot.open_captured_at IS NULL
THEN excluded.captured_at
ELSE snapshot.open_captured_at
END
""",
(
stat_date,
int(aid),
values.get("view"),
view_value,
values.get("like"),
values.get("favorite"),
values.get("coin"),
@@ -720,9 +735,11 @@ class Repository:
values.get("reply"),
values.get("danmaku"),
int(captured_at),
1 if ok else 0,
ok_flag,
err,
src,
open_view,
open_at,
),
)
self.db.write(run)
@@ -741,18 +758,23 @@ class Repository:
"""
INSERT INTO snapshot(
stat_date, aid, "view", "like", favorite, coin,
share, reply, danmaku, captured_at, ok, err, src
share, reply, danmaku, captured_at, ok, err, src,
view_open, open_captured_at
)
SELECT
?, aid, "view", "like", favorite, coin,
share, reply, danmaku, ?, 1, NULL, ?
share, reply, danmaku, ?, 1, NULL, ?,
"view", ?
FROM snapshot
WHERE stat_date = ? AND ok = 1
AND aid NOT IN (
SELECT aid FROM snapshot WHERE stat_date = ?
)
""",
(target_date, int(captured_at), src, source_date, target_date),
(
target_date, int(captured_at), src, int(captured_at),
source_date, target_date,
),
)
return int(cur.rowcount or 0)
return self.db.write(run)
+42 -5
View File
@@ -3,6 +3,9 @@
P0 只注册收藏夹同步任务;P1/P3 会在同一调度器上追加快照、粉丝和发现任务。
调度只负责“什么时候运行”,具体任务保持幂等,并在自己的服务层处理重试。
任务 ``fn`` 可返回 :class:`TaskControl` 以覆盖 ``ok`` / ``next_run``
(用于静默窗延后、失败补跑等);未返回时仍按 ``next_run_fn`` / ``interval`` 排期。
"""
from __future__ import annotations
@@ -12,6 +15,15 @@ from dataclasses import dataclass
from typing import Any, Callable, Sequence
@dataclass
class TaskControl:
"""任务可选返回值:覆盖成功标记与下次运行时间。"""
ok: bool = True
next_run: float | None = None
error: str | None = None
@dataclass
class ScheduledTask:
name: str
@@ -70,23 +82,48 @@ class Scheduler:
continue
started = current
error: str | None = None
control: TaskControl | None = None
try:
task.fn()
raw = task.fn()
except Exception as exc: # noqa: BLE001 单任务失败不终止调度
error = str(exc)
self._log("error", "scheduler.task_error", task=task.name, error=error)
else:
self._log("info", "scheduler.task_ok", task=task.name)
if isinstance(raw, TaskControl):
control = raw
finished = self._clock()
if task.next_run_fn is not None:
if control is not None and control.next_run is not None:
task.next_run = float(control.next_run)
elif task.next_run_fn is not None:
task.next_run = task.next_run_fn(started)
else:
task.next_run = started + task.interval
ok = error is None
if control is not None:
ok = bool(control.ok) and error is None
if control.error and error is None:
error = control.error
# Skip/fail ≠ success:先定 ok/error,再记 scheduler 日志。
if ok:
self._log("info", "scheduler.task_ok", task=task.name)
elif control is not None:
self._log(
"warning",
"scheduler.task_fail",
task=task.name,
error=error,
)
else:
self._log(
"error",
"scheduler.task_error",
task=task.name,
error=error,
)
result = TaskResult(
name=task.name,
started_at=started,
finished_at=finished,
ok=error is None,
ok=ok,
error=error,
)
results.append(result)
+19 -10
View File
@@ -51,6 +51,7 @@ class SnapshotSummary:
completed: bool = False
already_completed: bool = False
finished_at: int = 0
reason: str | None = None
errors: list[str] = field(default_factory=list)
def as_dict(self) -> dict[str, Any]:
@@ -73,6 +74,7 @@ class SnapshotSummary:
"backup_error": self.backup_error,
"completed": self.completed,
"already_completed": self.already_completed,
"reason": self.reason,
"errors": self.errors,
}
@@ -310,7 +312,8 @@ class SnapshotService:
"""刷新近 N 天新视频的当日快照;不碰完成日标记与日备份。
「近 N 天」按 pubdate 墙钟窗口:``now - recent_video_days * 86400``,
不是上海日历日差。允许覆盖同日已有 ``ok=1`` 行以便小时级刷新。
不是上海日历日差。当天入库但发布更早的旧片不进小时任务。
允许覆盖同日已有 ``ok=1`` 行以便小时级刷新。
"""
cfg = self.config_store.get()
client = self._client()
@@ -535,15 +538,8 @@ class SnapshotService:
raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}")
summary = SnapshotSummary(stat_date=stat_date, started_at=now)
source = self.repo.kv_get("snapshot:last_completed")
if not source or source >= stat_date:
self._log(
"info", "snapshot.seed_skip",
stat_date=stat_date, reason="no_source", source=source,
)
summary.finished_at = self._now()
summary.completed = True
return summary
# 目标日已完成:先于 no_source。全站快照当天完成后
# last_completed == stat_date,不可误判为无源再 10 分钟重试。
if self.repo.kv_get(self.completed_key(stat_date)) == "1":
self._log(
"info", "snapshot.seed_skip",
@@ -552,6 +548,17 @@ class SnapshotService:
summary.finished_at = self._now()
summary.completed = True
summary.already_completed = True
summary.reason = "already_completed"
return summary
source = self.repo.kv_get("snapshot:last_completed")
if not source or source >= stat_date:
self._log(
"info", "snapshot.seed_skip",
stat_date=stat_date, reason="no_source", source=source,
)
summary.finished_at = self._now()
summary.completed = False
summary.reason = "no_source"
return summary
if self.repo.kv_get(self.seeded_key(stat_date)) == "1":
self._log(
@@ -561,6 +568,7 @@ class SnapshotService:
summary.finished_at = self._now()
summary.completed = True
summary.already_completed = True
summary.reason = "already_seeded"
return summary
copied = self.repo.copy_ok_snapshots(
@@ -582,6 +590,7 @@ class SnapshotService:
self.repo.bump_cache_version("fans")
summary.finished_at = self._now()
summary.completed = True # seed 任务本身成功;不等于快照落盘完成
summary.reason = "seeded"
self._log(
"info", "snapshot.seed_done",
stat_date=stat_date, source=source,
+36
View File
@@ -57,3 +57,39 @@ def day_bounds(date: str) -> tuple[float, float]:
raise ValueError(f"日期格式必须是 YYYY-MM-DD: {date!r}")
start = datetime.strptime(date, "%Y-%m-%d").replace(tzinfo=SHANGHAI_TZ)
return start.timestamp(), (start + timedelta(days=1)).timestamp()
# 全站快照静默窗:上海时区 [23:00, 01:00)。此间 discovery / recent / cleanup 不执行。
SNAPSHOT_QUIET_START_HOUR = 23
SNAPSHOT_QUIET_END_HOUR = 1
def in_snapshot_quiet_window(ts: float | None = None) -> bool:
"""是否处于全站快照静默窗 [23:00, 01:00)(上海时区,跨日)。"""
dt = shanghai_datetime(ts)
return dt.hour >= SNAPSHOT_QUIET_START_HOUR or dt.hour < SNAPSHOT_QUIET_END_HOUR
def next_after_snapshot_quiet_window(ts: float) -> float:
"""若 ``ts`` 落在静默窗内,推到窗结束后的 01:00;否则原样返回。"""
if not in_snapshot_quiet_window(ts):
return ts
dt = shanghai_datetime(ts)
if dt.hour >= SNAPSHOT_QUIET_START_HOUR:
target = (dt + timedelta(days=1)).replace(
hour=SNAPSHOT_QUIET_END_HOUR, minute=0, second=0, microsecond=0,
)
else:
target = dt.replace(
hour=SNAPSHOT_QUIET_END_HOUR, minute=0, second=0, microsecond=0,
)
if target.timestamp() <= ts:
target += timedelta(days=1)
return target.timestamp()
def interval_next_run(now: float, interval: float) -> float:
"""``now + interval``,若落入静默窗则推到 01:00 之后。"""
if interval <= 0:
raise ValueError("interval 必须大于 0")
return next_after_snapshot_quiet_window(now + interval)
+184 -24
View File
@@ -20,6 +20,7 @@ from ..analytics import (
METRICS,
MIN_VIEW_FOR_GROWTH_RATE,
compute_growth,
smoothed_growth_rate,
content_hours,
eb_rate,
effective_shrink_minutes,
@@ -33,7 +34,14 @@ from ..analytics import (
)
from ..config import ConfigStore
from ..database import Database
from ..timeutil import day_bounds, next_daily_run, shanghai_date, shanghai_datetime
from ..timeutil import (
day_bounds,
interval_next_run,
next_after_snapshot_quiet_window,
next_daily_run,
shanghai_date,
shanghai_datetime,
)
from .cache import DASHBOARD_CACHE, TTL_FANS, TTL_VIDEOS, build_cache_key
TRACKED_STATUSES = ("active", "unfaved")
@@ -58,6 +66,19 @@ def _chunks(values: Sequence[int], size: int = 400) -> Iterable[Sequence[int]]:
yield values[index:index + size]
def _growth_baseline_view(item: Mapping[str, Any]) -> int | None:
"""增速分母:当前播放减去区间增量。zero 基线的增量等于当前播放,结果为 0。"""
growth = item.get("growth") or {}
current = (item.get("metrics") or {}).get("view")
delta = growth.get("delta")
if current is None or delta is None:
return None
try:
return int(current) - int(delta)
except (TypeError, ValueError):
return None
def _date_minus(day: str, days: int) -> str:
return (date.fromisoformat(day) - timedelta(days=int(days))).isoformat()
@@ -385,6 +406,18 @@ class DashboardQueries:
percentiles = score_percentiles(scores)
return videos, snapshots, scores, percentiles
@staticmethod
def _publish_date(pubdate: object) -> str | None:
if pubdate is None or isinstance(pubdate, bool):
return None
try:
ts = int(pubdate)
except (TypeError, ValueError):
return None
if ts <= 0:
return None
return shanghai_date(ts)
@staticmethod
def _growth_for(
snapshot: Mapping[str, Any] | None,
@@ -406,6 +439,70 @@ class DashboardQueries:
gap = None
return compute_growth(current, before, gap)
@classmethod
def _same_day_growth(
cls,
snapshot: Mapping[str, Any] | None,
video: Mapping[str, Any],
metric: str = "view",
) -> dict[str, Any] | None:
"""近 1 日、没有更早快照时:当日发布相对 0;否则用 view_open。"""
if snapshot is None:
return None
current = snapshot.get(metric)
if current is None:
return None
stat_date = str(snapshot.get("stat_date") or "")
published_on = cls._publish_date(video.get("pubdate"))
if published_on is not None and published_on == stat_date:
growth = compute_growth(current, 0, 0, allow_zero_gap=True)
payload = growth.as_dict()
payload["baseline"] = "zero"
# previous=0 时 rate = view/1,会按原始播放碾压真日环比。
payload["rate"] = None
return payload
try:
open_at = snapshot.get("open_captured_at")
captured_at = snapshot.get("captured_at")
view_open = snapshot.get("view_open")
open_ready = (
open_at is not None
and captured_at is not None
and int(open_at) < int(captured_at)
and view_open is not None
)
except (TypeError, ValueError):
open_ready = False
if not open_ready:
return None
growth = compute_growth(current, view_open, 0, allow_zero_gap=True)
payload = growth.as_dict()
payload["baseline"] = "open"
return payload
@classmethod
def _growth_result(
cls,
video: Mapping[str, Any],
snapshot: Mapping[str, Any] | None,
previous: Mapping[str, Any] | None,
*,
days_back: int,
metric: str = "view",
) -> dict[str, Any]:
if previous is not None:
growth = cls._growth_for(snapshot, previous, metric=metric)
payload = growth.as_dict()
payload["baseline"] = "snapshot" if growth.available else None
return payload
if days_back == 1:
fallback = cls._same_day_growth(snapshot, video, metric=metric)
if fallback is not None:
return fallback
payload = compute_growth(None, None, None).as_dict()
payload["baseline"] = None
return payload
def _item(
self,
video: Mapping[str, Any],
@@ -415,6 +512,7 @@ class DashboardQueries:
percentile: float | None = None,
previous: Mapping[str, Any] | None = None,
include_growth: bool = False,
growth_days: int | None = None,
) -> dict[str, Any]:
metrics = _metric_map(snapshot)
duration = video.get("duration")
@@ -455,7 +553,9 @@ class DashboardQueries:
),
}
if include_growth:
item["growth"] = self._growth_for(snapshot, previous).as_dict()
item["growth"] = self._growth_result(
video, snapshot, previous, days_back=int(growth_days or 1),
)
return item
# ------------------------------------------------------------ overview
@@ -474,7 +574,7 @@ class DashboardQueries:
videos = self._videos(conn)
aids = [int(video["aid"]) for video in videos]
snapshots = self._snapshot_map(conn, aids)
# 播放增量:按片相对当前点的相邻成功日,避免 seed/recent 拉到 T-2。
# 播放增量:近 1 日。有更早日用相邻成功日;当日发布相对 0;否则用 view_open。
previous = self._snapshot_previous_relative(conn, snapshots, days_back=1)
totals = {metric: 0 for metric in METRICS}
for video in videos:
@@ -489,12 +589,13 @@ class DashboardQueries:
for video in videos:
aid = int(video["aid"])
snapshot = snapshots.get(aid)
prior = previous.get(aid)
if snapshot is None or prior is None:
if snapshot is None:
continue
growth = self._growth_for(snapshot, prior)
if growth.available and growth.delta is not None:
view_delta_sum += growth.delta
growth = self._growth_result(
video, snapshot, previous.get(aid), days_back=1,
)
if growth.get("available") and growth.get("delta") is not None:
view_delta_sum += int(growth["delta"])
available_deltas += 1
if available_deltas:
view_delta = view_delta_sum
@@ -881,6 +982,7 @@ class DashboardQueries:
percentile=percentiles.get(int(video["aid"])),
previous=previous.get(int(video["aid"])),
include_growth=sort in {"growth", "growth_rate"},
growth_days=growth_days,
)
for video in videos
]
@@ -952,9 +1054,13 @@ class DashboardQueries:
if sort == "growth_rate":
items = [
item for item in items
if (item["metrics"] or {}).get("view", 0) >= MIN_VIEW_FOR_GROWTH_RATE
if item.get("growth", {}).get("baseline") != "zero"
and item.get("growth", {}).get("rate") is not None
and (_growth_baseline_view(item) or 0) >= MIN_VIEW_FOR_GROWTH_RATE
]
rate_prior = max(0, int(analytics.growth_rate_prior))
def sort_key(item: Mapping[str, Any]):
if sort == "added":
return (int(item["first_seen"]), int(item["pubdate"]))
@@ -971,9 +1077,18 @@ class DashboardQueries:
)
value = item.get(field)
return value if value 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)
if sort == "growth":
value = item.get("growth", {}).get("delta")
return value if value is not None else float("-inf")
if sort == "growth_rate":
growth = item.get("growth") or {}
delta = growth.get("delta")
current = (item.get("metrics") or {}).get("view")
if delta is None or current is None:
return float("-inf")
value = smoothed_growth_rate(
delta, int(current) - int(delta), rate_prior,
)
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
@@ -990,9 +1105,20 @@ class DashboardQueries:
reason = None
if sort in {"growth", "growth_rate"} and not growth_available:
available = False
reason = (
f"近 {growth_days} 日增长需要至少 {growth_days + 1} 天快照"
)
if growth_days == 1 and sort == "growth_rate":
reason = (
"近 1 日增速需要基线播放至少 "
f"{MIN_VIEW_FOR_GROWTH_RATE};当日发布不计入增速"
)
elif growth_days == 1:
reason = (
"近 1 日增长需要前一日快照;当日发布有一次成功采样即可;"
"更早发布的新入库需要当天至少两次成功采样"
)
else:
reason = (
f"近 {growth_days} 日增长需要至少 {growth_days + 1} 天快照"
)
duration_filter = None
if sort in duration_sorts:
duration_filter = {
@@ -1023,6 +1149,7 @@ class DashboardQueries:
"available": available,
"reason": reason,
"duration_filter": duration_filter,
"growth_rate_prior": rate_prior if sort == "growth_rate" else None,
}
return self.db.read(run)
@@ -1044,16 +1171,19 @@ class DashboardQueries:
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)
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["growth"] = self._growth_result(
video, snapshot, previous, days_back=1,
)
item["snapshot_count"] = int(conn.execute(
"SELECT COUNT(*) FROM snapshot WHERE aid = ?", (aid,)
).fetchone()[0])
@@ -1666,6 +1796,7 @@ class DashboardQueries:
"video_shrink_minutes": cfg.analytics.video_shrink_minutes,
"shrink_minutes": cfg.analytics.shrink_minutes,
"home_growth_days": cfg.analytics.home_growth_days,
"growth_rate_prior": cfg.analytics.growth_rate_prior,
},
"heat_weights": {
metric: float(value)
@@ -1847,9 +1978,11 @@ class DashboardQueries:
"fans": int(next_daily_run(now, cfg.schedule.snapshot_at)),
"snapshot": int(next_daily_run(now, cfg.schedule.snapshot_at)),
"snapshot_seed": int(next_daily_run(now, cfg.schedule.seed_at)),
"discover": int(now + discover_interval),
"snapshot_recent": int(now + recent_interval),
"discover": int(interval_next_run(now, float(discover_interval))),
"snapshot_recent": int(interval_next_run(now, float(recent_interval))),
}
retry_seconds = 300
seed_retry_seconds = 600
for name, key, at in (
("fans", "fans", cfg.schedule.snapshot_at),
("snapshot", "snapshot", cfg.schedule.snapshot_at),
@@ -1861,19 +1994,46 @@ class DashboardQueries:
if not (isinstance(record, dict) and record.get("finished_at")):
continue
finished_at = int(record["finished_at"])
summary = record.get("summary") if isinstance(record.get("summary"), dict) else {}
skipped = bool(record.get("skipped") or summary.get("skipped"))
failed = record.get("ok") is False
if name in {"discover", "snapshot_recent"}:
# 只按上次完成推下一档;进行中由 running 展示,不按墙钟抢跑。
# 上次完成推下一档;skip/失败短补跑;静默窗推到 01:00 后。
interval = (
recent_interval if name == "snapshot_recent"
else discover_interval
)
next_runs[name] = finished_at + interval
if skipped:
next_runs[name] = int(
next_after_snapshot_quiet_window(float(finished_at))
)
elif failed:
next_runs[name] = int(finished_at + retry_seconds)
else:
next_runs[name] = int(
interval_next_run(float(finished_at), float(interval))
)
else:
next_runs[name] = int(next_daily_run(finished_at, at))
if skipped or failed:
# 日更任务失败/skip:短间隔补跑,绝不按成功推到次日。
# snapshot_seed:全站快照占用 / no_source 时 10 分钟补跑。
gap = (
seed_retry_seconds if name == "snapshot_seed"
else retry_seconds
)
next_runs[name] = int(finished_at + gap)
else:
next_runs[name] = int(next_daily_run(finished_at, at))
errors = [
{"task": name, **record}
for name, record in tasks.items()
if isinstance(record, dict) and record.get("ok") is False
if isinstance(record, dict)
and record.get("ok") is False
and not record.get("skipped")
and not (
isinstance(record.get("summary"), dict)
and record["summary"].get("skipped")
)
]
counts = {str(row["status"]): int(row["n"]) for row in conn.execute(
"SELECT status, COUNT(*) AS n FROM video GROUP BY status"
+17 -5
View File
@@ -1,6 +1,6 @@
import {
coverUrl, escapeAttr, escapeHtml, formatCount, formatDate,
formatDelta, formatDuration, formatPercent, metricLabel, statusLabel,
formatDelta, formatDuration, formatPercent, formatSignedPercent, metricLabel, statusLabel,
} from "./format.js";
export function cover(item, className = "thumb") {
@@ -30,9 +30,15 @@ export function avatar(item, className = "avatar") {
loading="lazy" referrerpolicy="no-referrer"></div>`;
}
function growthBaselineTag(growth) {
if (growth?.baseline === "zero") return "<small>今日</small>";
if (growth?.baseline === "open") return "<small>入库后</small>";
return "";
}
export function videoCard(item, {
rank = null, primary = "view", primaryLabel = "", primaryText = null,
showStatus = false,
showStatus = false, growthField = "delta",
} = {}) {
const metrics = item.metrics || {};
const hasMetric = Object.prototype.hasOwnProperty.call(metrics, primary);
@@ -43,7 +49,13 @@ export function videoCard(item, {
const primaryValue = typeof raw === "number" ? formatCount(raw) : raw;
const label = primaryLabel || metricLabel(primary);
const growth = item.growth;
const delta = growth?.available ? growth.delta : null;
const showRate = growthField === "rate";
const rate = showRate && growth?.available ? growth.rate : null;
const delta = !showRate && growth?.available ? growth.delta : null;
const change = showRate
? (rate === null || rate === undefined ? null : formatSignedPercent(rate))
: (delta === null || delta === undefined ? null : formatDelta(delta));
const changeUp = showRate ? Number(rate) >= 0 : Number(delta) >= 0;
const title = escapeHtml(item.title || "(无标题)");
const up = escapeHtml(item.up_name || "未知 UP");
const upMid = item.up_mid;
@@ -73,7 +85,7 @@ export function videoCard(item, {
<span class="primary-stat">
<b>${escapeHtml(primaryValue ?? "—")}</b><small>${escapeHtml(label)}</small>
</span>
${delta !== null ? `<span class="delta ${delta >= 0 ? "up" : "down"}">${formatDelta(delta)}</span>` : ""}
${change !== null ? `<span class="delta ${changeUp ? "up" : "down"}">${change}${growthBaselineTag(growth)}</span>` : ""}
<span class="secondary-stat">互动 ${formatPercent(item.interaction_rate)}</span>
</div>
</div>
@@ -118,7 +130,7 @@ export function metricGrid(item) {
return `<div class="metric-grid">
${["view", "like", "coin", "favorite", "share", "reply", "danmaku"].map((metric) => {
const extra = metric === "view" && growth?.available
? `<span class="metric-delta ${growth.delta >= 0 ? "up" : "down"}">${formatDelta(growth.delta)}</span>`
? `<span class="metric-delta ${growth.delta >= 0 ? "up" : "down"}">${formatDelta(growth.delta)}${growthBaselineTag(growth)}</span>`
: "";
return `<div class="metric-cell">
<span class="metric-label">${escapeHtml(metricLabel(metric))}</span>
+6
View File
@@ -46,6 +46,12 @@ export function formatPercent(value, digits = 2) {
return `${(Number(value) * 100).toFixed(digits)}%`;
}
export function formatSignedPercent(value, digits = 2) {
const text = formatPercent(value, digits);
if (text === "—" || Number(value) <= 0) return text;
return `+${text}`;
}
export function formatDate(timestamp) {
if (!timestamp) return "—";
const date = new Date(Number(timestamp) * 1000);
+3 -3
View File
@@ -353,7 +353,7 @@ a.live-chip:hover { border-color: color-mix(in srgb, var(--accent) 55%, var(--bo
transition: border-color 0.15s cubic-bezier(0.22, 1, 0.36, 1),
background-color 0.15s cubic-bezier(0.22, 1, 0.36, 1);
}
.video-card { grid-template-columns: 128px minmax(0, 1fr); }
.video-card { grid-template-columns: 160px minmax(0, 1fr); }
.up-card { grid-template-columns: 64px minmax(0, 1fr); align-items: center; }
.video-card:hover, .up-card:hover {
border-color: color-mix(in srgb, var(--accent) 45%, var(--border));
@@ -995,7 +995,7 @@ a.live-chip:hover { border-color: color-mix(in srgb, var(--accent) 55%, var(--bo
.up-stats { grid-template-columns: repeat(5, minmax(0, 1fr)); }
.detail-hero { flex-direction: row; align-items: flex-start; }
.hero-cover { width: 300px; flex: 0 0 300px; }
.video-card { grid-template-columns: 144px minmax(0, 1fr); }
.video-card { grid-template-columns: 160px minmax(0, 1fr); }
.secondary-stat.optional { display: inline; }
}
@@ -1048,7 +1048,7 @@ a.live-chip:hover { border-color: color-mix(in srgb, var(--accent) 55%, var(--bo
}
@media (max-width: 380px) {
.video-card { grid-template-columns: 104px minmax(0, 1fr); gap: 8px; }
.video-card { gap: 8px; }
.metric-grid { grid-template-columns: repeat(2, minmax(0, 1fr)); }
.candidate-card { grid-template-columns: 80px minmax(0, 1fr); }
.candidate-thumb { width: 80px; }
+2 -1
View File
@@ -50,6 +50,7 @@ export async function renderAbout(root) {
<li><strong>播放 / 投币 / 收藏 / 评论</strong>:直接取当日快照;弹幕与转发不单列,已并入“互动”。</li>
<li><strong>互动</strong>:点赞 + 投币 + 收藏。综合热度也使用全部七项指标,按配置权重加权。</li>
<li><strong>互动率</strong>:(点赞 + 投币 + 收藏) / 播放;播放不足 100 不计算。</li>
<li><strong>播放增速</strong>:卡片上的百分比是增量 / 基线播放。基线播放不足 1000 不进增速榜。排序用增量 / (基线 + ${analytics.growth_rate_prior})。当日发布没有基线,不进增速榜。</li>
<li><strong>综合热度</strong>:每项先 <code>log1p</code>,再按中位数与 IQR 做稳健标准化,最后加权求和。分数只用于同一批数据内排序,不能跨日期比较。</li>
<li><strong>内容效率</strong>:播放 / 内容小时。内容小时 = 视频总时长(多 P 为各 P 之和)÷ 3600。榜单里的单条视频用收缩后的估计:<code>(播放 + κ·μ₀) / (小时 + κ)</code>,κ 默认 ${analytics.video_shrink_minutes} 分钟、μ₀ 为全库播放/小时;详情页保留原始播放/小时。</li>
<li><strong>互动效率</strong>:互动 / 内容小时,互动口径与收缩方式同上。</li>
@@ -92,7 +93,7 @@ export async function renderAbout(root) {
<li>B 站不公开完播率与单 P 播放,内容小时是“供给时长”,不是“实际观看时长”。</li>
<li>多 P 可能让同一用户产生多次播放计数,本站不单独校正,属于已知偏差。</li>
<li>粉丝数可能包含不活跃账号;粉丝快照暂不参与任何榜单。</li>
<li>样本不足(少于 2 天快照、单视频 UP)会显示提示,不伪造 0。</li>
<li>近 7/30 日增长仍要跨日快照。近 1 日:当日发布按 0 起算;更早发布的新入库按入库时播放量起算,需当天至少两次采样。样本不足时显示提示,不伪造 0。</li>
</ul>
</section>
+1 -1
View File
@@ -69,7 +69,7 @@ export async function renderHome(root) {
const growthEmptyTitle = growth.reason
|| `近 ${growthDays} 日增长暂不可用`;
const growthEmptyHint = growthDays <= 1
? "当前快照不足,完成至少 2 天快照后即可显示播放增量。"
? "当日发布按 0 起算;更早发布的新入库需要当天至少两次采样。"
: `需要至少 ${growthDays + 1} 天快照才能计算近 ${growthDays} 日增长。`;
const updatedLabel = formatClock(overview.updated_at);
const dataDateLabel = formatDataDate(overview.data_date);
@@ -166,6 +166,7 @@ function listHtml(data, scope, sort, startRank) {
return videoCard(item, {
rank: startRank + index, primary: primary.primary,
primaryLabel: primary.label, primaryText: primary.text,
growthField: sort === "growth_rate" ? "rate" : "delta",
});
}).join("")}</div>`;
}
@@ -179,6 +180,21 @@ function durationNote(data) {
return `<div class="section-note">内容效率与互动效率:时长 ≥ ${filter.min_duration_seconds} 秒、发布满 ${filter.maturity_days} 天、播放 ≥ ${filter.min_view}${shrink};时长缺失不参与。</div>`;
}
function growthNote(sort, growthDays, prior) {
if (sort === "growth_rate") {
const strength = Number(prior);
const smooth = Number.isFinite(strength)
? `百分比是真实相对增量;排序按平滑增速(基线 + ${strength})。`
: "百分比是真实相对增量;排序按平滑增速。";
const cold = Number(growthDays) === 1
? "基线播放不足 1000、以及当日发布,不进增速榜。"
: "基线播放不足 1000 不进增速榜。";
return `<div class="section-note">${smooth}${cold}</div>`;
}
if (sort !== "growth" || Number(growthDays) !== 1) return "";
return `<div class="section-note">近 1 日:当日发布按 0 起算;更早发布的新入库按入库时播放量起算,需要当天至少两次采样。</div>`;
}
function bindControls(root) {
root.querySelector("#rank-sort")?.addEventListener("change", (event) => {
replaceQuery({ sort: event.target.value, page: null });
@@ -245,6 +261,7 @@ async function renderScope(root, { query }, scope) {
</header>
${controls({ scope, sort, within, year, growthDays, q, years })}
${durationNote(data)}
${growthNote(sort, growthDays, data.growth_rate_prior)}
${data.available === false ? `<div class="notice-row warning">
<strong>${sort.startsWith("growth") ? "增长数据尚不可用" : "榜单筛选后为空"}</strong>
<span>${escapeHtml(data.reason || "")}</span>
+16 -3
View File
@@ -5,6 +5,15 @@ import {
escapeHtml, formatCount, formatDateTime, formatNumber,
} from "../format.js";
function taskResultLabel(record) {
const summary = record.summary && typeof record.summary === "object" ? record.summary : {};
if (record.skipped || summary.skipped) {
return `<span class="muted">跳过</span>`;
}
if (record.ok) return "成功";
return `<span class="danger">失败</span>`;
}
function taskRows(tasks) {
const entries = Object.entries(tasks || {});
if (!entries.length) return emptyState("暂无任务记录", "运行一次同步、快照或发现后会写入最近结果。");
@@ -15,12 +24,16 @@ function taskRows(tasks) {
${entries.map(([name, record]) => {
const summary = record.summary && typeof record.summary === "object"
? Object.entries(record.summary)
.filter(([key]) => ["added", "removed", "candidates", "notified", "ok", "failed", "total", "completed", "backup_error"].includes(key))
.filter(([key]) => [
"added", "removed", "candidates", "notified", "ok", "failed",
"total", "completed", "backup_error", "skipped", "already_completed",
"reason", "elapsed", "success",
].includes(key))
.map(([key, value]) => `${key}=${value}`).join(" ")
: "";
return `<tr>
<td>${escapeHtml(name)}</td>
<td>${record.ok ? "成功" : `<span class="danger">失败</span>`}</td>
<td>${taskResultLabel(record)}</td>
<td>${escapeHtml(formatDateTime(record.finished_at))}</td>
<td class="summary-cell">${escapeHtml(summary || record.error || "—")}</td>
</tr>`;
@@ -100,7 +113,7 @@ export async function renderSystem(root) {
<div><span>新视频小时同步</span><strong>${running.snapshot_recent ? "运行中" : escapeHtml(formatDateTime(next.snapshot_recent))}</strong></div>
<div><span>新视频搜索</span><strong>${running.discover || running.discovery ? "运行中" : escapeHtml(formatDateTime(next.discover))}</strong></div>
</div>
<p class="section-note">进行中显示「运行中」;空闲时按该任务上次完成时间推算下次运行,不按墙钟抢跑。新视频小时同步与全站快照互斥。</p>
<p class="section-note">进行中显示「运行中」;空闲时按该任务上次完成时间推算下次运行。23:00–01:00(上海)为全站快照窗:发现 / 小时同步 / 候选清理会延后到 01:00 后;粉丝更新不避开。跳过或失败不会记成成功,会短间隔补跑。</p>
</section>
<section class="panel">
+1 -1
View File
@@ -62,7 +62,7 @@ export async function renderVideo(root, { params, query }) {
<span>${escapeHtml(formatDuration(detail.duration))}</span>
${detail.pages ? `<span>${formatNumber(detail.pages)} P</span>` : ""}
${badge(statusLabel(detail.status), `status-${detail.status}`)}
${samples < 2 ? badge("样本不足", "warning") : ""}
${samples < 2 && !detail.growth?.available ? badge("样本不足", "warning") : ""}
</div>
${detail.blacklisted ? `<p class="section-note warning-text">该 UP 在黑名单中;黑名单只影响未来发现,不影响当前视频追踪。</p>` : ""}
${detail.up_follower_backfilled ? `<p class="section-note">UP 粉丝数为回填值,不是真实的入库时点。</p>` : ""}
+27
View File
@@ -12,6 +12,7 @@ from reaction_board.analytics import ( # noqa: E402
METRICS,
age_days,
compute_growth,
smoothed_growth_rate,
content_hours,
eb_rate,
effective_shrink_minutes,
@@ -131,6 +132,18 @@ def test_compute_growth():
no_gap = compute_growth(100, 90, 0)
assert no_gap.available is False
same_day = compute_growth(100, 90, 0, allow_zero_gap=True)
assert same_day.available is True
assert same_day.delta == 10
assert same_day.per_day is None
assert same_day.gap_days == 0
assert abs(same_day.rate - 10 / 90) < 1e-12
from_zero_same_day = compute_growth(50, 0, 0, allow_zero_gap=True)
assert from_zero_same_day.delta == 50
assert from_zero_same_day.rate == 50.0
assert from_zero_same_day.per_day is None
growth = compute_growth(1000, 800, 7)
assert growth.available is True
assert growth.delta == 200
@@ -149,6 +162,19 @@ def test_compute_growth():
assert negative.per_day == -5.0
def test_smoothed_growth_rate():
# 80→512:真实增速 5.4;prior=500 时排序分 432/580
assert abs(smoothed_growth_rate(432, 80, 500) - 432 / 580) < 1e-12
assert abs(smoothed_growth_rate(432, 5000, 500) - 432 / 5500) < 1e-12
assert abs(smoothed_growth_rate(432, 80, 1000) - 432 / 1080) < 1e-12
# prior=0 时排序分退回真实相对增速(基线 ≥ 1)
assert smoothed_growth_rate(432, 80, 0) == 432 / 80
assert smoothed_growth_rate(None, 80, 500) is None
assert smoothed_growth_rate(432, None, 500) is None
assert smoothed_growth_rate(10, 0, 0) is None
assert smoothed_growth_rate(10, 80, -1) is None
def test_normalize_cover_and_hosts():
assert normalize_cover(None) is None
assert normalize_cover("") is None
@@ -239,6 +265,7 @@ def main_test():
test_score_percentiles()
test_interaction_rate_thresholds()
test_compute_growth()
test_smoothed_growth_rate()
test_normalize_cover_and_hosts()
test_content_efficiency_helpers()
test_age_and_maturity()
+429 -2
View File
@@ -51,7 +51,7 @@ def test_init_and_status():
path = write_config(tmp)
code, output = run_cli(["init", "--config", str(path)])
assert code == 0
assert '"schema_version": 7' in output
assert '"schema_version": 8' in output
db = Database(Path(tmp) / "app.db")
db.initialize()
@@ -332,6 +332,422 @@ def test_backfill_followers_dry_run():
assert '"total": 0' in output
def test_help_lists_snapshot_seed():
from reaction_board.cli import _build_parser
parser = _build_parser()
help_text = parser.format_help()
assert "snapshot-seed" in help_text
args = parser.parse_args(["snapshot-seed", "--date", "2026-10-31"])
assert args.command == "snapshot-seed"
assert args.date == "2026-10-31"
def test_snapshot_seed_cli_clones_offline():
from reaction_board.bilibili.models import FavResource
with tempfile.TemporaryDirectory() as tmp:
path = write_config(tmp, cookie="", csrf="")
code, _ = run_cli(["init", "--config", str(path)])
assert code == 0
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
media_id = 11
repo.upsert_from_fav(
FavResource(
aid=1, bvid="BV1xx411c7mD", title="t", up_mid=9,
up_name="u", pubdate=1, duration=10, cover=None,
attr=0, fav_time=None,
),
media_id,
was_added=True,
)
yday, today = "2026-10-30", "2026-10-31"
repo.upsert_snapshot(
yday, 1,
metrics={"view": 10, "like": 1, "coin": 0, "favorite": 0,
"share": 0, "reply": 0, "danmaku": 0},
ok=True, err=None, captured_at=1_800_000_000,
)
repo.kv_set("snapshot:last_completed", yday)
repo.kv_set("snapshot:completed:" + yday, "1")
finally:
db.close()
code, output = run_cli([
"snapshot-seed", "--config", str(path), "--date", today,
])
assert code == 0, output
assert '"reason": "seeded"' in output
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
assert repo.kv_get("snapshot:seeded:" + today) == "1"
row = repo.get_snapshot(today, 1)
assert row is not None and int(row["view"]) == 10
assert row["src"] == "seed"
finally:
db.close()
def test_snapshot_seed_defers_when_snapshot_running():
with tempfile.TemporaryDirectory() as tmp:
path = write_config(tmp, cookie="", csrf="")
code, _ = run_cli(["init", "--config", str(path)])
assert code == 0
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
repo.kv_set(
"task:running:snapshot",
json.dumps({"name": "snapshot", "started_at": 1}),
)
repo.kv_set("snapshot:last_completed", "2026-10-30")
finally:
db.close()
code, output = run_cli([
"snapshot-seed", "--config", str(path), "--date", "2026-10-31",
])
assert code == 1, output
assert "snapshot_running" in output
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
last = json.loads(repo.kv_get("task:last:snapshot_seed"))
assert last["ok"] is False
assert last["skipped"] is True
assert last["error"] == "snapshot_running"
assert repo.kv_get("task:running:snapshot_seed") is None
assert repo.kv_get("task:running:snapshot") is not None
finally:
db.close()
def test_snapshot_seed_no_source_is_non_success():
with tempfile.TemporaryDirectory() as tmp:
path = write_config(tmp, cookie="", csrf="")
code, _ = run_cli(["init", "--config", str(path)])
assert code == 0
code, output = run_cli([
"snapshot-seed", "--config", str(path), "--date", "2026-10-31",
])
assert code == 1, output
assert "no_source" in output
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
last = json.loads(repo.kv_get("task:last:snapshot_seed"))
assert last["ok"] is False and last["skipped"] is True
assert last["error"] == "no_source"
finally:
db.close()
def test_defer_snapshot_seed_schedules_ten_minutes():
"""调度辅助:skip 记非成功,next_run = finished + 600。"""
from reaction_board.logging_json import JsonLogger
with tempfile.TemporaryDirectory() as tmp:
path = write_config(tmp, cookie="", csrf="")
code, _ = run_cli(["init", "--config", str(path)])
assert code == 0
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
logger = JsonLogger()
started = 1_800_000_000
finished = started + 5
ctrl = cli._defer_snapshot_seed(
repo, logger, started_at=started, reason="snapshot_running",
finished_at=finished,
)
assert ctrl.ok is False
assert ctrl.error == "snapshot_running"
assert abs(ctrl.next_run - (finished + cli._SEED_RETRY_SECONDS)) < 0.01
last = json.loads(repo.kv_get("task:last:snapshot_seed"))
assert last["skipped"] is True and last["ok"] is False
finally:
db.close()
def test_snapshot_seed_same_day_completed_succeeds():
"""当日已完成时手动 snapshot-seed 应成功 already_completed,不走 no_source。"""
with tempfile.TemporaryDirectory() as tmp:
path = write_config(tmp, cookie="", csrf="")
code, _ = run_cli(["init", "--config", str(path)])
assert code == 0
today = "2026-10-31"
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
repo.kv_set("snapshot:last_completed", today)
repo.kv_set("snapshot:completed:" + today, "1")
finally:
db.close()
code, output = run_cli([
"snapshot-seed", "--config", str(path), "--date", today,
])
assert code == 0, output
assert '"reason": "already_completed"' in output
assert "no_source" not in output
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
last = json.loads(repo.kv_get("task:last:snapshot_seed"))
assert last["ok"] is True
finally:
db.close()
def test_candidate_cleanup_failure_retries_soon():
"""candidate_cleanup 异常应记失败任务并约 5 分钟补跑,而非落到 86400 间隔。"""
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_client = cli._client
original_discovery = cli.DiscoveryService
original_delete = Repository.delete_expired_candidates
captured = {}
def boom(self, *args, **kwargs):
raise RuntimeError("cleanup db boom")
def fake_client(credentials, *args, **kwargs):
return FakeClient()
# 只跑 cleanup:把其它任务 next_run 推远;通过 Scheduler 包装拦截。
original_scheduler = cli.Scheduler
class CaptureScheduler(original_scheduler):
def __init__(self, tasks, **kwargs):
far = 1_000_000_000_000.0
for task in tasks:
if task.name != "candidate_cleanup":
task.next_run = far
else:
task.next_run = 0.0
super().__init__(tasks, **kwargs)
captured["scheduler"] = self
cli._client = fake_client
cli.DiscoveryService = FakeDiscovery
cli.Scheduler = CaptureScheduler
Repository.delete_expired_candidates = boom
try:
code, output = run_cli(["run", "--once", "--config", str(path)])
finally:
cli._client = original_client
cli.DiscoveryService = original_discovery
cli.Scheduler = original_scheduler
Repository.delete_expired_candidates = original_delete
assert code == 0, output
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
last = json.loads(repo.kv_get("task:last:candidate_cleanup"))
assert last["ok"] is False
assert "cleanup db boom" in last["error"]
finally:
db.close()
scheduler = captured["scheduler"]
cleanup_task = next(t for t in scheduler.tasks if t.name == "candidate_cleanup")
# next_run ≈ finished + 300;允许几秒时钟误差
assert cleanup_task.next_run > 0
delta = cleanup_task.next_run - __import__("time").time()
assert 250 <= delta <= 360, delta
def test_initial_daily_next_run_uses_wall_clock_not_zero():
"""常驻启动:无上次记录时应对齐下一档墙钟,不得为 0。"""
from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ
from datetime import datetime
with tempfile.TemporaryDirectory() as tmp:
path = write_config(tmp, cookie="", csrf="")
assert run_cli(["init", "--config", str(path)])[0] == 0
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
# 2026-11-03 10:00 上海
now = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp()
nxt = cli._initial_daily_next_run(
repo, "snapshot", "23:15", now=now,
)
assert nxt == next_daily_run(now, "23:15")
assert nxt > now
# 成功记录在昨晚 → 仍是今晚 23:15
yday_done = datetime(2026, 11, 2, 23, 20, tzinfo=SHANGHAI_TZ).timestamp()
repo.kv_set(
"task:last:snapshot",
json.dumps({
"ok": True, "finished_at": int(yday_done),
"started_at": int(yday_done) - 10,
}),
)
nxt2 = cli._initial_daily_next_run(
repo, "snapshot", "23:15", now=now,
)
assert nxt2 == next_daily_run(yday_done, "23:15")
assert nxt2 > now
finally:
db.close()
def test_reached_daily_slot_gates_force_seed():
from reaction_board.timeutil import SHANGHAI_TZ
from datetime import datetime
before = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp()
at = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp()
after = datetime(2026, 11, 3, 23, 16, tzinfo=SHANGHAI_TZ).timestamp()
assert cli._reached_daily_slot("23:15", before) is False
assert cli._reached_daily_slot("23:15", at) is True
assert cli._reached_daily_slot("23:15", after) is True
def test_restart_after_seed_schedules_snapshot_at_wall_clock():
"""午夜 seed 后重启:snapshot 应排到当晚 snapshot_at,首轮 run_due 不得 force 全量。"""
from reaction_board.scheduler import ScheduledTask, Scheduler
from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ, shanghai_date
from datetime import datetime
with tempfile.TemporaryDirectory() as tmp:
path = write_config(tmp, cookie="", csrf="")
assert run_cli(["init", "--config", str(path)])[0] == 0
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
# 白天:假定今日已 seed
now = datetime(2026, 11, 3, 10, 30, tzinfo=SHANGHAI_TZ).timestamp()
today = shanghai_date(now)
repo.kv_set(f"snapshot:seeded:{today}", "1")
repo.kv_set(
"task:last:snapshot_seed",
json.dumps({
"ok": True,
"finished_at": int(
datetime(2026, 11, 3, 0, 1, tzinfo=SHANGHAI_TZ).timestamp()
),
"started_at": int(
datetime(2026, 11, 3, 0, 0, tzinfo=SHANGHAI_TZ).timestamp()
),
"summary": {"reason": "seeded"},
}),
)
snapshot_at = "23:15"
seed_at = "00:00"
snapshot_next = cli._initial_daily_next_run(
repo, "snapshot", snapshot_at, now=now,
)
seed_next = cli._initial_daily_next_run(
repo, "snapshot_seed", seed_at, now=now,
)
fans_next = cli._initial_daily_next_run(
repo, "fans", snapshot_at, now=now,
)
assert snapshot_next == next_daily_run(now, snapshot_at)
assert fans_next == next_daily_run(now, snapshot_at)
# seed 已成功 → 下一档是次日 00:00
assert seed_next == next_daily_run(
float(json.loads(repo.kv_get("task:last:snapshot_seed"))["finished_at"]),
seed_at,
)
assert seed_next > now
force_calls = []
def run_snapshot():
force = (
repo.kv_get(f"snapshot:seeded:{today}") == "1"
and cli._reached_daily_slot(snapshot_at, now)
)
force_calls.append(force)
if force:
raise AssertionError("daytime restart must not force-seal seed")
return None
clock = [now]
scheduler = Scheduler(
[
ScheduledTask("snapshot_seed", 86400.0, lambda: None, next_run=seed_next),
ScheduledTask("snapshot", 86400.0, run_snapshot, next_run=snapshot_next),
ScheduledTask("fans", 86400.0, lambda: None, next_run=fans_next),
],
clock=lambda: clock[0],
)
assert scheduler.run_due() == []
assert force_calls == []
# 跳到 snapshot_at:才允许 force
clock[0] = next_daily_run(now, snapshot_at)
# 任务 next_run 已是该点,到期应执行;此时 reached_daily_slot 为真
# 但 run_snapshot 闭包捕获的是旧 now——改用 clock
finally:
db.close()
# 单独验证:到达 slot 后 force 门打开
slot = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp()
assert cli._reached_daily_slot("23:15", slot) is True
def test_daytime_seed_marker_does_not_force_without_slot():
"""即使 seeded=1,未到 snapshot_at 也不得 force(防守:即便误触发 run_snapshot)。"""
from reaction_board.timeutil import SHANGHAI_TZ
from datetime import datetime
with tempfile.TemporaryDirectory() as tmp:
path = write_config(tmp, cookie="", csrf="")
assert run_cli(["init", "--config", str(path)])[0] == 0
db = Database(Path(tmp) / "app.db")
db.initialize()
try:
repo = Repository(db)
now = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp()
today = "2026-11-03"
repo.kv_set(f"snapshot:seeded:{today}", "1")
force_seed = (
repo.kv_get(f"snapshot:seeded:{today}") == "1"
and cli._reached_daily_slot("23:15", now)
)
assert force_seed is False
finally:
db.close()
def main_test():
test_init_and_status()
test_missing_config()
@@ -345,7 +761,18 @@ def main_test():
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/错误路径测试全部通过")
test_help_lists_snapshot_seed()
test_snapshot_seed_cli_clones_offline()
test_snapshot_seed_defers_when_snapshot_running()
test_snapshot_seed_no_source_is_non_success()
test_snapshot_seed_same_day_completed_succeeds()
test_defer_snapshot_seed_schedules_ten_minutes()
test_candidate_cleanup_failure_retries_soon()
test_initial_daily_next_run_uses_wall_clock_not_zero()
test_reached_daily_slot_gates_force_seed()
test_restart_after_seed_schedules_snapshot_at_wall_clock()
test_daytime_seed_marker_does_not_force_without_slot()
print("OK: CLI init/sync/snapshot/snapshot-seed/auth-check/run/status/backfill/错误路径测试全部通过")
if __name__ == "__main__":
+4
View File
@@ -146,6 +146,7 @@ def test_analytics_defaults_and_validation():
assert cfg.analytics.video_shrink_minutes == 60
assert cfg.analytics.shrink_minutes is None
assert cfg.analytics.home_growth_days == "auto"
assert cfg.analytics.growth_rate_prior == 1000
assert cfg.schedule.recent_snapshot_minutes == 60
assert cfg.schedule.recent_video_days == 2
assert cfg.schedule.snapshot_at == "23:15"
@@ -159,6 +160,7 @@ def test_analytics_defaults_and_validation():
"video_shrink_minutes": 30,
"shrink_minutes": 90,
"home_growth_days": 1,
"growth_rate_prior": 2000,
}
data["schedule"] = {
"discover_minutes": 30,
@@ -176,6 +178,7 @@ def test_analytics_defaults_and_validation():
assert cfg.analytics.video_shrink_minutes == 30
assert cfg.analytics.shrink_minutes == 90
assert cfg.analytics.home_growth_days == 1
assert cfg.analytics.growth_rate_prior == 2000
assert cfg.schedule.recent_snapshot_minutes == 15
assert cfg.schedule.recent_video_days == 3
@@ -194,6 +197,7 @@ def test_analytics_defaults_and_validation():
({"home_growth_days": 3}, "home_growth_days"),
({"home_growth_days": "7"}, "home_growth_days"),
({"home_growth_days": 0}, "home_growth_days"),
({"growth_rate_prior": -1}, "growth_rate_prior"),
):
data = base_config()
data["analytics"] = bad
+4 -4
View File
@@ -24,7 +24,7 @@ def test_write_read_transaction():
with tempfile.TemporaryDirectory() as tmp:
db = make_db(tmp)
try:
assert db.schema_version == 7
assert db.schema_version == 8
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"
@@ -94,8 +94,8 @@ def test_close_and_reinitialize():
db2 = make_db(tmp)
try:
assert db2.schema_version == 7
assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 7
assert db2.schema_version == 8
assert db2.read(lambda c: c.execute("SELECT MAX(version) FROM schema_version").fetchone()[0]) == 8
finally:
db2.close()
@@ -123,7 +123,7 @@ def test_migration_backup_on_existing_db():
db = Database(path, backup_dir=Path(tmp) / "backup")
try:
applied = db.initialize()
assert applied == [1, 2, 3, 4, 5, 6, 7]
assert applied == [1, 2, 3, 4, 5, 6, 7, 8]
assert len(list((Path(tmp) / "backup").glob("app-pre-v0-*.db"))) == 1
finally:
db.close()
+46 -8
View File
@@ -38,8 +38,8 @@ def test_fresh_and_idempotent():
conn = sqlite3.connect(Path(tmp) / "app.db")
try:
applied = run_migrations(conn)
assert applied == [1, 2, 3, 4, 5, 6, 7]
assert current_version(conn) == 7
assert applied == [1, 2, 3, 4, 5, 6, 7, 8]
assert current_version(conn) == 8
assert EXPECTED_TABLES <= table_names(conn)
assert "schema_version" in table_names(conn)
columns = {
@@ -70,10 +70,14 @@ def test_fresh_and_idempotent():
assert {
"stat_date", "ip", "path", "hits", "first_at", "last_at",
} <= visit_columns
snapshot_columns = {
row[1] for row in conn.execute("PRAGMA table_info(snapshot)")
}
assert {"view_open", "open_captured_at"} <= snapshot_columns
# 再跑一次不应重复应用
assert run_migrations(conn) == []
row = conn.execute("SELECT COUNT(*) FROM schema_version").fetchone()
assert row[0] == 7
assert row[0] == 8
finally:
conn.close()
@@ -88,15 +92,15 @@ def test_backup_before_upgrade():
conn.execute("INSERT INTO kv(k, v) VALUES ('hello', 'world')")
conn.commit()
v8 = Migration(8, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",))
v9 = Migration(9, "add probe", ("CREATE TABLE probe (id INTEGER PRIMARY KEY)",))
applied = run_migrations(
conn,
backup_dir=backup_dir,
now=1_800_000_000,
migrations=MIGRATIONS + (v8,),
migrations=MIGRATIONS + (v9,),
)
assert applied == [8]
backups = list(backup_dir.glob("app-pre-v7-*.db"))
assert applied == [9]
backups = list(backup_dir.glob("app-pre-v8-*.db"))
assert len(backups) == 1
check = sqlite3.connect(backups[0])
try:
@@ -181,13 +185,46 @@ def test_v5_normalizes_unfaved():
)
conn.commit()
applied = run_migrations(conn)
assert applied == [5, 6, 7]
assert applied == [5, 6, 7, 8]
rows = dict(conn.execute("SELECT aid, status FROM video").fetchall())
assert rows == {1: "active", 2: "dropped"}
finally:
conn.close()
def test_v8_backfills_view_open():
with tempfile.TemporaryDirectory() as tmp:
conn = sqlite3.connect(Path(tmp) / "app.db")
try:
run_migrations(conn, migrations=MIGRATIONS[:7])
conn.execute(
"INSERT INTO video(aid, title, pubdate, status, first_seen) "
"VALUES (1, 't', 1, 'active', 1)"
)
conn.execute(
'INSERT INTO snapshot(stat_date, aid, "view", captured_at, ok) '
"VALUES ('2026-10-31', 1, 500, 100, 1)"
)
conn.execute(
'INSERT INTO snapshot(stat_date, aid, "view", captured_at, ok) '
"VALUES ('2026-10-30', 1, NULL, 90, 0)"
)
conn.commit()
assert run_migrations(conn) == [8]
ok_row = conn.execute(
"SELECT view_open, open_captured_at FROM snapshot "
"WHERE stat_date='2026-10-31'"
).fetchone()
assert ok_row[0] == 500 and ok_row[1] == 100
fail_row = conn.execute(
"SELECT view_open, open_captured_at FROM snapshot "
"WHERE stat_date='2026-10-30'"
).fetchone()
assert fail_row[0] is None and fail_row[1] is None
finally:
conn.close()
def main():
test_fresh_and_idempotent()
test_backup_before_upgrade()
@@ -195,6 +232,7 @@ def main():
test_prune_backups()
test_validate_migration_order()
test_v5_normalizes_unfaved()
test_v8_backfills_view_open()
print("OK: SQLite 迁移、备份与回滚测试全部通过")
+54
View File
@@ -296,6 +296,59 @@ def test_pending_snapshot_aids():
db.close()
def test_upsert_snapshot_preserves_view_open():
with tempfile.TemporaryDirectory() as tmp:
db, repo = setup(tmp)
try:
metrics = {
"view": 10, "like": 1, "favorite": 0, "coin": 0,
"share": 0, "reply": 0, "danmaku": 0,
}
repo.upsert_snapshot(
"2026-10-31", 1, metrics=metrics, ok=True, err=None, captured_at=10,
)
first = repo.get_snapshot("2026-10-31", 1)
assert first["view"] == 10
assert first["view_open"] == 10
assert first["open_captured_at"] == 10
metrics["view"] = 40
repo.upsert_snapshot(
"2026-10-31", 1, metrics=metrics, ok=True, err=None, captured_at=20,
)
second = repo.get_snapshot("2026-10-31", 1)
assert second["view"] == 40
assert second["view_open"] == 10
assert second["open_captured_at"] == 10
assert second["captured_at"] == 20
repo.upsert_snapshot(
"2026-10-31", 2, metrics=None, ok=False, err="x", captured_at=1,
)
failed = repo.get_snapshot("2026-10-31", 2)
assert failed["view_open"] is None
metrics["view"] = 7
repo.upsert_snapshot(
"2026-10-31", 2, metrics=metrics, ok=True, err=None, captured_at=5,
)
recovered = repo.get_snapshot("2026-10-31", 2)
assert recovered["view"] == 7
assert recovered["view_open"] == 7
assert recovered["open_captured_at"] == 5
copied = repo.copy_ok_snapshots(
"2026-10-31", "2026-11-01", captured_at=30, src="seed",
)
assert copied == 2
seeded = repo.get_snapshot("2026-11-01", 1)
assert seeded["view"] == 40
assert seeded["view_open"] == 40
assert seeded["open_captured_at"] == 30
assert seeded["src"] == "seed"
finally:
db.close()
def test_up_follower_snapshot_and_backfill():
with tempfile.TemporaryDirectory() as tmp:
db, repo = setup(tmp)
@@ -367,6 +420,7 @@ def main():
test_primary_folder_is_order_independent()
test_dropped_invalid_preserves_original_title()
test_pending_snapshot_aids()
test_upsert_snapshot_preserves_view_open()
test_up_follower_snapshot_and_backfill()
print("OK: 收藏夹/视频状态机与黑名单测试全部通过")
+103 -1
View File
@@ -8,7 +8,7 @@ import threading
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.stdout.reconfigure(encoding="utf-8")
from reaction_board.scheduler import ScheduledTask, Scheduler # noqa: E402
from reaction_board.scheduler import ScheduledTask, Scheduler, TaskControl # noqa: E402
def test_run_when_due_and_reschedule():
@@ -128,9 +128,111 @@ def test_next_run_fn_controls_reschedule():
assert calls == [1_000.0, 1_123.0]
def test_task_control_overrides_ok_and_next_run():
clock = [1_000.0]
task = ScheduledTask(
"defer",
60,
lambda: TaskControl(ok=False, next_run=1_500.0, error="quiet"),
)
scheduler = Scheduler([task], clock=lambda: clock[0])
results = scheduler.run_due()
assert len(results) == 1
assert results[0].ok is False
assert results[0].error == "quiet"
assert task.next_run == 1_500.0
# 未到 next_run 不重跑
clock[0] = 1_499.0
assert scheduler.run_due() == []
clock[0] = 1_500.0
task.fn = lambda: TaskControl(ok=True, next_run=2_000.0)
results = scheduler.run_due()
assert results[0].ok is True
assert task.next_run == 2_000.0
class _CaptureLogger:
def __init__(self):
self.events = []
def info(self, event, **fields):
self.events.append(("info", event, fields))
def warning(self, event, **fields):
self.events.append(("warning", event, fields))
def error(self, event, **fields):
self.events.append(("error", event, fields))
def test_task_control_ok_false_logs_fail_not_ok():
"""TaskControl(ok=False) 不得记 scheduler.task_ok(Skip ≠ success)。"""
clock = [1_000.0]
logger = _CaptureLogger()
task = ScheduledTask(
"quiet",
60,
lambda: TaskControl(ok=False, next_run=1_500.0, error="snapshot_quiet_window"),
)
scheduler = Scheduler([task], clock=lambda: clock[0], logger=logger)
results = scheduler.run_due()
assert results[0].ok is False
assert results[0].error == "snapshot_quiet_window"
assert ("info", "scheduler.task_ok", {"task": "quiet"}) not in logger.events
fail_events = [e for e in logger.events if e[1] == "scheduler.task_fail"]
assert len(fail_events) == 1
assert fail_events[0][0] == "warning"
assert fail_events[0][2]["task"] == "quiet"
assert fail_events[0][2]["error"] == "snapshot_quiet_window"
def test_task_control_ok_true_still_logs_task_ok():
clock = [1_000.0]
logger = _CaptureLogger()
task = ScheduledTask(
"ok_task",
60,
lambda: TaskControl(ok=True, next_run=1_200.0),
)
scheduler = Scheduler([task], clock=lambda: clock[0], logger=logger)
results = scheduler.run_due()
assert results[0].ok is True
assert ("info", "scheduler.task_ok", {"task": "ok_task"}) in logger.events
assert not any(e[1] == "scheduler.task_fail" for e in logger.events)
def test_raised_exception_logs_task_fail():
clock = [100.0]
logger = _CaptureLogger()
def boom():
raise RuntimeError("db down")
scheduler = Scheduler(
[ScheduledTask("bad", 10, boom)],
clock=lambda: clock[0],
logger=logger,
)
results = scheduler.run_due()
assert results[0].ok is False
assert results[0].error == "db down"
assert not any(e[1] == "scheduler.task_ok" for e in logger.events)
assert not any(e[1] == "scheduler.task_fail" for e in logger.events)
err_events = [e for e in logger.events if e[1] == "scheduler.task_error"]
assert len(err_events) == 1
assert err_events[0][0] == "error"
assert err_events[0][2]["error"] == "db down"
def main():
test_run_when_due_and_reschedule()
test_next_run_fn_controls_reschedule()
test_task_control_overrides_ok_and_next_run()
test_task_control_ok_false_logs_fail_not_ok()
test_task_control_ok_true_still_logs_task_ok()
test_raised_exception_logs_task_fail()
test_task_error_does_not_block_other_tasks()
test_run_forever_with_stop_event()
test_max_iterations_and_validation()
+42 -3
View File
@@ -444,19 +444,22 @@ def test_run_recent_videos_refreshes_without_completion_markers():
1: view_for(1, view=111),
2: view_for(2, view=222),
3: view_for(3, view=333),
4: view_for(4, view=444),
})
db, repo, service = make_service(tmp, transport, track_fans=())
try:
# 1/2 在墙钟近 2 天内;3 为旧片
# 1/2 在墙钟近 2 天内;3 为旧片;4 发布已久但当天入库,不进小时任务
add_video(repo, 1)
add_video(repo, 2)
add_video(repo, 3)
add_video(repo, 4)
repo.db.write(lambda conn: conn.executemany(
"UPDATE video SET pubdate = ? WHERE aid = ?",
[
(clock - 3600, 1),
(clock - 86400, 2),
(clock - 10 * 86400, 3),
(clock - 10 * 86400, 4),
],
))
# 预置同日 ok=1,验证小时任务可覆盖刷新
@@ -477,23 +480,27 @@ def test_run_recent_videos_refreshes_without_completion_markers():
)) == {1, 2}
row = repo.db.read(lambda conn: dict(conn.execute(
"SELECT view, ok, captured_at FROM snapshot WHERE stat_date=? AND aid=1",
"SELECT view, ok, captured_at, view_open FROM snapshot "
"WHERE stat_date=? AND aid=1",
(DATE,),
).fetchone()))
assert row["view"] == 111
assert row["ok"] == 1
assert row["captured_at"] == clock
assert row["view_open"] == 1
# 完成日 / 备份标记保持原样
assert repo.kv_get(service.completed_key(DATE)) is None
assert repo.kv_get("snapshot:last_completed") is None
assert repo.kv_get(service.backup_key(DATE)) is None
assert transport.count("/web-interface/view") == 2
assert repo.get_snapshot(DATE, 4) is None
# 全量任务随后才写完成日;且已 ok=1 的跳过
full = service.run_videos(date=DATE)
assert full.skipped >= 2 # 1、2 已被小时任务写过
assert full.success >= 1 # 至少旧片 3
assert full.success >= 1 # 至少旧片 3、4
assert repo.get_snapshot(DATE, 4) is not None
assert repo.kv_get(service.completed_key(DATE)) == "1"
assert repo.kv_get("snapshot:last_completed") == DATE
finally:
@@ -530,6 +537,36 @@ def test_full_snapshot_skips_ok_rows_including_hourly_refresh():
def test_seed_same_day_completed_is_already_completed_not_no_source():
"""全站完成后 last_completed==target:应 already_completed,不误判 no_source。"""
clock = 1_800_000_000
with tempfile.TemporaryDirectory() as tmp:
transport = SnapshotTransport({1: view_for(1)})
db, repo, service = make_service(tmp, transport, track_fans=())
try:
add_video(repo, 1)
today = "2026-10-31"
repo.upsert_snapshot(
today, 1,
metrics={"view": 10, "like": 1, "coin": 0, "favorite": 0,
"share": 0, "reply": 0, "danmaku": 0},
ok=True, err=None, captured_at=clock,
)
# 正常收尾态:last_completed 已推进到当日,且 completed 标记在位
repo.kv_set("snapshot:last_completed", today)
repo.kv_set(service.completed_key(today), "1")
summary = service.run_seed_from_last_completed(date=today)
assert summary.reason == "already_completed"
assert summary.already_completed is True
assert summary.completed is True
assert summary.reason != "no_source"
assert transport.count("/web-interface/view") == 0
finally:
db.close()
def test_seed_from_last_completed_does_not_mark_completed():
clock = 1_800_000_000
with tempfile.TemporaryDirectory() as tmp:
@@ -562,6 +599,7 @@ def test_seed_from_last_completed_does_not_mark_completed():
row = repo.get_snapshot(today, 1)
assert row is not None
assert int(row["view"]) == 10
assert int(row["view_open"]) == 10
assert row["src"] == "seed"
fan = repo.get_up_stat(today, 37754047)
assert fan is not None and int(fan["follower"]) == 100
@@ -638,6 +676,7 @@ def main():
test_backup_failure_retries_without_api()
test_run_recent_videos_refreshes_without_completion_markers()
test_full_snapshot_skips_ok_rows_including_hourly_refresh()
test_seed_same_day_completed_is_already_completed_not_no_source()
test_seed_from_last_completed_does_not_mark_completed()
test_seeded_day_skips_without_force_and_seals_with_force()
print("OK: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过")
+32
View File
@@ -11,7 +11,10 @@ sys.stdout.reconfigure(encoding="utf-8")
from reaction_board.timeutil import ( # noqa: E402
SHANGHAI_TZ,
day_bounds,
in_snapshot_quiet_window,
interval_next_run,
is_date,
next_after_snapshot_quiet_window,
next_daily_run,
parse_hhmm,
shanghai_date,
@@ -72,11 +75,40 @@ def test_day_bounds_and_date_validation():
raise AssertionError("非法日期应失败")
def test_snapshot_quiet_window():
assert in_snapshot_quiet_window(ts("2026-11-02 23:00:00"))
assert in_snapshot_quiet_window(ts("2026-11-02 23:59:59"))
assert in_snapshot_quiet_window(ts("2026-11-03 00:00:00"))
assert in_snapshot_quiet_window(ts("2026-11-03 00:59:59"))
assert not in_snapshot_quiet_window(ts("2026-11-03 01:00:00"))
assert not in_snapshot_quiet_window(ts("2026-11-02 22:59:59"))
assert not in_snapshot_quiet_window(ts("2026-11-02 12:00:00"))
assert next_after_snapshot_quiet_window(ts("2026-11-02 22:00:00")) == ts(
"2026-11-02 22:00:00"
)
assert next_after_snapshot_quiet_window(ts("2026-11-02 23:15:00")) == ts(
"2026-11-03 01:00:00"
)
assert next_after_snapshot_quiet_window(ts("2026-11-03 00:30:00")) == ts(
"2026-11-03 01:00:00"
)
# 22:50 + 60min = 23:50 → 推到次日 01:00
assert interval_next_run(ts("2026-11-02 22:50:00"), 3600) == ts(
"2026-11-03 01:00:00"
)
assert interval_next_run(ts("2026-11-02 12:00:00"), 3600) == ts(
"2026-11-02 13:00:00"
)
def main():
test_shanghai_date_boundary()
test_parse_hhmm()
test_next_daily_run()
test_day_bounds_and_date_validation()
test_snapshot_quiet_window()
print("OK: 上海时区与每日时间计算测试全部通过")
+282 -10
View File
@@ -15,7 +15,7 @@ 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
from tests.web_fixtures import DATES, FOLDER, build_db, resource, write_config # noqa: E402
FIXED_NOW = 1_793_459_000 # 2026-10-31 前后
@@ -46,7 +46,7 @@ def test_overview():
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["view_delta"] == (100 + 100 + 120) # 101/102 跨日 + 103 当日发布
assert overview["snapshot_health"]["total"] == 3
assert overview["folders"][0]["media_count"] == 3
assert overview["folders"][0]["ratio"] < 1
@@ -484,7 +484,7 @@ def test_about_payload():
assert about["snapshot_days"] == 3
assert about["snapshot_first_date"] == DATES[0]
assert about["fans_start_date"] == DATES[0]
assert about["schema_version"] == 7
assert about["schema_version"] == 8
assert about["analytics"] == {
"min_duration_seconds": 600,
"maturity_days": 2,
@@ -492,6 +492,7 @@ def test_about_payload():
"video_shrink_minutes": 60,
"shrink_minutes": None,
"home_growth_days": "auto",
"growth_rate_prior": 1000,
}
assert about["heat_weights"]["view"] == 0.30
assert about["followers"]["videos_total"] == 3
@@ -577,9 +578,13 @@ def test_growth_auto_falls_back_to_one_day():
assert auto["available"] is True
assert auto["growth_days"] == 1
assert auto["days"] == 1
assert {item["aid"] for item in auto["items"]} == {101, 102}
assert {item["aid"] for item in auto["items"]} == {101, 102, 103}
first = next(item for item in auto["items"] if item["aid"] == 101)
assert first["growth"]["delta"] == 100 # 相邻日
assert first["growth"]["baseline"] == "snapshot"
today = next(item for item in auto["items"] if item["aid"] == 103)
assert today["growth"]["baseline"] == "zero"
assert today["growth"]["delta"] == 120
forced7 = q.list_videos(sort="growth", growth_days=7)
assert forced7["available"] is False
@@ -605,12 +610,15 @@ def test_growth_auto_falls_back_to_one_day():
def test_snapshot_recent_next_run_follows_completion():
import json
from datetime import datetime
from reaction_board.timeutil import SHANGHAI_TZ, interval_next_run
with tempfile.TemporaryDirectory() as tmp:
db, q = setup(tmp)
try:
repo = Repository(db)
finished = FIXED_NOW - 10
# 午后完成:间隔后仍在静默窗外,应严格 finished+3600
finished = int(datetime(2026, 10, 31, 15, 0, 0, tzinfo=SHANGHAI_TZ).timestamp())
repo.kv_set(
"task:last:snapshot_recent",
json.dumps({
@@ -622,6 +630,128 @@ def test_snapshot_recent_next_run_follows_completion():
DASHBOARD_CACHE.clear()
system = q.system()
assert system["next_runs"]["snapshot_recent"] == finished + 3600
assert system["next_runs"]["snapshot_recent"] == int(
interval_next_run(float(finished), 3600)
)
finally:
db.close()
def test_snapshot_recent_next_run_defers_quiet_window_and_retry_on_skip():
import json
from datetime import datetime
from reaction_board.timeutil import SHANGHAI_TZ, next_after_snapshot_quiet_window
with tempfile.TemporaryDirectory() as tmp:
db, q = setup(tmp)
try:
repo = Repository(db)
# 22:50 成功 → 下一档 23:50 落入静默窗 → 推到 01:00
finished = int(datetime(2026, 10, 31, 22, 50, 0, tzinfo=SHANGHAI_TZ).timestamp())
repo.kv_set(
"task:last:snapshot_recent",
json.dumps({
"name": "snapshot_recent", "ok": True,
"started_at": finished - 5, "finished_at": finished,
"error": None, "summary": {},
}),
)
DASHBOARD_CACHE.clear()
system = q.system()
expected = int(datetime(2026, 11, 1, 1, 0, 0, tzinfo=SHANGHAI_TZ).timestamp())
assert system["next_runs"]["snapshot_recent"] == expected
# skip 记录:补跑到窗后 01:00,不得当成功推进
quiet = int(datetime(2026, 10, 31, 23, 20, 0, tzinfo=SHANGHAI_TZ).timestamp())
repo.kv_set(
"task:last:snapshot_recent",
json.dumps({
"name": "snapshot_recent", "ok": False, "skipped": True,
"started_at": quiet, "finished_at": quiet,
"error": "snapshot_quiet_window",
"summary": {"skipped": True, "reason": "snapshot_quiet_window"},
}),
)
DASHBOARD_CACHE.clear()
system = q.system()
assert system["next_runs"]["snapshot_recent"] == int(
next_after_snapshot_quiet_window(float(quiet))
)
assert not any(e["task"] == "snapshot_recent" for e in system["errors"])
finally:
db.close()
def test_snapshot_failed_next_run_retries_soon_not_next_day():
import json
from datetime import datetime
from reaction_board.timeutil import SHANGHAI_TZ, next_daily_run
with tempfile.TemporaryDirectory() as tmp:
db, q = setup(tmp)
try:
repo = Repository(db)
finished = int(datetime(2026, 10, 31, 23, 15, 0, tzinfo=SHANGHAI_TZ).timestamp())
repo.kv_set(
"task:last:snapshot",
json.dumps({
"name": "snapshot", "ok": False,
"started_at": finished - 1, "finished_at": finished,
"error": "boom", "summary": {"completed": False},
}),
)
DASHBOARD_CACHE.clear()
system = q.system()
assert system["next_runs"]["snapshot"] == finished + 300
assert system["next_runs"]["snapshot"] != int(
next_daily_run(finished, "23:15")
)
assert any(e["task"] == "snapshot" for e in system["errors"])
finally:
db.close()
def test_snapshot_seed_skip_next_run_retries_ten_minutes():
import json
from datetime import datetime
from reaction_board.timeutil import SHANGHAI_TZ, next_daily_run
with tempfile.TemporaryDirectory() as tmp:
db, q = setup(tmp)
try:
repo = Repository(db)
finished = int(datetime(2026, 11, 1, 0, 0, 0, tzinfo=SHANGHAI_TZ).timestamp())
repo.kv_set(
"task:last:snapshot_seed",
json.dumps({
"name": "snapshot_seed", "ok": False, "skipped": True,
"started_at": finished - 1, "finished_at": finished,
"error": "snapshot_running",
"summary": {"skipped": True, "reason": "snapshot_running"},
}),
)
DASHBOARD_CACHE.clear()
system = q.system()
assert system["next_runs"]["snapshot_seed"] == finished + 600
assert system["next_runs"]["snapshot_seed"] != int(
next_daily_run(finished, "00:00")
)
# skip 不算 errors
assert not any(e["task"] == "snapshot_seed" for e in system["errors"])
# no_source 同样 10 分钟
repo.kv_set(
"task:last:snapshot_seed",
json.dumps({
"name": "snapshot_seed", "ok": False, "skipped": True,
"started_at": finished - 1, "finished_at": finished,
"error": "no_source",
"summary": {"skipped": True, "reason": "no_source"},
}),
)
DASHBOARD_CACHE.clear()
system = q.system()
assert system["next_runs"]["snapshot_seed"] == finished + 600
finally:
db.close()
@@ -667,9 +797,10 @@ def test_overview_adjacent_previous_with_seed_and_dual_fields():
+ overview["counts"]["unfaved"]
+ overview["counts"]["removed"]
)
# 当前=seed(=完成日数值),相邻 previous=完成日本身 → 增量 0
# 101/102:当前=seed(=完成日数值),相邻 previous=完成日本身 → 增量 0
# 103:当日发布,近 1 日相对 0,计入当前播放
assert overview["view_delta_available"] is True
assert overview["view_delta"] == 0
assert overview["view_delta"] == 120
growth = q.list_videos(sort="growth", growth_days=1)
assert growth["available"] is True
@@ -678,6 +809,141 @@ def test_overview_adjacent_previous_with_seed_and_dual_fields():
assert item["growth"]["available"] is True
assert item["growth"]["delta"] == 0
assert item["growth"]["gap_days"] == 1
assert item["growth"]["baseline"] == "snapshot"
finally:
db.close()
def _view_metrics(view: int) -> dict[str, int]:
return {
"view": view, "like": 0, "coin": 0, "favorite": 0,
"share": 0, "reply": 0, "danmaku": 0,
}
def test_same_day_growth_baselines():
"""近 1 日:当日发布相对 0;更早发布用 view_open;7 日不用这两条退路。"""
with tempfile.TemporaryDirectory() as tmp:
db, q = setup(tmp)
try:
repo = Repository(db)
today = DATES[-1]
start = int(day_bounds(today)[0])
old_pub = int(day_bounds("2026-09-01")[0])
repo.upsert_snapshot(
today, 103, metrics=_view_metrics(200),
ok=True, err=None, captured_at=start + 3600,
)
repo.upsert_from_fav(
resource(105, up_mid=1, up_name="UP A", pubdate=old_pub),
FOLDER.media_id, was_added=True, now=start + 10,
)
repo.upsert_snapshot(
today, 105, metrics=_view_metrics(500),
ok=True, err=None, captured_at=start + 10,
)
repo.upsert_snapshot(
today, 105, metrics=_view_metrics(800),
ok=True, err=None, captured_at=start + 20,
)
repo.upsert_from_fav(
resource(106, up_mid=1, up_name="UP A", pubdate=old_pub),
FOLDER.media_id, was_added=True, now=start + 30,
)
repo.upsert_snapshot(
today, 106, metrics=_view_metrics(900),
ok=True, err=None, captured_at=start + 30,
)
repo.upsert_from_fav(
resource(107, up_mid=1, up_name="UP A", pubdate=old_pub),
FOLDER.media_id, was_added=True, now=start + 40,
)
repo.upsert_snapshot(
today, 107, metrics=_view_metrics(50),
ok=True, err=None, captured_at=start + 40,
)
db.write(lambda conn: conn.execute(
"UPDATE video SET pubdate = 0 WHERE aid = 107"
))
DASHBOARD_CACHE.clear()
one = q.list_videos(sort="growth", growth_days=1, page_size=20)
by_aid = {item["aid"]: item for item in one["items"]}
assert 103 in by_aid and 105 in by_aid
assert 106 not in by_aid and 107 not in by_aid
assert by_aid[103]["growth"]["baseline"] == "zero"
assert by_aid[103]["growth"]["delta"] == 200
assert by_aid[103]["growth"]["rate"] is None
assert by_aid[103]["growth"]["gap_days"] == 0
assert by_aid[105]["growth"]["baseline"] == "open"
assert by_aid[105]["growth"]["delta"] == 300
assert by_aid[101]["growth"]["baseline"] == "snapshot"
assert by_aid[101]["growth"]["delta"] == 100
rate = q.list_videos(sort="growth_rate", growth_days=1, page_size=20)
rate_aids = {item["aid"] for item in rate["items"]}
assert 103 not in rate_aids and 105 not in rate_aids
assert 101 in rate_aids
seven = q.list_videos(sort="growth", growth_days=7)
seven_aids = {item["aid"] for item in seven["items"]}
assert 103 not in seven_aids
assert 105 not in seven_aids
detail103 = q.video_detail(103)
assert detail103["growth"]["baseline"] == "zero"
assert detail103["growth"]["delta"] == 200
detail105 = q.video_detail(105)
assert detail105["growth"]["baseline"] == "open"
assert detail105["growth"]["delta"] == 300
assert q.video_detail(106)["growth"]["available"] is False
assert q.video_detail(107)["growth"]["available"] is False
overview = q.overview()
assert overview["view_delta"] == 100 + 100 + 200 + 300
finally:
db.close()
def test_growth_rate_sort_uses_prior():
"""基线不足 1000 不进增速榜;达标后按基线 + prior 排序,不按 raw rate。"""
with tempfile.TemporaryDirectory() as tmp:
db, q = setup(tmp)
try:
repo = Repository(db)
latest = DATES[-1]
previous_day = DATES[-2]
start = int(day_bounds(latest)[0])
pub = int(day_bounds("2026-09-01")[0])
# 301:80→512,真实 +540%,基线低于 1000,只留在增量榜。
# 304:1000→2500,真实增速高于 302,平滑后应排在 302 后面。
for aid, views in ((301, (80, 512)), (302, (5000, 10000)), (304, (1000, 2500))):
repo.upsert_from_fav(
resource(aid, up_mid=1, up_name="UP A", pubdate=pub),
FOLDER.media_id, was_added=True, now=start,
)
repo.upsert_snapshot(
previous_day, aid, metrics=_view_metrics(views[0]),
ok=True, err=None, captured_at=start - 86400,
)
repo.upsert_snapshot(
latest, aid, metrics=_view_metrics(views[1]),
ok=True, err=None, captured_at=start + 60,
)
DASHBOARD_CACHE.clear()
ranked = q.list_videos(sort="growth_rate", growth_days=1, page_size=20)
order = [item["aid"] for item in ranked["items"]]
assert ranked["growth_rate_prior"] == 1000
assert 301 not in order
assert order.index(302) < order.index(304)
modest = next(item for item in ranked["items"] if item["aid"] == 304)
assert abs(modest["growth"]["rate"] - 1500 / 1000) < 1e-12
growth = q.list_videos(sort="growth", growth_days=1, page_size=20)
grown = next(item for item in growth["items"] if item["aid"] == 301)
assert grown["growth"]["delta"] == 432
assert abs(grown["growth"]["rate"] - 432 / 80) < 1e-12
finally:
db.close()
@@ -741,7 +1007,12 @@ def main_test():
test_system_tasks_sorted_by_started_at_desc()
test_growth_auto_falls_back_to_one_day()
test_snapshot_recent_next_run_follows_completion()
test_snapshot_recent_next_run_defers_quiet_window_and_retry_on_skip()
test_snapshot_failed_next_run_retries_soon_not_next_day()
test_snapshot_seed_skip_next_run_retries_ten_minutes()
test_overview_adjacent_previous_with_seed_and_dual_fields()
test_same_day_growth_baselines()
test_growth_rate_sort_uses_prior()
print("OK: Web 查询层总览/榜单/详情/UP/时长效率/粉丝/候选/系统/访问统计测试全部通过")
@@ -805,13 +1076,15 @@ def test_system_status_uses_completed_date_and_running_flags():
def test_system_next_runs_follow_last_completion_not_wall_clock():
import json
from datetime import datetime
from reaction_board.timeutil import SHANGHAI_TZ
with tempfile.TemporaryDirectory() as tmp:
db, q = setup(tmp)
try:
repo = Repository(db)
# 发现任务刚完成不久:下次 = finished + 60min,即使 now 已超过也不跳档
finished = FIXED_NOW - 30
# 午后完成:下次 = finished + 60min,不跳档、不误入静默窗推算
finished = int(datetime(2026, 10, 31, 15, 0, 0, tzinfo=SHANGHAI_TZ).timestamp())
repo.kv_set(
"task:last:discovery",
json.dumps({
@@ -823,7 +1096,6 @@ def test_system_next_runs_follow_last_completion_not_wall_clock():
DASHBOARD_CACHE.clear()
system = q.system()
assert system["next_runs"]["discover"] == finished + 3600
# 仍小于/等于 now 也可以(表示本轮后的下一档已到期或未到),但绝不能跳到 +7200
assert system["next_runs"]["discover"] != finished + 7200
finally:
db.close()
+1 -1
View File
@@ -75,7 +75,7 @@ def test_readonly_routes_and_static():
assert status == 200 and payload["ok"] is True
status, _, payload = running.json("/api/about")
assert status == 200
assert payload["data"]["schema_version"] == 7
assert payload["data"]["schema_version"] == 8
assert payload["data"]["analytics"]["min_duration_seconds"] == 600
status, _, overview = running.json("/api/overview")
+1 -1
View File
@@ -138,7 +138,7 @@ def build_db(tmp: str | Path) -> Path:
day, 37754047, follower=1000 + index * 10, following=5,
ok=True, captured_at=ts(day) + 60,
)
# 103 只有最后一天快照,用于验证样本不足;104 另加一条失败快照
# 103 只有最后一天快照(发布日=快照日,近 1 日相对 0);104 另加一条失败快照
repo.upsert_snapshot(
DATES[-1], 103,
metrics={"view": 120, "like": 5, "coin": 1, "favorite": 2, "share": 0, "reply": 1, "danmaku": 1},
+4 -2
View File
@@ -237,8 +237,8 @@ def _insert_snapshot(
"""
INSERT OR REPLACE INTO snapshot(
stat_date, aid, "view", "like", favorite, coin, share, reply,
danmaku, captured_at, ok, err, src
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'preview')
danmaku, captured_at, ok, err, src, view_open, open_captured_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'preview', ?, ?)
""",
(
stat_date,
@@ -253,6 +253,8 @@ def _insert_snapshot(
int(captured_at),
1 if ok else 0,
err,
values.get("view") if ok else None,
int(captured_at) if ok else None,
),
)