diff --git a/docs/自动化与资料收集规划.md b/docs/自动化与资料收集规划.md index 7f4b7b1..d449f00 100644 --- a/docs/自动化与资料收集规划.md +++ b/docs/自动化与资料收集规划.md @@ -416,6 +416,16 @@ lookup_order(order_no) -> { 如果误把作业周期设成 `interval_seconds`，就变成「200 条/6 小时 = 800 条/天」，全库 12,881 条要 16 天才能过一轮 —— 等于没巡检（P1a 实现时踩过一次，已修）； - **排除**：`auto_managed = false`；`attempt_status in (pending, running)`（正在被批次处理）；最近 10 分钟内刚检测过（防重入）； - **动作**：调用现有 `check_douyu_cookie()`（两个只读接口，直连），写回 `ck_check_status / ck_check_result / ck_checked_at`； +- **⚠️ 三类判定（关键防护）**：检测失败不等于 CK 失效，结果必须分三类处理，否则斗鱼接口抖动一小时就能把上千个好账号标成失效、让客服误判成「CK 大面积挂掉」： + + | 判定 | 依据 | 处理 | + | --- | --- | --- | + | `valid` | 两个接口都通过 | 写 `ck_check_status='valid'` | + | `invalid` | 斗鱼**明确**回「未登录 / 登陆异常 / 请重新登陆」 | 写 `ck_check_status='invalid'` | + | `error` | 其余失败（`请求失败:` 超时、`响应异常`、未知结构、走异常兜底） | **不覆盖该行原结论**，只更新 `ck_check_result`（带 `verdict`）与 `ck_checked_at`，计入 `skipped` | + +- **自动暂停**：一轮里 `error` 占比 > 30% 且条数 ≥ 20 时，作业把自己 `enabled` 置 false 并在 `message` 写明原因（`detail.paused=true`）—— 大面积「查不到」通常是接口抖动或被风控，继续跑只会污染更多账号； + > 注意：**真实死号占比高不触发暂停**。首轮 100% `invalid` 是合法且预期的情况（选行是「从没检测过的优先」，积压的多是两个月前批量导入、从未登录的陈旧账号）。 - **并发**：默认 5，沿用现有 `ThreadPoolExecutor` 模式； - **不做**：不改 CK、不建重登批次、不动 `login_tasks.status`。 @@ -991,7 +1001,8 @@ P0 ──┬──> P2 ────┤ | --- | --- | --- | --- | | 只读查 order_site 打慢对方生产库 | 影响自动发货 | `statement_timeout=3s`、单行索引查询、2 条连接、本地缓存 | 关 `ORDER_SITE_LOOKUP_ENABLED` | | order_site 改表结构 | 适配层报错 | 只读 4 列且 88 个迁移零变更；SQL 集中在一个文件 | 同上（降级为手工） | -| **巡检本身触发斗鱼风控** | 检测接口被限流/封禁，误判大面积 invalid | 巡检走**只读接口**且不并发过高；失败率超阈值自动暂停；必要时给请求加随机抖动、降 `concurrency` | 关 `ck_health` 作业（回到人工检测） | +| **巡检把「查不到」当成「失效」** | 接口抖动/被风控时批量误标失效，客服误判 | 已做三类判定：只有斗鱼明确回「未登录」才写 invalid，其余只记 `error`、保留原结论；`error` 占比 > 30% 自动暂停作业 | 恢复接口后重新打开作业开关 | +| 巡检本身触发斗鱼风控 | 检测接口被限流/封禁，误判大面积 invalid | 巡检走**只读接口**且不并发过高；失败率超阈值自动暂停；必要时给请求加随机抖动、降 `concurrency` | 关 `ck_health` 作业（回到人工检测） | | 冷启动一次性全量扫 | 短时高并发被风控 | 首轮小批量（50）观察后再放量 | 调小 `batch_size` | | 自动重登烧掉邮箱配额 | 该账号当天无法登录 | 单账号退避到次日 + 每日上限 + 失败率熔断 | 关 `ck_relogin` 作业 | | 自动重登与人工操作撞车 | 状态互相覆盖 | `attempt_status` 在途判断 + 作业锁 | 关作业 | diff --git a/tests/test_automation_ck_health.py b/tests/test_automation_ck_health.py index 8cc416e..b594a82 100644 --- a/tests/test_automation_ck_health.py +++ b/tests/test_automation_ck_health.py @@ -183,7 +183,11 @@ class TestCkHealthNeverTouchesCookie(_Base): def test_invalid_result_marks_invalid_but_keeps_cookie(self): job = self._job() task = self.make_task(self.make_account("acc")) - ck_health.check_douyu_cookie = _fake_check(valid=False) + # 必须是鉴权类失败（斗鱼说未登录）才会被标成 invalid + ck_health.check_douyu_cookie = _fake_check( + valid=False, + message="鱼丸接口: 账户登陆异常，请重新登陆；等级接口: 用户未登录", + ) try: counters = ck_health.run_ck_health(self.session, job) finally: @@ -194,16 +198,23 @@ class TestCkHealthNeverTouchesCookie(_Base): assert task.cookie == "acf_uid=1" def test_detection_exception_does_not_abort_round(self): + """单个账号检测抛异常不影响整轮，但也不能把它标成失效。""" job = self._job() - self.make_task(self.make_account("a")) - self.make_task(self.make_account("b")) + first = self.make_task(self.make_account("a"), ck_check_status="valid") + second = self.make_task(self.make_account("b"), ck_check_status="valid") ck_health.check_douyu_cookie = _fake_check(raises=True) try: counters = ck_health.run_ck_health(self.session, job) finally: ck_health.check_douyu_cookie = _REAL_CHECK assert counters["scanned"] == 2 - assert counters["invalid"] == 2 + # 查不到不等于失效：两行都保留原结论，只记录这次异常 + assert counters["invalid"] == 0 + assert counters["skipped"] == 2 + for task in (first, second): + self.session.refresh(task) + assert task.ck_check_status == "valid" + assert task.ck_check_result["verdict"] == "error" def _job(self): return AutomationJob( @@ -363,13 +374,15 @@ def _aware(value): _REAL_CHECK = ck_health.check_douyu_cookie -def _fake_check(*, valid=True, raises=False): +def _fake_check(*, valid=True, raises=False, message=None): def fake(cookie: str) -> dict: if raises: raise RuntimeError("网络异常") return { "valid": valid, - "message": "ok" if valid else "Cookie 已失效", + "message": message + if message is not None + else ("ok" if valid else "Cookie 已失效"), "fish_ball": 1, "nickname": "测试", "level": 30, @@ -467,3 +480,104 @@ class TestAutomationApi(_Base): ) assert len(payload["items"]) == 1 assert payload["items"][0]["job_key"] == ck_health.JOB_KEY + + +class TestFailureClassification(_Base): + """「查不到」不等于「失效」：非鉴权类失败不能覆盖原来的结论。 + + 斗鱼接口抖动/被风控时，如果不加区分地写 invalid，一小时就能把上千个好账号 + 标成失效，客服会误判成 CK 大面积挂掉。这里盯三件事：分类正确、error 不改 + 结论、失败占比过高时自动暂停作业。 + """ + + def test_auth_failure_message_counts_as_invalid(self): + assert ( + ck_health.classify_result( + { + "valid": False, + "message": "鱼丸接口: 账户登陆异常，请重新登陆；等级接口: 用户未登录", + } + ) + == "invalid" + ) + + def test_network_failure_counts_as_error_not_invalid(self): + assert ( + ck_health.classify_result( + { + "valid": False, + "message": "鱼丸接口: 请求失败: timeout；等级接口: 请求失败: timeout", + } + ) + == "error" + ) + assert ( + ck_health.classify_result({"valid": False, "message": "响应异常"}) + == "error" + ) + + def test_valid_still_valid(self): + assert ck_health.classify_result({"valid": True, "message": "有效"}) == "valid" + + def test_error_keeps_previous_verdict_but_marks_checked(self): + job = self._job() + task = self.make_task(self.make_account("acc"), ck_check_status="valid") + ck_health.check_douyu_cookie = _fake_check( + valid=False, message="请求失败: timeout" + ) + try: + counters = ck_health.run_ck_health(self.session, job) + finally: + ck_health.check_douyu_cookie = _REAL_CHECK + self.session.refresh(task) + assert counters["valid"] == 0 and counters["invalid"] == 0 + assert counters["skipped"] == 1 # error 计入 skipped + assert task.ck_check_status == "valid" # 原结论保留 + assert task.ck_check_result["verdict"] == "error" # 但记录了这次查不到 + assert task.ck_checked_at is not None + + def test_mass_network_failure_pauses_the_job(self): + job = self._job() + job.enabled = True + job.batch_size = 40 + for index in range(40): + self.make_task(self.make_account(f"a{index}")) + ck_health.check_douyu_cookie = _fake_check( + valid=False, message="请求失败: Connection refused" + ) + try: + counters = ck_health.run_ck_health(self.session, job) + finally: + ck_health.check_douyu_cookie = _REAL_CHECK + assert counters["skipped"] == 40 + assert counters["detail"]["paused"] is True + assert job.enabled is False # 自动暂停，不再继续污染 + assert "已自动暂停" in counters["message"] + + def test_normal_invalid_does_not_pause(self): + """真的都是死号时不能暂停 —— 首轮 100% invalid 是合法且预期的情况。""" + job = self._job() + job.enabled = True + job.batch_size = 40 + for index in range(40): + self.make_task(self.make_account(f"a{index}")) + ck_health.check_douyu_cookie = _fake_check( + valid=False, + message="鱼丸接口: 账户登陆异常，请重新登陆；等级接口: 用户未登录", + ) + try: + counters = ck_health.run_ck_health(self.session, job) + finally: + ck_health.check_douyu_cookie = _REAL_CHECK + assert counters["invalid"] == 40 + assert counters["skipped"] == 0 + assert job.enabled is True + assert counters["detail"]["paused"] is False + + def _job(self): + return AutomationJob( + key=ck_health.JOB_KEY, + interval_seconds=6 * 3600, + batch_size=200, + concurrency=3, + ) diff --git a/web/backend/services/automation/jobs/ck_health.py b/web/backend/services/automation/jobs/ck_health.py index fa44f9b..c6d9663 100644 --- a/web/backend/services/automation/jobs/ck_health.py +++ b/web/backend/services/automation/jobs/ck_health.py @@ -23,6 +23,41 @@ from ...douyu_service import cookie_operable_filter JOB_KEY = "ck_health" FLUSH_EVERY = 25 +# 斗鱼明确表示「这个 CK 不能用」的说法。只有命中这些才算真的失效。 +AUTH_FAILURE_MARKERS = ( + "登陆异常", + "登录异常", + "未登录", + "登陆失效", + "登录失效", + "重新登陆", + "重新登录", + "账户登陆", +) + +# 非鉴权类失败（网络超时、响应异常、被判风控）占比超过它就自动暂停作业。 +# 这类失败说明「我们查不到」，不是「这个 CK 失效了」；如果不加区分地写 +# invalid，斗鱼接口抖动一小时就能把上千个好账号标成失效，客服会误判成 +# CK 大面积挂掉。 +ERROR_PAUSE_RATIO = 0.3 +ERROR_PAUSE_MIN_SCANNED = 20 + + +def classify_result(result: dict[str, Any]) -> str: + """把检测结果分成三类：valid / invalid（真失效）/ error（查不到）。 + + - ``valid``：斗鱼两个接口都通过； + - ``invalid``：斗鱼明确回复「未登录 / 登陆异常」之类的鉴权失败； + - ``error``：其余失败（请求失败、超时、响应异常、未知结构）—— + **不能当成失效**，也不覆盖该行原来的结论。 + """ + if result.get("valid"): + return "valid" + message = str(result.get("message") or "") + if any(marker in message for marker in AUTH_FAILURE_MARKERS): + return "invalid" + return "error" + def build_select_query(db: Session, job: AutomationJob, now: datetime): """构造本轮要检测记录的查询（与取行分开，便于按方言编译做回归）。 @@ -88,22 +123,27 @@ def run_ck_health(db: Session, job: AutomationJob) -> dict[str, Any]: results[task.id] = future.result() # 只写检测结果三件套；cookie / status / attempt_* 一律不动。 - valid = invalid = 0 + valid = invalid = errors = 0 for index, task in enumerate(tasks, start=1): result = results.get(task.id) if result is None: continue - is_valid = bool(result.get("valid")) - if is_valid: + verdict = classify_result(result) + if verdict == "valid": valid += 1 - else: + elif verdict == "invalid": invalid += 1 - task.ck_check_status = "valid" if is_valid else "invalid" + else: + errors += 1 + # error 时保留该行原来的结论：查不到不等于失效，宁可维持上一次判断。 + if verdict != "error": + task.ck_check_status = verdict task.ck_check_result = { "fish_ball": result.get("fish_ball"), "nickname": result.get("nickname"), "level": result.get("level"), "message": result.get("message", ""), + "verdict": verdict, } task.ck_checked_at = now if index % FLUSH_EVERY == 0: @@ -125,12 +165,30 @@ def run_ck_health(db: Session, job: AutomationJob) -> dict[str, Any]: ) .count() ) + error_ratio = errors / len(tasks) if tasks else 0.0 + paused = False + if errors >= ERROR_PAUSE_MIN_SCANNED and error_ratio > ERROR_PAUSE_RATIO: + # 大面积「查不到」通常是斗鱼接口抖动或被风控，继续跑只会污染更多账号。 + job.enabled = False + job.updated_at = datetime.now(UTC) + paused = True + message = f"本轮检测 {len(tasks)} 条，剩余待检 {remaining} 条" + if errors: + message += f"；非鉴权类失败 {errors} 条（{error_ratio:.0%}）" + if paused: + message += "；失败占比过高，已自动暂停作业" return { "scanned": len(tasks), "valid": valid, "invalid": invalid, - "skipped": 0, + "skipped": errors, "message": message, - "detail": {"remaining": remaining, "concurrency": concurrency}, + "detail": { + "remaining": remaining, + "concurrency": concurrency, + "errors": errors, + "error_ratio": round(error_ratio, 4), + "paused": paused, + }, }