diff --git a/docs/自动化与资料收集规划.md b/docs/自动化与资料收集规划.md index 7887599..5611ab3 100644 --- a/docs/自动化与资料收集规划.md +++ b/docs/自动化与资料收集规划.md @@ -374,8 +374,8 @@ lookup_order(order_no) -> { | `key` (PK) | 作业标识:`ck_health` / `ck_relogin` | | `title` | 展示名 | | `enabled` | 开关(**默认 false**) | -| `interval_seconds` | 检测周期(CK 巡检默认 6h) | -| `batch_size` | 单轮处理上限(`ck_health` 默认 **200**,按 6h 周期反推,见 §7.3;`ck_relogin` 默认 20) | +| `interval_seconds` | **单账号重检间隔**(CK 巡检默认 6h)—— 注意不是作业运行周期,作业每个 tick 都跑(见 §7.3) | +| `batch_size` | 单轮处理上限(`ck_health` 默认 **200**,按「5 分钟一轮 × 200 条 = 40 条/分钟 ≥ 需要的 36 条/分钟」反推,见 §7.3;`ck_relogin` 默认 20) | | `concurrency` | 单轮并发 | | `config` (JSON) | 作业私有配置(退避表、每日上限、失败率阈值等) | | `last_run_at` / `next_run_at` | 调度用 | @@ -391,8 +391,10 @@ lookup_order(order_no) -> { ### 7.3 P1a · 作业一:`ck_health`(巡检,只检测) - **选取范围**:从**持有 CK 的 `login_tasks` 行**里挑(即现有 `cookie_holding_filter()` 的口径:`cookie is not null and cookie != ''`)—— CK 是挂在登录记录上的,不是挂在 `accounts` 上; -- **调度**:每 5 分钟 tick 一次;每轮挑「距上次检测超过 `interval_seconds`(默认 6h)」的行,最多 `batch_size`(默认 50)个。 +- **调度**:**每 5 分钟跑一轮**(`TICK_SECONDS = 300`);每轮挑「距上次检测超过 `interval_seconds`(默认 6h)」的行,最多 `batch_size`(默认 200)个。 这样全库是**滚动覆盖**的,负载平稳,不会出现「整点扫 1.2 万个账号」的尖峰; +- **⚠️ 两个「间隔」不能混**:`TICK_SECONDS` 是**作业多久跑一轮**;`job.interval_seconds` 是**同一行数据多久重新检测一次**(只用于选行 cutoff)。 + 如果误把作业周期设成 `interval_seconds`,就变成「200 条/6 小时 = 800 条/天」,全库 12,881 条要 16 天才能过一轮 —— 等于没巡检(P1a 实现时踩过一次,已修); - **排除**:`auto_managed = false`;`attempt_status in (pending, running)`(正在被批次处理);最近 10 分钟内刚检测过(防重入); - **动作**:调用现有 `check_douyu_cookie()`(两个只读接口,直连),写回 `ck_check_status / ck_check_result / ck_checked_at`; - **并发**:默认 5,沿用现有 `ThreadPoolExecutor` 模式; @@ -406,18 +408,19 @@ lookup_order(order_no) -> { | 当前状态分布 | valid **10,079** / invalid **905** / **从未检测 1,897** | | 历史检测时间 | 最早 2026-08-07,最新 2026-10-02(人工零星跑的) | -按 6 小时巡检周期折算: +按「单账号 6 小时内重检一次」折算: ``` -需要吞吐 = 12,881 ÷ 6h ≈ 2,147 行/小时 ≈ 179 行/5 分钟 tick -单行耗时 ≈ 0.5~1s(两个只读接口),并发 5 → 179 行约 18~36 秒/tick ← 可行 +需要吞吐 = 12,881 ÷ 6h ≈ 36 行/分钟 +一个 tick = 5 分钟 → 每轮需要 ≈ 180 行 +单行耗时 ≈ 0.13s(实测 3 个账号 0.4s,并发 3)→ 180 行并发 5 约 5 秒/轮 ← 很轻松 ``` -**所以 `batch_size` 默认值要按周期反推,不能拍一个 50**:50 的话实际巡检周期会被拉长到 ≈ 21 小时(12,881 ÷ (50/5min))。建议: +**`batch_size` 默认 200**(比需要的 180 留一点余量): -- 默认目标周期 6h → `batch_size` 默认 **200**(留余量),`concurrency` 默认 **5**; -- **冷启动**:首次启用时先跑小批量(如 50)观察一轮,确认无异常和风控反应,再调到目标值; -- 取行时按 `ck_checked_at ASC`(最久没查的优先),自然错峰,不会同一时刻全量到期。 +- **冷启动**:全库积压 12,881 条,按 200 条/5 分钟 ≈ **5.4 小时**跑完第一遍,随后进入稳态; +- **稳态**:每分钟约 36 条到期,每轮扫到一百多条,负载很轻; +- 取行按 `ck_checked_at ASC`(最久没查的优先,`NULLS FIRST`),自然错峰,不会同一时刻全量到期。 ### 7.4 P1b · 作业二:`ck_relogin`(自愈,默认关闭) @@ -947,7 +950,10 @@ pending_material(待用户补资料) ← 模板有必填字段时的 - **P1a 先上**:只做「帮你把失效 CK 找出来」,零风险、立刻省掉大批量人工点检测;观察一到两周、确认检测结果与人工判断一致后,再打开 P1b; - **P0 随时可做**:只动本系统,order_site 侧代码改动量为 0; - **P1b 代码可与 P1a 一起写完**,但开关默认关闭,等观察结论再开; -- **P1a 上线按三段走**:先 `batch_size=50` 跑一轮看风控反应 → 调到 200 达到 6h 周期 → 观察一到两周确认检测结果与人工判断一致。 +- **P1a 上线按三段走**: + ① 打开 `AUTOMATION_ENABLED=true` 并重启(此时作业开关仍关着,调度线程空转,不会有任何检测); + ② 在 `/automation` 页点「立即跑一次」并先把「每轮条数」设 **50** —— 手动触发不受开关限制,正好用来观察斗鱼风控与失败率; + ③ 确认无异常后把「每轮条数」调 **200** 并打开「启用」,调度器接管(5 分钟一轮,约 5.4 小时覆盖全库一遍),观察一到两周确认检测结果与人工判断一致。 依赖关系: diff --git a/tests/test_automation_ck_health.py b/tests/test_automation_ck_health.py index 1655b05..45a1fff 100644 --- a/tests/test_automation_ck_health.py +++ b/tests/test_automation_ck_health.py @@ -258,7 +258,7 @@ class TestScheduler(_Base): run = self.session.query(AutomationRun).one() assert run.status == "failed" and "接口挂了" in run.message - def test_next_run_follows_interval_and_counters_persist(self): + def test_next_run_follows_tick_and_counters_persist(self): from web.backend.services.automation import ensure_jobs calls: list[str] = [] @@ -274,7 +274,12 @@ class TestScheduler(_Base): self.session.query(AutomationJob).filter(AutomationJob.key == "probe").one() ) assert job.last_status == "success" - assert job.next_run_at - job.last_run_at == timedelta(seconds=3600) + # 下一轮在一个 tick 之后(作业每个 tick 都跑),与 interval_seconds 无关 + from web.backend.services.automation.scheduler import TICK_SECONDS + + assert _aware(job.next_run_at) - _aware(job.last_run_at) == timedelta( + seconds=TICK_SECONDS + ) run = self.session.query(AutomationRun).one() assert (run.scanned, run.valid, run.invalid) == (3, 2, 1) @@ -334,6 +339,11 @@ class TestAutomationPermissions(_Base): assert checker(current_user=operator).id == operator.id +def _aware(value): + """库里读出的是 naive UTC,比较前补时区。""" + return value if value.tzinfo else value.replace(tzinfo=UTC) + + _REAL_CHECK = ck_health.check_douyu_cookie @@ -364,7 +374,7 @@ class TestAutomationApi(_Base): def test_list_jobs_exposes_scheduler_and_job_state(self): api = self._api() payload = api.list_jobs(db=self.session, current=self.admin) - assert payload["scheduler"]["tick_seconds"] == 60 + assert payload["scheduler"]["tick_seconds"] == 300 assert payload["scheduler"]["running"] is False keys = [item["key"] for item in payload["jobs"]] assert keys == [ck_health.JOB_KEY] diff --git a/web/backend/services/automation/__init__.py b/web/backend/services/automation/__init__.py index 342708e..66daaa3 100644 --- a/web/backend/services/automation/__init__.py +++ b/web/backend/services/automation/__init__.py @@ -46,7 +46,9 @@ def register_default_jobs() -> None: JobSpec( key=CK_HEALTH_KEY, title="CK 自动巡检", - # 6 小时一轮;全库 12,881 条按此反推 batch_size=200(见规划附录 C)。 + # interval_seconds = 单账号多久重新检测一次(不是作业周期; + # 作业每个 tick 都跑)。全库 12,881 条按 5 分钟一轮、单轮 200 条 + # 反推,吞吐 40 条/分钟,够 6 小时内覆盖一遍(规划 §7.3)。 interval_seconds=6 * 3600, batch_size=200, concurrency=5, diff --git a/web/backend/services/automation/scheduler.py b/web/backend/services/automation/scheduler.py index 7271eeb..aea822b 100644 --- a/web/backend/services/automation/scheduler.py +++ b/web/backend/services/automation/scheduler.py @@ -26,8 +26,16 @@ from sqlalchemy.orm import Session from ...database import SessionLocal from ...models import AutomationJob, AutomationRun -# 调度轮询间隔:作业间隔都以它为单位对齐(默认 6h 的作业不会差这一分钟)。 -TICK_SECONDS = 60 +# 调度轮询间隔,也是作业的运行节奏:每个 tick 跑一轮。 +# +# 注意区分两个「间隔」: +# - **本常量**:作业多久跑一轮(5 分钟); +# - **job.interval_seconds**:同一行数据多久重新检测一次(默认 6h),只用于选行 cutoff。 +# +# 分开的原因:全库 12,881 条的 CK 若想「6 小时内都被查一遍」,需要约 36 条/分钟的 +# 吞吐;按 5 分钟一轮、单轮 200 条算 = 40 条/分钟,刚好够。若把作业周期也设成 6h, +# 就变成 200 条/6h = 800 条/天,全库要 16 天才能过一轮,等于没巡检。 +TICK_SECONDS = 300 TRUTHY = {"1", "true", "yes", "on"} @@ -169,9 +177,9 @@ def _run_one(db: Session, job: AutomationJob, spec: JobSpec) -> dict[str, Any]: run.detail = counters.get("detail") or None job.last_run_at = finished - job.next_run_at = finished + timedelta( - seconds=max(int(job.interval_seconds or spec.interval_seconds), 60) - ) + # 下一轮固定在一个 tick 之后;interval_seconds 是「同一行多久重检」, + # 由作业自己在选行时用,不参与调度。 + job.next_run_at = finished + timedelta(seconds=TICK_SECONDS) job.last_status = status job.last_message = message job.updated_at = finished diff --git a/web/frontend/src/pages/AutomationPage.tsx b/web/frontend/src/pages/AutomationPage.tsx index a41163b..602bab3 100644 --- a/web/frontend/src/pages/AutomationPage.tsx +++ b/web/frontend/src/pages/AutomationPage.tsx @@ -178,7 +178,7 @@ export default function AutomationPage() { showIcon style={{ marginBottom: 16 }} message="巡检只检测、不改 CK" - description="本作业只把失效账号找出来并标记,不会替换任何 Cookie、也不会改账号状态;重登仍由人工在「CK 检测与重登」页面操作。" + description="每 5 分钟跑一轮,每轮按「每轮条数」上限挑最久没检测的账号;「单账号重检间隔」指同一个账号最多多久重新检测一次。本作业只标记失效账号,不会替换任何 Cookie、也不会改账号状态,重登仍由人工在「CK 检测与重登」页面操作。" /> {jobs.map((job) => ( @@ -212,11 +212,11 @@ export default function AutomationPage() {