"""账号生命周期状态：推导、收口写入与人工态切换。 ## 为什么需要「唯一推导函数」 账号的底层判据有三层 —— **有没有 CK**（`cookie`）、**CK 好不好** （`ck_check_status`）、**救不救得回来**（`relogin_*`）。`status` 是它们的 「人话摘要」，让客服一眼看出该处理谁、让列表能按状态筛选。 摘要一旦落库，就存在**两份真相**：底层判据 + 摘要。谁忘了同步，页面就撒谎 —— `accounts.bind_status` 正是这么烂掉的（36 种取值互相覆盖，最后没人知道绑定状态是啥）。 所以这里把推导收成**一个纯函数** `derive_status()`，所有写入点只调用 `sync_status()`（推导 + 变了才写），**不允许各处自己判断该写什么状态**。 ## 三层防线 1. **收口**：状态只由本模块写（`sync_status` / `set_manual_state`）； 2. **穷举单测**：`derive_status` 是真值表，全生命周期走查见 `test_account_lifecycle.py`； 3. **幂等**：状态没变时 `sync_status` 返回 False，**不写库、不发事件**。 ## 与 `auto_managed` 的分工 - `status=disabled`：**账号不可用**（封号、退单、归还）； - `auto_managed=False`：账号可用，但**自动化别碰**（敏感账号人工维护）。 两者都由「是否参与自动化」消费，但语义不同，所以保留两个开关。 """ from __future__ import annotations from datetime import UTC, datetime from typing import Any from loguru import logger from sqlalchemy.orm import Session from ..models import Account from .account_events_service import ( CATEGORY_AUTH, CATEGORY_LIFECYCLE, EVENT_CK_CLEARED, EVENT_CK_INVALID, EVENT_STATUS_CHANGED, emit, ) # ── 状态取值域（6 个）────────────────────────────────────── # 刻意**不做** `relogin`（退避重试中）：它完全等价于 # `invalid + relogin_next_at is not None`，独立成状态就多出一个必须与 # `relogin_next_at` 同步的字段 —— 正是本模块要消灭的东西。 STATUS_IMPORTED = "imported" # 没有 CK，跑不了业务 STATUS_ACTIVE = "active" # 有 CK 且未判失效 STATUS_INVALID = "invalid" # CK 判失效，还会自动救 STATUS_MANUAL = "manual" # CK 判失效，已放弃自动救，要人处理 STATUS_DISABLED = "disabled" # 人工停用 STATUS_ARCHIVED = "archived" # 人工归档（软删除） # 由底层判据推导出来的状态。 DERIVED_STATES: tuple[str, ...] = ( STATUS_IMPORTED, STATUS_ACTIVE, STATUS_INVALID, STATUS_MANUAL, ) # 人工设定的状态：优先级最高，`derive_status` **不会**覆盖它们。 MANUAL_STATES: tuple[str, ...] = (STATUS_DISABLED, STATUS_ARCHIVED) ALL_STATES: tuple[str, ...] = DERIVED_STATES + MANUAL_STATES # 参与自动化的状态：disabled / archived 一律跳过。 AUTOMATABLE_STATES: tuple[str, ...] = DERIVED_STATES # 重登失败分型 → 需要人工（自动重试只会加重风控 / 重试多少次都一样）。 # 放在这里而不是 job 里：`derive_status` 要用它，service 不该反过来依赖 job。 MANUAL_KINDS = frozenset({"captcha", "credential", "must_bind_mobile"}) STATUS_LABELS = { STATUS_IMPORTED: "未登录", STATUS_ACTIVE: "正常", STATUS_INVALID: "CK 失效", STATUS_MANUAL: "待人工", STATUS_DISABLED: "已停用", STATUS_ARCHIVED: "已归档", } def derive_status(account: Account) -> str: """按账号当前的底层判据推导应有的生命周期状态（**纯函数**）。 人工态（disabled / archived）优先级最高：即使账号还有 CK、CK 也还健康， 只要人说了停用/归档，就一直保持 —— 自动推导不能把人的决定冲掉。 """ if account.status in MANUAL_STATES: return account.status # 人工态不被覆盖 if not (account.cookie or ""): # 没有 CK 就跑不了业务：从未登录成功，或被清过 CK。 return STATUS_IMPORTED if (account.ck_check_status or "") != "invalid": # 有 CK，且没被判失效（含「还没检测过」）。 return STATUS_ACTIVE kind = account.relogin_failure_kind or "" if kind in MANUAL_KINDS: return STATUS_MANUAL # 极验 / 凭证问题，自动重试无意义 if (account.relogin_attempts or 0) > 0 and account.relogin_next_at is None: # 试过了、却没排下一次 —— 次数用尽，等于已放弃自动重登。 return STATUS_MANUAL return STATUS_INVALID def _label(status: str) -> str: return STATUS_LABELS.get(status, status or "未知") def _apply( db: Session, account: Account, target: str, *, reason: str = "", actor_type: str = "system", actor_id: int | None = None, occurred_at: datetime | None = None, ) -> bool: """真正的状态写入（唯一落点）：改字段 + 记事件。状态没变则什么都不做。""" previous = account.status if previous == target: return False now = occurred_at or datetime.now(UTC) account.status = target account.status_reason = (reason or "")[:64] account.status_changed_at = now if target == STATUS_DISABLED: account.disabled_at = now elif target == STATUS_ARCHIVED: account.archived_at = now else: account.disabled_at = None account.archived_at = None # 状态变迁进时间线：`status` 是覆盖式字段，覆盖之后就再也看不出 # 「什么时候从正常变成失效的」。 # # ⚠️ **刻意不传 `source_ref`**。一开始用了 # `status:{id}:{target}:{秒级时间戳}` 作幂等键，结果是：同一个账号在同一秒内 # 两次切到同一状态（active→invalid→active→…，甚至只是测试里时间被固定） # 就会撞唯一键，`emit` 静默吞掉异常 —— **状态改了、时间线却少一条**， # 正是本模块要消灭的那种「悄悄不一致」。 # # 而且这里根本不需要幂等键：`sync_status` 本身就是幂等的（状态没变直接返回 # False，压根不会走到这儿），所以不存在「同一次变迁被写两次」的路径。 emit( db, account_id=account.id, event_type=EVENT_STATUS_CHANGED, title=f"状态：{_label(previous)} → {_label(target)}", category=CATEGORY_LIFECYCLE, occurred_at=now, detail={ "from": previous or "", "to": target, "reason": reason or "", }, source="lifecycle", actor_type=actor_type, actor_id=actor_id, ) return True def sync_status( db: Session, account: Account, *, occurred_at: datetime | None = None, actor_type: str = "system", actor_id: int | None = None, ) -> bool: """按底层判据推导并写入状态。**返回状态是否真的变了**。 调用时机：任何改动 `cookie` / `ck_check_status` / `relogin_*` 之后。 幂等 —— 状态没变时既不写库也不发事件。 """ target = derive_status(account) if target == account.status: return False if account.id is None: # 还没落库（新建账号）—— 直接给字段，等 flush 后自然带上去。 account.status = target account.status_changed_at = occurred_at or datetime.now(UTC) return True return _apply( db, account, target, reason="", actor_type=actor_type, actor_id=actor_id, occurred_at=occurred_at, ) def set_manual_state( db: Session, account: Account, state: str, *, reason: str = "", actor_id: int | None = None, actor_type: str = "user", occurred_at: datetime | None = None, ) -> bool: """人工设定状态：`disabled` / `archived` / `auto`（恢复自动推导）。 传 `auto` 表示解除人工态，立刻按底层判据重新推导。 """ if state == "auto": account.status = STATUS_IMPORTED # 先把人工态摘掉，否则 derive 会原样返回 target = derive_status(account) if target == STATUS_IMPORTED: # 推导结果本来就是 imported：直接落字段，不必发「假」的变迁事件 account.status = STATUS_IMPORTED account.status_changed_at = occurred_at or datetime.now(UTC) account.disabled_at = None account.archived_at = None return True return _apply( db, account, target, reason=reason, actor_type=actor_type, actor_id=actor_id, occurred_at=occurred_at, ) if state not in MANUAL_STATES: raise ValueError(f"不支持的人工状态：{state}") # 从人工态切到另一个人工态时，previous 要按真实值取，事件才准 return _apply( db, account, state, reason=reason, actor_type=actor_type, actor_id=actor_id, occurred_at=occurred_at, ) def on_ck_verdict( db: Session, account: Account, verdict: str, *, detail: dict[str, Any] | None = None, occurred_at: datetime | None = None, actor_type: str = "system", actor_id: int | None = None, ) -> None: """一次 CK 检测结论落定后的统一处理（**顺序固定**，避免各处自己拼）。 1. 写 `ck_check_status` / `ck_check_result` / `ck_checked_at`； `error` = 「查不到」，**保留上一次结论** —— 查不到不等于失效； 2. **由好变坏**的那一刻记一条 `ck_invalid` 事件； 3. 重算生命周期状态。 第 2 步只认「转变」：巡检 6 小时一轮，如果每轮都记，一个失效一周的账号会攒下 28 条一模一样的「CK 检测失效」把时间线淹掉。判定失效是**一次状态转变**， 不是每轮读数。（旧实现注释里写着「避免每轮刷屏」，代码却没做到。） """ previous = account.ck_check_status or "" now = occurred_at or datetime.now(UTC) if verdict != "error": account.ck_check_status = verdict data = detail or {} account.ck_check_result = { "fish_ball": data.get("fish_ball"), "nickname": data.get("nickname"), "level": data.get("level"), "message": data.get("message", ""), "verdict": verdict, } account.ck_checked_at = now if verdict == "invalid" and previous != "invalid": emit( db, account_id=account.id, event_type=EVENT_CK_INVALID, title="CK 检测失效", category=CATEGORY_AUTH, occurred_at=now, detail={ "message": data.get("message", ""), "nickname": data.get("nickname"), }, source="login", actor_type=actor_type, actor_id=actor_id, ) sync_status(db, account, occurred_at=now, actor_type=actor_type, actor_id=actor_id) def on_login_success( db: Session, account: Account, *, occurred_at: datetime | None = None, ) -> None: """登录成功对账号意味着什么：**这一轮重登的退避痕迹作废，然后重算状态**。 只在这里定义一次，而不是散在 runner 与结算两处 —— 两处各写一遍就会出现 「谁先谁后」的差异。 人工手动登录也走这条，于是**「待人工」的账号被人工登录救回来时，状态会正确 回到正常**，而不是继续挂在 manual 上（`relogin_failure_kind` 清空后 `derive_status` 不再判它 manual）。 """ account.relogin_attempts = 0 account.relogin_next_at = None account.relogin_failure_kind = "" account.relogin_last_error = "" sync_status(db, account, occurred_at=occurred_at) def on_ck_cleared( db: Session, account: Account, *, occurred_at: datetime | None = None, actor_id: int | None = None, reason: str = "manual_clear", ) -> None: """清空账号的 CK：**CK 本体与健康状态一起清**，记一条 `ck_cleared`，再重算状态。 ⚠️ 这是**不可逆**操作（重登当前成功率极低，清掉很可能补不回来），所以时间线 必须留痕。单个清除与批量清除共用这一条路径 —— 以前是两处各写一遍， 改一处漏一处只是时间问题。 """ now = occurred_at or datetime.now(UTC) account.cookie = "" account.cookie_updated_at = now account.ck_check_status = "" account.ck_check_result = None account.ck_checked_at = None emit( db, account_id=account.id, event_type=EVENT_CK_CLEARED, title="CK 被清除", category=CATEGORY_AUTH, occurred_at=now, source="audit", actor_type="user" if actor_id else "system", actor_id=actor_id, detail={"reason": reason}, ) # 没有 CK 就跑不了业务 → 状态回到「未登录」 sync_status( db, account, occurred_at=now, actor_type="user" if actor_id else "system", actor_id=actor_id, ) def status_payload(account: Account) -> dict[str, Any]: """给接口用的状态片段（含中文标签，前端不必再维护一份映射）。""" return { "status": account.status or "", "status_label": _label(account.status or ""), "status_reason": account.status_reason or "", "status_changed_at": account.status_changed_at, } def warn_if_drift(account: Account) -> None: """漂移告警：落库状态与推导结果不一致时记一条 warning（不写库）。 作为「有人新增了底层判据写入点却忘了调 sync_status」的兜底观测手段， 由巡检作业顺手调用。真正的修复是补上 sync_status，不是在这里硬改。 """ expected = derive_status(account) if (account.status or "") != expected: logger.warning( f"账号状态漂移：account={account.id} 落库={account.status} 推导={expected}" )