diff --git a/tests/test_account_overview_refresh_job.py b/tests/test_account_overview_refresh_job.py new file mode 100644 index 0000000..5b13fc2 --- /dev/null +++ b/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, "样本不足时不该熔断" diff --git a/web/backend/services/automation/__init__.py b/web/backend/services/automation/__init__.py index 92a983a..12bcd41 100644 --- a/web/backend/services/automation/__init__.py +++ b/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, diff --git a/web/backend/services/automation/jobs/account_overview_refresh.py b/web/backend/services/automation/jobs/account_overview_refresh.py new file mode 100644 index 0000000..53cf7d7 --- /dev/null +++ b/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, + }