yml2213
19f1cbceba
1. 修复 DouyuBatchRunner 在 run 结束时私自关闭外部 Session 导致作业状态无法回写并死锁在 running 的问题（增加 close_db 参数）； 2. 修复资料补全作业返回值映射，将 succeeded/failed 对齐到 valid/invalid，并在全库暂无待查账号时给出清晰说明； 3. 增强调度器 _run_one 边界防御：支持字段名兜底与 Session 崩溃时独立连接兜底回写，防止作业无限重复发起； 4. 修复巡检作业在全库无到期账号时的说明信息，如实展示最久检测时间与重检间隔； 5. 修复 AutomationPage 布局：移除外层 Form 的 layout=inline，Card 恢复 100% 满宽纵向排列，内部字段横向弹性自适应； 6. 修复未启用作业因残留历史 next_run_at 而错误显示「排队中」的问题。
480 lines
18 KiB
Python
480 lines
18 KiB
Python
"""进程内自动化调度器。
|
||
|
||
背景：后台的日常动作原本全靠人工点（判断 CK 失效、重登、收集资料、记台账）。
|
||
自动化按「渐进上线」推进，因此：
|
||
|
||
- 调度器总开关 ``AUTOMATION_ENABLED`` **默认关闭**；
|
||
- 每个作业自己还有开关，**默认关闭**，在 ``/automation`` 页面打开；
|
||
- 作业状态全部落库（``automation_jobs`` / ``automation_runs``），进程重启不丢
|
||
调度进度，也不会因为重启而重复跑一轮。
|
||
|
||
部署是单容器 ``--workers 1``，所以一个进程内线程就够，不需要分布式锁。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import threading
|
||
from collections.abc import Callable
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
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
|
||
|
||
# 调度轮询间隔，也是作业的运行节奏：每个 tick 跑一轮。
|
||
#
|
||
# 注意区分两个「间隔」：
|
||
# - **本常量**：作业多久跑一轮（5 分钟）；
|
||
# - **job.interval_seconds**：同一行数据多久重新检测一次（默认 6h），只用于选行 cutoff。
|
||
#
|
||
# 分开的原因：全库 12,881 条的 CK 若想「6 小时内都被查一遍」，需要约 36 条/分钟的
|
||
# 吞吐；按 5 分钟一轮、单轮 200 条算 = 40 条/分钟，刚好够。若把作业周期也设成 6h，
|
||
# 就变成 200 条/6h = 800 条/天，全库要 16 天才能过一轮，等于没巡检。
|
||
# 心跳间隔：只用来「看一眼有没有到点的作业」，几乎不耗资源，所以设得比作业
|
||
# 周期密得多。作业周期由各自的 next_run_at（= 自己结束时刻 + JOB_INTERVAL）决定。
|
||
#
|
||
# 这里曾经把心跳周期直接当作业周期用（心跳 300s + next_run_at = 结束 + 300s），
|
||
# 两者同频会导致系统性跳轮：作业 19 秒跑完 → next = +5 分钟，而下一次心跳正好
|
||
# 在 5 分钟整，常常差几十秒没到点，于是整轮空转，实际变成每 10 分钟才跑一次。
|
||
# 表现就是「检测明明只要 19 秒，却要排队等很久」。
|
||
TICK_SECONDS = 30
|
||
|
||
# 作业自身的运行周期：跑完一轮后隔多久再跑下一轮。
|
||
JOB_INTERVAL_SECONDS = 300
|
||
|
||
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]]
|
||
#: **多久跑一次**（调度频率）。默认 5 分钟，与改动前的实际行为一致。
|
||
#: ⚠️ 必须放在**有默认值**的位置（dataclass 不允许默认值字段排在无默认值之前）。
|
||
run_interval_seconds: int = 300
|
||
|
||
|
||
_REGISTRY: dict[str, JobSpec] = {}
|
||
|
||
#: 同时最多几个作业在跑。作业之间互不依赖，并行是为了**慢作业不挡住快作业**
|
||
#: （以前串行：一个作业跑 4 分钟，CK 巡检就得等 4 分钟）。
|
||
_JOB_MAX_WORKERS = 4
|
||
|
||
#: 运行间隔的下限：太短会变成忙轮询，没有意义
|
||
MIN_RUN_INTERVAL_SECONDS = 30
|
||
_EXECUTOR: ThreadPoolExecutor | None = None
|
||
_EXECUTOR_LOCK = threading.Lock()
|
||
|
||
#: **在途守卫**：正在跑的作业 key。
|
||
#: 必须要有 —— `next_run_at` 是在作业**跑完之后**才更新的，所以一个跑得比 tick
|
||
#: 还久的作业，下一个 tick 看它仍然是「到期」状态，会被重复派发，
|
||
#: 等于把速率翻倍（对斗鱼接口来说就是"并发放开"，正是风控最敏感的地方）。
|
||
_IN_FLIGHT: set[str] = set()
|
||
_IN_FLIGHT_LOCK = threading.Lock()
|
||
_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,
|
||
run_interval_seconds=spec.run_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 cleanup_orphan_runs(db: Session, message: str = "服务重启中断") -> int:
|
||
"""把上次进程留下的 ``running`` 运行记录收尾。
|
||
|
||
作业是在调度线程里同步跑的：进程一重启（部署、崩溃、手工重启），那轮作业就
|
||
再没机会写回结果，``automation_runs`` 会永远停在 ``running``，页面上一直显示
|
||
「执行中」。登录任务有自己的孤儿清理，运行记录原先漏了。
|
||
|
||
收尾为 ``failed`` 而不是 ``success``：那轮确实没跑完，如实记录；被中断的账号
|
||
由各自的孤儿清理负责标记，并会按退避重试。
|
||
"""
|
||
orphans = db.query(AutomationRun).filter(AutomationRun.status == "running").all()
|
||
for run in orphans:
|
||
# 用 interrupted 而不是 failed：这不是作业本身的失败（账号侧 CK 没丢、
|
||
# 被中断的也会按退避重试），只是进程没了导致那轮没跑完。混进 failed
|
||
# 会让页面一片红，掩盖真正的失败。
|
||
run.status = "interrupted"
|
||
run.finished_at = datetime.now(UTC)
|
||
run.message = message
|
||
if orphans:
|
||
db.commit()
|
||
return len(orphans)
|
||
|
||
|
||
def _get_executor() -> ThreadPoolExecutor:
|
||
global _EXECUTOR
|
||
with _EXECUTOR_LOCK:
|
||
if _EXECUTOR is None:
|
||
_EXECUTOR = ThreadPoolExecutor(
|
||
max_workers=_JOB_MAX_WORKERS, thread_name_prefix="automation-job"
|
||
)
|
||
return _EXECUTOR
|
||
|
||
|
||
def _run_job_isolated(job_key: str) -> dict[str, Any] | None:
|
||
"""在**自己的 session** 里跑一个作业。
|
||
|
||
必须独立 session：SQLAlchemy 的 Session **不是线程安全的**，
|
||
多个作业共用一个 session 会互相踩事务（而他们现在是真的并行了）。
|
||
重新按 key 取行，是因为调度线程里的那行可能已经过期。
|
||
"""
|
||
db = SessionLocal()
|
||
try:
|
||
job = db.query(AutomationJob).filter(AutomationJob.key == job_key).first()
|
||
spec = _REGISTRY.get(job_key)
|
||
if job is None or spec is None:
|
||
return None
|
||
return _run_one(db, job, spec)
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
def run_due_jobs(
|
||
db: Session, *, now: datetime | None = None, wait: bool = True
|
||
) -> list[dict[str, Any]]:
|
||
"""跑一轮到期的作业。
|
||
|
||
**作业之间并发**（以前是串行）：慢作业不该让快作业干等。
|
||
两个正交的保护：
|
||
|
||
- **在途守卫**：同一个作业不会重复派发（`next_run_at` 要等跑完才更新，
|
||
否则跑得比 tick 久的作业每个 tick 都会被再派发一次 = 速率翻倍）
|
||
- `wait=False`（调度循环用）：**派发完就返回**，tick 不被慢作业拖住；
|
||
`wait=True`（默认，测试与手动触发用）：跑完并返回 outcomes
|
||
|
||
⚠️ 每**行**多久重检由作业自己按 `interval_seconds` 决定（见 `_run_one` 的注释），
|
||
这里的 `next_run_at` 固定是「跑完 + 一个 tick」。
|
||
"""
|
||
now = now or datetime.now(UTC)
|
||
jobs = (
|
||
db.query(AutomationJob)
|
||
.filter(AutomationJob.enabled.is_(True))
|
||
.order_by(AutomationJob.key.asc())
|
||
.all()
|
||
)
|
||
|
||
dispatched: list[str] = []
|
||
for job in jobs:
|
||
spec = _REGISTRY.get(job.key)
|
||
if spec is None or not _due(job, now):
|
||
continue
|
||
with _IN_FLIGHT_LOCK:
|
||
if job.key in _IN_FLIGHT:
|
||
# 上一轮还没跑完 —— 跳过这一轮，别叠加（那等于把速率翻倍）
|
||
logger.info("自动化作业 {} 仍在执行，本轮跳过", job.key)
|
||
continue
|
||
_IN_FLIGHT.add(job.key)
|
||
dispatched.append(job.key)
|
||
|
||
if wait:
|
||
# 同步路径（测试 / 手动触发）：**顺序跑、用调用方的 session**。
|
||
# 这样调用方跑完就能直接读到自己 session 里的最新状态。
|
||
# 生产路径（下面是 wait=False）才走线程池 + 独立 session。
|
||
outcomes_sync: list[dict[str, Any]] = []
|
||
for key in dispatched:
|
||
try:
|
||
job = db.query(AutomationJob).filter(AutomationJob.key == key).first()
|
||
spec = _REGISTRY.get(key)
|
||
if job is not None and spec is not None:
|
||
outcomes_sync.append(_run_one(db, job, spec))
|
||
except Exception: # noqa: BLE001 作业边界要兜住，不能拖垮调度
|
||
logger.exception("自动化作业 {} 执行异常", key)
|
||
finally:
|
||
_release(key)
|
||
|
||
# 到期但没注册实现的作业只留日志，不静默消失
|
||
for job in jobs:
|
||
if job.key not in _REGISTRY:
|
||
logger.warning("自动化作业 {} 没有注册实现，已跳过", job.key)
|
||
return outcomes_sync
|
||
|
||
futures = {}
|
||
executor = _get_executor()
|
||
for key in dispatched:
|
||
futures[executor.submit(_run_job_isolated, key)] = key
|
||
|
||
# 到期但没注册实现的作业（例如降级回滚后残留的行）只留日志，不静默消失。
|
||
for job in jobs:
|
||
if job.key not in _REGISTRY:
|
||
logger.warning("自动化作业 {} 没有注册实现，已跳过", job.key)
|
||
|
||
# 派发完就返回；在途守卫会在各自跑完后清掉标记。
|
||
for future, key in futures.items():
|
||
future.add_done_callback(lambda _f, k=key: _release(k))
|
||
return []
|
||
|
||
|
||
def _release(job_key: str) -> None:
|
||
"""清掉在途标记（无论成功失败都要清，否则这个作业以后再也不跑了）。"""
|
||
with _IN_FLIGHT_LOCK:
|
||
_IN_FLIGHT.discard(job_key)
|
||
|
||
|
||
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)
|
||
scanned = int(counters.get("scanned", 0) or 0)
|
||
valid = int(counters.get("valid", counters.get("succeeded", 0)) or 0)
|
||
invalid = int(counters.get("invalid", counters.get("failed", 0)) or 0)
|
||
relogin_started = int(counters.get("relogin_started", 0) or 0)
|
||
relogin_success = int(counters.get("relogin_success", 0) or 0)
|
||
skipped = int(counters.get("skipped", 0) or 0)
|
||
detail = counters.get("detail") or None
|
||
|
||
try:
|
||
run.finished_at = finished
|
||
run.status = status
|
||
run.message = message
|
||
run.scanned = scanned
|
||
run.valid = valid
|
||
run.invalid = invalid
|
||
run.relogin_started = relogin_started
|
||
run.relogin_success = relogin_success
|
||
run.skipped = skipped
|
||
run.detail = detail
|
||
|
||
job.last_run_at = finished
|
||
# 下一轮 = 跑完 + **运行间隔**（可配置）。
|
||
# ⚠️ 别再用 `interval_seconds` —— 那是「同一行多久重检」，
|
||
# 由作业自己的选行逻辑使用，**不参与调度**（两个概念以前挤在一个字段里）。
|
||
job.next_run_at = finished + timedelta(
|
||
seconds=max(int(job.run_interval_seconds or 0), MIN_RUN_INTERVAL_SECONDS)
|
||
)
|
||
job.last_status = status
|
||
job.last_message = message
|
||
job.updated_at = finished
|
||
db.commit()
|
||
except Exception: # noqa: BLE001
|
||
# 如果 commit 失败（例如外部把 session 关了或连接抖动），用独立 session 挽救
|
||
db.rollback()
|
||
logger.exception("自动化作业 {} 提交状态失败，尝试独立连接回写", job.key)
|
||
try:
|
||
rescue_db = SessionLocal()
|
||
try:
|
||
r = (
|
||
rescue_db.query(AutomationRun)
|
||
.filter(AutomationRun.id == run.id)
|
||
.first()
|
||
)
|
||
if r:
|
||
r.finished_at = finished
|
||
r.status = status
|
||
r.message = message
|
||
r.scanned = scanned
|
||
r.valid = valid
|
||
r.invalid = invalid
|
||
j = (
|
||
rescue_db.query(AutomationJob)
|
||
.filter(AutomationJob.key == job.key)
|
||
.first()
|
||
)
|
||
if j:
|
||
j.last_run_at = finished
|
||
j.next_run_at = finished + timedelta(
|
||
seconds=max(
|
||
int(j.run_interval_seconds or 0), MIN_RUN_INTERVAL_SECONDS
|
||
)
|
||
)
|
||
j.last_status = status
|
||
j.last_message = message
|
||
j.updated_at = finished
|
||
rescue_db.commit()
|
||
finally:
|
||
rescue_db.close()
|
||
except Exception: # noqa: BLE001
|
||
logger.exception("自动化作业 {} 独立连接回写仍失败", job.key)
|
||
|
||
logger.info(
|
||
"自动化作业 {} 完成：status={} scanned={} valid={} invalid={} skipped={}",
|
||
job.key,
|
||
status,
|
||
scanned,
|
||
valid,
|
||
invalid,
|
||
skipped,
|
||
)
|
||
return {
|
||
"job_key": job.key,
|
||
"status": status,
|
||
"message": message,
|
||
"scanned": scanned,
|
||
"valid": valid,
|
||
"invalid": invalid,
|
||
"skipped": 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)
|
||
with _IN_FLIGHT_LOCK:
|
||
if key in _IN_FLIGHT:
|
||
raise RuntimeError(f"作业 {key} 正在执行中，请勿重复触发")
|
||
_IN_FLIGHT.add(key)
|
||
try:
|
||
return _run_one(db, job, spec)
|
||
finally:
|
||
_release(key)
|
||
|
||
|
||
def _loop() -> None:
|
||
db = SessionLocal()
|
||
try:
|
||
ensure_jobs(db)
|
||
except Exception: # noqa: BLE001 初始化失败不应让线程静默退出
|
||
logger.exception("自动化作业初始化失败")
|
||
finally:
|
||
db.close()
|
||
|
||
logger.info(
|
||
"自动化调度器已启动：心跳 {}s，作业周期 {}s", TICK_SECONDS, JOB_INTERVAL_SECONDS
|
||
)
|
||
while not _stop_event.wait(TICK_SECONDS):
|
||
db = SessionLocal()
|
||
try:
|
||
# wait=False：派发完就返回 —— tick 不再被慢作业拖住
|
||
run_due_jobs(db, wait=False)
|
||
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()
|
||
global _EXECUTOR
|
||
with _EXECUTOR_LOCK:
|
||
if _EXECUTOR is not None:
|
||
# 不 cancel_futures：让已经在跑的作业自然跑完，硬砍会让账号停在半路
|
||
_EXECUTOR.shutdown(wait=False)
|
||
_EXECUTOR = None
|
||
_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,
|
||
"job_interval_seconds": JOB_INTERVAL_SECONDS,
|
||
"registered": sorted(_REGISTRY),
|
||
}
|