diff --git a/docs/自动化与资料收集规划.md b/docs/自动化与资料收集规划.md index 4b11687..4a3af0e 100644 --- a/docs/自动化与资料收集规划.md +++ b/docs/自动化与资料收集规划.md @@ -1,7 +1,7 @@ # 自动化与资料收集规划 更新时间：2026-10-02 -状态：**规划中（未开工）**。产品决策已与负责人确认（见 §4），代码实现待排期。 +状态：**P1a（CK 自动巡检）已落地**，P0/P2/P3/P4/P1b 待排期。产品决策已确认（见 §4），施工单见附录 C。 --- @@ -356,6 +356,15 @@ lookup_order(order_no) -> { - **P1a（先做）**：把「哪些账号 CK 失效了」自动找出来并展示 —— **只检测、只标记、只提示，不改动任何 CK**；重登仍然人工点。 - **P1b（观察后再开）**：把「失效账号自动重登」也自动化 —— 有退避、有熔断、有开关，**默认关闭**。 +> **P1a 已落地（2026-10-02）**，实际落点： +> `services/automation/scheduler.py`（调度线程 + 到点判定 + 运行记录）、 +> `services/automation/jobs/ck_health.py`（巡检作业）、 +> `routers/automation.py`（4 个接口）、`permissions.py`（`automation:view` / `automation:manage`）、 +> `migrations/versions/20261002_0037_automation_jobs.py`、 +> 前端 `pages/AutomationPage.tsx`（菜单「自动巡检」）、 +> `tests/test_automation_ck_health.py`（22 条）。 +> 端到端已用 3 个真实账号验证：巡检结果与直接检测逐条一致，且 `cookie` / `status` / `message` / `attempt_status` 全程未变。 + ### 7.2 数据模型（迁移 1 个，P1a/P1b 共用） **新表 `automation_jobs`** @@ -1065,7 +1074,7 @@ OSS_PREFIX=collect --- -## 附录 A · 迁移清单（共 5 个，均为加表/加字段，不动存量数据） +## 附录 A · 迁移清单（均为加表/加字段，不动存量数据；编号按落地顺序取） | 阶段 | 迁移内容 | | --- | --- | @@ -1126,15 +1135,20 @@ OSS_PREFIX=collect 战略与数据模型在正文已定；这里只写**按什么顺序动手、每步怎么验证**。 -### C.1 迁移编号分配（现状 head = `20261001_0036`） +### C.1 迁移编号（按落地顺序取，**不预留**） -| 阶段 | 文件名 | revision | +现状 head = `20261001_0036`。**谁先落地谁取下一个号**，不要按阶段预分配。 + +| 阶段 | revision | 说明 | | --- | --- | --- | -| P0 | `20261002_0037_external_order_cache.py` | `20261002_0037` | -| P1 | `20261002_0038_automation_jobs.py` | `20261002_0038` | -| P2 | `20261002_0039_collect_links_submissions.py` | `20261002_0039` | -| P3 | `20261002_0040_skin_catalog_exchange_records.py` | `20261002_0040` | -| P4 | `20261002_0041_exchange_order_tables.py` | `20261002_0041` | +| P1a（已落地） | `20261002_0037` | `automation_jobs` / `automation_runs` / `login_tasks` +3 / `accounts.auto_managed` | +| P0 | 落地时取下一个 | `external_order_cache` | +| P2 / P3 / P4 | 各自落地时依次取号 | 见附录 A 的表清单 | + +> **为什么不能预留编号**：Alembic 的 `down_revision` 必须形成单链。若先给 P1 占 0038、 +> 把 0037 留给 P0，等 P0 真落地时它的 `down_revision` 只能指向 0038（否则与 0038 平级， +> 出现**两个 head**，`tests/test_migrations.py` 的单头断言会直接失败），编号比父提交还小。 +> 所以按落地顺序递增，才是唯一不别扭的做法。 约定：每个迁移的 `down_revision` 指向前一个；**写成幂等的**（照 `20261001_0036` 的写法，先 `sa.inspect` 查存在再改）；每加一个迁移，`tests/test_migrations.py` 的 `HEAD_REVISION` 要同步改。 @@ -1143,7 +1157,11 @@ OSS_PREFIX=collect - 测试文件按功能命名：`tests/test_<域>_<主题>.py`（现有 38 个文件、235 条）； - **纯函数优先**：匹配打分、别名归一、365 天判定、订单号归一、模板必填校验都做成纯函数重点覆盖； - **写操作不进自动回归**：兑换/重登/支付用 mock 客户端做契约测试（沿用 `tests/test_douyu_write_path_contracts.py`）； -- 提交门禁：pre-commit 跑 `ruff format` + `tsc -b`；pre-push 跑全量 pytest； +- 提交门禁：pre-commit 跑 `ruff format --check`（仅暂存文件）+ `tsc -b`；pre-push 跑全量 pytest； +- **DB 读出来的时间是 naive UTC**：`datetime` 列取回后不带时区，直接和 `datetime.now(UTC)` 比较会 + `TypeError: can't compare offset-naive and offset-aware`（P1a 踩过一次）。比较前一律用 + `_as_utc()`（`services/yyb_service.py` 与 `services/automation/scheduler.py` 各有一份）补时区。 + SQL 里用绑定参数比较不受影响，只有 Python 侧比较需要注意 —— P0 判缓存过期时会再遇到； - **每个阶段独立可发布**，新作业/新功能一律**开关默认关闭**。 ### C.3 已核实的环境事实 @@ -1158,11 +1176,11 @@ OSS_PREFIX=collect | OSS | 桶 `live-hub-assets`（私有+版本控制）、RAM `live-hub-oss`、`.env` 已配 | `oss2`；**不要用 HeadBucket** | | 迁移 head / 测试基线 | `20261001_0036` / 235 条 | 见 C.1 | -### C.4 P1a 施工单（下一个要做的，8 步） +### C.4 P1a 施工单（✅ 已按此完成，8 步） | # | 动作 | 文件 | 验证 | | --- | --- | --- | --- | -| 1 | 迁移 0038：`automation_jobs`、`automation_runs`；`login_tasks` +3 字段；`accounts.auto_managed` | `migrations/versions/20261002_0038_*.py`、`models.py` | `pytest tests/test_migrations.py`；本地建表通过 | +| 1 | 迁移 0037：`automation_jobs`、`automation_runs`；`login_tasks` +3 字段；`accounts.auto_managed` | `migrations/versions/20261002_0037_automation_jobs.py`、`models.py` | `pytest tests/test_migrations.py`；本地建表通过 | | 2 | 调度器骨架：作业注册表、60s tick、进程内锁、开关、`automation_runs` 记录、优雅停止 | `services/automation/scheduler.py` | 单测：假时钟驱动 tick，断言「作业 enabled=false 时不执行」「不重叠执行」 | | 3 | `ck_health` 作业：选行（`cookie_holding_filter` + 超期 + 排除在途/`auto_managed=false`）、并发 5、调 `check_douyu_cookie()`、写回 `ck_check_*` | `services/automation/jobs/ck_health.py` | 单测：选行条件正确；**断言全程不写 `cookie` / `status`**（P1a 的核心约束） | | 4 | 接口 + 权限 + 注册：4 个 `/api/automation/*`；`automation:view` / `automation:manage`；lifespan 启停 | `routers/automation.py`、`permissions.py`、`main.py` | `tests/test_automation_api.py`：权限校验、开关生效、手动触发 | @@ -1177,7 +1195,7 @@ OSS_PREFIX=collect | 阶段 | 顺序 | | --- | --- | -| **P0** | ① `pyproject.toml` 加 `psycopg[binary]` → ② 迁移 0037 + `models.py` → ③ `external_order_source.py`（唯一跨库 SQL；URL 归一成 `postgresql+psycopg://`；`statement_timeout`；2 条连接池） → ④ `tests/test_external_order_source.py`（用真实只读账号跑只读查询，覆盖 A/B/C 三桶与不存在订单） → ⑤ 降级开关验证 | -| **P2** | ① 迁移 0039 → ② `object_storage.py`（`oss2`，Put/Get/Delete/List，不用 HeadBucket） → ③ `collect_service.py`（token/限流/图片校验+Pillow 重编码+缩略图） → ④ 公开路由 `/api/public/collect/*` + 内部 `/api/collect/*` → ⑤ 前端公开页（放 `App.tsx` 的 `/login` 同级）+ 内部两页 → ⑥ 测试（含"宿主机找不到图片副本"的验收） | -| **P3** | ① 迁移 0040 → ② `skin_catalog` 从 `douyu_goods_snapshot` 同步 + `sku_catalog_sync` 作业（挂 P1 调度器） → ③ `skin_service.py`（别名归一 + `can_exchange` 365 天 + 落账） → ④ `/api/skins/*` → ⑤ `/skins` 页 → ⑥ 把 62 条唯一映射 + 14 条多选映射导入 | -| **P4** | ① 迁移 0041 → ② `exchange_order_service.py`（`resolve_skin_rule` 按 SKU 四级匹配 + 状态机 + 事件） → ③ 前置校验（CK/365 天/绑定/积分） → ④ 工单接口 + 与 `collect_submissions` 衔接 → ⑤ `/exchange-rules` + `/exchange-orders` 两页 → ⑥ 端到端联调（发链接 → 提交 → 校验 → 执行 → 台账） | +| **P0** | ① `pyproject.toml` 加 `psycopg[binary]` → ② 迁移（取下一个号）+ `models.py` → ③ `external_order_source.py`（唯一跨库 SQL；URL 归一成 `postgresql+psycopg://`；`statement_timeout`；2 条连接池） → ④ `tests/test_external_order_source.py`（用真实只读账号跑只读查询，覆盖 A/B/C 三桶与不存在订单） → ⑤ 降级开关验证 | +| **P2** | ① 迁移（取下一个号）→ ② `object_storage.py`（`oss2`，Put/Get/Delete/List，不用 HeadBucket） → ③ `collect_service.py`（token/限流/图片校验+Pillow 重编码+缩略图） → ④ 公开路由 `/api/public/collect/*` + 内部 `/api/collect/*` → ⑤ 前端公开页（放 `App.tsx` 的 `/login` 同级）+ 内部两页 → ⑥ 测试（含"宿主机找不到图片副本"的验收） | +| **P3** | ① 迁移（取下一个号）→ ② `skin_catalog` 从 `douyu_goods_snapshot` 同步 + `sku_catalog_sync` 作业（挂 P1 调度器） → ③ `skin_service.py`（别名归一 + `can_exchange` 365 天 + 落账） → ④ `/api/skins/*` → ⑤ `/skins` 页 → ⑥ 把 62 条唯一映射 + 14 条多选映射导入 | +| **P4** | ① 迁移（取下一个号）→ ② `exchange_order_service.py`（`resolve_skin` 两步匹配 + 状态机 + 事件） → ③ 前置校验（CK/365 天/绑定/积分） → ④ 工单接口 + 与 `collect_submissions` 衔接 → ⑤ `/exchange-rules` + `/exchange-orders` 两页 → ⑥ 端到端联调（发链接 → 提交 → 校验 → 执行 → 台账） | diff --git a/tests/test_automation_ck_health.py b/tests/test_automation_ck_health.py new file mode 100644 index 0000000..1655b05 --- /dev/null +++ b/tests/test_automation_ck_health.py @@ -0,0 +1,443 @@ +"""P1a 回归：CK 自动巡检。 + +两块约束必须一起守住： + +1. **调度器行为**：只跑到点的、打开开关的作业；作业抛异常只记 failed，不能 + 把调度线程带崩；跑完排下一轮时间；已有配置不能被重启重置。 +2. **本作业绝不改 CK**：P1a 的定位是「只检测、只标记」，所以巡检后 + ``cookie`` / ``status`` / ``message`` / ``attempt_*`` 必须一字未动 —— + 线上全靠这条保证「自动化上线了但人工流程照旧」，回归时一旦有人顺手写了 + ``status``，这里要立刻红。 +""" + +from datetime import UTC, datetime, timedelta + +import pytest +from fastapi import HTTPException +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from web.backend.database import Base +from web.backend.deps import require_permission +from web.backend.models import Account, AutomationJob, AutomationRun, LoginTask, User +from web.backend.services.automation import ( + JobSpec, + register_default_jobs, + register_job, + reset_registry, + run_due_jobs, + run_job_now, +) +from web.backend.services.automation.jobs import ck_health + + +class _Base: + def setup_method(self): + self.engine = create_engine("sqlite://") + Base.metadata.create_all(self.engine) + self.session = sessionmaker(bind=self.engine)() + self.admin = User( + username="admin", password_hash="h", role="super_admin", is_active=True + ) + self.support = User( + username="kf", password_hash="h", role="support", is_active=True + ) + self.session.add_all([self.admin, self.support]) + self.session.commit() + # 注册表是模块级的，不重置会被上一个用例注册的探针作业污染。 + reset_registry() + register_default_jobs() + + def teardown_method(self): + self.session.close() + + def make_account(self, name="acc", *, auto_managed=True, assigned_to=None): + account = Account( + username=name, + password="p", + email=f"{name}@example.com", + email_password="ep", + assigned_to=assigned_to, + auto_managed=auto_managed, + ) + self.session.add(account) + self.session.commit() + return account + + def make_task( + self, + account, + *, + cookie="acf_uid=1", + status="success", + attempt_status="", + ck_checked_at=None, + ck_check_status="", + ): + task = LoginTask( + batch_id="b1", + account_id=account.id, + created_by=self.admin.id, + status=status, + cookie=cookie, + attempt_status=attempt_status, + ck_checked_at=ck_checked_at, + ck_check_status=ck_check_status, + ) + self.session.add(task) + self.session.commit() + return task + + +class TestCkHealthSelection(_Base): + def _selected_ids(self, job): + return [ + t.id for t in ck_health._select_tasks(self.session, job, datetime.now(UTC)) + ] + + def test_picks_only_rows_that_still_hold_cookie(self): + job = self._job() + with_cookie = self.make_task(self.make_account("a")) + self.make_task(self.make_account("b"), cookie="") + assert self._selected_ids(job) == [with_cookie.id] + + def test_skips_accounts_with_inflight_attempt(self): + job = self._job() + idle = self.make_task(self.make_account("a")) + self.make_task(self.make_account("b"), attempt_status="running") + self.make_task(self.make_account("c"), attempt_status="pending") + assert self._selected_ids(job) == [idle.id] + + def test_skips_accounts_opted_out_of_automation(self): + job = self._job() + managed = self.make_task(self.make_account("a")) + self.make_task(self.make_account("b", auto_managed=False)) + assert self._selected_ids(job) == [managed.id] + + def test_respects_interval_and_prefers_stalest(self): + job = self._job() + fresh = self.make_task( + self.make_account("fresh"), ck_checked_at=datetime.now(UTC) + ) + stale = self.make_task( + self.make_account("stale"), + ck_checked_at=datetime.now(UTC) - timedelta(hours=30), + ) + never = self.make_task(self.make_account("never"), ck_checked_at=None) + ids = self._selected_ids(job) + assert fresh.id not in ids + # 从没检测过的排最前，其次是久未检测的 + assert ids[0] == never.id + assert stale.id in ids + + def test_batch_size_caps_the_round(self): + job = self._job() + job.batch_size = 2 + for index in range(5): + self.make_task(self.make_account(f"a{index}")) + assert len(self._selected_ids(job)) == 2 + + def _job(self): + return AutomationJob( + key=ck_health.JOB_KEY, interval_seconds=6 * 3600, batch_size=200 + ) + + +class TestCkHealthNeverTouchesCookie(_Base): + def test_patrol_only_writes_check_fields(self): + job = self._job() + account = self.make_account("acc") + task = self.make_task(account, attempt_status="failed", ck_check_status="") + before = (task.cookie, task.status, task.message, task.attempt_status) + + ck_health.check_douyu_cookie = _fake_check(valid=True) + try: + counters = ck_health.run_ck_health(self.session, job) + finally: + ck_health.check_douyu_cookie = _REAL_CHECK + + self.session.refresh(task) + assert counters["scanned"] == 1 and counters["valid"] == 1 + assert task.ck_check_status == "valid" + assert task.ck_checked_at is not None + assert task.ck_check_result["nickname"] == "测试" + # 核心断言：CK 本体与尝试状态一字未动 + assert (task.cookie, task.status, task.message, task.attempt_status) == before + + def test_invalid_result_marks_invalid_but_keeps_cookie(self): + job = self._job() + task = self.make_task(self.make_account("acc")) + ck_health.check_douyu_cookie = _fake_check(valid=False) + try: + counters = ck_health.run_ck_health(self.session, job) + finally: + ck_health.check_douyu_cookie = _REAL_CHECK + self.session.refresh(task) + assert counters["invalid"] == 1 + assert task.ck_check_status == "invalid" + assert task.cookie == "acf_uid=1" + + def test_detection_exception_does_not_abort_round(self): + job = self._job() + self.make_task(self.make_account("a")) + self.make_task(self.make_account("b")) + ck_health.check_douyu_cookie = _fake_check(raises=True) + try: + counters = ck_health.run_ck_health(self.session, job) + finally: + ck_health.check_douyu_cookie = _REAL_CHECK + assert counters["scanned"] == 2 + assert counters["invalid"] == 2 + + def _job(self): + return AutomationJob( + key=ck_health.JOB_KEY, interval_seconds=6 * 3600, batch_size=200 + ) + + +class TestScheduler(_Base): + def _register_probe(self, calls): + def run(db, job): + calls.append(job.key) + return {"scanned": 3, "valid": 2, "invalid": 1, "message": "ok"} + + register_job( + JobSpec( + key="probe", + title="探针", + interval_seconds=3600, + batch_size=1, + concurrency=1, + run=run, + ) + ) + + def test_only_enabled_jobs_run(self): + from web.backend.services.automation import ensure_jobs + + calls: list[str] = [] + self._register_probe(calls) + ensure_jobs(self.session) + self.session.query(AutomationJob).filter(AutomationJob.key == "probe").update( + {"enabled": True} + ) + self.session.commit() + + run_due_jobs(self.session) + assert calls == ["probe"] + + calls.clear() + run_due_jobs(self.session) + # 上一轮已排了 next_run_at，未到点不再跑 + assert calls == [] + + def test_job_failure_is_recorded_not_raised(self): + from web.backend.services.automation import ensure_jobs + + def boom(db, job): + raise RuntimeError("接口挂了") + + register_job( + JobSpec( + key="boom", + title="会炸的作业", + interval_seconds=3600, + batch_size=1, + concurrency=1, + run=boom, + ) + ) + ensure_jobs(self.session) + self.session.query(AutomationJob).filter(AutomationJob.key == "boom").update( + {"enabled": True} + ) + self.session.commit() + + outcomes = run_due_jobs(self.session) + assert outcomes[0]["status"] == "failed" + run = self.session.query(AutomationRun).one() + assert run.status == "failed" and "接口挂了" in run.message + + def test_next_run_follows_interval_and_counters_persist(self): + from web.backend.services.automation import ensure_jobs + + calls: list[str] = [] + self._register_probe(calls) + ensure_jobs(self.session) + self.session.query(AutomationJob).filter(AutomationJob.key == "probe").update( + {"enabled": True} + ) + self.session.commit() + + run_due_jobs(self.session) + job = ( + 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) + run = self.session.query(AutomationRun).one() + assert (run.scanned, run.valid, run.invalid) == (3, 2, 1) + + def test_ensure_jobs_keeps_operator_settings(self): + from web.backend.services.automation import ensure_jobs + + ensure_jobs(self.session) + job = ( + self.session.query(AutomationJob) + .filter(AutomationJob.key == ck_health.JOB_KEY) + .one() + ) + job.enabled = True + job.batch_size = 7 + self.session.commit() + + ensure_jobs(self.session) # 模拟重启 + self.session.refresh(job) + assert job.enabled is True and job.batch_size == 7 + + def test_run_job_now_ignores_switch(self): + from web.backend.services.automation import ensure_jobs + + ensure_jobs(self.session) + outcome = run_job_now(self.session, ck_health.JOB_KEY) + assert outcome["job_key"] == ck_health.JOB_KEY + assert self.session.query(AutomationRun).count() == 1 + + def test_run_job_now_unknown_key(self): + with pytest.raises(KeyError): + run_job_now(self.session, "nope") + + +class TestAutomationPermissions(_Base): + def test_view_allows_operation_denies_support(self): + from web.backend.permissions import ROLE_PERMISSIONS + + assert "automation:view" in ROLE_PERMISSIONS["operation"] + assert "automation:view" not in ROLE_PERMISSIONS["support"] + assert "automation:manage" in ROLE_PERMISSIONS["super_admin"] + assert "automation:manage" not in ROLE_PERMISSIONS["operation"] + + def test_manage_dependency_rejects_operator(self): + operator = User( + username="op", password_hash="h", role="operation", is_active=True + ) + checker = require_permission("automation:manage") + with pytest.raises(HTTPException) as exc: + checker(current_user=operator) + assert exc.value.status_code == 403 + + def test_view_dependency_accepts_operator(self): + operator = User( + username="op", password_hash="h", role="operation", is_active=True + ) + checker = require_permission("automation:view") + assert checker(current_user=operator).id == operator.id + + +_REAL_CHECK = ck_health.check_douyu_cookie + + +def _fake_check(*, valid=True, raises=False): + def fake(cookie: str) -> dict: + if raises: + raise RuntimeError("网络异常") + return { + "valid": valid, + "message": "ok" if valid else "Cookie 已失效", + "fish_ball": 1, + "nickname": "测试", + "level": 30, + "checked_at": datetime.now(UTC), + } + + return fake + + +class TestAutomationApi(_Base): + """接口层：序列化字段、开关落库、手动触发不受开关限制。""" + + def _api(self): + from web.backend.routers import automation as api + + return api + + 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"]["running"] is False + keys = [item["key"] for item in payload["jobs"]] + assert keys == [ck_health.JOB_KEY] + job = payload["jobs"][0] + assert job["enabled"] is False + assert job["interval_seconds"] == 6 * 3600 + assert job["batch_size"] == 200 + assert job["latest_run"] is None + + def test_update_job_persists_and_schedules_next_run(self): + api = self._api() + api.update_job( + ck_health.JOB_KEY, + api.JobUpdate( + enabled=True, interval_seconds=3600, batch_size=5, concurrency=2 + ), + db=self.session, + current=self.admin, + ) + job = self.session.query(AutomationJob).one() + assert job.enabled is True + assert job.interval_seconds == 3600 + assert job.batch_size == 5 + assert job.next_run_at is not None # 打开开关就排了一轮 + + # 审计留痕 + from web.backend.models import AuditLog + + entry = ( + self.session.query(AuditLog) + .filter(AuditLog.action == "automation:update") + .one() + ) + assert entry.target == ck_health.JOB_KEY + + def test_run_endpoint_records_run_and_audit(self): + api = self._api() + self.make_task(self.make_account("a")) + ck_health.check_douyu_cookie = _fake_check(valid=True) + try: + outcome = api.run_job( + ck_health.JOB_KEY, db=self.session, current=self.admin + ) + finally: + ck_health.check_douyu_cookie = _REAL_CHECK + assert outcome["status"] == "success" + assert outcome["scanned"] == 1 + assert self.session.query(AutomationRun).count() == 1 + + from web.backend.models import AuditLog + + assert ( + self.session.query(AuditLog) + .filter(AuditLog.action == "automation:run") + .count() + == 1 + ) + + def test_run_endpoint_unknown_job_returns_404(self): + api = self._api() + with pytest.raises(HTTPException) as exc: + api.run_job("nope", db=self.session, current=self.admin) + assert exc.value.status_code == 404 + + def test_runs_endpoint_filters_by_job_key(self): + api = self._api() + self.session.add( + AutomationRun(job_key=ck_health.JOB_KEY, status="success", scanned=1) + ) + self.session.add(AutomationRun(job_key="other", status="failed")) + self.session.commit() + payload = api.list_runs( + job_key=ck_health.JOB_KEY, limit=50, db=self.session, current=self.admin + ) + assert len(payload["items"]) == 1 + assert payload["items"][0]["job_key"] == ck_health.JOB_KEY diff --git a/tests/test_migrations.py b/tests/test_migrations.py index cd993e1..aa76df6 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -22,7 +22,7 @@ TESTS_DIR = Path(__file__).resolve().parent PROJECT_ROOT = TESTS_DIR.parent VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions" -HEAD_REVISION = "20261001_0036" +HEAD_REVISION = "20261002_0037" # 迁移新增、但不声明在模型里的复合查询索引。 EXTRA_INDEXES = ( diff --git a/web/backend/main.py b/web/backend/main.py index 1add633..7dc8ce6 100644 --- a/web/backend/main.py +++ b/web/backend/main.py @@ -20,6 +20,7 @@ from .routers import ( accounts, audit, auth, + automation, cookies, dashboard, douyu, @@ -75,7 +76,20 @@ async def lifespan(app: FastAPI): logger.info(f"启动清理应用宝残留任务: {cleaned_yyb} 条") finally: db.close() - yield + + # 自动化调度器：AUTOMATION_ENABLED 未开启时不会起线程（默认关闭）。 + from .services.automation import ( + register_default_jobs, + start_scheduler, + stop_scheduler, + ) + + register_default_jobs() + start_scheduler() + try: + yield + finally: + stop_scheduler() app = FastAPI( @@ -135,6 +149,7 @@ app.include_router(huya.router) app.include_router(douyu.router) app.include_router(yyb.router) app.include_router(audit.router) +app.include_router(automation.router) @app.get("/api/health") diff --git a/web/backend/migrations/versions/20261002_0037_automation_jobs.py b/web/backend/migrations/versions/20261002_0037_automation_jobs.py new file mode 100644 index 0000000..5d35935 --- /dev/null +++ b/web/backend/migrations/versions/20261002_0037_automation_jobs.py @@ -0,0 +1,135 @@ +"""自动化作业：作业定义表、运行记录表，以及 CK 巡检所需的退避字段。 + +P1a 只落地「CK 自动巡检」需要的表结构；重登退避（relogin_*）与账号级开关 +（auto_managed）也一并建好，供 P1b 使用，避免之后再改一次表。 + +上一版登录任务把 attempt_* 与 CK 本体拆开，这里延续同一原则：巡检只写 +ck_check_*，绝不碰 cookie / status / attempt_*。 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "20261002_0037" +down_revision: str | None = "20261001_0036" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def _columns(bind, table: str) -> set[str]: + return {item["name"] for item in sa.inspect(bind).get_columns(table)} + + +def _tables(bind) -> set[str]: + return set(sa.inspect(bind).get_table_names()) + + +def upgrade() -> None: + bind = op.get_bind() + + existing_tables = _tables(bind) + if "automation_jobs" not in existing_tables: + op.create_table( + "automation_jobs", + sa.Column("key", sa.String(length=64), primary_key=True), + sa.Column( + "title", sa.String(length=128), nullable=False, server_default="" + ), + sa.Column( + "enabled", sa.Boolean(), nullable=False, server_default=sa.text("0") + ), + sa.Column( + "interval_seconds", + sa.Integer(), + nullable=False, + server_default="21600", + ), + sa.Column("batch_size", sa.Integer(), nullable=False, server_default="200"), + sa.Column("concurrency", sa.Integer(), nullable=False, server_default="5"), + sa.Column("config", sa.JSON(), nullable=True), + sa.Column("last_run_at", sa.DateTime(), nullable=True), + sa.Column("next_run_at", sa.DateTime(), nullable=True), + sa.Column( + "last_status", sa.String(length=16), nullable=False, server_default="" + ), + sa.Column( + "last_message", sa.String(length=512), nullable=False, server_default="" + ), + sa.Column("updated_at", sa.DateTime(), nullable=False), + ) + if "automation_runs" not in existing_tables: + op.create_table( + "automation_runs", + sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True), + sa.Column("job_key", sa.String(length=64), nullable=False), + sa.Column("started_at", sa.DateTime(), nullable=False), + sa.Column("finished_at", sa.DateTime(), nullable=True), + sa.Column( + "status", sa.String(length=16), nullable=False, server_default="running" + ), + sa.Column("scanned", sa.Integer(), nullable=False, server_default="0"), + sa.Column("valid", sa.Integer(), nullable=False, server_default="0"), + sa.Column("invalid", sa.Integer(), nullable=False, server_default="0"), + sa.Column( + "relogin_started", sa.Integer(), nullable=False, server_default="0" + ), + sa.Column( + "relogin_success", sa.Integer(), nullable=False, server_default="0" + ), + sa.Column("skipped", sa.Integer(), nullable=False, server_default="0"), + sa.Column( + "message", sa.String(length=512), nullable=False, server_default="" + ), + sa.Column("detail", sa.JSON(), nullable=True), + ) + op.create_index("ix_automation_runs_job_key", "automation_runs", ["job_key"]) + + login_columns = _columns(bind, "login_tasks") + if "relogin_attempts" not in login_columns: + op.add_column( + "login_tasks", + sa.Column( + "relogin_attempts", sa.Integer(), nullable=False, server_default="0" + ), + ) + if "relogin_next_at" not in login_columns: + op.add_column( + "login_tasks", sa.Column("relogin_next_at", sa.DateTime(), nullable=True) + ) + if "relogin_last_error" not in login_columns: + op.add_column( + "login_tasks", + sa.Column( + "relogin_last_error", + sa.String(length=512), + nullable=False, + server_default="", + ), + ) + + account_columns = _columns(bind, "accounts") + if "auto_managed" not in account_columns: + op.add_column( + "accounts", + sa.Column( + "auto_managed", + sa.Boolean(), + nullable=False, + server_default=sa.text("1"), + ), + ) + + +def downgrade() -> None: + bind = op.get_bind() + for table in ("automation_runs", "automation_jobs"): + if table in _tables(bind): + op.drop_table(table) + login_columns = _columns(bind, "login_tasks") + for name in ("relogin_last_error", "relogin_next_at", "relogin_attempts"): + if name in login_columns: + op.drop_column("login_tasks", name) + if "auto_managed" in _columns(bind, "accounts"): + op.drop_column("accounts", "auto_managed") diff --git a/web/backend/models.py b/web/backend/models.py index 566c84d..692776f 100644 --- a/web/backend/models.py +++ b/web/backend/models.py @@ -111,6 +111,10 @@ class Account(Base): esports_can_change_time: Mapped[int | None] = mapped_column(Integer, nullable=True) xpd_game_name: Mapped[str] = mapped_column(String(128), default="") xpd_openid: Mapped[str] = mapped_column(String(128), default="") + # 自动化后门：false 时巡检与自动重登都跳过这个账号（特殊账号人工维护）。 + auto_managed: Mapped[bool] = mapped_column( + Boolean, default=True, server_default="1" + ) xpd_role_id: Mapped[str] = mapped_column(String(64), default="") xpd_plat_id: Mapped[int | None] = mapped_column(Integer, nullable=True) xpd_area_id: Mapped[int | None] = mapped_column(Integer, nullable=True) @@ -171,6 +175,14 @@ class LoginTask(Base): String(64), default="", server_default="" ) attempt_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + # 自动重登（P1b）用的退避状态。P1a 的巡检不碰这几个字段。 + relogin_attempts: Mapped[int] = mapped_column( + Integer, default=0, server_default="0" + ) # 连续重登尝试次数，成功即清零 + relogin_next_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + relogin_last_error: Mapped[str] = mapped_column( + String(512), default="", server_default="" + ) account: Mapped[Account] = relationship("Account", back_populates="login_tasks") @@ -673,3 +685,51 @@ class AuditLog(Base): # 旧记录未标记结果时为 NULL，避免把历史操作误判成失败。 success: Mapped[bool | None] = mapped_column(Boolean, nullable=True, index=True) created_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow) + + +class AutomationJob(Base): + """自动化作业定义。状态落库，进程重启不丢调度进度。""" + + __tablename__ = "automation_jobs" + + key: Mapped[str] = mapped_column(String(64), primary_key=True) + title: Mapped[str] = mapped_column(String(128), default="") + # 默认关闭：按「渐进上线」原则，先观察再打开。 + enabled: Mapped[bool] = mapped_column(Boolean, default=False, server_default="0") + interval_seconds: Mapped[int] = mapped_column( + Integer, default=21600, server_default="21600" + ) + batch_size: Mapped[int] = mapped_column(Integer, default=200, server_default="200") + concurrency: Mapped[int] = mapped_column(Integer, default=5, server_default="5") + config: Mapped[dict | None] = mapped_column(JSON, nullable=True) + last_run_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + next_run_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + last_status: Mapped[str] = mapped_column(String(16), default="", server_default="") + last_message: Mapped[str] = mapped_column( + String(512), default="", server_default="" + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime, default=_utcnow, onupdate=_utcnow + ) + + +class AutomationRun(Base): + """自动化作业每一轮的执行记录，用于页面展示与排障。""" + + __tablename__ = "automation_runs" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + job_key: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + started_at: Mapped[datetime] = mapped_column(DateTime, default=_utcnow) + finished_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True) + status: Mapped[str] = mapped_column( + String(16), default="running", server_default="running" + ) # running / success / failed / skipped + scanned: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + valid: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + invalid: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + relogin_started: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + relogin_success: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + skipped: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + message: Mapped[str] = mapped_column(String(512), default="", server_default="") + detail: Mapped[dict | None] = mapped_column(JSON, nullable=True) diff --git a/web/backend/permissions.py b/web/backend/permissions.py index 6882c18..6f5f754 100644 --- a/web/backend/permissions.py +++ b/web/backend/permissions.py @@ -51,6 +51,9 @@ PERMISSIONS = { "proxy:manage": "代理配置管理", "whitelist:manage": "白名单配置管理", "whitelist:test": "测试白名单", + # 自动化 + "automation:view": "查看自动化作业与运行历史", + "automation:manage": "修改自动化配置、手动触发作业", # 系统 "system:settings": "系统设置", "audit:view": "查看审计日志", @@ -90,6 +93,7 @@ ROLE_PERMISSIONS = { "proxy:manage", "whitelist:manage", "whitelist:test", + "automation:view", ], "support": [ "account:view_assigned", diff --git a/web/backend/routers/automation.py b/web/backend/routers/automation.py new file mode 100644 index 0000000..5d434e8 --- /dev/null +++ b/web/backend/routers/automation.py @@ -0,0 +1,191 @@ +"""自动化作业接口。 + +设计上「先能看、再能开」：查看需要 ``automation:view``（运营也有），改配置与 +手动触发需要 ``automation:manage``（仅超管）。作业默认关闭，打开后由进程内 +调度器按间隔执行。 +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel, Field +from sqlalchemy.orm import Session + +from ..database import get_db +from ..deps import require_permission +from ..models import AutomationJob, AutomationRun, User +from ..services.audit_service import record_audit +from ..services.automation import ( + ensure_jobs, + registered_jobs, + run_job_now, + scheduler_status, +) + +router = APIRouter(prefix="/api/automation", tags=["自动化"]) + +MAX_BATCH_SIZE = 2000 +MIN_INTERVAL_SECONDS = 300 + + +class JobUpdate(BaseModel): + """作业配置更新；只传要改的字段。""" + + enabled: bool | None = None + interval_seconds: int | None = Field(default=None, ge=MIN_INTERVAL_SECONDS) + batch_size: int | None = Field(default=None, ge=1, le=MAX_BATCH_SIZE) + concurrency: int | None = Field(default=None, ge=1, le=16) + + +def _job_out(job: AutomationJob) -> dict[str, Any]: + return { + "key": job.key, + "title": job.title, + "enabled": bool(job.enabled), + "interval_seconds": int(job.interval_seconds or 0), + "batch_size": int(job.batch_size or 0), + "concurrency": int(job.concurrency or 0), + "last_run_at": job.last_run_at.isoformat() if job.last_run_at else None, + "next_run_at": job.next_run_at.isoformat() if job.next_run_at else None, + "last_status": job.last_status or "", + "last_message": job.last_message or "", + "registered": job.key in registered_jobs(), + } + + +def _run_out(run: AutomationRun) -> dict[str, Any]: + return { + "id": run.id, + "job_key": run.job_key, + "started_at": run.started_at.isoformat() if run.started_at else None, + "finished_at": run.finished_at.isoformat() if run.finished_at else None, + "status": run.status or "", + "scanned": int(run.scanned or 0), + "valid": int(run.valid or 0), + "invalid": int(run.invalid or 0), + "skipped": int(run.skipped or 0), + "message": run.message or "", + "detail": run.detail or None, + } + + +@router.get("/jobs") +def list_jobs( + db: Session = Depends(get_db), + current: User = Depends(require_permission("automation:view")), +): + """列出作业配置、最近一次结果与调度器状态。""" + ensure_jobs(db) + jobs = db.query(AutomationJob).order_by(AutomationJob.key.asc()).all() + latest: dict[str, AutomationRun] = {} + for run in ( + db.query(AutomationRun).order_by(AutomationRun.id.desc()).limit(200).all() + ): + latest.setdefault(run.job_key, run) + return { + "scheduler": scheduler_status(), + "jobs": [ + { + **_job_out(job), + "latest_run": _run_out(latest[job.key]) if job.key in latest else None, + } + for job in jobs + ], + } + + +@router.patch("/jobs/{key}") +def update_job( + key: str, + payload: JobUpdate, + db: Session = Depends(get_db), + current: User = Depends(require_permission("automation:manage")), +): + """改开关 / 间隔 / 批量 / 并发。打开开关时排定下一轮时间。""" + ensure_jobs(db) + job = db.query(AutomationJob).filter(AutomationJob.key == key).first() + if job is None: + raise HTTPException(status_code=404, detail="作业不存在") + + changed: dict[str, Any] = {} + if payload.enabled is not None and bool(payload.enabled) != bool(job.enabled): + job.enabled = bool(payload.enabled) + changed["enabled"] = job.enabled + # 刚打开时排一轮：不早于一个 tick，避免打开瞬间就压满 CPU。 + job.next_run_at = datetime.now(UTC) + _tick_delta() if job.enabled else None + if payload.interval_seconds is not None: + changed["interval_seconds"] = payload.interval_seconds + job.interval_seconds = payload.interval_seconds + if payload.batch_size is not None: + changed["batch_size"] = payload.batch_size + job.batch_size = payload.batch_size + if payload.concurrency is not None: + changed["concurrency"] = payload.concurrency + job.concurrency = payload.concurrency + + if changed: + job.updated_at = datetime.now(UTC) + record_audit( + db, + current, + action="automation:update", + target=key, + detail=changed, + ) + db.commit() + return _job_out(job) + + +@router.post("/jobs/{key}/run") +def run_job( + key: str, + db: Session = Depends(get_db), + current: User = Depends(require_permission("automation:manage")), +): + """立刻跑一轮（不看 enabled）。用于正式打开开关前先试跑。""" + ensure_jobs(db) + try: + outcome = run_job_now(db, key) + except KeyError: + raise HTTPException(status_code=404, detail="作业不存在") from None + record_audit( + db, + current, + action="automation:run", + target=key, + detail={ + "status": outcome.get("status"), + "scanned": outcome.get("scanned"), + "valid": outcome.get("valid"), + "invalid": outcome.get("invalid"), + "message": outcome.get("message"), + }, + ) + db.commit() + return outcome + + +@router.get("/runs") +def list_runs( + job_key: str | None = Query(default=None), + limit: int = Query(default=50, ge=1, le=200), + db: Session = Depends(get_db), + current: User = Depends(require_permission("automation:view")), +): + """运行历史，最近的在前。""" + query = db.query(AutomationRun) + if job_key: + query = query.filter(AutomationRun.job_key == job_key) + runs = query.order_by(AutomationRun.id.desc()).limit(limit).all() + return {"items": [_run_out(run) for run in runs]} + + +def _tick_delta(): + from datetime import timedelta + + from ..services.automation.scheduler import TICK_SECONDS + + return timedelta(seconds=TICK_SECONDS) diff --git a/web/backend/routers/login.py b/web/backend/routers/login.py index c19b4df..36cf3a4 100644 --- a/web/backend/routers/login.py +++ b/web/backend/routers/login.py @@ -13,6 +13,7 @@ from ..models import Account, LoginTask, User from ..models import ProxyConfig as ProxyConfigModel from ..permissions import get_user_permissions, user_has_permission from ..schemas import LoginBatchRequest, LoginTaskOut +from ..services.audit_service import record_audit from ..services.login_service import LoginBatchRunner, batch_registry router = APIRouter(prefix="/api/login", tags=["登录任务"]) @@ -111,6 +112,26 @@ async def create_batch( thread = threading.Thread(target=runner.run, daemon=True) thread.start() + # 审计：批次是谁在什么时候发起、多少个账号、什么模式。 + # 自动重登与人工登录共用同一条链路，没有这条记录就无法追溯来源。 + try: + record_audit( + db, + current, + action="login:batch", + target=batch_id, + detail={ + "mode": req.mode, + "requested": len(req.account_ids), + "accepted": len(valid_ids), + "concurrency": req.concurrency, + "api_strategy": req.api_strategy, + }, + ) + db.commit() + except Exception: # noqa: BLE001 审计失败不应影响已启动的批次 + db.rollback() + return {"batch_id": batch_id, "count": len(valid_ids), "success": True} diff --git a/web/backend/services/automation/__init__.py b/web/backend/services/automation/__init__.py new file mode 100644 index 0000000..342708e --- /dev/null +++ b/web/backend/services/automation/__init__.py @@ -0,0 +1,55 @@ +"""自动化调度：作业注册、调度线程与各作业实现。 + +登记作业请放在 ``register_default_jobs()``，由 ``main.py`` 在启动时调用一次， +避免 import 顺序影响注册结果。 +""" + +from __future__ import annotations + +from .jobs.ck_health import JOB_KEY as CK_HEALTH_KEY +from .jobs.ck_health import run_ck_health +from .scheduler import ( + JobSpec, + automation_enabled, + ensure_jobs, + register_job, + registered_jobs, + reset_registry, + run_due_jobs, + run_job_now, + scheduler_status, + start_scheduler, + stop_scheduler, +) + +__all__ = [ + "CK_HEALTH_KEY", + "JobSpec", + "automation_enabled", + "ensure_jobs", + "register_default_jobs", + "register_job", + "registered_jobs", + "reset_registry", + "run_ck_health", + "run_due_jobs", + "run_job_now", + "scheduler_status", + "start_scheduler", + "stop_scheduler", +] + + +def register_default_jobs() -> None: + """登记内置作业（幂等）。""" + register_job( + JobSpec( + key=CK_HEALTH_KEY, + title="CK 自动巡检", + # 6 小时一轮；全库 12,881 条按此反推 batch_size=200（见规划附录 C）。 + interval_seconds=6 * 3600, + batch_size=200, + concurrency=5, + run=run_ck_health, + ) + ) diff --git a/web/backend/services/automation/jobs/__init__.py b/web/backend/services/automation/jobs/__init__.py new file mode 100644 index 0000000..cbda5a4 --- /dev/null +++ b/web/backend/services/automation/jobs/__init__.py @@ -0,0 +1 @@ +"""自动化作业实现。""" diff --git a/web/backend/services/automation/jobs/ck_health.py b/web/backend/services/automation/jobs/ck_health.py new file mode 100644 index 0000000..93fe1d5 --- /dev/null +++ b/web/backend/services/automation/jobs/ck_health.py @@ -0,0 +1,125 @@ +"""CK 巡检作业（P1a）。 + +只做一件事：把「持有 CK 但久未检测」的账号挑出来，调现成的 +``check_douyu_cookie()`` 检测，把结果写回 ``ck_check_*``。 + +**本作业不修改 CK 本身**：不写 ``cookie`` / ``status`` / ``message``，也不碰 +``attempt_*``。P1a 上线后人工照旧可以点检测和重登，自动重登是 P1b 的事。 +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import UTC, datetime, timedelta +from typing import Any + +from sqlalchemy import or_ +from sqlalchemy.orm import Session + +from ....models import Account, AutomationJob, LoginTask +from ...cookie_check_service import check_douyu_cookie +from ...douyu_service import cookie_operable_filter + +JOB_KEY = "ck_health" +FLUSH_EVERY = 25 + + +def _select_tasks(db: Session, job: AutomationJob, now: datetime) -> list[LoginTask]: + """挑本轮要检测的记录。 + + 口径与人工点检测一致（``cookie_operable_filter``：持有 CK 且没有在途尝试）， + 再加三条：距上次检测超过作业间隔、账号允许自动管理、最久没查的优先。 + """ + cutoff = now - timedelta(seconds=max(int(job.interval_seconds or 0), 60)) + return ( + db.query(LoginTask) + .join(Account, LoginTask.account_id == Account.id) + .filter(cookie_operable_filter()) + .filter(Account.auto_managed.is_(True)) + .filter( + or_( + LoginTask.ck_checked_at.is_(None), + LoginTask.ck_checked_at < cutoff, + ) + ) + .order_by(LoginTask.ck_checked_at.asc().nullsfirst(), LoginTask.id.asc()) + .limit(max(int(job.batch_size or 0), 1)) + .all() + ) + + +def _detect(task: LoginTask) -> dict[str, Any]: + try: + return check_douyu_cookie(task.cookie or "") + except Exception as exc: # noqa: BLE001 单个账号失败不影响整轮 + return { + "valid": False, + "message": f"检测异常: {exc}", + "fish_ball": None, + "nickname": None, + "level": None, + "checked_at": datetime.now(UTC), + } + + +def run_ck_health(db: Session, job: AutomationJob) -> dict[str, Any]: + now = datetime.now(UTC) + tasks = _select_tasks(db, job, now) + if not tasks: + return {"scanned": 0, "valid": 0, "invalid": 0, "skipped": 0} + + results: dict[int, dict[str, Any]] = {} + concurrency = max(min(int(job.concurrency or 1), 16), 1) + with ThreadPoolExecutor(max_workers=concurrency) as pool: + futures = {pool.submit(_detect, task): task for task in tasks} + for future in as_completed(futures): + task = futures[future] + results[task.id] = future.result() + + # 只写检测结果三件套；cookie / status / attempt_* 一律不动。 + valid = invalid = 0 + for index, task in enumerate(tasks, start=1): + result = results.get(task.id) + if result is None: + continue + is_valid = bool(result.get("valid")) + if is_valid: + valid += 1 + else: + invalid += 1 + task.ck_check_status = "valid" if is_valid else "invalid" + task.ck_check_result = { + "fish_ball": result.get("fish_ball"), + "nickname": result.get("nickname"), + "level": result.get("level"), + "message": result.get("message", ""), + } + task.ck_checked_at = now + if index % FLUSH_EVERY == 0: + db.commit() + db.commit() + + # 剩余待检数量只用于提示，不参与调度。 + remaining = ( + db.query(LoginTask) + .join(Account, LoginTask.account_id == Account.id) + .filter(cookie_operable_filter()) + .filter(Account.auto_managed.is_(True)) + .filter( + or_( + LoginTask.ck_checked_at.is_(None), + LoginTask.ck_checked_at + < now - timedelta(seconds=max(int(job.interval_seconds or 0), 60)), + ) + ) + .count() + ) + message = f"本轮检测 {len(tasks)} 条，剩余待检 {remaining} 条" + return { + "scanned": len(tasks), + "valid": valid, + "invalid": invalid, + "skipped": 0, + "message": message, + "detail": {"remaining": remaining, "concurrency": concurrency}, + } diff --git a/web/backend/services/automation/scheduler.py b/web/backend/services/automation/scheduler.py new file mode 100644 index 0000000..7271eeb --- /dev/null +++ b/web/backend/services/automation/scheduler.py @@ -0,0 +1,267 @@ +"""进程内自动化调度器。 + +背景：后台的日常动作原本全靠人工点（判断 CK 失效、重登、收集资料、记台账）。 +自动化按「渐进上线」推进，因此： + +- 调度器总开关 ``AUTOMATION_ENABLED`` **默认关闭**； +- 每个作业自己还有开关，**默认关闭**，在 ``/automation`` 页面打开； +- 作业状态全部落库（``automation_jobs`` / ``automation_runs``），进程重启不丢 + 调度进度，也不会因为重启而重复跑一轮。 + +部署是单容器 ``--workers 1``，所以一个进程内线程就够，不需要分布式锁。 +""" + +from __future__ import annotations + +import os +import threading +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Any + +from loguru import logger +from sqlalchemy.orm import Session + +from ...database import SessionLocal +from ...models import AutomationJob, AutomationRun + +# 调度轮询间隔：作业间隔都以它为单位对齐（默认 6h 的作业不会差这一分钟）。 +TICK_SECONDS = 60 + +TRUTHY = {"1", "true", "yes", "on"} + + +@dataclass(frozen=True) +class JobSpec: + """一个可调度的作业。``run`` 接收会话与作业行，返回本轮计数。""" + + key: str + title: str + interval_seconds: int + batch_size: int + concurrency: int + run: Callable[[Session, AutomationJob], dict[str, Any]] + + +_REGISTRY: dict[str, JobSpec] = {} +_thread: threading.Thread | None = None +_stop_event = threading.Event() +_lock = threading.Lock() + + +def register_job(spec: JobSpec) -> None: + _REGISTRY[spec.key] = spec + + +def registered_jobs() -> dict[str, JobSpec]: + return dict(_REGISTRY) + + +def reset_registry() -> None: + """清空作业注册表（测试隔离用；正常启动只调用 register_default_jobs）。""" + _REGISTRY.clear() + + +def automation_enabled() -> bool: + return os.getenv("AUTOMATION_ENABLED", "").strip().lower() in TRUTHY + + +def ensure_jobs(db: Session) -> None: + """把注册表里的作业补进 ``automation_jobs``（幂等）。 + + 只补缺失的行，**不覆盖已有配置** —— 运维在页面上改过的开关、间隔、批量 + 不能因为重启被重置回默认值。 + """ + existing = {row.key for row in db.query(AutomationJob.key).all()} + for spec in _REGISTRY.values(): + if spec.key in existing: + continue + db.add( + AutomationJob( + key=spec.key, + title=spec.title, + enabled=False, + interval_seconds=spec.interval_seconds, + batch_size=spec.batch_size, + concurrency=spec.concurrency, + config=None, + next_run_at=None, + last_status="", + last_message="", + updated_at=datetime.now(UTC), + ) + ) + db.commit() + + +def _as_utc(value: datetime | None) -> datetime | None: + """DB 时间列存的是 naive UTC，读出来要补时区标记才能跟 now 比较。""" + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +def _due(job: AutomationJob, now: datetime) -> bool: + """到点判定：没排过时间视为到点（刚打开开关就会跑一轮）。""" + next_run_at = _as_utc(job.next_run_at) + return next_run_at is None or next_run_at <= now + + +def run_due_jobs(db: Session, *, now: datetime | None = None) -> list[dict[str, Any]]: + """跑一轮「到点的作业」，返回每轮结果。 + + 抽成普通函数是为了可测：测试直接调用即可，不必启动线程或等 60 秒。 + 作业串行执行 —— 目前只有 CK 巡检，单轮约 20~40 秒，远小于 tick，串行 + 反而省掉了并发调度的复杂度。 + """ + now = now or datetime.now(UTC) + outcomes: list[dict[str, Any]] = [] + jobs = ( + db.query(AutomationJob) + .filter(AutomationJob.enabled.is_(True)) + .order_by(AutomationJob.key.asc()) + .all() + ) + for job in jobs: + spec = _REGISTRY.get(job.key) + if spec is None or not _due(job, now): + continue + outcomes.append(_run_one(db, job, spec)) + + # 到期但没注册实现的作业（例如降级回滚后残留的行）只留日志，不静默消失。 + for job in jobs: + if job.key not in _REGISTRY: + logger.warning("自动化作业 {} 没有注册实现，已跳过", job.key) + return outcomes + + +def _run_one(db: Session, job: AutomationJob, spec: JobSpec) -> dict[str, Any]: + started = datetime.now(UTC) + run = AutomationRun(job_key=job.key, started_at=started, status="running") + db.add(run) + db.commit() + + counters: dict[str, Any] = {} + status = "success" + message = "" + try: + counters = spec.run(db, job) or {} + message = str(counters.pop("message", "") or "") + except Exception as exc: # noqa: BLE001 作业边界要兜住任何异常，不能拖垮调度线程 + db.rollback() + status = "failed" + message = f"{type(exc).__name__}: {exc}"[:500] + logger.exception("自动化作业 {} 执行失败", job.key) + + finished = datetime.now(UTC) + run.finished_at = finished + run.status = status + run.message = message + run.scanned = int(counters.get("scanned", 0) or 0) + run.valid = int(counters.get("valid", 0) or 0) + run.invalid = int(counters.get("invalid", 0) or 0) + run.relogin_started = int(counters.get("relogin_started", 0) or 0) + run.relogin_success = int(counters.get("relogin_success", 0) or 0) + run.skipped = int(counters.get("skipped", 0) or 0) + 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) + ) + job.last_status = status + job.last_message = message + job.updated_at = finished + db.commit() + + logger.info( + "自动化作业 {} 完成：status={} scanned={} valid={} invalid={} skipped={}", + job.key, + status, + run.scanned, + run.valid, + run.invalid, + run.skipped, + ) + return { + "job_key": job.key, + "status": status, + "message": message, + "scanned": run.scanned, + "valid": run.valid, + "invalid": run.invalid, + "skipped": run.skipped, + } + + +def run_job_now(db: Session, key: str) -> dict[str, Any]: + """立刻跑一个作业，**不看 enabled 开关**。 + + 页面上的「立即跑一次」用这个：运维需要能在正式打开开关之前先试一轮。 + """ + job = db.query(AutomationJob).filter(AutomationJob.key == key).first() + spec = _REGISTRY.get(key) + if job is None or spec is None: + raise KeyError(key) + return _run_one(db, job, spec) + + +def _loop() -> None: + db = SessionLocal() + try: + ensure_jobs(db) + except Exception: # noqa: BLE001 初始化失败不应让线程静默退出 + logger.exception("自动化作业初始化失败") + finally: + db.close() + + logger.info("自动化调度器已启动，tick={}s", TICK_SECONDS) + while not _stop_event.wait(TICK_SECONDS): + db = SessionLocal() + try: + run_due_jobs(db) + except Exception: # noqa: BLE001 单轮异常不能让调度线程死掉 + db.rollback() + logger.exception("自动化调度轮询异常") + finally: + db.close() + logger.info("自动化调度器已停止") + + +def start_scheduler() -> bool: + """启动调度线程。返回是否真的启动了（未开启总开关时返回 False）。""" + global _thread + if not automation_enabled(): + logger.info("AUTOMATION_ENABLED 未开启，自动化调度器不启动") + return False + with _lock: + if _thread is not None and _thread.is_alive(): + return False + _stop_event.clear() + _thread = threading.Thread( + target=_loop, name="automation-scheduler", daemon=True + ) + _thread.start() + return True + + +def stop_scheduler() -> None: + global _thread + with _lock: + if _thread is None: + return + _stop_event.set() + _thread.join(timeout=5) + _thread = None + + +def scheduler_status() -> dict[str, Any]: + return { + "enabled": automation_enabled(), + "running": _thread is not None and _thread.is_alive(), + "tick_seconds": TICK_SECONDS, + "registered": sorted(_REGISTRY), + } diff --git a/web/frontend/src/App.tsx b/web/frontend/src/App.tsx index 5a0166e..1206daa 100644 --- a/web/frontend/src/App.tsx +++ b/web/frontend/src/App.tsx @@ -15,6 +15,7 @@ const AccountCheckPage = lazy(() => import('./pages/AccountCheckPage')); const AssignmentsPage = lazy(() => import('./pages/AssignmentsPage')); const LoginTasksPage = lazy(() => import('./pages/LoginTasksPage')); const ProxyPage = lazy(() => import('./pages/ProxyPage')); +const AutomationPage = lazy(() => import('./pages/AutomationPage')); const UsersPage = lazy(() => import('./pages/UsersPage')); const CookiePage = lazy(() => import('./pages/CookiePage')); const CookieOperationsPage = lazy(() => import('./pages/CookieOperationsPage')); @@ -100,6 +101,7 @@ function AppContent() { )} /> )} /> )} /> + )} /> )} /> ) : } /> diff --git a/web/frontend/src/api/automation.ts b/web/frontend/src/api/automation.ts new file mode 100644 index 0000000..0ceb720 --- /dev/null +++ b/web/frontend/src/api/automation.ts @@ -0,0 +1,18 @@ +import api from './client'; +import type { + AutomationJobItem, + AutomationJobUpdate, + AutomationJobsResponse, + AutomationRunOutcome, + AutomationRunsResponse, +} from './types'; + +export const automationApi = { + jobs: () => api.get('/automation/jobs'), + updateJob: (key: string, payload: AutomationJobUpdate) => + api.patch(`/automation/jobs/${key}`, payload), + runJob: (key: string) => + api.post(`/automation/jobs/${key}/run`), + runs: (params: { job_key?: string; limit?: number }) => + api.get('/automation/runs', { params }), +}; diff --git a/web/frontend/src/api/modules.ts b/web/frontend/src/api/modules.ts index 51bf8b5..4fce821 100644 --- a/web/frontend/src/api/modules.ts +++ b/web/frontend/src/api/modules.ts @@ -4,6 +4,7 @@ export { accountApi } from './accounts'; export { appApi } from './app'; export { auditApi } from './audit'; export { authApi } from './auth'; +export { automationApi } from './automation'; export { cookieApi } from './cookies'; export { dashboardApi } from './dashboard'; export { douyuApi } from './douyu'; diff --git a/web/frontend/src/api/types.ts b/web/frontend/src/api/types.ts index f2ee7c7..b6a046c 100644 --- a/web/frontend/src/api/types.ts +++ b/web/frontend/src/api/types.ts @@ -827,3 +827,67 @@ export interface PermissionsListResult { permissions: Record; role_permissions: Record; } + +// ==================== 自动化作业 ==================== + +export interface AutomationRunItem { + id: number; + job_key: string; + started_at: string | null; + finished_at: string | null; + status: string; + scanned: number; + valid: number; + invalid: number; + skipped: number; + message: string; + detail: Record | null; +} + +export interface AutomationJobItem { + key: string; + title: string; + enabled: boolean; + interval_seconds: number; + batch_size: number; + concurrency: number; + last_run_at: string | null; + next_run_at: string | null; + last_status: string; + last_message: string; + registered: boolean; + latest_run: AutomationRunItem | null; +} + +export interface AutomationSchedulerStatus { + enabled: boolean; + running: boolean; + tick_seconds: number; + registered: string[]; +} + +export interface AutomationJobsResponse { + scheduler: AutomationSchedulerStatus; + jobs: AutomationJobItem[]; +} + +export interface AutomationRunsResponse { + items: AutomationRunItem[]; +} + +export interface AutomationJobUpdate { + enabled?: boolean; + interval_seconds?: number; + batch_size?: number; + concurrency?: number; +} + +export interface AutomationRunOutcome { + job_key: string; + status: string; + message: string; + scanned: number; + valid: number; + invalid: number; + skipped: number; +} diff --git a/web/frontend/src/layouts/MainLayout.tsx b/web/frontend/src/layouts/MainLayout.tsx index e410f2d..ff656c6 100644 --- a/web/frontend/src/layouts/MainLayout.tsx +++ b/web/frontend/src/layouts/MainLayout.tsx @@ -7,7 +7,7 @@ import { MenuFoldOutlined, MenuUnfoldOutlined, SwapOutlined, SunOutlined, MoonOutlined, DesktopOutlined, GiftOutlined, ApartmentOutlined, MobileOutlined, ShoppingCartOutlined, BookOutlined, TrophyOutlined, ShopOutlined, GlobalOutlined, - SafetyCertificateOutlined, FileSearchOutlined, + SafetyCertificateOutlined, FileSearchOutlined, RobotOutlined, } from '@ant-design/icons'; import { useNavigate, useLocation, Outlet } from 'react-router-dom'; import { getUser, clearAuth, type AuthUser } from '../store/auth'; @@ -75,6 +75,7 @@ export default function MainLayout({ onLogout }: { onLogout?: () => void }) { } if (can('cookie:operate')) { douyuItems.push({ key: '/cookie-operations', label: 'CK 检测与重登', icon: }); + douyuItems.push({ key: '/automation', label: '自动巡检', icon: }); } if (can('douyu:task')) { douyuItems.push({ key: '/douyu/elite', label: '精英宝典', icon: }); diff --git a/web/frontend/src/pages/AutomationPage.tsx b/web/frontend/src/pages/AutomationPage.tsx new file mode 100644 index 0000000..a41163b --- /dev/null +++ b/web/frontend/src/pages/AutomationPage.tsx @@ -0,0 +1,281 @@ +import { useCallback, useEffect, useState } from 'react'; +import { + Alert, + Button, + Card, + Descriptions, + Form, + InputNumber, + Popconfirm, + Space, + Switch, + Table, + Tag, + Typography, +} from 'antd'; +import type { ColumnsType } from 'antd/es/table'; +import { ReloadOutlined, ThunderboltOutlined } from '@ant-design/icons'; +import { message } from '../utils/antdMessage'; +import { getErrorMessage } from '../utils/error'; +import { automationApi, type AutomationJobItem, type AutomationRunItem } from '../api/modules'; +import { usePermissions } from '../hooks/usePermissions'; +import { formatTime } from '../utils/time'; + +const STATUS_TAG: Record = { + success: { color: 'green', text: '成功' }, + failed: { color: 'red', text: '失败' }, + running: { color: 'blue', text: '执行中' }, + skipped: { color: 'default', text: '已跳过' }, +}; + +function secondsToHours(seconds: number): number { + return Math.round((seconds / 3600) * 10) / 10; +} + +export default function AutomationPage() { + const [jobs, setJobs] = useState([]); + const [schedulerEnabled, setSchedulerEnabled] = useState(true); + const [schedulerRunning, setSchedulerRunning] = useState(false); + const [runs, setRuns] = useState([]); + const [loading, setLoading] = useState(false); + const [savingKey, setSavingKey] = useState(null); + const [runningKey, setRunningKey] = useState(null); + const [forms] = Form.useForm(); + const { can } = usePermissions(); + const canManage = can('automation:manage'); + + const load = useCallback(async () => { + setLoading(true); + try { + const data = await automationApi.jobs(); + setJobs(data.jobs); + setSchedulerEnabled(data.scheduler.enabled); + setSchedulerRunning(data.scheduler.running); + const history = await automationApi.runs({ limit: 50 }); + setRuns(history.items); + // 统一回填表单：initialValue 只在首次渲染生效，刷新后不会跟着变。 + forms.setFieldsValue( + Object.fromEntries( + data.jobs.map((job) => [ + job.key, + { + enabled: job.enabled, + interval_hours: secondsToHours(job.interval_seconds), + batch_size: job.batch_size, + concurrency: job.concurrency, + }, + ]), + ), + ); + } catch (e: unknown) { + message.error(getErrorMessage(e)); + } finally { + setLoading(false); + } + }, [forms]); + + useEffect(() => { + load(); + }, [load]); + + const saveJob = async (job: AutomationJobItem) => { + const values = forms.getFieldValue(job.key) || {}; + setSavingKey(job.key); + try { + await automationApi.updateJob(job.key, { + enabled: values.enabled, + interval_seconds: Math.max(Math.round((values.interval_hours ?? 6) * 3600), 300), + batch_size: values.batch_size, + concurrency: values.concurrency, + }); + message.success(`${job.title} 配置已保存`); + await load(); + } catch (e: unknown) { + message.error(getErrorMessage(e)); + } finally { + setSavingKey(null); + } + }; + + const runOnce = async (job: AutomationJobItem) => { + setRunningKey(job.key); + try { + const outcome = await automationApi.runJob(job.key); + message.success( + `跑完一轮：扫描 ${outcome.scanned} 条，有效 ${outcome.valid}，失效 ${outcome.invalid}`, + ); + await load(); + } catch (e: unknown) { + message.error(getErrorMessage(e)); + } finally { + setRunningKey(null); + } + }; + + const runColumns: ColumnsType = [ + { + title: '开始时间', + dataIndex: 'started_at', + width: 180, + render: (value: string | null) => formatTime(value), + }, + { + title: '作业', + dataIndex: 'job_key', + width: 120, + }, + { + title: '结果', + dataIndex: 'status', + width: 90, + render: (value: string) => { + const tag = STATUS_TAG[value] ?? { color: 'default', text: value || '-' }; + return {tag.text}; + }, + }, + { title: '扫描', dataIndex: 'scanned', width: 80 }, + { title: '有效', dataIndex: 'valid', width: 80 }, + { + title: '失效', + dataIndex: 'invalid', + width: 80, + render: (value: number) => (value > 0 ? {value} : 0), + }, + { title: '耗时', width: 90, render: (_, row) => formatDuration(row.started_at, row.finished_at) }, + { title: '说明', dataIndex: 'message', ellipsis: true }, + ]; + + return ( + + } onClick={load} loading={loading}> + 刷新 + + } + > + {!schedulerEnabled && ( + + )} + {schedulerEnabled && !schedulerRunning && ( + + )} + + + {jobs.map((job) => ( + + + + {formatTime(job.last_run_at)} + + + {formatTime(job.next_run_at)} + + + {job.last_status ? ( + + {STATUS_TAG[job.last_status]?.text ?? job.last_status} + + ) : ( + '-' + )} + + {job.last_message || '-'} + + +
saveJob(job)}> + + + + + + + + + + + + + + + + runOnce(job)} + disabled={!canManage} + > + + + + +
+ {!canManage && ( + + 当前账号只能查看，修改配置需要 automation:manage 权限。 + + )} + + ))} + + + + + + + ); +} + +function formatDuration(start: string | null, end: string | null): string { + if (!start || !end) return '-'; + const ms = new Date(end).getTime() - new Date(start).getTime(); + if (!Number.isFinite(ms) || ms < 0) return '-'; + if (ms < 1000) return `${ms}ms`; + return `${(ms / 1000).toFixed(1)}s`; +}