feat(automation): CK 自动巡检（P1a），只检测不改 CK
后台原本靠人工逐批点「CK 检测」发现失效账号，1.2 万个账号根本盯不过来。
P1a 把「哪些账号失效了」自动化：自动挑久未检测的账号、调现成检测接口、
把结果写回 ck_check_*。**本作业不替换任何 CK**，重登仍由人工点，符合规划里
「先做只读/只提示，确认无误再做自动动作」的渐进上线原则。
实现：
- 迁移 20261002_0037：automation_jobs / automation_runs 两张表，
login_tasks 加 relogin_attempts / relogin_next_at / relogin_last_error
（为 P1b 备好），accounts 加 auto_managed（特殊账号不参与自动化）；
- services/automation/scheduler.py：进程内调度线程，60s tick，状态全部落库
（重启不丢调度进度、不重复跑轮）；AUTOMATION_ENABLED 默认关闭，作业开关
也默认关闭；ensure_jobs 只补缺失行、不覆盖运维改过的配置；
作业异常只记 failed，不拖垮调度线程；单容器 --workers 1，无需分布式锁；
- services/automation/jobs/ck_health.py：选行口径与人工检测一致
（cookie_holding_filter：持有 CK 且无在途尝试），叠加久未检测、
auto_managed、batch_size，按 ck_checked_at 升序（最久没查的优先）；
只写 ck_check_status / ck_check_result / ck_checked_at；
- routers/automation.py：GET /jobs、PATCH /jobs/{key}、POST /jobs/{key}/run、
GET /runs；新增权限点 automation:view（运营可看）与 automation:manage
（仅超管），改配置与手动触发都写审计；
- 顺带补 POST /api/login/batch 的审计（此前批次是谁发起、多少账号无据可查）；
- 前端 /automation 页：作业卡片（开关/周期/单轮条数/并发）、运行历史、
「立即跑一次」；总开关未打开、调度线程未运行时给明确提示。
验证：
- 22 条新测试覆盖调度器行为（只跑到点的、异常不抛出、下轮时间、配置不被重置）、
选行条件（在途尝试/auto_managed/间隔/batch_size）、接口与权限，其中一条
专门断言巡检后 cookie/status/message/attempt_status 一字未动；
- 全量 257 条通过；前端 tsc 与构建通过；
- 端到端用 3 个真实账号跑真实斗鱼接口：巡检结果与直接调 check_douyu_cookie
逐条一致（3/3 valid），CK 与尝试状态全程未变。
顺带修掉一处会炸生产的坑：DB 时间列读出来是 naive UTC，直接与
datetime.now(UTC) 比较会 TypeError，现统一走 _as_utc() 补时区。
This commit is contained in:
yml2213
+34
-16
@@ -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` 两页 → ⑥ 端到端联调（发链接 → 提交 → 校验 → 执行 → 台账） |
|
||||
|
||||
@@ -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
|
||||
@@ -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 = (
|
||||
|
||||
+16
-1
@@ -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")
|
||||
|
||||
@@ -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")
|
||||
@@ -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)
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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)
|
||||
@@ -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}
|
||||
|
||||
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
)
|
||||
@@ -0,0 +1 @@
|
||||
"""自动化作业实现。"""
|
||||
@@ -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},
|
||||
}
|
||||
@@ -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),
|
||||
}
|
||||
@@ -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() {
|
||||
<Route path="huya/browser-exchange" element={lazyRoute(<HuyaBrowserExchangePage />)} />
|
||||
<Route path="yyb/recharge" element={lazyRoute(<YybRechargePage />)} />
|
||||
<Route path="proxy" element={lazyRoute(<ProxyPage />)} />
|
||||
<Route path="automation" element={lazyRoute(<AutomationPage />)} />
|
||||
<Route path="users" element={lazyRoute(<UsersPage />)} />
|
||||
<Route path="audit-logs" element={isSuperAdmin ? lazyRoute(<AuditLogsPage />) : <Navigate to="/" replace />} />
|
||||
</Route>
|
||||
|
||||
@@ -0,0 +1,18 @@
|
||||
import api from './client';
|
||||
import type {
|
||||
AutomationJobItem,
|
||||
AutomationJobUpdate,
|
||||
AutomationJobsResponse,
|
||||
AutomationRunOutcome,
|
||||
AutomationRunsResponse,
|
||||
} from './types';
|
||||
|
||||
export const automationApi = {
|
||||
jobs: () => api.get<AutomationJobsResponse, AutomationJobsResponse>('/automation/jobs'),
|
||||
updateJob: (key: string, payload: AutomationJobUpdate) =>
|
||||
api.patch<AutomationJobItem, AutomationJobItem>(`/automation/jobs/${key}`, payload),
|
||||
runJob: (key: string) =>
|
||||
api.post<AutomationRunOutcome, AutomationRunOutcome>(`/automation/jobs/${key}/run`),
|
||||
runs: (params: { job_key?: string; limit?: number }) =>
|
||||
api.get<AutomationRunsResponse, AutomationRunsResponse>('/automation/runs', { params }),
|
||||
};
|
||||
@@ -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';
|
||||
|
||||
@@ -827,3 +827,67 @@ export interface PermissionsListResult {
|
||||
permissions: Record<string, string>;
|
||||
role_permissions: Record<string, string[]>;
|
||||
}
|
||||
|
||||
// ==================== 自动化作业 ====================
|
||||
|
||||
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<string, unknown> | 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;
|
||||
}
|
||||
|
||||
@@ -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: <SafetyCertificateOutlined /> });
|
||||
douyuItems.push({ key: '/automation', label: '自动巡检', icon: <RobotOutlined /> });
|
||||
}
|
||||
if (can('douyu:task')) {
|
||||
douyuItems.push({ key: '/douyu/elite', label: '精英宝典', icon: <BookOutlined /> });
|
||||
|
||||
@@ -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<string, { color: string; text: string }> = {
|
||||
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<AutomationJobItem[]>([]);
|
||||
const [schedulerEnabled, setSchedulerEnabled] = useState(true);
|
||||
const [schedulerRunning, setSchedulerRunning] = useState(false);
|
||||
const [runs, setRuns] = useState<AutomationRunItem[]>([]);
|
||||
const [loading, setLoading] = useState(false);
|
||||
const [savingKey, setSavingKey] = useState<string | null>(null);
|
||||
const [runningKey, setRunningKey] = useState<string | null>(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<AutomationRunItem> = [
|
||||
{
|
||||
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 color={tag.color}>{tag.text}</Tag>;
|
||||
},
|
||||
},
|
||||
{ title: '扫描', dataIndex: 'scanned', width: 80 },
|
||||
{ title: '有效', dataIndex: 'valid', width: 80 },
|
||||
{
|
||||
title: '失效',
|
||||
dataIndex: 'invalid',
|
||||
width: 80,
|
||||
render: (value: number) => (value > 0 ? <Typography.Text type="warning">{value}</Typography.Text> : 0),
|
||||
},
|
||||
{ title: '耗时', width: 90, render: (_, row) => formatDuration(row.started_at, row.finished_at) },
|
||||
{ title: '说明', dataIndex: 'message', ellipsis: true },
|
||||
];
|
||||
|
||||
return (
|
||||
<Space direction="vertical" size="middle" style={{ width: '100%' }}>
|
||||
<Card
|
||||
title="自动巡检"
|
||||
extra={
|
||||
<Button icon={<ReloadOutlined />} onClick={load} loading={loading}>
|
||||
刷新
|
||||
</Button>
|
||||
}
|
||||
>
|
||||
{!schedulerEnabled && (
|
||||
<Alert
|
||||
type="warning"
|
||||
showIcon
|
||||
style={{ marginBottom: 16 }}
|
||||
message="调度器总开关未打开"
|
||||
description="服务端环境变量 AUTOMATION_ENABLED 未开启，打开下面的作业开关不会真正执行。需要先打开总开关并重启服务，再回到本页打开作业。"
|
||||
/>
|
||||
)}
|
||||
{schedulerEnabled && !schedulerRunning && (
|
||||
<Alert
|
||||
type="warning"
|
||||
showIcon
|
||||
style={{ marginBottom: 16 }}
|
||||
message="调度线程未运行"
|
||||
description="总开关已开但调度线程不在运行状态，请检查服务日志或重启服务。"
|
||||
/>
|
||||
)}
|
||||
<Alert
|
||||
type="info"
|
||||
showIcon
|
||||
style={{ marginBottom: 16 }}
|
||||
message="巡检只检测、不改 CK"
|
||||
description="本作业只把失效账号找出来并标记，不会替换任何 Cookie、也不会改账号状态；重登仍由人工在「CK 检测与重登」页面操作。"
|
||||
/>
|
||||
|
||||
{jobs.map((job) => (
|
||||
<Card key={job.key} type="inner" title={job.title} style={{ marginBottom: 16 }}>
|
||||
<Descriptions size="small" column={2} style={{ marginBottom: 16 }}>
|
||||
<Descriptions.Item label="最近运行">
|
||||
{formatTime(job.last_run_at)}
|
||||
</Descriptions.Item>
|
||||
<Descriptions.Item label="下次运行">
|
||||
{formatTime(job.next_run_at)}
|
||||
</Descriptions.Item>
|
||||
<Descriptions.Item label="最近结果">
|
||||
{job.last_status ? (
|
||||
<Tag color={STATUS_TAG[job.last_status]?.color ?? 'default'}>
|
||||
{STATUS_TAG[job.last_status]?.text ?? job.last_status}
|
||||
</Tag>
|
||||
) : (
|
||||
'-'
|
||||
)}
|
||||
</Descriptions.Item>
|
||||
<Descriptions.Item label="最近说明">{job.last_message || '-'}</Descriptions.Item>
|
||||
</Descriptions>
|
||||
|
||||
<Form form={forms} layout="inline" onFinish={() => saveJob(job)}>
|
||||
<Form.Item
|
||||
name={[job.key, 'enabled']}
|
||||
label="启用"
|
||||
valuePropName="checked"
|
||||
>
|
||||
<Switch disabled={!canManage} />
|
||||
</Form.Item>
|
||||
<Form.Item
|
||||
name={[job.key, 'interval_hours']}
|
||||
label="巡检周期(小时)"
|
||||
>
|
||||
<InputNumber min={0.1} step={1} disabled={!canManage} />
|
||||
</Form.Item>
|
||||
<Form.Item name={[job.key, 'batch_size']} label="单轮条数">
|
||||
<InputNumber min={1} max={2000} disabled={!canManage} />
|
||||
</Form.Item>
|
||||
<Form.Item name={[job.key, 'concurrency']} label="并发">
|
||||
<InputNumber min={1} max={16} disabled={!canManage} />
|
||||
</Form.Item>
|
||||
<Form.Item>
|
||||
<Space>
|
||||
<Button
|
||||
type="primary"
|
||||
htmlType="submit"
|
||||
disabled={!canManage}
|
||||
loading={savingKey === job.key}
|
||||
>
|
||||
保存
|
||||
</Button>
|
||||
<Popconfirm
|
||||
title="立刻跑一轮？"
|
||||
description="不看启用开关，立即执行一次巡检。"
|
||||
onConfirm={() => runOnce(job)}
|
||||
disabled={!canManage}
|
||||
>
|
||||
<Button
|
||||
icon={<ThunderboltOutlined />}
|
||||
disabled={!canManage}
|
||||
loading={runningKey === job.key}
|
||||
>
|
||||
立即跑一次
|
||||
</Button>
|
||||
</Popconfirm>
|
||||
</Space>
|
||||
</Form.Item>
|
||||
</Form>
|
||||
{!canManage && (
|
||||
<Typography.Text type="secondary">
|
||||
当前账号只能查看，修改配置需要 automation:manage 权限。
|
||||
</Typography.Text>
|
||||
)}
|
||||
</Card>
|
||||
))}
|
||||
</Card>
|
||||
|
||||
<Card title="运行历史">
|
||||
<Table
|
||||
rowKey="id"
|
||||
size="small"
|
||||
loading={loading}
|
||||
columns={runColumns}
|
||||
dataSource={runs}
|
||||
pagination={{ pageSize: 20, showSizeChanger: false }}
|
||||
/>
|
||||
</Card>
|
||||
</Space>
|
||||
);
|
||||
}
|
||||
|
||||
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`;
|
||||
}
|
||||
Reference in New Issue
Block a user