feat(automation): CK 自动重登（P1b）+ 失败分型与待人工队列

P1a 的巡检只产出「谁失效了」的清单，不兑现修复：1,356 个失效账号仍要人工逐个选集→点重登→过滑块。线上实测 20 个失效账号重登 20/20 成功（2 分 13 秒、零极验），证实失效是会话级过期而非凭证问题，可救量约 1,350。后端： - 迁移 20261002_0038：login_tasks 加 relogin_failure_kind（枚举），供作业决定「继续退避重试」还是「转人工」；不用 attempt_message 做字符串匹配（文案一改就失效，页面也要按类型筛选）； - jobs/ck_relogin.py： - 候选 = 巡检判定 invalid + 允许自动管理 + 无在途尝试 + 退避到期 + 未超每日上限 + **未转人工**（极验/凭证问题不再自动重登）； - 复用 LoginBatchRunner(mode="relogin")，成功才替换 CK（沿用既有安全语义）。自动化身份取最早的超管做 created_by，权限给 login:view_all 以免受归属限制；总时长上限压到 240 秒，避免长时间占住调度线程； - settle_results() 独立成纯函数便于单测：成功清空退避状态；失败按分型分流 —— captcha/credential 转人工（不排期）、email_quota 退避到次日零点+随机抖动、 email_code/interrupted/network 走 1h→3h→12h→24h 阶梯；次数用尽也停下等人工； - 未知文案默认按可重试处理（有每日上限兜底），别把可能只是抖动的账号推给人工； - 注册第二个作业「CK 自动重登」（batch 10 / 并发 3，**默认关闭**）； - 新增 GET /api/automation/manual-queue：按原因分组列出待人工处理的账号。可重试类型但已无下次排期的（次数用尽）单独成组 —— 否则会混进「验证码失败」里，客服看不出这个已经不会再自动试了。前端： - /automation 页新增「待人工处理」卡片，按分型分组展示（极验要过滑块、凭证要核对、次数用尽要人工确认），各带账号与最近失败原因。测试：新增 26 条（分型顺序陷阱「上限」优先于「验证码」、退避阶梯、配额退避次日、候选筛选、MySQL 方言编译、结算、待人工队列分组）。提交前全量 297 通过、ruff 全绿、前端 tsc 通过。注意：提交时另一路「测试改用 MySQL」的改造正在同一工作区进行，本次刻意只暂存本功能相关文件，未触碰 tests/conftest.py、tests/support/ 及对方已转换的测试文件。
2026-10-02 20:51:19 +08:00
parent 5645f19102
commit 8fc8e623b9
11 changed files with 1000 additions and 65 deletions
@@ -8,6 +8,8 @@ from __future__ import annotations

 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
+from .jobs.ck_relogin import run_ck_relogin
 from .scheduler import (
    JobSpec,
    automation_enabled,
@@ -24,6 +26,7 @@ from .scheduler import (

 __all__ = [
    "CK_HEALTH_KEY",
+    "CK_RELOGIN_KEY",
    "JobSpec",
    "automation_enabled",
    "ensure_jobs",
@@ -32,6 +35,7 @@ __all__ = [
    "registered_jobs",
    "reset_registry",
    "run_ck_health",
+    "run_ck_relogin",
    "run_due_jobs",
    "run_job_now",
    "scheduler_status",
@@ -42,6 +46,18 @@ __all__ = [

 def register_default_jobs() -> None:
    """登记内置作业（幂等）。"""
+    register_job(
+        JobSpec(
+            key=CK_RELOGIN_KEY,
+            title="CK 自动重登",
+            # interval_seconds 同样是「同一行多久重试一次」的兜底（真实节奏由
+            # relogin_next_at 的退避表决定）；批次小、跑得慢，单轮 10 条。
+            interval_seconds=6 * 3600,
+            batch_size=10,
+            concurrency=3,
+            run=run_ck_relogin,
+        )
+    )
    register_job(
        JobSpec(
            key=CK_HEALTH_KEY,
@@ -0,0 +1,274 @@
+"""CK 自动重登作业（P1b）。
+
+和巡检（P1a）是一对：巡检只产出「谁失效了」的清单，本作业负责把能救的救回来。
+线上实测（2026-10-02）20 个失效账号重登 **20/20 成功** —— 失效是会话级过期，
+不是凭证问题。
+
+**核心设计：失败必须分型。** 线上数据表明 415 个重登失败里只有 35% 是真正需要
+人工的（极验滑块），另外 64%（取码失败 + 重启中断）其实可自动重试；不分型就会
+把可救名额一起推给人工。分型结果落 ``login_tasks.relogin_failure_kind``。
+
+本作业**默认关闭**，且按规划 §7.4.7 分三段上线（小规模实测已完成 → 限量 → 全量）。
+"""
+
+from __future__ import annotations
+
+import random
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from sqlalchemy import func, or_
+from sqlalchemy.orm import Session
+
+from ....database import SessionLocal
+from ....models import Account, AutomationJob, LoginTask, User
+from ....models import ProxyConfig as ProxyConfigModel
+from ...douyu_service import LOGIN_ATTEMPT_IN_FLIGHT_STATUSES, cookie_holding_filter
+from ...login_service import LoginBatchRunner, get_relogin_limits
+
+JOB_KEY = "ck_relogin"
+
+# 失败分型 → 后续动作。
+# 需要人工的：极验滑块（自动重试只会加重风控）、凭证问题（重试多少次都一样）。
+MANUAL_KINDS = frozenset({"captcha", "credential"})
+# 可自动重试的：取码失败/超时（多为邮件延迟）、中断（服务重启或人工停止）、
+# 网络异常。email_quota 单列，因为它要退避到次日而不是按小时退避。
+RETRYABLE_KINDS = frozenset({"email_code", "interrupted", "network"})
+
+# 退避阶梯（分钟）：1h → 3h → 12h → 24h，超出取最后一级。
+BACKOFF_MINUTES = (60, 180, 720, 1440)
+
+# 自动重登时给 runner 的总时长上限。默认的 600 秒会长时间占住调度线程，
+# 自动化场景压到 240 秒：一轮跑不完的账号留着下一轮（它们本来就有退避）。
+AUTO_MAX_TOTAL_TIME = 240
+
+# 邮箱配额打满后退避到次日零点，再叠加随机抖动，避免整点齐发。
+QUOTA_JITTER_MINUTES = 30
+
+
+def classify_failure(message: str | None) -> str:
+    """把重登失败消息映射成分型（见规划 §7.4.3）。
+
+    顺序很重要：「邮箱验证码发送次数已达上限」既含「验证码」也含「上限」，
+    必须先在 ``email_quota`` 命中，否则会被归到可重试的 ``email_code`` 上，
+    导致次日才该重试的账号被按小时反复撞配额。
+    """
+    text = message or ""
+    if "上限" in text or "配额" in text:
+        return "email_quota"
+    if "极验" in text or "验证" in text and "滑块" in text:
+        return "captcha"
+    if "验证码" in text:
+        return "email_code"
+    if "中断" in text or "已停止" in text or "任务已停止" in text:
+        return "interrupted"
+    if "密码" in text or "账号不存在" in text or "缺少密码或邮箱" in text:
+        return "credential"
+    # 请求失败/超时/连接异常，以及未知文案：都按可重试处理。
+    # 宁可多给一次机会（有每日上限兜底），也不要把可能只是抖动的账号推给人工。
+    return "network"
+
+
+def _cutoff(now: datetime) -> datetime:
+    return now
+
+
+def build_select_query(
+    db: Session,
+    job: AutomationJob,
+    now: datetime,
+    *,
+    max_attempts: int = 2,
+    system_user_id: int | None = None,
+):
+    """构造本轮重登候选的查询（与取行分开，便于按方言编译做回归）。"""
+    query = (
+        db.query(LoginTask)
+        .join(Account, LoginTask.account_id == Account.id)
+        .filter(LoginTask.ck_check_status == "invalid")
+        .filter(cookie_holding_filter())
+        .filter(
+            func.coalesce(LoginTask.attempt_status, "").notin_(
+                LOGIN_ATTEMPT_IN_FLIGHT_STATUSES
+            )
+        )
+        .filter(Account.auto_managed.is_(True))
+        .filter(LoginTask.relogin_attempts < max_attempts)
+        .filter(
+            or_(
+                LoginTask.relogin_next_at.is_(None),
+                LoginTask.relogin_next_at <= _cutoff(now),
+            )
+        )
+    )
+    # 已判定需要人工的类型不再自动重试（极验/凭证）。空值用 coalesce 兜住历史行。
+    query = query.filter(
+        func.coalesce(LoginTask.relogin_failure_kind, "").notin_(tuple(MANUAL_KINDS))
+    )
+    if system_user_id is not None:
+        pass  # 预留：将来若需要限制到某个归属人，在这里加过滤
+    return query.order_by(
+        LoginTask.relogin_next_at.is_(None).desc(),
+        LoginTask.relogin_next_at.asc(),
+        LoginTask.id.asc(),
+    ).limit(max(int(job.batch_size or 0), 1))
+
+
+def _system_user_id(db: Session) -> int | None:
+    """自动重登要用一个真实用户身份落 ``created_by``（外键非空，且便于追责）。
+
+    取最早创建的超管作为「系统操作人」；库是空的就返回 None，作业会跳过本轮。
+    """
+    row = (
+        db.query(User.id)
+        .filter(User.role == "super_admin")
+        .order_by(User.id.asc())
+        .first()
+    )
+    if row:
+        return int(row[0])
+    row = db.query(User.id).order_by(User.id.asc()).first()
+    return int(row[0]) if row else None
+
+
+def _next_retry_at(kind: str, attempts: int, now: datetime) -> datetime:
+    if kind == "email_quota":
+        tomorrow = (now + timedelta(days=1)).replace(
+            hour=0, minute=0, second=0, microsecond=0
+        )
+        return tomorrow + timedelta(minutes=random.uniform(0, QUOTA_JITTER_MINUTES))
+    step = BACKOFF_MINUTES[min(max(attempts, 1), len(BACKOFF_MINUTES)) - 1]
+    return now + timedelta(minutes=step)
+
+
+def run_ck_relogin(db: Session, job: AutomationJob) -> dict[str, Any]:
+    now = datetime.now(UTC)
+    config = job.config or {}
+    max_attempts = int(config.get("max_attempts_per_day", 2) or 2)
+
+    system_user_id = _system_user_id(db)
+    if system_user_id is None:
+        return {
+            "scanned": 0,
+            "valid": 0,
+            "invalid": 0,
+            "skipped": 0,
+            "message": "库里没有可用用户，无法以系统身份发起重登",
+        }
+
+    tasks = build_select_query(db, job, now, max_attempts=max_attempts).all()
+    if not tasks:
+        return {"scanned": 0, "valid": 0, "invalid": 0, "skipped": 0}
+
+    task_ids = [task.id for task in tasks]
+    proxy = db.query(ProxyConfigModel).first()
+    _, max_total_time = get_relogin_limits()
+
+    # 批次用独立会话（runner 会在自己的线程/会话里写库），跑完再回来读结果。
+    runner_db = SessionLocal()
+    runner = LoginBatchRunner(
+        db=runner_db,
+        account_ids=[],
+        created_by=system_user_id,
+        # 自动化要对全库生效，不受「只能操作分配给自己的账号」限制。
+        creator_permissions=["login:view_all", "cookie:operate"],
+        max_login_retries=int(config.get("max_login_retries", 5) or 5),
+        max_total_time=min(int(max_total_time), AUTO_MAX_TOTAL_TIME),
+        proxy_config=proxy,
+        log_queue=None,
+        loop=None,
+        concurrency=max(min(int(job.concurrency or 3), 8), 1),
+        mode="relogin",
+        relogin_task_ids=task_ids,
+    )
+    batch_id = runner.batch_id
+    try:
+        runner.run()
+    finally:
+        runner_db.close()
+
+    # runner 用的是另一个会话，这里必须重新查，不能读旧对象。
+    db.expire_all()
+    fresh = {
+        row.id: row
+        for row in db.query(LoginTask).filter(LoginTask.id.in_(task_ids)).all()
+    }
+
+    counters = settle_results(fresh, task_ids, now=now, max_attempts=max_attempts)
+    db.commit()
+
+    parts = [
+        f"本轮重登 {counters['scanned']} 条",
+        f"成功 {counters['valid']}",
+        f"失败 {counters['invalid']}",
+    ]
+    if counters["retry"]:
+        parts.append(f"待重试 {counters['retry']}")
+    if counters["email_quota"]:
+        parts.append(f"其中配额满 {counters['email_quota']}（退避到次日）")
+    if counters["manual"]:
+        parts.append(f"转人工 {counters['manual']}")
+    return {
+        "scanned": counters["scanned"],
+        "valid": counters["valid"],
+        "invalid": counters["invalid"],
+        "relogin_started": counters["scanned"],
+        "relogin_success": counters["valid"],
+        "skipped": 0,
+        "message": "，".join(parts),
+        "detail": {
+            "batch_id": batch_id,
+            "manual": counters["manual"],
+            "retry": counters["retry"],
+            "email_quota": counters["email_quota"],
+        },
+    }
+
+
+def settle_results(
+    fresh: dict[int, LoginTask],
+    task_ids: list[int],
+    *,
+    now: datetime,
+    max_attempts: int = 2,
+) -> dict[str, int]:
+    """批次跑完后结算：成功的清零，失败的按分型决定重试还是转人工。
+
+    单独抽出来是因为这段是 P1b 的核心决策逻辑，必须能脱离真实登录做单测。
+    只改 ``relogin_*`` 四个字段，**不碰 CK 本体**（CK 由 runner 在成功时替换）。
+    """
+    counters = {
+        "scanned": len(task_ids),
+        "valid": 0,
+        "invalid": 0,
+        "manual": 0,
+        "retry": 0,
+        "email_quota": 0,
+    }
+    for task_id in task_ids:
+        task = fresh.get(task_id)
+        if task is None:
+            continue
+        if task.attempt_status == "success":
+            counters["valid"] += 1
+            task.relogin_attempts = 0
+            task.relogin_next_at = None
+            task.relogin_failure_kind = ""
+            task.relogin_last_error = ""
+            continue
+
+        counters["invalid"] += 1
+        kind = classify_failure(task.attempt_message)
+        task.relogin_failure_kind = kind
+        task.relogin_last_error = (task.attempt_message or "")[:500]
+        task.relogin_attempts = int(task.relogin_attempts or 0) + 1
+        if kind in MANUAL_KINDS or task.relogin_attempts >= max_attempts:
+            # 转人工：极验/凭证问题重试无意义，次数用尽也不该无限撞。
+            counters["manual"] += 1
+            task.relogin_next_at = None
+            continue
+        if kind == "email_quota":
+            counters["email_quota"] += 1
+        counters["retry"] += 1
+        task.relogin_next_at = _next_retry_at(kind, task.relogin_attempts, now)
+    return counters