fix(scheduler): no false skip for full snapshot; quiet window + retry #7
No files matched your search
@@ -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 与入库操作
|
||||
|
||||
+1
-1
@@ -800,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,单写线程
|
||||
|
||||
+820
-98
File diff suppressed because it is too large.
Load diff
@@ -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
|
||||
@@ -47,6 +59,7 @@ class Scheduler:
|
||||
clock=time.time,
|
||||
sleeper=time.sleep,
|
||||
logger=None,
|
||||
due_reader: Callable[[str], float | None] | None = None,
|
||||
) -> None:
|
||||
if not tasks:
|
||||
raise ValueError("至少需要一个调度任务")
|
||||
@@ -57,6 +70,8 @@ class Scheduler:
|
||||
self._clock = clock
|
||||
self._sleeper = sleeper
|
||||
self.logger = logger
|
||||
# 每轮从落盘 due_at 刷新 next_run(seed 挂当晚 seal 等同进程可见)
|
||||
self._due_reader = due_reader
|
||||
|
||||
def _log(self, level: str, event: str, **fields) -> None:
|
||||
if self.logger is not None:
|
||||
@@ -66,27 +81,56 @@ class Scheduler:
|
||||
current = self._clock() if now is None else now
|
||||
results: list[TaskResult] = []
|
||||
for task in self.tasks:
|
||||
if self._due_reader is not None:
|
||||
disk_due = self._due_reader(task.name)
|
||||
if disk_due is not None:
|
||||
task.next_run = float(disk_due)
|
||||
if task.next_run > current:
|
||||
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)
|
||||
|
||||
@@ -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,
|
||||
}
|
||||
|
||||
@@ -140,12 +142,20 @@ class SnapshotService:
|
||||
return f"fans:seeded:{stat_date}"
|
||||
|
||||
def run_if_needed(
|
||||
self, *, date: str | None = None, force: bool = False
|
||||
self,
|
||||
*,
|
||||
date: str | None = None,
|
||||
force: bool = False,
|
||||
keep_scraped: bool | None = None,
|
||||
) -> SnapshotSummary:
|
||||
"""视频快照(含 UP 头像补齐);粉丝采集请用 ``run_fans_if_needed``。
|
||||
|
||||
当日若仅有午夜 seed 壳且未 ``force``,直接返回等待晚间落盘,
|
||||
避免把「全员 skip」误标成 completed。
|
||||
|
||||
``force`` + ``keep_scraped``:
|
||||
- 晚间 seal / 断点续跑:``keep_scraped=True``(默认)保留已采 ok=1;
|
||||
- CLI ``--force``:显式 ``keep_scraped=False`` 全量重采。
|
||||
"""
|
||||
now = self._now()
|
||||
stat_date = date or shanghai_date(now)
|
||||
@@ -153,7 +163,10 @@ class SnapshotService:
|
||||
raise ValueError(f"日期格式必须是 YYYY-MM-DD: {stat_date!r}")
|
||||
seeded = self.repo.kv_get(self.seeded_key(stat_date)) == "1"
|
||||
if force:
|
||||
return self.run_videos(date=stat_date, force=True)
|
||||
ks = True if keep_scraped is None else bool(keep_scraped)
|
||||
return self.run_videos(
|
||||
date=stat_date, force=True, keep_scraped=ks,
|
||||
)
|
||||
|
||||
completed = self.repo.kv_get(self.completed_key(stat_date)) == "1"
|
||||
if seeded and not completed:
|
||||
@@ -167,6 +180,7 @@ class SnapshotService:
|
||||
finished_at=now,
|
||||
completed=False,
|
||||
already_completed=False,
|
||||
reason="seeded_pending_seal",
|
||||
)
|
||||
pending = self.repo.pending_snapshot_aids(stat_date)
|
||||
if completed and not pending:
|
||||
@@ -203,8 +217,19 @@ class SnapshotService:
|
||||
summary = self.run_videos(date=date, force=force)
|
||||
return self.run_fans(date=summary.stat_date, force=force, summary=summary)
|
||||
|
||||
def run_videos(self, *, date: str | None = None, force: bool = False) -> SnapshotSummary:
|
||||
"""采集视频七项指标并在全部成功后标记完成日 / 备份。"""
|
||||
def run_videos(
|
||||
self,
|
||||
*,
|
||||
date: str | None = None,
|
||||
force: bool = False,
|
||||
keep_scraped: bool = False,
|
||||
) -> SnapshotSummary:
|
||||
"""采集视频七项指标并在全部成功后标记完成日 / 备份。
|
||||
|
||||
``force``:忽略完成标记,重跑。
|
||||
``keep_scraped``:force 时仍跳过非 seed 的 ok=1(seal 断点续跑);
|
||||
CLI ``--force`` 不传此标志,会重采全部成功行。
|
||||
"""
|
||||
cfg = self.config_store.get()
|
||||
client = self._client()
|
||||
now = self._now()
|
||||
@@ -225,11 +250,19 @@ class SnapshotService:
|
||||
summary.total = len(videos)
|
||||
for index, aid in enumerate(videos, 1):
|
||||
existing = self.repo.get_snapshot(stat_date, aid)
|
||||
if existing is not None and int(existing.get("ok") or 0) == 1 and not force:
|
||||
# 断点续跑:同日已成功则跳过。小时任务刷新过的行也是 ok=1,
|
||||
# 自然覆盖「1 小时内 captured_at 已更新则不再次更新」的要求。
|
||||
if existing is not None and int(existing.get("ok") or 0) == 1:
|
||||
src = str(existing.get("src") or "")
|
||||
if keep_scraped and src == "seed":
|
||||
# seal 续跑:必须打穿克隆壳
|
||||
pass
|
||||
elif keep_scraped and src != "seed":
|
||||
summary.skipped += 1
|
||||
continue
|
||||
elif not force:
|
||||
# 普通续跑:同日已成功则跳过(含小时任务刚刷新的)
|
||||
summary.skipped += 1
|
||||
continue
|
||||
# force 且非 keep_scraped:CLI 全量重采,不 skip
|
||||
|
||||
stat, error, unavailable = self._fetch_view(client, aid)
|
||||
if unavailable:
|
||||
@@ -423,6 +456,7 @@ class SnapshotService:
|
||||
completed=False,
|
||||
already_completed=False,
|
||||
fans_skipped=len(cfg.track_fans),
|
||||
reason="seeded_pending_seal",
|
||||
)
|
||||
return self.run_fans(date=stat_date, force=True)
|
||||
if not self._pending_fan_mids(stat_date, cfg.track_fans):
|
||||
@@ -536,15 +570,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",
|
||||
@@ -553,6 +580,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(
|
||||
@@ -562,6 +600,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(
|
||||
@@ -583,6 +622,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,
|
||||
|
||||
@@ -0,0 +1,438 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""调度任务状态持久化(跨进程重启 / 容器重建)。
|
||||
|
||||
每条任务一条 kv 记录 ``task:state:{name}``,字段见 :class:`TaskState`。
|
||||
调度器只认落盘的 ``due_at`` / ``status`` / ``stat_date`` / ``progress``,
|
||||
不再靠启动时把 ``next_run=0`` 猜「要不要立刻跑」。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import time
|
||||
from dataclasses import asdict, dataclass
|
||||
from datetime import timedelta
|
||||
from typing import Any, Mapping
|
||||
|
||||
from .repository import Repository
|
||||
from .timeutil import parse_hhmm, shanghai_date, shanghai_datetime
|
||||
|
||||
STATUS_PENDING = "pending"
|
||||
STATUS_RUNNING = "running"
|
||||
STATUS_DONE = "done"
|
||||
STATUS_FAILED = "failed"
|
||||
STATUS_DEFERRED = "deferred"
|
||||
|
||||
_ACTIVE = frozenset({STATUS_PENDING, STATUS_RUNNING, STATUS_FAILED, STATUS_DEFERRED})
|
||||
|
||||
|
||||
@dataclass
|
||||
class TaskState:
|
||||
name: str
|
||||
status: str
|
||||
due_at: float
|
||||
stat_date: str | None = None
|
||||
progress: dict[str, Any] | None = None
|
||||
updated_at: float = 0.0
|
||||
error: str | None = None
|
||||
|
||||
def as_dict(self) -> dict[str, Any]:
|
||||
payload = asdict(self)
|
||||
# JSON 友好:去掉 None progress 也可保留
|
||||
return payload
|
||||
|
||||
|
||||
def state_key(name: str) -> str:
|
||||
return f"task:state:{name}"
|
||||
|
||||
|
||||
def load_state(repo: Repository, name: str) -> TaskState | None:
|
||||
raw = repo.kv_get(state_key(name))
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
except (TypeError, ValueError, json.JSONDecodeError):
|
||||
return None
|
||||
if not isinstance(data, dict) or not data.get("name"):
|
||||
return None
|
||||
try:
|
||||
due_at = float(data.get("due_at") or 0)
|
||||
updated_at = float(data.get("updated_at") or 0)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
progress = data.get("progress")
|
||||
if progress is not None and not isinstance(progress, dict):
|
||||
progress = None
|
||||
return TaskState(
|
||||
name=str(data.get("name") or name),
|
||||
status=str(data.get("status") or STATUS_PENDING),
|
||||
due_at=due_at,
|
||||
stat_date=(str(data["stat_date"]) if data.get("stat_date") else None),
|
||||
progress=progress,
|
||||
updated_at=updated_at,
|
||||
error=(str(data["error"]) if data.get("error") else None),
|
||||
)
|
||||
|
||||
|
||||
def save_state(repo: Repository, state: TaskState) -> None:
|
||||
repo.kv_set(state_key(state.name), json.dumps(state.as_dict(), ensure_ascii=False))
|
||||
|
||||
|
||||
def _now(now: float | None) -> float:
|
||||
return time.time() if now is None else float(now)
|
||||
|
||||
|
||||
def recover_stale_running(
|
||||
repo: Repository,
|
||||
name: str,
|
||||
*,
|
||||
now: float | None = None,
|
||||
) -> TaskState | None:
|
||||
"""进程已死但仍为 ``running``:改为 ``pending``,保留 progress / stat_date。
|
||||
|
||||
``due_at`` 取 ``min(原 due_at, now)``,使过期/中断任务可立即续跑,
|
||||
未到期的(极少见)仍尊重原 due_at。
|
||||
"""
|
||||
current = _now(now)
|
||||
state = load_state(repo, name)
|
||||
if state is None or state.status != STATUS_RUNNING:
|
||||
return state
|
||||
state.status = STATUS_PENDING
|
||||
state.due_at = min(float(state.due_at or current), current)
|
||||
state.updated_at = current
|
||||
save_state(repo, state)
|
||||
return state
|
||||
|
||||
|
||||
def mark_running(
|
||||
repo: Repository,
|
||||
name: str,
|
||||
*,
|
||||
stat_date: str | None,
|
||||
now: float | None = None,
|
||||
progress: Mapping[str, Any] | None = None,
|
||||
due_at: float | None = None,
|
||||
) -> TaskState:
|
||||
current = _now(now)
|
||||
prev = load_state(repo, name)
|
||||
state = TaskState(
|
||||
name=name,
|
||||
status=STATUS_RUNNING,
|
||||
due_at=float(due_at if due_at is not None else (prev.due_at if prev else current)),
|
||||
stat_date=stat_date if stat_date is not None else (prev.stat_date if prev else None),
|
||||
progress=dict(progress) if progress is not None else (prev.progress if prev else None),
|
||||
updated_at=current,
|
||||
error=None,
|
||||
)
|
||||
save_state(repo, state)
|
||||
return state
|
||||
|
||||
|
||||
def mark_done(
|
||||
repo: Repository,
|
||||
name: str,
|
||||
*,
|
||||
due_at: float,
|
||||
now: float | None = None,
|
||||
stat_date: str | None = None,
|
||||
progress: Mapping[str, Any] | None = None,
|
||||
) -> TaskState:
|
||||
"""成功:记 done,并写入下一次 ``due_at``(下一条墙钟/间隔)。"""
|
||||
current = _now(now)
|
||||
state = TaskState(
|
||||
name=name,
|
||||
status=STATUS_DONE,
|
||||
due_at=float(due_at),
|
||||
stat_date=stat_date,
|
||||
progress=dict(progress) if progress is not None else None,
|
||||
updated_at=current,
|
||||
error=None,
|
||||
)
|
||||
save_state(repo, state)
|
||||
return state
|
||||
|
||||
|
||||
def mark_failed(
|
||||
repo: Repository,
|
||||
name: str,
|
||||
*,
|
||||
due_at: float,
|
||||
now: float | None = None,
|
||||
stat_date: str | None = None,
|
||||
progress: Mapping[str, Any] | None = None,
|
||||
error: str | None = None,
|
||||
) -> TaskState:
|
||||
current = _now(now)
|
||||
prev = load_state(repo, name)
|
||||
state = TaskState(
|
||||
name=name,
|
||||
status=STATUS_FAILED,
|
||||
due_at=float(due_at),
|
||||
stat_date=stat_date if stat_date is not None else (prev.stat_date if prev else None),
|
||||
progress=(
|
||||
dict(progress) if progress is not None
|
||||
else (prev.progress if prev else None)
|
||||
),
|
||||
updated_at=current,
|
||||
error=error,
|
||||
)
|
||||
save_state(repo, state)
|
||||
return state
|
||||
|
||||
|
||||
def mark_deferred(
|
||||
repo: Repository,
|
||||
name: str,
|
||||
*,
|
||||
due_at: float,
|
||||
now: float | None = None,
|
||||
stat_date: str | None = None,
|
||||
progress: Mapping[str, Any] | None = None,
|
||||
error: str | None = None,
|
||||
) -> TaskState:
|
||||
"""主动延后(静默窗 / seed 避让等):保留进度,写未来 due_at。"""
|
||||
current = _now(now)
|
||||
prev = load_state(repo, name)
|
||||
state = TaskState(
|
||||
name=name,
|
||||
status=STATUS_DEFERRED,
|
||||
due_at=float(due_at),
|
||||
stat_date=stat_date if stat_date is not None else (prev.stat_date if prev else None),
|
||||
progress=(
|
||||
dict(progress) if progress is not None
|
||||
else (prev.progress if prev else None)
|
||||
),
|
||||
updated_at=current,
|
||||
error=error,
|
||||
)
|
||||
save_state(repo, state)
|
||||
return state
|
||||
|
||||
|
||||
def ensure_pending(
|
||||
repo: Repository,
|
||||
name: str,
|
||||
*,
|
||||
due_at: float,
|
||||
stat_date: str | None,
|
||||
now: float | None = None,
|
||||
overwrite_done: bool = True,
|
||||
) -> TaskState:
|
||||
"""保证存在一条 pending(如 seed 成功后挂上当晚 seal)。
|
||||
|
||||
同日未完成:取更早的 due_at。
|
||||
跨日:不得把新一天的 due 粘到昨天的 ``stat_date`` 上——
|
||||
若已有更早日期的未完成记录,保留它(跨天续跑优先);
|
||||
若已有更新日期的未完成,则重置为调用方的 due/stat_date。
|
||||
``overwrite_done`` 时覆盖已完成记录。
|
||||
"""
|
||||
current = _now(now)
|
||||
prev = load_state(repo, name)
|
||||
if prev is not None and prev.status in _ACTIVE and prev.status != STATUS_RUNNING:
|
||||
if (
|
||||
stat_date is not None
|
||||
and prev.stat_date
|
||||
and prev.stat_date != stat_date
|
||||
):
|
||||
# 未完成的更早归属日优先保留,避免 seed 挂今晚时冲掉昨天续跑。
|
||||
if prev.stat_date < stat_date:
|
||||
return prev
|
||||
# prev 归属日更新:按调用方重置,禁止跨日字段拼接。
|
||||
state = TaskState(
|
||||
name=name,
|
||||
status=STATUS_PENDING,
|
||||
due_at=float(due_at),
|
||||
stat_date=stat_date,
|
||||
progress=None,
|
||||
updated_at=current,
|
||||
error=None,
|
||||
)
|
||||
save_state(repo, state)
|
||||
return state
|
||||
# 同日(或缺一边 stat_date):取更早 due_at,补上缺失的 stat_date
|
||||
due = min(float(prev.due_at or due_at), float(due_at))
|
||||
sd = prev.stat_date or stat_date
|
||||
if due == prev.due_at and sd == prev.stat_date and prev.status == STATUS_PENDING:
|
||||
return prev
|
||||
prev.due_at = due
|
||||
prev.stat_date = sd
|
||||
prev.status = STATUS_PENDING
|
||||
prev.updated_at = current
|
||||
save_state(repo, prev)
|
||||
return prev
|
||||
if prev is not None and prev.status == STATUS_RUNNING:
|
||||
return prev
|
||||
if prev is not None and prev.status == STATUS_DONE and not overwrite_done:
|
||||
return prev
|
||||
state = TaskState(
|
||||
name=name,
|
||||
status=STATUS_PENDING,
|
||||
due_at=float(due_at),
|
||||
stat_date=stat_date,
|
||||
progress=None,
|
||||
updated_at=current,
|
||||
error=None,
|
||||
)
|
||||
save_state(repo, state)
|
||||
return state
|
||||
|
||||
|
||||
def bootstrap_from_last(
|
||||
repo: Repository,
|
||||
name: str,
|
||||
*,
|
||||
wall_due_at: float,
|
||||
now: float | None = None,
|
||||
retry_seconds: float = 300.0,
|
||||
) -> TaskState | None:
|
||||
"""升级/冷启动:无 ``task:state`` 时从 ``task:last`` 回填。
|
||||
|
||||
- 成功 → ``done`` + ``due_at=wall_due_at``
|
||||
- 失败/跳过 → ``failed``/``deferred``,``due_at`` 为补跑点(已过则立刻)
|
||||
"""
|
||||
current = _now(now)
|
||||
existing = load_state(repo, name)
|
||||
if existing is not None:
|
||||
return existing
|
||||
raw = repo.kv_get(f"task:last:{name}")
|
||||
if not raw:
|
||||
return None
|
||||
try:
|
||||
data = json.loads(raw)
|
||||
except (TypeError, ValueError, json.JSONDecodeError):
|
||||
return None
|
||||
if not isinstance(data, dict):
|
||||
return None
|
||||
ok = bool(data.get("ok"))
|
||||
skipped = bool(data.get("skipped"))
|
||||
try:
|
||||
finished = float(data.get("finished_at") or data.get("started_at") or current)
|
||||
except (TypeError, ValueError):
|
||||
finished = current
|
||||
summary = data.get("summary")
|
||||
stat_date = None
|
||||
if isinstance(summary, dict) and summary.get("stat_date"):
|
||||
stat_date = str(summary["stat_date"])
|
||||
error = data.get("error")
|
||||
error_s = str(error) if error else None
|
||||
if ok and not skipped:
|
||||
state = TaskState(
|
||||
name=name,
|
||||
status=STATUS_DONE,
|
||||
due_at=float(wall_due_at),
|
||||
stat_date=stat_date,
|
||||
updated_at=current,
|
||||
error=None,
|
||||
)
|
||||
save_state(repo, state)
|
||||
return state
|
||||
due = finished + float(retry_seconds)
|
||||
if due > current:
|
||||
pass
|
||||
else:
|
||||
due = current
|
||||
state = TaskState(
|
||||
name=name,
|
||||
status=STATUS_DEFERRED if skipped else STATUS_FAILED,
|
||||
due_at=float(due),
|
||||
stat_date=stat_date,
|
||||
updated_at=current,
|
||||
error=error_s,
|
||||
)
|
||||
save_state(repo, state)
|
||||
return state
|
||||
|
||||
|
||||
def bootstrap_seed_markers(
|
||||
repo: Repository,
|
||||
name: str,
|
||||
*,
|
||||
slot_hhmm: str,
|
||||
wall_due_at: float,
|
||||
now: float | None = None,
|
||||
) -> TaskState | None:
|
||||
"""无 state 时:若存在未 seal 的 seed 壳,挂 pending(跨天优先昨天)。
|
||||
|
||||
已过当日 ``slot_hhmm``(或归属日早于今天)→ ``due_at=now`` 尽快补跑。
|
||||
"""
|
||||
del wall_due_at # 未完成 seed 不走纯墙钟
|
||||
current = _now(now)
|
||||
existing = load_state(repo, name)
|
||||
if existing is not None:
|
||||
return existing
|
||||
if name not in {"snapshot", "fans"}:
|
||||
return None
|
||||
today = shanghai_date(current)
|
||||
yday = (shanghai_datetime(current) - timedelta(days=1)).strftime("%Y-%m-%d")
|
||||
hour, minute = parse_hhmm(slot_hhmm)
|
||||
today_slot = shanghai_datetime(current).replace(
|
||||
hour=hour, minute=minute, second=0, microsecond=0,
|
||||
).timestamp()
|
||||
|
||||
def _seeded_unfinished(day: str) -> bool:
|
||||
if name == "snapshot":
|
||||
if repo.kv_get(f"snapshot:seeded:{day}") != "1":
|
||||
return False
|
||||
return repo.kv_get(f"snapshot:completed:{day}") != "1"
|
||||
# fans:seed 键在成功 seal 后删除,有键即未完成
|
||||
return repo.kv_get(f"fans:seeded:{day}") == "1"
|
||||
|
||||
for day in (yday, today):
|
||||
if not _seeded_unfinished(day):
|
||||
continue
|
||||
if day < today or current >= today_slot:
|
||||
due = current
|
||||
else:
|
||||
due = float(today_slot)
|
||||
return ensure_pending(
|
||||
repo, name, due_at=due, stat_date=day, now=current,
|
||||
)
|
||||
return None
|
||||
|
||||
|
||||
def startup_next_run(
|
||||
repo: Repository,
|
||||
name: str,
|
||||
*,
|
||||
wall_due_at: float,
|
||||
now: float | None = None,
|
||||
slot_hhmm: str | None = None,
|
||||
retry_seconds: float = 300.0,
|
||||
) -> float:
|
||||
"""常驻启动时的 ``ScheduledTask.next_run``:只认落盘状态。
|
||||
|
||||
- 无状态时先 ``bootstrap_from_last`` / seed 标记回填
|
||||
- 恢复残留 ``running`` → ``pending``
|
||||
- ``due_at`` 是唯一时间门闩(失败延时不会被提前)
|
||||
"""
|
||||
current = _now(now)
|
||||
if load_state(repo, name) is None:
|
||||
bootstrap_from_last(
|
||||
repo, name, wall_due_at=wall_due_at, now=current,
|
||||
retry_seconds=retry_seconds,
|
||||
)
|
||||
if slot_hhmm:
|
||||
bootstrap_seed_markers(
|
||||
repo, name, slot_hhmm=slot_hhmm,
|
||||
wall_due_at=wall_due_at, now=current,
|
||||
)
|
||||
state = recover_stale_running(repo, name, now=current)
|
||||
if state is None:
|
||||
return float(wall_due_at)
|
||||
due = float(state.due_at or 0)
|
||||
if due <= 0:
|
||||
return float(wall_due_at)
|
||||
# 无论 status:due_at 是唯一时间门闩
|
||||
if due > current:
|
||||
return due
|
||||
# 已到期:未完成要续跑;done 也表示下一轮已到期
|
||||
return due
|
||||
|
||||
|
||||
def resume_stat_date(repo: Repository, name: str, *, fallback: str) -> str:
|
||||
"""续跑时优先用状态里的归属日,避免跨日重启跑错天。"""
|
||||
state = load_state(repo, name)
|
||||
if state is not None and state.stat_date and state.status in _ACTIVE:
|
||||
return state.stat_date
|
||||
return fallback
|
||||
@@ -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)
|
||||
@@ -34,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")
|
||||
@@ -1971,9 +1978,27 @@ 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
|
||||
# 优先 task:state.due_at(跨重启唯一时间门闩);无则回退 task:last 推导。
|
||||
state_rows = conn.execute(
|
||||
"SELECT k, v FROM kv WHERE k LIKE 'task:state:%'"
|
||||
).fetchall()
|
||||
for row in state_rows:
|
||||
key = str(row["k"]).removeprefix("task:state:")
|
||||
ui_name = "discover" if key == "discovery" else key
|
||||
if ui_name not in next_runs:
|
||||
continue
|
||||
try:
|
||||
state = json.loads(str(row["v"]))
|
||||
due = float(state.get("due_at") or 0)
|
||||
except (TypeError, ValueError, json.JSONDecodeError):
|
||||
continue
|
||||
if due > 0:
|
||||
next_runs[ui_name] = int(due)
|
||||
for name, key, at in (
|
||||
("fans", "fans", cfg.schedule.snapshot_at),
|
||||
("snapshot", "snapshot", cfg.schedule.snapshot_at),
|
||||
@@ -1981,23 +2006,58 @@ class DashboardQueries:
|
||||
("discover", "discovery", None),
|
||||
("snapshot_recent", "snapshot_recent", None),
|
||||
):
|
||||
# 已有落盘 due_at 则不再用 last 覆盖
|
||||
state_raw = self._kv(conn, f"task:state:{key}")
|
||||
if state_raw:
|
||||
try:
|
||||
if float(json.loads(state_raw).get("due_at") or 0) > 0:
|
||||
continue
|
||||
except (TypeError, ValueError, json.JSONDecodeError):
|
||||
pass
|
||||
record = tasks.get(key)
|
||||
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:
|
||||
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"
|
||||
|
||||
@@ -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">
|
||||
|
||||
+617
-1
@@ -332,6 +332,607 @@ 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;落盘 task:state=deferred。"""
|
||||
from reaction_board.logging_json import JsonLogger
|
||||
from reaction_board.task_state import STATUS_DEFERRED, load_state
|
||||
|
||||
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"
|
||||
expected_due = finished + cli._SEED_RETRY_SECONDS
|
||||
assert abs(ctrl.next_run - expected_due) < 0.01
|
||||
last = json.loads(repo.kv_get("task:last:snapshot_seed"))
|
||||
assert last["skipped"] is True and last["ok"] is False
|
||||
state = load_state(repo, "snapshot_seed")
|
||||
assert state is not None
|
||||
assert state.status == STATUS_DEFERRED
|
||||
assert abs(state.due_at - expected_due) < 0.01
|
||||
assert state.due_at > finished
|
||||
assert state.error == "snapshot_running"
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_quiet_window_defer_persists_deferred_state():
|
||||
"""H8: 静默窗 defer 落盘 status=deferred + due_at=窗后 01:00,不得记成功。"""
|
||||
from datetime import datetime
|
||||
from unittest import mock
|
||||
|
||||
from reaction_board.logging_json import JsonLogger
|
||||
from reaction_board.task_state import STATUS_DEFERRED, load_state
|
||||
from reaction_board.timeutil import (
|
||||
SHANGHAI_TZ,
|
||||
next_after_snapshot_quiet_window,
|
||||
)
|
||||
|
||||
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()
|
||||
# 静默窗内:2026-11-03 23:30 上海
|
||||
quiet = datetime(2026, 11, 3, 23, 30, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
expected_due = next_after_snapshot_quiet_window(quiet)
|
||||
assert expected_due > quiet
|
||||
with mock.patch("reaction_board.cli.time.time", return_value=quiet):
|
||||
ctrl = cli._quiet_window_defer(
|
||||
repo, logger, "discovery", started_at=int(quiet) - 1,
|
||||
)
|
||||
assert ctrl.ok is False
|
||||
assert ctrl.error == cli._QUIET_SKIP_REASON
|
||||
assert abs(ctrl.next_run - expected_due) < 0.01
|
||||
last = json.loads(repo.kv_get("task:last:discovery"))
|
||||
assert last["ok"] is False and last["skipped"] is True
|
||||
assert last["error"] == cli._QUIET_SKIP_REASON
|
||||
# 不得推成成功完成类 kv
|
||||
assert repo.kv_get("snapshot:last_completed") is None
|
||||
assert repo.kv_get("discovery:last_completed") is None
|
||||
state = load_state(repo, "discovery")
|
||||
assert state is not None
|
||||
assert state.status == STATUS_DEFERRED
|
||||
assert abs(state.due_at - expected_due) < 0.01
|
||||
assert state.due_at > quiet
|
||||
assert state.error == cli._QUIET_SKIP_REASON
|
||||
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
|
||||
from reaction_board.task_state import STATUS_FAILED, load_state
|
||||
|
||||
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"]
|
||||
state = load_state(repo, "candidate_cleanup")
|
||||
assert state is not None
|
||||
assert state.status == STATUS_FAILED
|
||||
assert state.due_at > 0
|
||||
delta_state = state.due_at - __import__("time").time()
|
||||
assert 250 <= delta_state <= 360, delta_state
|
||||
assert "cleanup db boom" in (state.error or "")
|
||||
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():
|
||||
"""常驻启动:无状态时对齐墙钟;有 task:state.due_at 时认落盘。"""
|
||||
from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ
|
||||
from reaction_board.task_state import mark_done
|
||||
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
|
||||
# 昨晚成功 → state.due_at=今晚 23:15
|
||||
yday_done = datetime(2026, 11, 2, 23, 20, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
due = next_daily_run(yday_done, "23:15")
|
||||
mark_done(repo, "snapshot", due_at=due, now=yday_done, stat_date="2026-11-02")
|
||||
nxt2 = cli._initial_daily_next_run(
|
||||
repo, "snapshot", "23:15", now=now,
|
||||
)
|
||||
assert nxt2 == due
|
||||
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 后重启:认 task:state pending due_at=当晚,首轮不得跑/force。"""
|
||||
from reaction_board.scheduler import ScheduledTask, Scheduler
|
||||
from reaction_board.timeutil import next_daily_run, SHANGHAI_TZ, shanghai_date
|
||||
from reaction_board.task_state import ensure_pending, mark_done
|
||||
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, 10, 30, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
today = shanghai_date(now)
|
||||
repo.kv_set(f"snapshot:seeded:{today}", "1")
|
||||
slot = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
seed_done_at = datetime(2026, 11, 3, 0, 1, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
mark_done(
|
||||
repo, "snapshot_seed",
|
||||
due_at=next_daily_run(seed_done_at, "00:00"),
|
||||
now=seed_done_at,
|
||||
stat_date=today,
|
||||
)
|
||||
ensure_pending(
|
||||
repo, "snapshot", due_at=slot, stat_date=today, now=seed_done_at,
|
||||
)
|
||||
ensure_pending(
|
||||
repo, "fans", due_at=slot, stat_date=today, now=seed_done_at,
|
||||
)
|
||||
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 == slot
|
||||
assert fans_next == slot
|
||||
assert seed_next == next_daily_run(seed_done_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, clock[0])
|
||||
)
|
||||
force_calls.append(force)
|
||||
if force and clock[0] < slot:
|
||||
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 == []
|
||||
clock[0] = slot
|
||||
results = scheduler.run_due()
|
||||
assert len(results) >= 1
|
||||
assert force_calls == [True]
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
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 test_off_plan_completion_uses_started_at_not_finished_at():
|
||||
"""规则2:完成记录 started_at 早于今日计划点 → off-plan;禁用 finished_at。"""
|
||||
from datetime import datetime
|
||||
from reaction_board.timeutil import SHANGHAI_TZ
|
||||
|
||||
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)
|
||||
slot = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
# 白天开始、晚上才结束:started_at < slot,finished_at > slot
|
||||
day_start = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
day_end = datetime(2026, 11, 3, 23, 30, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
repo.kv_set(
|
||||
"task:last:snapshot",
|
||||
json.dumps({
|
||||
"name": "snapshot",
|
||||
"ok": True,
|
||||
"skipped": False,
|
||||
"started_at": int(day_start),
|
||||
"finished_at": int(day_end),
|
||||
"error": None,
|
||||
"summary": {"stat_date": "2026-11-03"},
|
||||
}, ensure_ascii=False),
|
||||
)
|
||||
assert cli._last_success_started_at(repo, "snapshot") == float(int(day_start))
|
||||
assert cli._off_plan_completion(repo, "snapshot", "23:15", slot) is True
|
||||
# 若误用 finished_at 会判成计划内;started_at 必须仍为 off-plan
|
||||
assert day_end >= slot
|
||||
assert day_start < slot
|
||||
|
||||
# 计划内:started_at == slot → 不可 off-plan
|
||||
repo.kv_set(
|
||||
"task:last:snapshot",
|
||||
json.dumps({
|
||||
"name": "snapshot",
|
||||
"ok": True,
|
||||
"skipped": False,
|
||||
"started_at": int(slot),
|
||||
"finished_at": int(slot + 600),
|
||||
"error": None,
|
||||
"summary": {"stat_date": "2026-11-03"},
|
||||
}, ensure_ascii=False),
|
||||
)
|
||||
assert cli._off_plan_completion(repo, "snapshot", "23:15", slot) is False
|
||||
assert cli._off_plan_completion(
|
||||
repo, "snapshot", "23:15", slot + 60,
|
||||
) is False
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_cross_midnight_force_seal_ignores_todays_slot_gate():
|
||||
"""规则1:昨天 seed 未完成,跨日后即便未到今日 23:15 也 force。"""
|
||||
from datetime import datetime
|
||||
from reaction_board.timeutil import SHANGHAI_TZ
|
||||
|
||||
now = datetime(2026, 11, 3, 0, 10, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
assert cli._reached_daily_slot("23:15", now) is False
|
||||
assert cli._should_force_daily_seal(
|
||||
seeded=True,
|
||||
stat_date="2026-11-02",
|
||||
today="2026-11-03",
|
||||
slot_hhmm="23:15",
|
||||
now=now,
|
||||
) is True
|
||||
# 同日白天仍不得 force
|
||||
day = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
assert cli._should_force_daily_seal(
|
||||
seeded=True,
|
||||
stat_date="2026-11-03",
|
||||
today="2026-11-03",
|
||||
slot_hhmm="23:15",
|
||||
now=day,
|
||||
) is False
|
||||
night = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
assert cli._should_force_daily_seal(
|
||||
seeded=True,
|
||||
stat_date="2026-11-03",
|
||||
today="2026-11-03",
|
||||
slot_hhmm="23:15",
|
||||
now=night,
|
||||
) is True
|
||||
|
||||
|
||||
def test_initial_daily_bootstraps_failed_last_without_state():
|
||||
"""A: 无 state、仅 task:last 失败 → 启动 next_run 立刻可补,不退墙钟。"""
|
||||
from datetime import datetime
|
||||
from reaction_board.timeutil import SHANGHAI_TZ, next_daily_run
|
||||
from reaction_board.task_state import STATUS_FAILED, load_state
|
||||
|
||||
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, 18, 0, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
repo.kv_set(
|
||||
"task:last:snapshot",
|
||||
json.dumps({
|
||||
"name": "snapshot",
|
||||
"ok": False,
|
||||
"skipped": False,
|
||||
"started_at": int(now - 900),
|
||||
"finished_at": int(now - 800),
|
||||
"error": "net",
|
||||
"summary": {"stat_date": "2026-11-03"},
|
||||
}, ensure_ascii=False),
|
||||
)
|
||||
wall = next_daily_run(now, "23:15")
|
||||
nxt = cli._initial_daily_next_run(
|
||||
repo, "snapshot", "23:15", now=now,
|
||||
)
|
||||
assert nxt <= now
|
||||
assert nxt != wall
|
||||
state = load_state(repo, "snapshot")
|
||||
assert state is not None
|
||||
assert state.status == STATUS_FAILED
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def main_test():
|
||||
test_init_and_status()
|
||||
test_missing_config()
|
||||
@@ -345,7 +946,22 @@ 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_quiet_window_defer_persists_deferred_state()
|
||||
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()
|
||||
test_off_plan_completion_uses_started_at_not_finished_at()
|
||||
test_cross_midnight_force_seal_ignores_todays_slot_gate()
|
||||
test_initial_daily_bootstraps_failed_last_without_state()
|
||||
print("OK: CLI init/sync/snapshot/snapshot-seed/auth-check/run/status/backfill/错误路径测试全部通过")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
+103
-1
@@ -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()
|
||||
|
||||
@@ -537,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:
|
||||
@@ -631,6 +661,117 @@ def test_seeded_day_skips_without_force_and_seals_with_force():
|
||||
db.close()
|
||||
|
||||
|
||||
def test_seal_keep_scraped_skips_view_progress_punches_seed():
|
||||
"""晚间 seal(keep_scraped):已 view 的 ok=1 作进度跳过,seed 壳仍打穿。"""
|
||||
clock = 1_800_000_000
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
transport = SnapshotTransport({
|
||||
1: view_for(1, view=111),
|
||||
2: view_for(2, view=222),
|
||||
})
|
||||
db, repo, service = make_service(tmp, transport)
|
||||
try:
|
||||
add_video(repo, 1)
|
||||
add_video(repo, 2)
|
||||
today = DATE
|
||||
# 两条 seed 壳
|
||||
for aid, view in ((1, 10), (2, 20)):
|
||||
repo.upsert_snapshot(
|
||||
today, aid,
|
||||
metrics={"view": view, "like": 1, "coin": 0, "favorite": 0,
|
||||
"share": 0, "reply": 0, "danmaku": 0},
|
||||
ok=True, err=None, captured_at=clock, src="seed",
|
||||
)
|
||||
# 模拟 seal 中途:aid=1 已真实采集
|
||||
repo.upsert_snapshot(
|
||||
today, 1,
|
||||
metrics={"view": 111, "like": 1, "coin": 0, "favorite": 0,
|
||||
"share": 0, "reply": 0, "danmaku": 0},
|
||||
ok=True, err=None, captured_at=clock + 1, src="view",
|
||||
)
|
||||
repo.kv_set(service.seeded_key(today), "1")
|
||||
|
||||
sealed = service.run_if_needed(date=today, force=True)
|
||||
assert sealed.completed is True
|
||||
assert sealed.skipped >= 1 # aid=1 进度跳过
|
||||
assert sealed.success >= 1 # aid=2 seed 打穿
|
||||
# 只应请求 aid=2 的 view(aid=1 已是 view 进度)
|
||||
assert transport.count("/web-interface/view") == 1
|
||||
assert int(repo.get_snapshot(today, 1)["view"]) == 111
|
||||
assert repo.get_snapshot(today, 1)["src"] == "view"
|
||||
assert int(repo.get_snapshot(today, 2)["view"]) == 222
|
||||
assert repo.get_snapshot(today, 2)["src"] == "view"
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
|
||||
def test_force_keep_scraped_false_rescrapes_view_progress():
|
||||
"""C: CLI --force 路径 keep_scraped=False 须重采已有 view 进度行。"""
|
||||
clock = 1_800_000_000
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
transport = SnapshotTransport({
|
||||
1: view_for(1, view=999),
|
||||
2: view_for(2, view=222),
|
||||
})
|
||||
db, repo, service = make_service(tmp, transport)
|
||||
try:
|
||||
add_video(repo, 1)
|
||||
add_video(repo, 2)
|
||||
today = DATE
|
||||
for aid, view, src in ((1, 111, "view"), (2, 20, "seed")):
|
||||
repo.upsert_snapshot(
|
||||
today, aid,
|
||||
metrics={"view": view, "like": 1, "coin": 0, "favorite": 0,
|
||||
"share": 0, "reply": 0, "danmaku": 0},
|
||||
ok=True, err=None, captured_at=clock, src=src,
|
||||
)
|
||||
repo.kv_set(service.seeded_key(today), "1")
|
||||
forced = service.run_if_needed(
|
||||
date=today, force=True, keep_scraped=False,
|
||||
)
|
||||
assert forced.completed is True
|
||||
# 两条都重采
|
||||
assert transport.count("/web-interface/view") == 2
|
||||
assert int(repo.get_snapshot(today, 1)["view"]) == 999
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_force_default_keep_scraped_skips_view_progress():
|
||||
"""C: force 默认 keep_scraped=True(seal)跳过已 view 行。"""
|
||||
clock = 1_800_000_000
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
transport = SnapshotTransport({
|
||||
1: view_for(1, view=999),
|
||||
2: view_for(2, view=222),
|
||||
})
|
||||
db, repo, service = make_service(tmp, transport)
|
||||
try:
|
||||
add_video(repo, 1)
|
||||
add_video(repo, 2)
|
||||
today = DATE
|
||||
repo.upsert_snapshot(
|
||||
today, 1,
|
||||
metrics={"view": 111, "like": 1, "coin": 0, "favorite": 0,
|
||||
"share": 0, "reply": 0, "danmaku": 0},
|
||||
ok=True, err=None, captured_at=clock, src="view",
|
||||
)
|
||||
repo.upsert_snapshot(
|
||||
today, 2,
|
||||
metrics={"view": 20, "like": 1, "coin": 0, "favorite": 0,
|
||||
"share": 0, "reply": 0, "danmaku": 0},
|
||||
ok=True, err=None, captured_at=clock, src="seed",
|
||||
)
|
||||
repo.kv_set(service.seeded_key(today), "1")
|
||||
sealed = service.run_if_needed(date=today, force=True)
|
||||
assert sealed.completed is True
|
||||
assert transport.count("/web-interface/view") == 1
|
||||
assert int(repo.get_snapshot(today, 1)["view"]) == 111
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def main():
|
||||
test_snapshot_success_resume_and_force()
|
||||
test_resume_skips_ok_and_retries_failed()
|
||||
@@ -646,8 +787,12 @@ 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()
|
||||
test_seal_keep_scraped_skips_view_progress_punches_seed()
|
||||
test_force_keep_scraped_false_rescrapes_view_progress()
|
||||
test_force_default_keep_scraped_skips_view_progress()
|
||||
print("OK: 每日快照、断点续跑、失败占位与粉丝采集测试全部通过")
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,398 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""task:state 持久化与启动续跑: uv run python tests/test_task_state.py"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import tempfile
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
|
||||
from reaction_board.database import Database # noqa: E402
|
||||
from reaction_board.repository import Repository # noqa: E402
|
||||
from reaction_board.scheduler import ScheduledTask, Scheduler # noqa: E402
|
||||
from reaction_board.task_state import ( # noqa: E402
|
||||
STATUS_DEFERRED,
|
||||
STATUS_DONE,
|
||||
STATUS_FAILED,
|
||||
STATUS_PENDING,
|
||||
STATUS_RUNNING,
|
||||
bootstrap_from_last,
|
||||
ensure_pending,
|
||||
load_state,
|
||||
mark_deferred,
|
||||
mark_done,
|
||||
mark_failed,
|
||||
mark_running,
|
||||
recover_stale_running,
|
||||
resume_stat_date,
|
||||
startup_next_run,
|
||||
)
|
||||
from reaction_board.timeutil import SHANGHAI_TZ, next_daily_run # noqa: E402
|
||||
|
||||
|
||||
def _repo(tmp: str) -> tuple[Database, Repository]:
|
||||
db = Database(Path(tmp) / "app.db")
|
||||
db.initialize()
|
||||
return db, Repository(db)
|
||||
|
||||
|
||||
def test_restart_mid_run_resumes_same_stat_date():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
now = datetime(2026, 11, 3, 23, 40, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
mark_running(
|
||||
repo, "snapshot", stat_date="2026-11-03", now=now,
|
||||
progress={"done": 10, "total": 100},
|
||||
)
|
||||
# 模拟进程崩溃:running 残留
|
||||
recovered = recover_stale_running(repo, "snapshot", now=now + 60)
|
||||
assert recovered is not None
|
||||
assert recovered.status == STATUS_PENDING
|
||||
assert recovered.stat_date == "2026-11-03"
|
||||
assert recovered.progress == {"done": 10, "total": 100}
|
||||
assert recovered.due_at <= now + 60
|
||||
assert resume_stat_date(repo, "snapshot", fallback="2026-11-04") == "2026-11-03"
|
||||
nxt = startup_next_run(
|
||||
repo, "snapshot",
|
||||
wall_due_at=next_daily_run(now + 60, "23:15"),
|
||||
now=now + 60,
|
||||
)
|
||||
assert nxt <= now + 60
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_seeded_before_snapshot_at_does_not_force_run():
|
||||
"""seed 后挂 pending;白天重启认落盘 due_at(刻意 ≠ 墙钟)且未到不跑。"""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
now = datetime(2026, 11, 3, 10, 30, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
# 刻意偏离墙钟 23:15,证明认落盘而非 fallback
|
||||
slot = datetime(2026, 11, 3, 22, 0, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
wall = next_daily_run(now, "23:15")
|
||||
assert slot != wall
|
||||
ensure_pending(
|
||||
repo, "snapshot", due_at=slot, stat_date="2026-11-03", now=now,
|
||||
)
|
||||
repo.kv_set("snapshot:seeded:2026-11-03", "1")
|
||||
nxt = startup_next_run(
|
||||
repo, "snapshot",
|
||||
wall_due_at=wall,
|
||||
now=now,
|
||||
)
|
||||
assert nxt == slot
|
||||
assert nxt != wall
|
||||
assert nxt > now
|
||||
calls = []
|
||||
|
||||
def run_snapshot():
|
||||
calls.append("ran")
|
||||
return None
|
||||
|
||||
clock = [now]
|
||||
sched = Scheduler(
|
||||
[ScheduledTask("snapshot", 86400.0, run_snapshot, next_run=nxt)],
|
||||
clock=lambda: clock[0],
|
||||
)
|
||||
assert sched.run_due() == []
|
||||
assert calls == []
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_failed_future_due_at_does_not_run_early():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
now = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
due = now + 600
|
||||
mark_failed(
|
||||
repo, "snapshot", due_at=due, now=now,
|
||||
stat_date="2026-11-03", error="boom",
|
||||
)
|
||||
nxt = startup_next_run(
|
||||
repo, "snapshot",
|
||||
wall_due_at=next_daily_run(now, "23:15"),
|
||||
now=now,
|
||||
)
|
||||
assert nxt == due
|
||||
clock = [now]
|
||||
calls = []
|
||||
sched = Scheduler(
|
||||
[
|
||||
ScheduledTask(
|
||||
"snapshot", 86400.0, lambda: calls.append(1), next_run=nxt,
|
||||
)
|
||||
],
|
||||
clock=lambda: clock[0],
|
||||
)
|
||||
assert sched.run_due() == []
|
||||
assert calls == []
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_due_at_passed_runs_including_failed_retry():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
now = datetime(2026, 11, 3, 12, 10, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
due = now - 30 # 已过重试点
|
||||
mark_failed(
|
||||
repo, "snapshot", due_at=due, now=due - 300,
|
||||
stat_date="2026-11-03", error="boom",
|
||||
)
|
||||
nxt = startup_next_run(
|
||||
repo, "snapshot",
|
||||
wall_due_at=next_daily_run(now, "23:15"),
|
||||
now=now,
|
||||
)
|
||||
# 不得退回今晚墙钟;应立刻可跑
|
||||
assert nxt == due
|
||||
assert nxt <= now
|
||||
wall = next_daily_run(now, "23:15")
|
||||
assert nxt != wall
|
||||
calls = []
|
||||
clock = [now]
|
||||
sched = Scheduler(
|
||||
[
|
||||
ScheduledTask(
|
||||
"snapshot", 86400.0, lambda: calls.append(1), next_run=nxt,
|
||||
)
|
||||
],
|
||||
clock=lambda: clock[0],
|
||||
)
|
||||
results = sched.run_due()
|
||||
assert len(results) == 1
|
||||
assert calls == [1]
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_done_writes_next_due_and_blocks_until_then():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
now = datetime(2026, 11, 3, 23, 40, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
next_due = next_daily_run(now, "23:15")
|
||||
mark_done(repo, "snapshot", due_at=next_due, now=now, stat_date="2026-11-03")
|
||||
state = load_state(repo, "snapshot")
|
||||
assert state is not None
|
||||
assert state.status == STATUS_DONE
|
||||
nxt = startup_next_run(
|
||||
repo, "snapshot", wall_due_at=next_due, now=now + 60,
|
||||
)
|
||||
assert nxt == next_due
|
||||
assert nxt > now + 60
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_no_state_falls_back_to_wall_clock():
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
now = datetime(2026, 11, 3, 10, 0, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
wall = next_daily_run(now, "23:15")
|
||||
nxt = startup_next_run(repo, "snapshot", wall_due_at=wall, now=now)
|
||||
assert nxt == wall
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
|
||||
def test_fail_or_defer_preserves_progress():
|
||||
"""H7: mark_failed / mark_deferred 不传 progress 时保留已有 progress 与 stat_date。"""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
now = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
progress = {"done": 42, "total": 100, "cursor": "aid:99"}
|
||||
mark_running(
|
||||
repo, "snapshot", stat_date="2026-11-03", now=now,
|
||||
progress=progress,
|
||||
)
|
||||
before = load_state(repo, "snapshot")
|
||||
assert before is not None
|
||||
assert before.progress == progress
|
||||
assert before.stat_date == "2026-11-03"
|
||||
|
||||
due_fail = now + 300
|
||||
mark_failed(
|
||||
repo, "snapshot", due_at=due_fail, now=now + 1,
|
||||
progress=None, error="boom",
|
||||
)
|
||||
failed = load_state(repo, "snapshot")
|
||||
assert failed is not None
|
||||
assert failed.status == STATUS_FAILED
|
||||
assert failed.due_at == due_fail
|
||||
assert failed.progress == progress
|
||||
assert failed.stat_date == "2026-11-03"
|
||||
assert failed.error == "boom"
|
||||
|
||||
due_defer = now + 600
|
||||
mark_deferred(
|
||||
repo, "snapshot", due_at=due_defer, now=now + 2,
|
||||
progress=None, error="snapshot_quiet_window",
|
||||
)
|
||||
deferred = load_state(repo, "snapshot")
|
||||
assert deferred is not None
|
||||
assert deferred.status == STATUS_DEFERRED
|
||||
assert deferred.due_at == due_defer
|
||||
assert deferred.progress == progress
|
||||
assert deferred.stat_date == "2026-11-03"
|
||||
assert deferred.error == "snapshot_quiet_window"
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
|
||||
def test_bootstrap_from_last_fail_due_soon():
|
||||
"""A: 仅有 task:last 失败、无 state → 回填 failed 且 due_at 可立刻补跑。"""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
now = datetime(2026, 11, 3, 12, 0, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
wall = next_daily_run(now, "23:15")
|
||||
repo.kv_set(
|
||||
"task:last:snapshot",
|
||||
json.dumps({
|
||||
"name": "snapshot",
|
||||
"ok": False,
|
||||
"skipped": False,
|
||||
"started_at": int(now - 600),
|
||||
"finished_at": int(now - 500),
|
||||
"error": "boom",
|
||||
"summary": {"stat_date": "2026-11-03"},
|
||||
}, ensure_ascii=False),
|
||||
)
|
||||
assert load_state(repo, "snapshot") is None
|
||||
boot = bootstrap_from_last(
|
||||
repo, "snapshot", wall_due_at=wall, now=now, retry_seconds=300.0,
|
||||
)
|
||||
assert boot is not None
|
||||
assert boot.status == STATUS_FAILED
|
||||
assert boot.stat_date == "2026-11-03"
|
||||
assert boot.due_at <= now
|
||||
nxt = startup_next_run(
|
||||
repo, "snapshot", wall_due_at=wall, now=now, slot_hhmm="23:15",
|
||||
)
|
||||
assert nxt <= now
|
||||
assert nxt != wall
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_ensure_pending_does_not_glue_yesterday_stat_date():
|
||||
"""B: 昨天未完成保留;不得把今晚 due 粘到昨天的 stat_date 上。"""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
now = datetime(2026, 11, 3, 0, 10, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
yday_due = now - 60
|
||||
ensure_pending(
|
||||
repo, "snapshot", due_at=yday_due,
|
||||
stat_date="2026-11-02", now=now - 120,
|
||||
)
|
||||
tonight = datetime(2026, 11, 3, 23, 15, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
# seed 想挂今晚:不得冲掉昨天未完成
|
||||
kept = ensure_pending(
|
||||
repo, "snapshot", due_at=tonight,
|
||||
stat_date="2026-11-03", now=now,
|
||||
)
|
||||
assert kept.stat_date == "2026-11-02"
|
||||
assert kept.due_at == yday_due
|
||||
assert resume_stat_date(
|
||||
repo, "snapshot", fallback="2026-11-03",
|
||||
) == "2026-11-02"
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_cross_midnight_mid_seal_resumes_yesterday():
|
||||
"""规则1 / B: 23:15 崩溃 → 00:10 重启续跑昨天,due 立刻可跑。"""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
crash = datetime(2026, 11, 2, 23, 40, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
mark_running(
|
||||
repo, "snapshot", stat_date="2026-11-02", now=crash,
|
||||
progress={"done": 3, "total": 10},
|
||||
)
|
||||
repo.kv_set("snapshot:seeded:2026-11-02", "1")
|
||||
restart = datetime(2026, 11, 3, 0, 10, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
wall = next_daily_run(restart, "23:15")
|
||||
nxt = startup_next_run(
|
||||
repo, "snapshot", wall_due_at=wall, now=restart, slot_hhmm="23:15",
|
||||
)
|
||||
assert nxt <= restart
|
||||
assert resume_stat_date(
|
||||
repo, "snapshot", fallback="2026-11-03",
|
||||
) == "2026-11-02"
|
||||
state = load_state(repo, "snapshot")
|
||||
assert state is not None
|
||||
assert state.status == STATUS_PENDING
|
||||
assert state.progress == {"done": 3, "total": 10}
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def test_scheduler_due_reader_picks_up_enqueued_seal():
|
||||
"""D: 同进程 seed 写盘 due_at=now 后,下一轮 run_due 须跑 seal。"""
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
db, repo = _repo(tmp)
|
||||
try:
|
||||
now = datetime(2026, 11, 3, 23, 20, tzinfo=SHANGHAI_TZ).timestamp()
|
||||
# snapshot 原 next_run 挂次日;盘上被 enqueue 成 now
|
||||
tomorrow = next_daily_run(now, "23:15")
|
||||
calls = []
|
||||
|
||||
def run_snap():
|
||||
calls.append("snap")
|
||||
return None
|
||||
|
||||
def due_reader(name: str):
|
||||
st = load_state(repo, name)
|
||||
return float(st.due_at) if st and st.due_at else None
|
||||
|
||||
clock = [now]
|
||||
sched = Scheduler(
|
||||
[ScheduledTask("snapshot", 86400.0, run_snap, next_run=tomorrow)],
|
||||
clock=lambda: clock[0],
|
||||
due_reader=due_reader,
|
||||
)
|
||||
assert sched.run_due() == []
|
||||
assert calls == []
|
||||
ensure_pending(
|
||||
repo, "snapshot", due_at=now, stat_date="2026-11-03", now=now,
|
||||
)
|
||||
results = sched.run_due()
|
||||
assert len(results) == 1
|
||||
assert calls == ["snap"]
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
|
||||
def main_test():
|
||||
test_restart_mid_run_resumes_same_stat_date()
|
||||
test_seeded_before_snapshot_at_does_not_force_run()
|
||||
test_failed_future_due_at_does_not_run_early()
|
||||
test_due_at_passed_runs_including_failed_retry()
|
||||
test_done_writes_next_due_and_blocks_until_then()
|
||||
test_no_state_falls_back_to_wall_clock()
|
||||
test_fail_or_defer_preserves_progress()
|
||||
test_bootstrap_from_last_fail_due_soon()
|
||||
test_ensure_pending_does_not_glue_yesterday_stat_date()
|
||||
test_cross_midnight_mid_seal_resumes_yesterday()
|
||||
test_scheduler_due_reader_picks_up_enqueued_seal()
|
||||
print("OK: task_state persistence / resume / due_at gates")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main_test()
|
||||
@@ -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: 上海时区与每日时间计算测试全部通过")
|
||||
|
||||
|
||||
|
||||
+133
-4
@@ -610,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({
|
||||
@@ -627,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()
|
||||
|
||||
@@ -882,6 +1007,9 @@ 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()
|
||||
@@ -948,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({
|
||||
@@ -966,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()
|
||||
|
||||
Reference in new issue
Block a user