From 8fc8e623b98bce3f5ffb3fd0d29f903208cca5c3 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Fri, 2 Oct 2026 20:51:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(automation):=20CK=20=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E9=87=8D=E7=99=BB=EF=BC=88P1b=EF=BC=89+=20=E5=A4=B1=E8=B4=A5?= =?UTF-8?q?=E5=88=86=E5=9E=8B=E4=B8=8E=E5=BE=85=E4=BA=BA=E5=B7=A5=E9=98=9F?= =?UTF-8?q?=E5=88=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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/ 及对方已转换的测试文件。 --- tests/test_automation_ck_health.py | 10 +- tests/test_automation_ck_relogin.py | 377 ++++++++++++++++++ tests/test_migrations.py | 2 +- .../20261002_0038_relogin_failure_kind.py | 41 ++ web/backend/models.py | 5 + web/backend/routers/automation.py | 80 +++- web/backend/services/automation/__init__.py | 16 + .../services/automation/jobs/ck_relogin.py | 274 +++++++++++++ web/frontend/src/api/automation.ts | 29 +- web/frontend/src/api/types.ts | 63 ++- web/frontend/src/pages/AutomationPage.tsx | 168 ++++++-- 11 files changed, 1000 insertions(+), 65 deletions(-) create mode 100644 tests/test_automation_ck_relogin.py create mode 100644 web/backend/migrations/versions/20261002_0038_relogin_failure_kind.py create mode 100644 web/backend/services/automation/jobs/ck_relogin.py diff --git a/tests/test_automation_ck_health.py b/tests/test_automation_ck_health.py index b594a82..b4d83f9 100644 --- a/tests/test_automation_ck_health.py +++ b/tests/test_automation_ck_health.py @@ -406,8 +406,8 @@ class TestAutomationApi(_Base): assert payload["scheduler"]["tick_seconds"] == 300 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 ck_health.JOB_KEY in keys + job = next(item for item in payload["jobs"] if item["key"] == ck_health.JOB_KEY) assert job["enabled"] is False assert job["interval_seconds"] == 6 * 3600 assert job["batch_size"] == 200 @@ -423,7 +423,11 @@ class TestAutomationApi(_Base): db=self.session, current=self.admin, ) - job = self.session.query(AutomationJob).one() + job = ( + self.session.query(AutomationJob) + .filter(AutomationJob.key == ck_health.JOB_KEY) + .one() + ) assert job.enabled is True assert job.interval_seconds == 3600 assert job.batch_size == 5 diff --git a/tests/test_automation_ck_relogin.py b/tests/test_automation_ck_relogin.py new file mode 100644 index 0000000..dfb63fc --- /dev/null +++ b/tests/test_automation_ck_relogin.py @@ -0,0 +1,377 @@ +"""P1b 回归：CK 自动重登的候选筛选、失败分型与结算。 + +两件事必须一起守住： + +1. **分型正确**：线上 415 个重登失败里只有 35%（极验）真的需要人工，另外 64% + （取码失败 + 重启中断）其实可自动重试。分错方向就是白扔可救名额，或者让 + 极验账号反复撞风控。 +2. **选行安全**：只挑「巡检判定失效 + 允许自动管理 + 没有在途尝试 + 退避到期 + + 没转人工」的行；转人工的类型绝不能被再次自动重登。 + +结算逻辑（``settle_results``）与选行查询（``build_select_query``）都刻意抽成了 +纯函数/可编译对象，这里不碰真实登录链路。 +""" + +from datetime import UTC, datetime, timedelta + +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker + +from web.backend.database import Base +from web.backend.models import Account, AutomationJob, LoginTask, User +from web.backend.services.automation.jobs import ck_relogin + + +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.session.add(self.admin) + self.session.commit() + + def teardown_method(self): + self.session.close() + + def make_task( + self, + name="acc", + *, + ck_check_status="invalid", + cookie="acf_uid=1", + attempt_status="", + attempt_message="", + relogin_attempts=0, + relogin_next_at=None, + relogin_failure_kind="", + relogin_last_error="", + auto_managed=True, + ): + account = Account( + username=name, + password="p", + email=f"{name}@e.com", + email_password="ep", + auto_managed=auto_managed, + ) + self.session.add(account) + self.session.commit() + task = LoginTask( + batch_id="b", + account_id=account.id, + created_by=self.admin.id, + status="success", + cookie=cookie, + ck_check_status=ck_check_status, + attempt_status=attempt_status, + attempt_message=attempt_message, + relogin_attempts=relogin_attempts, + relogin_next_at=relogin_next_at, + relogin_failure_kind=relogin_failure_kind, + relogin_last_error=relogin_last_error, + ) + self.session.add(task) + self.session.commit() + return task + + def job(self, batch_size=10): + return AutomationJob( + key=ck_relogin.JOB_KEY, + interval_seconds=6 * 3600, + batch_size=batch_size, + concurrency=3, + ) + + +class TestFailureClassification(_Base): + def test_quota_wins_over_code(self): + """「发送次数已达上限」既含验证码又含上限，必须优先判配额。 + + 判成 email_code 会按小时反复撞配额；判成 email_quota 才退避到次日。 + """ + msg = "重新登录失败: 提交验证码失败: 您的邮箱验证码发送次数已达上限，请改天再试！（旧 Cookie 已保留）" + assert ck_relogin.classify_failure(msg) == "email_quota" + + def test_captcha_is_manual(self): + assert ( + ck_relogin.classify_failure( + "重新登录失败: 极验验证失败: slide（旧 Cookie 已保留）" + ) + == "captcha" + ) + assert "captcha" in ck_relogin.MANUAL_KINDS + + def test_email_code_is_retryable(self): + assert ( + ck_relogin.classify_failure( + "重新登录失败: 提交验证码失败: 邮箱验证码错误，请重新输入" + ) + == "email_code" + ) + + def test_interrupted_is_retryable(self): + assert ( + ck_relogin.classify_failure("登录已中断（服务重启），旧 Cookie 已保留") + == "interrupted" + ) + assert ( + ck_relogin.classify_failure("重新登录已停止，旧 Cookie 已保留") + == "interrupted" + ) + + def test_credential_is_manual(self): + assert ( + ck_relogin.classify_failure("重新登录失败: 账号缺少密码或邮箱") + == "credential" + ) + assert "credential" in ck_relogin.MANUAL_KINDS + + def test_unknown_defaults_to_retryable(self): + """未知文案按可重试处理：有每日上限兜底，别把可能只是抖动的账号推给人工。""" + assert ck_relogin.classify_failure("某种没见过的报错") == "network" + assert ck_relogin.classify_failure("") == "network" + + +class TestBackoff(_Base): + def test_ladder_is_escalating(self): + now = datetime(2026, 10, 2, 12, 0, tzinfo=UTC) + first = ck_relogin._next_retry_at("email_code", 1, now) - now + second = ck_relogin._next_retry_at("email_code", 2, now) - now + third = ck_relogin._next_retry_at("email_code", 3, now) - now + assert first == timedelta(minutes=60) + assert second == timedelta(minutes=180) + assert third == timedelta(minutes=720) + + def test_quota_goes_to_next_day_with_jitter(self): + now = datetime(2026, 10, 2, 12, 0, tzinfo=UTC) + nxt = ck_relogin._next_retry_at("email_quota", 1, now) + assert nxt.date() == (now + timedelta(days=1)).date() + assert nxt.hour == 0 + assert 0 <= nxt.minute <= ck_relogin.QUOTA_JITTER_MINUTES + + +class TestSelection(_Base): + def _ids(self, job): + now = datetime.now(UTC) + return [ + t.id + for t in ck_relogin.build_select_query( + self.session, job, now, max_attempts=2 + ).all() + ] + + def test_picks_invalid_with_expired_backoff(self): + job = self.job() + ready = self.make_task("ready") + assert self._ids(job) == [ready.id] + + def test_skips_valid_and_empty_cookie_and_opted_out(self): + job = self.job() + keep = self.make_task("keep") + self.make_task("valid", ck_check_status="valid") + self.make_task("nocookie", cookie="") + self.make_task("optedout", auto_managed=False) + assert self._ids(job) == [keep.id] + + def test_skips_inflight_attempt(self): + job = self.job() + keep = self.make_task("keep") + self.make_task("running", attempt_status="running") + self.make_task("pending", attempt_status="pending") + assert self._ids(job) == [keep.id] + + def test_respects_backoff_window(self): + job = self.job() + due = self.make_task( + "due", relogin_next_at=datetime.now(UTC) - timedelta(minutes=1) + ) + self.make_task("later", relogin_next_at=datetime.now(UTC) + timedelta(hours=3)) + assert self._ids(job) == [due.id] + + def test_skips_when_attempts_exhausted(self): + job = self.job() + keep = self.make_task("keep") + self.make_task("exhausted", relogin_attempts=2) + assert self._ids(job) == [keep.id] + + def test_never_retries_manual_kinds(self): + """已判定极验/凭证问题的账号，绝不能再次自动重登。""" + job = self.job() + keep = self.make_task("keep") + self.make_task("captcha", relogin_failure_kind="captcha") + self.make_task("credential", relogin_failure_kind="credential") + assert self._ids(job) == [keep.id] + + def test_retryable_kinds_stay_eligible(self): + job = self.job() + code = self.make_task("code", relogin_failure_kind="email_code") + net = self.make_task("net", relogin_failure_kind="network") + assert sorted(self._ids(job)) == sorted([code.id, net.id]) + + def test_generated_sql_is_mysql_compatible(self): + from sqlalchemy.dialects import mysql + + query = ck_relogin.build_select_query( + self.session, self.job(), datetime.now(UTC), max_attempts=2 + ) + sql = str(query.statement.compile(dialect=mysql.dialect())).upper() + assert "NULLS FIRST" not in sql and "NULLS LAST" not in sql + + def test_batch_size_caps_the_round(self): + job = self.job(batch_size=3) + for index in range(6): + self.make_task(f"a{index}") + assert len(self._ids(job)) == 3 + + +class TestSettlement(_Base): + """结算：成功清零；失败按分型决定重试还是转人工。""" + + def _settle(self, tasks, *, max_attempts=2): + now = datetime.now(UTC) + fresh = {t.id: t for t in tasks} + counters = ck_relogin.settle_results( + fresh, [t.id for t in tasks], now=now, max_attempts=max_attempts + ) + self.session.commit() + return counters, now + + def test_success_resets_state(self): + task = self.make_task( + "ok", + attempt_status="success", + relogin_attempts=1, + relogin_failure_kind="email_code", + relogin_last_error="旧的报错", + relogin_next_at=datetime.now(UTC) + timedelta(hours=1), + ) + counters, _ = self._settle([task]) + self.session.refresh(task) + assert counters["valid"] == 1 and counters["invalid"] == 0 + assert task.relogin_attempts == 0 + assert task.relogin_next_at is None + assert task.relogin_failure_kind == "" + assert task.relogin_last_error == "" + + def test_retryable_failure_schedules_next_attempt(self): + task = self.make_task( + "retry", + attempt_status="failed", + attempt_message="等待验证码超时（旧 Cookie 已保留）", + ) + counters, now = self._settle([task]) + self.session.refresh(task) + assert counters["invalid"] == 1 and counters["retry"] == 1 + assert task.relogin_failure_kind == "email_code" + assert task.relogin_attempts == 1 + assert task.relogin_next_at is not None and task.relogin_next_at > now + assert "旧 Cookie" in task.relogin_last_error + + def test_captcha_goes_to_manual_without_schedule(self): + task = self.make_task( + "cap", attempt_status="failed", attempt_message="极验验证失败: slide" + ) + counters, _ = self._settle([task]) + self.session.refresh(task) + assert counters["manual"] == 1 and counters["retry"] == 0 + assert task.relogin_failure_kind == "captcha" + assert task.relogin_next_at is None # 不再自动排期 + + def test_quota_backs_off_to_next_day(self): + task = self.make_task( + "quota", + attempt_status="failed", + attempt_message="您的邮箱验证码发送次数已达上限，请改天再试！", + ) + counters, now = self._settle([task]) + self.session.refresh(task) + assert counters["email_quota"] == 1 and counters["retry"] == 1 + assert task.relogin_next_at.date() == (now + timedelta(days=1)).date() + + def test_attempts_exhausted_stops_retrying(self): + task = self.make_task( + "used", + attempt_status="failed", + attempt_message="等待验证码超时", + relogin_attempts=1, + ) + counters, _ = self._settle([task], max_attempts=2) + self.session.refresh(task) + assert counters["manual"] == 1 and counters["retry"] == 0 + assert task.relogin_next_at is None + + def test_settlement_never_touches_cookie(self): + task = self.make_task( + "keepck", attempt_status="failed", attempt_message="极验验证失败: slide" + ) + before = (task.cookie, task.status) + self._settle([task]) + self.session.refresh(task) + assert (task.cookie, task.status) == before + + +class TestManualQueueApi(_Base): + """待人工处理队列：只列自动重登救不了的，按原因分组。""" + + def _queue(self): + from web.backend.routers import automation as api + + return api.manual_queue(limit=20, db=self.session, current=self.admin) + + def test_groups_manual_kinds_only(self): + captcha = self.make_task( + "cap", + attempt_status="failed", + attempt_message="极验验证失败: slide", + relogin_failure_kind="captcha", + relogin_attempts=1, + relogin_last_error="极验验证失败: slide", + ) + self.make_task( + "cred", + attempt_status="failed", + attempt_message="账号缺少密码或邮箱", + relogin_failure_kind="credential", + relogin_attempts=1, + relogin_last_error="账号缺少密码或邮箱", + ) + # 可自动重试的不该出现在人工队列里 + self.make_task( + "retryable", + attempt_status="failed", + attempt_message="等待验证码超时", + relogin_failure_kind="email_code", + relogin_attempts=1, + relogin_next_at=datetime.now(UTC) + timedelta(hours=1), + relogin_last_error="等待验证码超时", + ) + payload = self._queue() + kinds = [group["kind"] for group in payload["groups"]] + assert sorted(kinds) == ["captcha", "credential"] + assert payload["total"] == 2 + captcha_group = next(g for g in payload["groups"] if g["kind"] == "captcha") + assert captcha_group["count"] == 1 + assert captcha_group["samples"][0]["task_id"] == captcha.id + assert "极验" in captcha_group["label"] + + def test_exhausted_retries_show_up_as_manual(self): + """次数用尽的（next_run_at 已清空）也要冒出来，否则会静默卡住。""" + self.make_task( + "used", + attempt_status="failed", + attempt_message="等待验证码超时", + relogin_failure_kind="email_code", + relogin_attempts=2, + relogin_next_at=None, + relogin_last_error="等待验证码超时", + ) + payload = self._queue() + assert payload["total"] == 1 + assert "重试次数用尽" in payload["groups"][0]["label"] + + def test_empty_queue(self): + payload = self._queue() + assert payload["total"] == 0 and payload["groups"] == [] diff --git a/tests/test_migrations.py b/tests/test_migrations.py index aa76df6..2acd3d6 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -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 = "20261002_0037" +HEAD_REVISION = "20261002_0038" # 迁移新增、但不声明在模型里的复合查询索引。 EXTRA_INDEXES = ( diff --git a/web/backend/migrations/versions/20261002_0038_relogin_failure_kind.py b/web/backend/migrations/versions/20261002_0038_relogin_failure_kind.py new file mode 100644 index 0000000..beb8e76 --- /dev/null +++ b/web/backend/migrations/versions/20261002_0038_relogin_failure_kind.py @@ -0,0 +1,41 @@ +"""登录记录增加自动重登的失败分型字段。 + +P1b 的自动重登不能把「失败」笼统对待：线上 415 个失败里，极验滑块（144）与 +凭证问题必须转人工，而取码失败（144）与中断（121）其实可自动重试 —— 不分型 +就会把 265 个可救名额一起推给人工。 + +这个字段落分型结果，供作业决定「继续退避重试」还是「转人工」；页面也按它分组 +展示待处理项。不要用 attempt_message 做字符串匹配：文案一改就失效。 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "20261002_0038" +down_revision: str | None = "20261002_0037" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + bind = op.get_bind() + columns = {item["name"] for item in sa.inspect(bind).get_columns("login_tasks")} + if "relogin_failure_kind" not in columns: + op.add_column( + "login_tasks", + sa.Column( + "relogin_failure_kind", + sa.String(length=16), + nullable=False, + server_default="", + ), + ) + + +def downgrade() -> None: + bind = op.get_bind() + columns = {item["name"] for item in sa.inspect(bind).get_columns("login_tasks")} + if "relogin_failure_kind" in columns: + op.drop_column("login_tasks", "relogin_failure_kind") diff --git a/web/backend/models.py b/web/backend/models.py index 695eac1..169236a 100644 --- a/web/backend/models.py +++ b/web/backend/models.py @@ -182,6 +182,11 @@ class LoginTask(Base): relogin_last_error: Mapped[str] = mapped_column( String(512), default="", server_default="" ) + # 失败分型：captcha / credential 转人工，email_code / email_quota / + # interrupted / network 自动重试（见规划 §7.4.3）。空 = 没失败过。 + relogin_failure_kind: Mapped[str] = mapped_column( + String(16), default="", server_default="" + ) account: Mapped[Account] = relationship("Account", back_populates="login_tasks") diff --git a/web/backend/routers/automation.py b/web/backend/routers/automation.py index 5d434e8..53f0149 100644 --- a/web/backend/routers/automation.py +++ b/web/backend/routers/automation.py @@ -12,11 +12,12 @@ from typing import Any from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel, Field +from sqlalchemy import and_, or_ from sqlalchemy.orm import Session from ..database import get_db from ..deps import require_permission -from ..models import AutomationJob, AutomationRun, User +from ..models import Account, AutomationJob, AutomationRun, LoginTask, User from ..services.audit_service import record_audit from ..services.automation import ( ensure_jobs, @@ -168,6 +169,83 @@ def run_job( return outcome +# 失败分型 → 页面展示用的中文说明。「转人工」的两种要写清为什么， +# 否则客服不知道该做什么。 +MANUAL_KIND_LABELS = { + "captcha": "极验滑块：需要人工过一次验证", + "credential": "凭证问题：需要核对账号密码 / 邮箱", + "": "重试次数用尽：需要人工确认账号是否还可用", +} + + +def _manual_kind_label(kind: str) -> str: + return MANUAL_KIND_LABELS.get(kind, f"待确认（{kind}）") + + +@router.get("/manual-queue") +def manual_queue( + limit: int = Query(default=20, ge=1, le=100), + db: Session = Depends(get_db), + current: User = Depends(require_permission("automation:view")), +): + """待人工处理的重登失败账号，按失败原因分组。 + + 自动重登只处理「还能自动救」的失败；极验滑块、凭证问题、以及重试次数用尽的 + 必须交给人 —— 这个接口就是那份清单（规划 §7.4.6）。 + """ + from ..services.automation.jobs.ck_relogin import MANUAL_KINDS + + # 转人工 = 分型属于需人工，或者次数用尽（next_run_at 已清空且失败过）。 + manual_filter = or_( + LoginTask.relogin_failure_kind.in_(tuple(MANUAL_KINDS)), + and_( + LoginTask.relogin_last_error != "", + LoginTask.relogin_next_at.is_(None), + LoginTask.relogin_attempts > 0, + ), + ) + rows = ( + db.query(LoginTask, Account.username) + .join(Account, LoginTask.account_id == Account.id) + .filter(LoginTask.ck_check_status == "invalid") + .filter(LoginTask.relogin_failure_kind != "") + .filter(manual_filter) + .order_by(LoginTask.attempt_at.desc(), LoginTask.id.desc()) + .all() + ) + grouped: dict[str, dict] = {} + for task, username in rows: + kind = task.relogin_failure_kind or "" + # 可重试的类型却已经没有下次排期 = 次数用尽，单独成组。 + # 否则它会混进「验证码失败」组里，客服看不出「这个已经不会再自动试了」。 + if kind not in MANUAL_KINDS: + kind = "" + group = grouped.setdefault( + kind, + { + "kind": kind, + "label": _manual_kind_label(kind), + "count": 0, + "samples": [], + }, + ) + group["count"] += 1 + if len(group["samples"]) < limit: + group["samples"].append( + { + "task_id": task.id, + "account_username": username, + "relogin_attempts": int(task.relogin_attempts or 0), + "last_error": task.relogin_last_error or task.attempt_message or "", + "attempt_at": task.attempt_at.isoformat() + if task.attempt_at + else None, + } + ) + groups = sorted(grouped.values(), key=lambda item: -item["count"]) + return {"total": len(rows), "groups": groups} + + @router.get("/runs") def list_runs( job_key: str | None = Query(default=None), diff --git a/web/backend/services/automation/__init__.py b/web/backend/services/automation/__init__.py index 66daaa3..45ac1ce 100644 --- a/web/backend/services/automation/__init__.py +++ b/web/backend/services/automation/__init__.py @@ -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, diff --git a/web/backend/services/automation/jobs/ck_relogin.py b/web/backend/services/automation/jobs/ck_relogin.py new file mode 100644 index 0000000..b399295 --- /dev/null +++ b/web/backend/services/automation/jobs/ck_relogin.py @@ -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 diff --git a/web/frontend/src/api/automation.ts b/web/frontend/src/api/automation.ts index 0ceb720..5ae06f8 100644 --- a/web/frontend/src/api/automation.ts +++ b/web/frontend/src/api/automation.ts @@ -1,18 +1,35 @@ -import api from './client'; +import api from "./client"; import type { AutomationJobItem, + AutomationManualQueue, AutomationJobUpdate, AutomationJobsResponse, AutomationRunOutcome, AutomationRunsResponse, -} from './types'; +} from "./types"; export const automationApi = { - jobs: () => api.get('/automation/jobs'), + jobs: () => + api.get("/automation/jobs"), updateJob: (key: string, payload: AutomationJobUpdate) => - api.patch(`/automation/jobs/${key}`, payload), + api.patch( + `/automation/jobs/${key}`, + payload, + ), runJob: (key: string) => - api.post(`/automation/jobs/${key}/run`), + api.post( + `/automation/jobs/${key}/run`, + ), runs: (params: { job_key?: string; limit?: number }) => - api.get('/automation/runs', { params }), + api.get( + "/automation/runs", + { params }, + ), + manualQueue: (limit = 20) => + api.get( + "/automation/manual-queue", + { + params: { limit }, + }, + ), }; diff --git a/web/frontend/src/api/types.ts b/web/frontend/src/api/types.ts index b6a046c..85cc4b1 100644 --- a/web/frontend/src/api/types.ts +++ b/web/frontend/src/api/types.ts @@ -174,7 +174,7 @@ export interface YybTask { created_by: number; created_by_username?: string; provider: string; - platform: 'android' | 'ios'; + platform: "android" | "ios"; points: number | null; price_fen?: number | null; product_id: string; @@ -185,7 +185,10 @@ export interface YybTask { status: string; phase: string; message: string; - result: { logs?: string[]; payment_status?: Record | null } | null; + result: { + logs?: string[]; + payment_status?: Record | null; + } | null; login_qr_data?: string; login_qr_mime_type?: string; payment_qr_data?: string; @@ -198,10 +201,22 @@ export interface YybTask { } export interface YybSelectionOptions { - products: Array<{ points: number; product_id: string; price_fen: number; offer_id: string; name: string }>; + products: Array<{ + points: number; + product_id: string; + price_fen: number; + offer_id: string; + name: string; + }>; zones: Array<{ zone_id: string; name: string }>; roles: Array<{ role_id: string; name: string; ban_status: string }>; - default_product: { points: number; product_id: string; price_fen: number; offer_id: string; name: string } | null; + default_product: { + points: number; + product_id: string; + price_fen: number; + offer_id: string; + name: string; + } | null; default_zone: { zone_id: string; name: string } | null; } @@ -280,7 +295,7 @@ export interface CookieOperationItem { ck_check_status: string; ck_checked_at: string | null; created_at: string | null; - relogin_status: '' | 'relogin_pending' | 'relogin_running' | 'relogin_failed'; + relogin_status: "" | "relogin_pending" | "relogin_running" | "relogin_failed"; relogin_message: string; relogin_batch_id: string; } @@ -379,7 +394,7 @@ export interface DouyuConfig { xpd_act_id: string; xpd_rid: string; gold_pay_type: number; - gold_recharge_channel: 'wechat_qr' | 'supplier_api'; + gold_recharge_channel: "wechat_qr" | "supplier_api"; gold_api_product_id: string; gold_api_account_template_name: string; gift_id: string; @@ -388,13 +403,13 @@ export interface DouyuConfig { } export interface DouyuRechargeChannel { - gold_recharge_channel: 'wechat_qr' | 'supplier_api'; + gold_recharge_channel: "wechat_qr" | "supplier_api"; } export interface DouyuTaskBatchRequest { account_ids: number[]; task_type: string; - handbook_scope: 'elite' | 'esports' | 'peace'; + handbook_scope: "elite" | "esports" | "peace"; concurrency?: number; payload?: Record; } @@ -413,7 +428,7 @@ export interface DouyuTaskItem { account_uid: string; account_nickname: string; task_type: string; - handbook_scope: 'elite' | 'esports' | 'peace' | 'legacy'; + handbook_scope: "elite" | "esports" | "peace" | "legacy"; status: string; message: string; result: Record | null; @@ -474,10 +489,10 @@ export interface HuyaFetchSmsCodeResult { message: string; } /** 工作台账号集合范围: elite=精英宝典 / browser_exchange=浏览器兑换 */ -export type HuyaWorkbenchScope = 'elite' | 'browser_exchange'; +export type HuyaWorkbenchScope = "elite" | "browser_exchange"; /** 浏览器兑换目标页: live_room=直播间 / elite=精英宝典 / esports=电竞宝典 */ -export type HuyaExchangeTargetKey = 'live_room' | 'elite' | 'esports'; +export type HuyaExchangeTargetKey = "live_room" | "elite" | "esports"; export interface HuyaExchangeTargetOption { key: HuyaExchangeTargetKey; @@ -592,7 +607,7 @@ export interface HuyaAutoRegisterRequest { export interface HuyaAutoRegisterRetryRequest { /** continue=从停止处继续（默认，不重试失败）；retry_failed=只重试失败；all_unfinished=失败+未完成 */ - mode?: 'continue' | 'retry_failed' | 'all_unfinished'; + mode?: "continue" | "retry_failed" | "all_unfinished"; concurrency?: number; wait_seconds?: number; poll_interval?: number; @@ -720,7 +735,7 @@ export interface HuyaConfig { export interface HuyaTaskBatchRequest { account_ids: number[]; task_type: string; - handbook_scope?: 'legacy' | 'elite'; + handbook_scope?: "legacy" | "elite"; concurrency?: number; payload?: Record; } @@ -754,7 +769,7 @@ export interface HuyaTaskItem { account_uid: string; account_nickname: string; task_type: string; - handbook_scope: 'legacy' | 'elite'; + handbook_scope: "legacy" | "elite"; status: string; message: string; result: Record | null; @@ -891,3 +906,23 @@ export interface AutomationRunOutcome { invalid: number; skipped: number; } + +export interface AutomationManualSample { + task_id: number; + account_username: string; + relogin_attempts: number; + last_error: string; + attempt_at: string | null; +} + +export interface AutomationManualGroup { + kind: string; + label: string; + count: number; + samples: AutomationManualSample[]; +} + +export interface AutomationManualQueue { + total: number; + groups: AutomationManualGroup[]; +} diff --git a/web/frontend/src/pages/AutomationPage.tsx b/web/frontend/src/pages/AutomationPage.tsx index 602bab3..2762c78 100644 --- a/web/frontend/src/pages/AutomationPage.tsx +++ b/web/frontend/src/pages/AutomationPage.tsx @@ -1,4 +1,4 @@ -import { useCallback, useEffect, useState } from 'react'; +import { useCallback, useEffect, useState } from "react"; import { Alert, Button, @@ -12,20 +12,26 @@ import { 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'; +} 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 AutomationManualGroup, + type AutomationManualSample, + type AutomationRunItem, +} from "../api/modules"; +import { usePermissions } from "../hooks/usePermissions"; +import { formatTime } from "../utils/time"; const STATUS_TAG: Record = { - success: { color: 'green', text: '成功' }, - failed: { color: 'red', text: '失败' }, - running: { color: 'blue', text: '执行中' }, - skipped: { color: 'default', text: '已跳过' }, + success: { color: "green", text: "成功" }, + failed: { color: "red", text: "失败" }, + running: { color: "blue", text: "执行中" }, + skipped: { color: "default", text: "已跳过" }, }; function secondsToHours(seconds: number): number { @@ -37,12 +43,14 @@ export default function AutomationPage() { const [schedulerEnabled, setSchedulerEnabled] = useState(true); const [schedulerRunning, setSchedulerRunning] = useState(false); const [runs, setRuns] = useState([]); + const [manualGroups, setManualGroups] = useState([]); + const [manualTotal, setManualTotal] = useState(0); const [loading, setLoading] = useState(false); const [savingKey, setSavingKey] = useState(null); const [runningKey, setRunningKey] = useState(null); const [forms] = Form.useForm(); const { can } = usePermissions(); - const canManage = can('automation:manage'); + const canManage = can("automation:manage"); const load = useCallback(async () => { setLoading(true); @@ -53,6 +61,9 @@ export default function AutomationPage() { setSchedulerRunning(data.scheduler.running); const history = await automationApi.runs({ limit: 50 }); setRuns(history.items); + const manual = await automationApi.manualQueue(); + setManualGroups(manual.groups); + setManualTotal(manual.total); // 统一回填表单：initialValue 只在首次渲染生效，刷新后不会跟着变。 forms.setFieldsValue( Object.fromEntries( @@ -84,7 +95,10 @@ export default function AutomationPage() { try { await automationApi.updateJob(job.key, { enabled: values.enabled, - interval_seconds: Math.max(Math.round((values.interval_hours ?? 6) * 3600), 300), + interval_seconds: Math.max( + Math.round((values.interval_hours ?? 6) * 3600), + 300, + ), batch_size: values.batch_size, concurrency: values.concurrency, }); @@ -112,41 +126,76 @@ export default function AutomationPage() { } }; + const manualColumns: ColumnsType = [ + { title: "记录", dataIndex: "task_id", width: 80 }, + { + title: "账号", + dataIndex: "account_username", + width: 160, + ellipsis: true, + }, + { title: "已重试", dataIndex: "relogin_attempts", width: 80 }, + { + title: "失败原因", + dataIndex: "last_error", + ellipsis: true, + render: (value: string) => value || "-", + }, + { + title: "最近尝试", + dataIndex: "attempt_at", + width: 160, + render: (value: string | null) => formatTime(value), + }, + ]; + const runColumns: ColumnsType = [ { - title: '开始时间', - dataIndex: 'started_at', + title: "开始时间", + dataIndex: "started_at", width: 180, render: (value: string | null) => formatTime(value), }, { - title: '作业', - dataIndex: 'job_key', + title: "作业", + dataIndex: "job_key", width: 120, }, { - title: '结果', - dataIndex: 'status', + title: "结果", + dataIndex: "status", width: 90, render: (value: string) => { - const tag = STATUS_TAG[value] ?? { color: 'default', text: value || '-' }; + const tag = STATUS_TAG[value] ?? { + color: "default", + text: value || "-", + }; return {tag.text}; }, }, - { title: '扫描', dataIndex: 'scanned', width: 80 }, - { title: '有效', dataIndex: 'valid', width: 80 }, + { title: "扫描", dataIndex: "scanned", width: 80 }, + { title: "有效", dataIndex: "valid", width: 80 }, { - title: '失效', - dataIndex: 'invalid', + title: "失效", + dataIndex: "invalid", width: 80, - render: (value: number) => (value > 0 ? {value} : 0), + render: (value: number) => + value > 0 ? ( + {value} + ) : ( + 0 + ), }, - { title: '耗时', width: 90, render: (_, row) => formatDuration(row.started_at, row.finished_at) }, - { title: '说明', dataIndex: 'message', ellipsis: true }, + { + title: "耗时", + width: 90, + render: (_, row) => formatDuration(row.started_at, row.finished_at), + }, + { title: "说明", dataIndex: "message", ellipsis: true }, ]; return ( - + {jobs.map((job) => ( - + {formatTime(job.last_run_at)} @@ -192,34 +246,36 @@ export default function AutomationPage() { {job.last_status ? ( - + {STATUS_TAG[job.last_status]?.text ?? job.last_status} ) : ( - '-' + "-" )} - {job.last_message || '-'} + + {job.last_message || "-"} +
saveJob(job)}> - + - + @@ -258,6 +314,38 @@ export default function AutomationPage() { ))} + + 自动重登救不了的类型（共 {manualTotal} 个） + + } + > + {manualGroups.length === 0 ? ( + + 暂无 —— 能自动重试的都会自己重试，需要人工的才出现在这里。 + + ) : ( + manualGroups.map((group) => ( + + + + )) + )} + +