From 19f1cbcebaec8cf4d4fd9a8fcda38ef18b8ce8b1 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Sun, 4 Oct 2026 22:11:44 +0800 Subject: [PATCH] =?UTF-8?q?fix(automation):=20=E4=BF=AE=E5=A4=8D=E4=BD=9C?= =?UTF-8?q?=E4=B8=9A=E5=AE=8C=E6=88=90=E7=8A=B6=E6=80=81=E5=9B=9E=E5=86=99?= =?UTF-8?q?=E5=B4=A9=E6=BA=83=E3=80=81=E7=BB=9F=E8=AE=A1=E5=AD=97=E6=AE=B5?= =?UTF-8?q?=E5=AF=B9=E9=BD=90=E5=8F=8A=E8=A1=A8=E5=8D=95=E5=B8=83=E5=B1=80?= =?UTF-8?q?=E6=8C=A4=E5=8E=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 而错误显示「排队中」的问题。 --- .../jobs/account_overview_refresh.py | 22 +- .../services/automation/jobs/ck_health.py | 18 +- web/backend/services/automation/scheduler.py | 115 ++++++--- web/backend/services/douyu_runner.py | 3 +- web/backend/services/douyu_runner_core.py | 2 + web/frontend/src/pages/AutomationPage.tsx | 231 +++++++++--------- 6 files changed, 245 insertions(+), 146 deletions(-) diff --git a/web/backend/services/automation/jobs/account_overview_refresh.py b/web/backend/services/automation/jobs/account_overview_refresh.py index c823c7e..144e889 100644 --- a/web/backend/services/automation/jobs/account_overview_refresh.py +++ b/web/backend/services/automation/jobs/account_overview_refresh.py @@ -145,7 +145,14 @@ def run_account_overview_refresh(db: Session, job: AutomationJob) -> dict[str, A now = datetime.now(UTC) accounts = build_select_query(db, job, now).all() if not accounts: - return {"scanned": 0, "succeeded": 0, "failed": 0, "message": "没有待查账号"} + return { + "scanned": 0, + "valid": 0, + "invalid": 0, + "succeeded": 0, + "failed": 0, + "message": "全库暂无待补全或到期的账号", + } account_ids = [a.id for a in accounts] batch_id, count = create_douyu_planned_tasks( @@ -158,14 +165,23 @@ def run_account_overview_refresh(db: Session, job: AutomationJob) -> dict[str, A {"source": "scheduled_refresh"}, ) if count == 0: - return {"scanned": 0, "succeeded": 0, "failed": 0, "message": "账号都不可查"} + return { + "scanned": 0, + "valid": 0, + "invalid": 0, + "succeeded": 0, + "failed": 0, + "message": "候选账号均不可查（可能正有在途任务）", + } # 复用现有执行器：并发、代理、进度、失败处理都是现成的 + # ⚠️ 必须 close_db=False，避免 runner 结束时把调度线程的 session 关闭 runner = DouyuBatchRunner( db, batch_id, "query_account_overview", concurrency=max(1, min(int(job.concurrency or 1), 10)), + close_db=False, ) runner.run() db.commit() @@ -200,6 +216,8 @@ def run_account_overview_refresh(db: Session, job: AutomationJob) -> dict[str, A logger.info(f"[overview] {message}") return { "scanned": len(rows), + "valid": succeeded, + "invalid": failed, "succeeded": succeeded, "failed": failed, "batch_id": batch_id, diff --git a/web/backend/services/automation/jobs/ck_health.py b/web/backend/services/automation/jobs/ck_health.py index bd00556..72902bc 100644 --- a/web/backend/services/automation/jobs/ck_health.py +++ b/web/backend/services/automation/jobs/ck_health.py @@ -13,13 +13,13 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import UTC, datetime, timedelta from typing import Any -from sqlalchemy import or_ +from sqlalchemy import func, or_ from sqlalchemy.orm import Session from ....models import Account, AutomationJob from ...account_lifecycle import MANUAL_STATES, on_ck_verdict from ...cookie_check_service import check_douyu_cookie -from ...douyu_service import cookie_operable_filter +from ...douyu_service import cookie_holding_filter, cookie_operable_filter JOB_KEY = "ck_health" FLUSH_EVERY = 25 @@ -135,7 +135,19 @@ def run_ck_health(db: Session, job: AutomationJob) -> dict[str, Any]: now = datetime.now(UTC) accounts = _select_accounts(db, job, now) if not accounts: - return {"scanned": 0, "valid": 0, "invalid": 0, "skipped": 0} + earliest_checked = ( + db.query(func.min(Account.ck_checked_at)) + .filter(cookie_holding_filter()) + .scalar() + ) + msg = "全库暂无到期待检账号" + if earliest_checked: + if earliest_checked.tzinfo is None: + earliest_checked = earliest_checked.replace(tzinfo=UTC) + hours_ago = (now - earliest_checked).total_seconds() / 3600 + recheck_hours = max(int(job.interval_seconds or 0), 60) / 3600 + msg = f"全库暂无到期待检账号（最久前检测于 {hours_ago:.1f} 小时前，重检间隔为 {recheck_hours:.1f} 小时）" + return {"scanned": 0, "valid": 0, "invalid": 0, "skipped": 0, "message": msg} results: dict[int, dict[str, Any]] = {} concurrency = max(min(int(job.concurrency or 1), 16), 1) diff --git a/web/backend/services/automation/scheduler.py b/web/backend/services/automation/scheduler.py index af2eaa1..18ea3bc 100644 --- a/web/backend/services/automation/scheduler.py +++ b/web/backend/services/automation/scheduler.py @@ -301,46 +301,94 @@ def _run_one(db: Session, job: AutomationJob, spec: JobSpec) -> dict[str, Any]: 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 + 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 - 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() + 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, - run.scanned, - run.valid, - run.invalid, - run.skipped, + scanned, + valid, + invalid, + skipped, ) return { "job_key": job.key, "status": status, "message": message, - "scanned": run.scanned, - "valid": run.valid, - "invalid": run.invalid, - "skipped": run.skipped, + "scanned": scanned, + "valid": valid, + "invalid": invalid, + "skipped": skipped, } @@ -353,7 +401,14 @@ def run_job_now(db: Session, key: str) -> dict[str, Any]: spec = _REGISTRY.get(key) if job is None or spec is None: raise KeyError(key) - return _run_one(db, job, spec) + 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: diff --git a/web/backend/services/douyu_runner.py b/web/backend/services/douyu_runner.py index 33f9e22..57ba3a7 100644 --- a/web/backend/services/douyu_runner.py +++ b/web/backend/services/douyu_runner.py @@ -192,4 +192,5 @@ class DouyuBatchRunner( self._push_log("info", f"斗鱼任务批次 {self.batch_id} 完成") self._push_log("result", "") finally: - self.db.close() + if getattr(self, "close_db", True): + self.db.close() diff --git a/web/backend/services/douyu_runner_core.py b/web/backend/services/douyu_runner_core.py index 4019b29..e4777a7 100644 --- a/web/backend/services/douyu_runner_core.py +++ b/web/backend/services/douyu_runner_core.py @@ -69,8 +69,10 @@ class DouyuBatchRunnerCore: log_queue: asyncio.Queue | None = None, loop: asyncio.AbstractEventLoop | None = None, concurrency: int = 3, + close_db: bool = True, ): self.db = db + self.close_db = close_db self.batch_id = batch_id self.task_type = task_type self.payload = payload or {} diff --git a/web/frontend/src/pages/AutomationPage.tsx b/web/frontend/src/pages/AutomationPage.tsx index a20b17e..5131f7d 100644 --- a/web/frontend/src/pages/AutomationPage.tsx +++ b/web/frontend/src/pages/AutomationPage.tsx @@ -38,15 +38,13 @@ const STATUS_TAG: Record = { skipped: { color: "default", text: "已跳过" }, }; -/** - * 下次运行时间：两个作业串行跑在同一个调度线程里，前一个作业耗时长时， - * 后一个会排队等待 —— 于是 next_run_at 会停在「已经过了」的时刻。 - * 直接显示一个过去时间看起来像故障，这里标明「排队中」。 - */ function nextRunLabel(job: AutomationJobItem): ReactNode { - if (!job.next_run_at) { + if (!job.enabled) { return 未启用; } + if (!job.next_run_at) { + return -; + } const next = new Date(job.next_run_at).getTime(); if (!Number.isFinite(next)) return formatTime(job.next_run_at); if (next > Date.now()) return formatTime(job.next_run_at); @@ -54,7 +52,7 @@ function nextRunLabel(job: AutomationJobItem): ReactNode { 排队中{" "} - （{formatTime(job.next_run_at)} 到期，等待前序作业让出线程） + （{formatTime(job.next_run_at)} 到期） ); @@ -273,116 +271,129 @@ export default function AutomationPage() { description="两个作业是联动的上下游：巡检把失效账号标记为 invalid，「CK 自动重登」的候选条件正是这个标记，所以巡检一发现失效，重登作业在下一个周期就会把它救回来（需打开重登开关）。刻意分成两个作业是为了止损粒度 —— 巡检只读、零风险；重登会真实登录（消耗邮箱验证码、可能遇到极验），出问题可单独关掉重登而不影响检测。巡检本身不会替换任何 Cookie、也不会改账号状态。" /> -
- {/* ⚠️ 必须有个纵向容器：`layout="inline"` 会把直接子元素排成一行， - 5 张 Card 会横着挤在一起且丢掉间距（真踩过）。 */} - - {jobs.map((job) => ( - - 上游：巡检标记为失效且有 CK 的账号，当前待重登{" "} - {pendingRelogin} 个 - - ) : job.key === "ck_health" ? ( - - 下游：CK 自动重登（发现失效后自动接手） - - ) : null - } - style={{ width: '100%' }} - > - - - {formatTime(job.last_run_at)} - - - {nextRunLabel(job)} - - - {job.last_status ? ( - - {STATUS_TAG[job.last_status]?.text ?? job.last_status} - - ) : ( - "-" - )} - - - {job.last_message || "-"} - - + + + {jobs.map((job) => ( + + 上游：巡检标记为失效且有 CK 的账号，当前待重登{" "} + {pendingRelogin} 个 + + ) : job.key === "ck_health" ? ( + + 下游：CK 自动重登（发现失效后自动接手） + + ) : null + } + style={{ width: "100%" }} + > + + + {formatTime(job.last_run_at)} + + + {nextRunLabel(job)} + + + {job.last_status ? ( + + {STATUS_TAG[job.last_status]?.text ?? job.last_status} + + ) : ( + "-" + )} + + + {job.last_message || "-"} + + - {/* ⚠️ 这里**不能**各自包一个 ： - 多个 绑同一个 form 实例时，`onFinish` 会被最后挂载的那个覆盖 —— - 点第一张卡的保存，跑的其实是**最后一张卡**的保存（真踩过： - 存「账号资料定时补全」提示的是「积分到期补记」，即存错了作业）。 - Form 统一提到这一层外面，各卡只用 Form.Item 的 [job.key, 字段] 区分。 */} - {/* 字段行：显式 gap 撑开，别让输入框挤在一起 */} - - - - - - - - - - - - - - - - - - - - runOnce(job)} - disabled={!canManage} + + + + + + + + + + + + + + + - - - {!canManage && ( - - 当前账号只能查看，修改配置需要 automation:manage 权限。 - - )} - - ))} + runOnce(job)} + disabled={!canManage} + > + + + + + {!canManage && ( +
+ + 当前账号只能查看，修改配置需要 automation:manage 权限。 + +
+ )} + + ))}