fix(automation): 巡检区分「查不到」与「失效」，并在大面积失败时自动暂停
P1a 原本把「检测没通过」一律写成 invalid。但 check_douyu_cookie 的失败分两类： 斗鱼明确回「未登录/登陆异常」（CK 真失效），以及请求超时/响应异常（我们查不到）。 一视同仁的后果是：斗鱼接口抖动或被风控时，每 5 分钟 200 个、一小时就能把两千多个 好账号标成失效，客服会误判成「CK 大面积挂掉」。规划里的失败率熔断只给了 P1b， P1a 漏了。 改动： - 新增 classify_result()：valid / invalid（命中斗鱼鉴权类说法）/ error（其余）； - error 时**不覆盖该行原来的 ck_check_status**，只更新 ck_check_result（带 verdict 字段）与 ck_checked_at，并计入本轮 skipped； - 一轮 error 占比 > 30% 且条数 >= 20 时自动把作业 enabled 置 false，message 与 detail.paused 写明原因，避免继续污染； - 真实死号占比高**不触发**暂停 —— 首轮 100% invalid 是预期情况（选行「从没检测过 的优先」，积压的是两个月前批量导入、attempt_status 为空、从未登录的陈旧账号， 生产实测 200 条错误信息一致为斗鱼侧「用户未登录」）。 新增 6 条测试：分类正确性、error 保留原结论、异常不再标失效、大面积失败自动暂停、 真实死号不暂停。全量 271 条通过。
This commit is contained in:
yml2213
+12
-1
@@ -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` 在途判断 + 作业锁 | 关作业 |
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user