feat(automation): 新增「账号资料定时补全」作业 —— 分批慢跑，不撞风控

## 为什么不能一次性跑全量

生产约 **8,000 个**「CK 有效但没查过」的账号。一次性发起 ≈ **8 万次斗鱼请求**，
很容易触发风控 —— 而风控一旦触发是**所有账号一起受影响**。试水 200 个时
0.65 秒/账号看着很香（全量 86 分钟），但那正是危险的地方：**跑得越快越容易被封**。

## 新作业 `account_overview_refresh`

照「自动巡检」（`ck_health`）那套做，并复用它的两道保护。

| 项 | 默认 | 说明 |
| --- | --- | --- |
| `interval_seconds` | 30 分钟 | 多久一轮 |
| `batch_size` | 30 | 每轮查多少（约 20 秒跑完）|
| `concurrency` | 3 | 并发 |
| `refresh_days` | 7（config 可调）| 查过的多久后再查 |
| `enabled` | **False** | 默认关，由运维手动开 |

按默认值：8,000 个约 **3 天**补完，全程不打满斗鱼接口。

## 两类账号**分开**（用户明确要求）

- **没查过的**（`overview_queried_at IS NULL`）**优先**
- 查过的只在超过 `refresh_days` 后重新入选

否则"周期性刷新老账号"会永远挤掉"从没查过的新账号"。
MySQL 没有 `NULLS FIRST`，用 `ORDER BY overview_queried_at IS NULL DESC` 等价实现。

## 只查有效 CK

`ck_check_status == 'valid'`。失效的查了也是白查（只会多一条失败记录），
交给「CK 检测与重登」处理。另外沿用 `cookie_operable_filter()`（有 CK 且无在途尝试）、
排除人工状态（停用/归档）与「不参与自动化」的账号。

## 风控熔断（沿用巡检口径）

一轮里失败率 > 50%（且样本 ≥ 10）就**自己把自己关掉**，交给人确认。
消息里写明可能原因，否则运维只看到"失败率高"会一头雾水：

    失败率过高，已自动关闭作业（大面积失败通常是**被风控**或接口抖动，请先确认再重开）

样本 < 10 不熔断 —— 1 个失败就是 100%，比例没意义。

## 查询记录

每轮都建真实的 `douyu_tasks` 行（payload 标 `source=scheduled_refresh`），
「哪个账号、什么时候、查出来什么」都有据可查，也能在任务列表里
区分「定时巡检」与「人工点查询」。账号上的 `overview_queried_at`
则是"下次该不该再查"的依据。

## 验证

- **镜像库实测**：`本轮查 3 个：成功 3 / 失败 0（剩 7784 个待查）`
- 本地 `TZ=Asia/Shanghai` **632 passed / 6 skipped**；ruff 全绿
- 新增 11 条用例：选择口径（只查有效 CK、没查过优先、refresh_days 可调、
  人工状态排除、batch_size 限流）+ 熔断（高失败率关闭、正常不关、小样本不熔断）
  + 一条回归：计数不能走带 LIMIT 的查询（作业第一版就是这么挂的）

tests/test_account_overview_refresh_job.py

@@ -0,0 +1,210 @@
+"""定时补全账号资料的作业（`account_overview_refresh`）。
+
+**为什么不能一次性跑全量**：生产约 8,000 个待查账号，一次性发起 ≈ 8 万次斗鱼请求，
+很容易触发风控 —— 而风控的代价是所有账号一起受影响。所以做成定时分批，
+并复用「自动巡检」的两道保护（`cookie_operable_filter` + 错误率熔断）。
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+from tests.support.db import get_test_engine, make_test_session
+from web.backend.models import Account, AutomationJob, User
+from web.backend.services.automation.jobs import account_overview_refresh as job
+
+
+class _Base:
+    def setup_method(self):
+        self.engine = get_test_engine()
+        self.session = make_test_session(self.engine)
+        self.admin = User(username="ovr-admin", password_hash="h", role="super_admin")
+        self.session.add(self.admin)
+        self.session.flush()
+        self.job = AutomationJob(
+            key=job.JOB_KEY,
+            title="账号资料定时补全",
+            interval_seconds=1800,
+            batch_size=10,
+            concurrency=3,
+            enabled=False,
+            config={"refresh_days": 7},
+        )
+        self.session.add(self.job)
+        self.session.commit()
+
+    def teardown_method(self):
+        self.session.close()
+
+    def account(
+        self,
+        name: str,
+        *,
+        ck: str = "valid",
+        queried_days_ago: int | None = None,
+        auto_managed: bool = True,
+        status: str = "active",
+    ) -> Account:
+        acc = Account(
+            username=name,
+            password="p",
+            email=f"{name}@e.com",
+            email_password="ep",
+            cookie="acf_uid=1; acf_ccn=c",
+            ck_check_status=ck,
+            auto_managed=auto_managed,
+            status=status,
+        )
+        if queried_days_ago is not None:
+            acc.overview_queried_at = datetime.now(UTC) - timedelta(
+                days=queried_days_ago
+            )
+        self.session.add(acc)
+        self.session.commit()
+        return acc
+
+    def selected(self) -> list[Account]:
+        return job.build_select_query(self.session, self.job, datetime.now(UTC)).all()
+
+
+class TestSelection(_Base):
+    def test_only_valid_ck_is_selected(self):
+        """只查有效 CK —— 失效的查了也是白查，只会多一条失败记录。"""
+        self.account("ovr-valid", ck="valid")
+        self.account("ovr-invalid", ck="invalid")
+        self.account("ovr-error", ck="error")
+        names = [a.username for a in self.selected()]
+        assert "ovr-valid" in names
+        assert "ovr-invalid" not in names
+        assert "ovr-error" not in names
+
+    def test_never_queried_comes_first(self):
+        """**没查过的优先** —— 否则周期性刷新会永远挤掉从没查过的新账号。"""
+        self.account("ovr-old", queried_days_ago=30)
+        self.account("ovr-never")
+        assert self.selected()[0].username == "ovr-never"
+
+    def test_recently_queried_is_skipped(self):
+        """刚查过的要跳过（`refresh_days` 之内不再查）。"""
+        self.account("ovr-fresh", queried_days_ago=1)
+        names = [a.username for a in self.selected()]
+        assert "ovr-fresh" not in names
+
+    def test_stale_is_selected_again(self):
+        self.account("ovr-stale", queried_days_ago=30)
+        assert "ovr-stale" in [a.username for a in self.selected()]
+
+    def test_refresh_days_is_configurable(self):
+        self.job.config = {"refresh_days": 100}
+        self.session.commit()
+        self.account("ovr-30d", queried_days_ago=30)
+        assert "ovr-30d" not in [a.username for a in self.selected()]
+
+    def test_manual_and_unmanaged_accounts_are_excluded(self):
+        """人工状态（停用/归档）与「不参与自动化」的账号不该被自动碰。"""
+        self.account("ovr-disabled", status="disabled")
+        self.account("ovr-archived", status="archived")
+        self.account("ovr-unmanaged", auto_managed=False)
+        names = [a.username for a in self.selected()]
+        for bad in ("ovr-disabled", "ovr-archived", "ovr-unmanaged"):
+            assert bad not in names
+
+    def test_batch_size_limits_the_round(self):
+        for i in range(5):
+            self.account(f"ovr-batch-{i}")
+        self.job.batch_size = 2
+        self.session.commit()
+        assert len(self.selected()) == 2
+
+    def test_remaining_count_works_with_limit_applied(self):
+        """回归：计数不能走 `build_select_query`。
+
+        那个查询已经 `LIMIT` 了，再 `order_by(None)` 会被 SQLAlchemy 拒绝：
+        "order_by() being called on a Query which already has LIMIT"。
+        （真踩过 —— 作业第一版就是这么挂的。）
+        """
+        for i in range(4):
+            self.account(f"ovr-count-{i}")
+        self.job.batch_size = 1
+        self.session.commit()
+        assert job.count_remaining(self.session, self.job, datetime.now(UTC)) >= 4
+
+
+class TestCircuitBreaker(_Base):
+    """失败率过高就**自己把自己关掉**（风控保护）。
+
+    与「自动巡检」同口径。大面积失败通常意味着被风控或斗鱼接口抖动 ——
+    这时候继续跑只会把更多账号写坏，停手交给人确认才是对的。
+    """
+
+    def _run_with(self, total: int, failed: int):
+        """打桩执行器，只关心失败率熔断的逻辑。"""
+        from web.backend.models import DouyuTask
+
+        for i in range(total):
+            self.account(f"ovr-cb-{i}")
+
+        def fake_create(db, ids, task_type, scope, created_by, payload=None):
+            batch_id = "cb-batch"
+            for account_id in ids:
+                self.session.add(
+                    DouyuTask(
+                        batch_id=batch_id,
+                        account_id=account_id,
+                        task_type=task_type,
+                        handbook_scope=scope,
+                        status="planned",
+                        created_by=created_by,
+                    )
+                )
+            self.session.commit()
+            return batch_id, len(ids)
+
+        session = self.session  # 闭包捕获：下面 _FakeRunner 里的 self 是 runner
+
+        class _FakeRunner:
+            def __init__(self, *a, **k):
+                pass
+
+            def run(self):
+                # 把前 `failed` 个标成失败，其余成功
+                rows = (
+                    session.query(DouyuTask)
+                    .filter(DouyuTask.batch_id == "cb-batch")
+                    .order_by(DouyuTask.id.asc())
+                    .all()
+                )
+                for index, row in enumerate(rows):
+                    row.status = "failed" if index < failed else "success"
+                session.commit()
+
+        import web.backend.services.automation.jobs.account_overview_refresh as mod
+
+        original_create = mod.create_douyu_planned_tasks
+        original_runner = mod.DouyuBatchRunner
+        mod.create_douyu_planned_tasks = fake_create
+        mod.DouyuBatchRunner = _FakeRunner
+        try:
+            return job.run_account_overview_refresh(self.session, self.job)
+        finally:
+            mod.create_douyu_planned_tasks = original_create
+            mod.DouyuBatchRunner = original_runner
+
+    def test_high_failure_ratio_disables_the_job(self):
+        out = self._run_with(total=20, failed=15)  # 75% 失败
+        self.session.refresh(self.job)
+        assert out["paused"] is True
+        assert self.job.enabled is False, "失败率过高应自动关闭作业"
+        assert "风控" in out["message"]
+
+    def test_normal_failure_ratio_keeps_job_running(self):
+        out = self._run_with(total=20, failed=1)  # 5% 失败（真实的 CK 失效率）
+        self.session.refresh(self.job)
+        assert out["paused"] is False
+        assert "风控" not in out["message"]
+
+    def test_small_sample_does_not_trigger_breaker(self):
+        """样本太少时比例没有意义（1 个失败就是 100%），不该熔断。"""
+        out = self._run_with(total=3, failed=3)
+        self.session.refresh(self.job)
+        assert out["paused"] is False, "样本不足时不该熔断"

web/backend/services/automation/__init__.py

@@ -6,6 +6,12 @@
 
 from __future__ import annotations
 
+from .jobs.account_overview_refresh import (
+    JOB_KEY as ACCOUNT_OVERVIEW_REFRESH_KEY,
+)
+from .jobs.account_overview_refresh import (
+    run_account_overview_refresh,
+)
 from .jobs.ck_health import JOB_KEY as CK_HEALTH_KEY
 from .jobs.ck_health import run_ck_health
 from .jobs.ck_relogin import JOB_KEY as CK_RELOGIN_KEY
@@ -33,6 +39,7 @@ from .scheduler import (
 )
 
 __all__ = [
+    "ACCOUNT_OVERVIEW_REFRESH_KEY",
     "CK_HEALTH_KEY",
     "CK_RELOGIN_KEY",
     "DOUYU_TASKS_RETENTION_KEY",
@@ -44,6 +51,7 @@ __all__ = [
     "register_job",
     "registered_jobs",
     "reset_registry",
+    "run_account_overview_refresh",
     "run_ck_health",
     "run_ck_relogin",
     "run_douyu_tasks_retention",
@@ -92,6 +100,18 @@ def register_default_jobs() -> None:
             run=run_ck_relogin,
         )
     )
+    register_job(
+        JobSpec(
+            key=ACCOUNT_OVERVIEW_REFRESH_KEY,
+            title="账号资料定时补全",
+            # 默认每 30 分钟一轮、每轮 30 个 —— 约 20 秒跑完，慢而稳。
+            # 8,000 个待查按这个速度约 3 天补完，全程不打满斗鱼接口。
+            interval_seconds=30 * 60,
+            batch_size=30,
+            concurrency=3,
+            run=run_account_overview_refresh,
+        )
+    )
     register_job(
         JobSpec(
             key=CK_HEALTH_KEY,

web/backend/services/automation/jobs/account_overview_refresh.py

@@ -0,0 +1,213 @@
+"""作业：**定时分批**补全账号基础资料（一键查询的内容）。
+
+## 为什么不能一次性跑全量
+
+生产有约 8,000 个「CK 有效但没查过」的账号。一次性发起等于
+**~8 万次斗鱼请求**，很容易触发风控 —— 而风控的代价是所有账号一起受影响。
+
+所以做成**定时分批**：每轮只取 `batch_size` 个（默认 30），
+跑完就结束，下一轮再来。慢一点没关系，**稳**最重要。
+
+口径与「自动巡检」（`ck_health`）一致，并复用它的两道保护：
+- `cookie_operable_filter()`：账号有 CK 且没有在途尝试
+- **错误率熔断**：一轮里失败比例过高（通常是风控或接口抖动）就
+  **自己把自己关掉**，不再继续污染账号
+
+## 两类账号分开
+
+| 类型 | 条件 | 优先级 |
+| --- | --- | --- |
+| **没查过的** | `overview_queried_at IS NULL` | **优先**（补历史欠账）|
+| 查过但过期了 | `overview_queried_at < now - refresh_days` | 其次（周期性刷新）|
+
+用户明确要求这两类分开 —— 否则"周期性刷新老账号"会永远挤掉"从没查过的新账号"。
+
+## 查询记录
+
+每次都会建真实的 `douyu_tasks` 行（`query_account_overview`），
+所以「哪个账号、什么时候、查出来什么」都有据可查；
+账号上的 `overview_queried_at` 则是"下次该不该再查"的依据。
+
+## 只查有效 CK
+
+`ck_check_status == 'valid'`。CK 失效的账号查了也是白查（只会多一条失败记录），
+交给「CK 检测与重登」去处理。
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from loguru import logger
+from sqlalchemy import or_
+from sqlalchemy.orm import Session
+
+from ....models import Account, AutomationJob
+from ...account_lifecycle import MANUAL_STATES
+from ...douyu_runner import DouyuBatchRunner
+from ...douyu_service import cookie_operable_filter, create_douyu_planned_tasks
+
+JOB_KEY = "account_overview_refresh"
+
+#: 默认多久刷新一次已查过的账号（天）
+DEFAULT_REFRESH_DAYS = 7
+
+#: 触发熔断的最小样本量：样本太少时比例没有意义（1 个失败就是 100%）
+ERROR_PAUSE_MIN_SCANNED = 10
+
+#: 失败比例超过这个值就熔断（与 ck_health 同口径）
+ERROR_PAUSE_RATIO = 0.5
+
+
+def _positive_int(value: Any, default: int) -> int:
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return default
+    return parsed if parsed > 0 else default
+
+
+def _load_config(job: AutomationJob) -> dict[str, Any]:
+    config = job.config
+    if isinstance(config, str):
+        try:
+            config = json.loads(config)
+        except ValueError:
+            config = {}
+    return dict(config) if isinstance(config, dict) else {}
+
+
+def build_select_query(db: Session, job: AutomationJob, now: datetime):
+    """选出本轮要查的账号。
+
+    - 只要 **CK 有效**（`ck_check_status == 'valid'`）—— 失效的查了也是白查
+    - **没查过的优先**：`overview_queried_at IS NULL` 排在最前
+      （MySQL 没有 `NULLS FIRST`，用 `IS NULL` 的布尔值降序等价实现）
+    - 查过的只在超过 `refresh_days` 后才重新入选
+    """
+    refresh_days = _positive_int(
+        _load_config(job).get("refresh_days"), DEFAULT_REFRESH_DAYS
+    )
+    cutoff = now - timedelta(days=refresh_days)
+    batch = _positive_int(job.batch_size, 30)
+    batch = _positive_int(job.batch_size, 30)
+    return (
+        _base_query(db, job, now)
+        # 没查过的（IS NULL = 1）排最前 —— MySQL 没有 NULLS FIRST，
+        # 用 `IS NULL` 的布尔值降序等价实现
+        .order_by(Account.overview_queried_at.is_(None).desc(), Account.id.asc())
+        .limit(batch)
+    )
+
+
+def _base_query(db: Session, job: AutomationJob, now: datetime):
+    """**不带排序和 LIMIT** 的基础查询。
+
+    单独抽出来是因为计数不能再走 `build_select_query`：
+    SQLAlchemy 不允许在已 `LIMIT` 的查询上再 `order_by(None)`
+    （抛 "order_by() being called on a Query which already has LIMIT"）。
+    """
+    refresh_days = _positive_int(
+        _load_config(job).get("refresh_days"), DEFAULT_REFRESH_DAYS
+    )
+    cutoff = now - timedelta(days=refresh_days)
+    return db.query(Account).filter(
+        cookie_operable_filter(),
+        Account.auto_managed.is_(True),
+        Account.status.notin_(MANUAL_STATES),
+        Account.ck_check_status == "valid",
+        or_(
+            Account.overview_queried_at.is_(None),
+            Account.overview_queried_at < cutoff,
+        ),
+    )
+
+
+def count_remaining(db: Session, job: AutomationJob, now: datetime) -> int:
+    """还剩多少待查（只用于提示，不参与调度）。"""
+    return _base_query(db, job, now).count()
+
+
+def _system_user_id(db: Session) -> int:
+    """任务行的 created_by —— 定时任务不是某个人的操作，记在超管名下。"""
+    from ....models import User
+
+    user = (
+        db.query(User)
+        .filter(User.role == "super_admin")
+        .order_by(User.id.asc())
+        .first()
+    )
+    if user is None:
+        raise RuntimeError("库中没有超管用户，无法为定时任务提供 created_by")
+    return int(user.id)
+
+
+def run_account_overview_refresh(db: Session, job: AutomationJob) -> dict[str, Any]:
+    """本轮：选一批 → 建任务 → 就地跑完 → 看失败率决定要不要熔断。"""
+    now = datetime.now(UTC)
+    accounts = build_select_query(db, job, now).all()
+    if not accounts:
+        return {"scanned": 0, "succeeded": 0, "failed": 0, "message": "没有待查账号"}
+
+    account_ids = [a.id for a in accounts]
+    batch_id, count = create_douyu_planned_tasks(
+        db,
+        account_ids,
+        "query_account_overview",
+        "elite",
+        _system_user_id(db),
+        # 标上来源，便于在任务列表里区分「定时巡检」与「人工点查询」
+        {"source": "scheduled_refresh"},
+    )
+    if count == 0:
+        return {"scanned": 0, "succeeded": 0, "failed": 0, "message": "账号都不可查"}
+
+    # 复用现有执行器：并发、代理、进度、失败处理都是现成的
+    runner = DouyuBatchRunner(
+        db,
+        batch_id,
+        "query_account_overview",
+        concurrency=max(1, min(int(job.concurrency or 1), 10)),
+    )
+    runner.run()
+    db.commit()
+
+    from ....models import DouyuTask
+
+    rows = db.query(DouyuTask).filter(DouyuTask.batch_id == batch_id).all()
+    succeeded = sum(1 for t in rows if t.status == "success")
+    failed = sum(1 for t in rows if t.status == "failed")
+
+    error_ratio = failed / len(rows) if rows else 0.0
+    paused = False
+    if len(rows) >= ERROR_PAUSE_MIN_SCANNED and error_ratio > ERROR_PAUSE_RATIO:
+        # 大面积失败通常是风控或斗鱼接口抖动。继续跑只会把更多账号写坏，
+        # 所以**自己把自己关掉**，交给人来确认（与 ck_health 同口径）。
+        job.enabled = False
+        db.commit()
+        paused = True
+        logger.warning(
+            f"[overview] 失败率 {error_ratio:.0%} 过高，已自动关闭作业"
+            f"（batch={batch_id}，failed={failed}/{len(rows)}）—— 请确认是否被风控"
+        )
+
+    message = (
+        f"本轮查 {len(rows)} 个：成功 {succeeded} / 失败 {failed}"
+        f"（剩 {count_remaining(db, job, now)} 个待查）"
+    )
+    if paused:
+        # 把**可能的原因**写进消息里：运维看作业历史时才知道该去查什么，
+        # 否则只看到"失败率高"会一头雾水。
+        message += "；失败率过高，已自动关闭作业（大面积失败通常是**被风控**或接口抖动，请先确认再重开）"
+    logger.info(f"[overview] {message}")
+    return {
+        "scanned": len(rows),
+        "succeeded": succeeded,
+        "failed": failed,
+        "batch_id": batch_id,
+        "paused": paused,
+        "message": message,
+    }