diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 3ec2c54..7cde956 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -26,7 +26,7 @@ TESTS_DIR = Path(__file__).resolve().parent PROJECT_ROOT = TESTS_DIR.parent VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions" -HEAD_REVISION = "20261003_0041" +HEAD_REVISION = "20261003_0042" # 迁移新增、但不声明在模型里的复合查询索引。 EXTRA_INDEXES = ( @@ -36,6 +36,9 @@ EXTRA_INDEXES = ( "ix_login_tasks_status_finished_at_id", # 账号时间线按「账号 + 时间倒序」查,模型里不声明这个复合索引。 "ix_account_events_account_occurred", + # CK 搬到账号后,「在途尝试」只能留在登录记录上,巡检/重登每轮按 + # (account_id, attempt_status) 过滤 —— 模型里不声明这个索引。 + "ix_login_tasks_account_attempt_status", ) diff --git a/web/backend/crypto_storage.py b/web/backend/crypto_storage.py index eed80f1..b620a96 100644 --- a/web/backend/crypto_storage.py +++ b/web/backend/crypto_storage.py @@ -150,6 +150,8 @@ _SENSITIVE_COLUMNS: tuple[tuple[str, str], ...] = ( ("accounts", "password"), ("accounts", "email"), ("accounts", "email_password"), + # CK 已由 login_tasks 迁到账号上;两边都列上,回填后任一侧出现明文都能被兜住。 + ("accounts", "cookie"), ("login_tasks", "cookie"), ("huya_accounts", "account_password"), ("huya_accounts", "cookie"), diff --git a/web/backend/migrations/versions/20261003_0042_account_lifecycle.py b/web/backend/migrations/versions/20261003_0042_account_lifecycle.py new file mode 100644 index 0000000..33fa98c --- /dev/null +++ b/web/backend/migrations/versions/20261003_0042_account_lifecycle.py @@ -0,0 +1,332 @@ +"""账号全生命周期:把登录态与生命周期状态落到 accounts,并扩展账号事件台账。 + +## 为什么 + +1. **CK 不该寄存在登录记录上**。`login_tasks.cookie` 是账号的**登录态**,不是「某次登录的 + 产物」。删一条历史记录就等于删掉账号的登录态 —— 这是 2026-10-03 误删事故的根因。 +2. **CK 健康与重登进度** 同样挂在「碰巧最后一次登录那一行」上;一个账号有多行持 CK 时 + (实测存在)谁是权威并不明确。 +3. **生命周期状态根本不存在**,只能靠「有没有持 CK 的记录」反推。 +4. `account_events` 除了时间线,还要承担**业务台账**(开通/积分/兑换/充值), + 需要可检索的客体键与索引来支撑「365 天同皮肤」这类规则。 + +设计依据:`docs/设计-账号全生命周期.md`。 + +## 安全性(本迁移只加列、只读源表,不删不改任何旧数据) + +- **分块回填**:1.4 万账号、单账号 CK 约 1.7KB,一次性 UPDATE 会长时间持锁; +- **幂等**:只在 `accounts.cookie` 为空时写,可重跑;也给回填后新写入的 CK 让路; +- **自校验 fail closed**:回填不完整就让迁移抛错(宁可起不来,也不要带着半份 CK 上线); +- **值级校验**:不只比条数 —— 条数相等仍可能张冠李戴(A 账号拿到 B 账号的 CK)。 + +旧列本次**全部保留**,下一次发版再 drop。 +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "20261003_0042" +down_revision: str | None = "20261003_0041" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +# 一次回填多少账号。太小则往返多,太大则单事务持锁久。 +BACKFILL_CHUNK = 200 + +_ACCOUNT_COLUMNS: tuple[tuple[str, sa.types.TypeEngine, bool], ...] = ( + # ── 登录态:CK 本体 ── + ("cookie", sa.Text(), False), + ("cookie_updated_at", sa.DateTime(), True), + ("cookie_source", sa.String(length=16), False), + # ── 登录态:CK 健康 ── + ("ck_check_status", sa.String(length=16), False), + ("ck_check_result", sa.JSON(), True), + ("ck_checked_at", sa.DateTime(), True), + # ── 登录态:重登退避进度 ── + ("relogin_attempts", sa.Integer(), False), + ("relogin_next_at", sa.DateTime(), True), + ("relogin_last_error", sa.String(length=512), False), + ("relogin_failure_kind", sa.String(length=16), False), + # ── 生命周期状态 ── + ("status", sa.String(length=16), False), + ("status_reason", sa.String(length=64), False), + ("status_changed_at", sa.DateTime(), True), + ("disabled_at", sa.DateTime(), True), + ("archived_at", sa.DateTime(), True), +) + +# 列名 → server_default(NOT NULL 的列必须有,否则已有行无法满足约束) +# ⚠️ `cookie` 是 TEXT,**不能**有 server_default(MySQL 报 1101: +# BLOB/TEXT/GEOMETRY/JSON column can't have a default value)。 +# 已有行由 ALTER TABLE 隐式填 '',这里不配默认值即可。 +_COLUMN_DEFAULTS: dict[str, str] = { + "cookie_source": "", + "ck_check_status": "", + "relogin_attempts": "0", + "relogin_last_error": "", + "relogin_failure_kind": "", + "status": "imported", + "status_reason": "", +} + +_EVENT_COLUMNS: tuple[tuple[str, sa.types.TypeEngine, bool], ...] = ( + ("category", sa.String(length=16), False), + ("subject_type", sa.String(length=16), False), + ("subject_key", sa.String(length=96), False), + ("actor_type", sa.String(length=16), False), + ("actor_id", sa.Integer(), True), +) + + +def _tables(bind) -> set[str]: + return set(sa.inspect(bind).get_table_names()) + + +def _columns(bind, table: str) -> set[str]: + return {item["name"] for item in sa.inspect(bind).get_columns(table)} + + +def _indexes(bind, table: str) -> set[str]: + return {item["name"] for item in sa.inspect(bind).get_indexes(table)} + + +def _add_columns(bind, table: str, columns) -> None: + """幂等加列:已存在就跳过。""" + existing = _columns(bind, table) + for name, col_type, nullable in columns: + if name in existing: + continue + default = _COLUMN_DEFAULTS.get(name) + op.add_column( + table, + sa.Column( + name, + col_type, + nullable=nullable, + server_default=default if not nullable else None, + ), + ) + + +def _add_index(bind, table: str, name: str, columns: list[str]) -> None: + if name in _indexes(bind, table): + return + op.create_index(name, table, columns) + + +def _backfill_login_state(bind) -> None: + """把每个账号「最近一条持有 CK 的记录」的 CK 与状态搬到 accounts。 + + 只读 `login_tasks`,不删不改。分块执行。 + """ + from sqlalchemy import bindparam + + pairs = bind.execute( + sa.text( + """ + SELECT account_id, MAX(id) AS last_id + FROM login_tasks + WHERE cookie IS NOT NULL AND cookie <> '' + GROUP BY account_id + ORDER BY account_id + """ + ) + ).all() + if not pairs: + return + + account_ids = [int(row[0]) for row in pairs] + stmt = sa.text( + """ + UPDATE accounts a + JOIN ( + SELECT account_id, cookie, finished_at, ck_check_status, + ck_check_result, ck_checked_at, relogin_attempts, + relogin_next_at, relogin_last_error, relogin_failure_kind + FROM ( + SELECT lt.account_id, lt.cookie, lt.finished_at, lt.ck_check_status, + lt.ck_check_result, lt.ck_checked_at, lt.relogin_attempts, + lt.relogin_next_at, lt.relogin_last_error, + lt.relogin_failure_kind, + ROW_NUMBER() OVER ( + PARTITION BY lt.account_id ORDER BY lt.id DESC + ) AS rn + FROM login_tasks lt + WHERE lt.cookie IS NOT NULL AND lt.cookie <> '' + AND lt.account_id IN :ids + ) ranked + WHERE ranked.rn = 1 + ) src ON src.account_id = a.id + SET a.cookie = src.cookie, + a.cookie_updated_at = src.finished_at, + a.ck_check_status = COALESCE(src.ck_check_status, ''), + a.ck_check_result = src.ck_check_result, + a.ck_checked_at = src.ck_checked_at, + a.relogin_attempts = COALESCE(src.relogin_attempts, 0), + a.relogin_next_at = src.relogin_next_at, + a.relogin_last_error = COALESCE(src.relogin_last_error, ''), + a.relogin_failure_kind = COALESCE(src.relogin_failure_kind, '') + WHERE a.cookie IS NULL OR a.cookie = '' + """ + ).bindparams(bindparam("ids", expanding=True)) + + for start in range(0, len(account_ids), BACKFILL_CHUNK): + chunk = account_ids[start : start + BACKFILL_CHUNK] + bind.execute(stmt, {"ids": chunk}) + + +def _backfill_lifecycle_status(bind) -> None: + """按登录态推导生命周期状态。 + + 取值域见设计文档 §3.4。这里只用**已有的确定信息**推导,不做猜测: + 有 CK 且未判失效 → active;判失效则按重登进度分流;无 CK → imported。 + """ + bind.execute( + sa.text( + """ + UPDATE accounts + SET status = CASE + WHEN cookie IS NULL OR cookie = '' THEN 'imported' + WHEN ck_check_status = 'invalid' + AND relogin_failure_kind IN ('captcha', 'credential') + THEN 'manual' + WHEN ck_check_status = 'invalid' + AND relogin_next_at IS NOT NULL THEN 'relogin' + WHEN ck_check_status = 'invalid' THEN 'invalid' + ELSE 'active' + END, + status_changed_at = COALESCE(status_changed_at, updated_at, created_at) + WHERE status = 'imported' OR status IS NULL + """ + ) + ) + + +def _verify_backfill(bind) -> None: + """自校验:不完整 / 张冠李戴都直接抛错,让应用起不来(fail closed)。 + + `expected` 必须运行时算 —— 这个数在长(重登在跑)。 + """ + expected = int( + bind.execute( + sa.text( + """ + SELECT COUNT(DISTINCT account_id) FROM login_tasks + WHERE cookie IS NOT NULL AND cookie <> '' + """ + ) + ).scalar() + or 0 + ) + actual = int( + bind.execute( + sa.text( + "SELECT COUNT(*) FROM accounts WHERE cookie IS NOT NULL AND cookie <> ''" + ) + ).scalar() + or 0 + ) + if actual < expected: + raise RuntimeError( + f"CK 回填不完整:期望 {expected} 个账号,实际 {actual} 个。" + "中止迁移 —— 宁可起不来,也不要带着半份 CK 上线。" + ) + + mismatched = int( + bind.execute( + sa.text( + """ + SELECT COUNT(*) FROM accounts a + JOIN ( + SELECT account_id, cookie, + ROW_NUMBER() OVER ( + PARTITION BY account_id ORDER BY id DESC + ) AS rn + FROM login_tasks + WHERE cookie IS NOT NULL AND cookie <> '' + ) src ON src.account_id = a.id AND src.rn = 1 + WHERE a.cookie <> src.cookie + """ + ) + ).scalar() + or 0 + ) + if mismatched: + raise RuntimeError( + f"CK 值级校验失败:{mismatched} 个账号的 CK 与其最近一条持 CK 记录不一致," + "疑似张冠李戴。中止迁移。" + ) + + +def upgrade() -> None: + bind = op.get_bind() + + # 1. accounts 加列 + 索引 + _add_columns(bind, "accounts", _ACCOUNT_COLUMNS) + _add_index(bind, "accounts", "ix_accounts_ck_checked_at", ["ck_checked_at"]) + _add_index( + bind, + "accounts", + "ix_accounts_status_relogin_next", + ["status", "relogin_next_at"], + ) + _add_index(bind, "accounts", "ix_accounts_status", ["status"]) + + # 2. account_events 扩展(台账列 + 复合索引) + if "account_events" in _tables(bind): + _add_columns(bind, "account_events", _EVENT_COLUMNS) + _add_index( + bind, + "account_events", + "ix_account_events_ledger", + ["account_id", "event_type", "subject_key", "occurred_at"], + ) + + # 3. login_tasks 补「在途判定」索引:CK 搬到账号后,「在途」只能留在登录记录上, + # 巡检/重登每轮都要按 (account_id, attempt_status) 过滤。 + if "login_tasks" in _tables(bind): + _add_index( + bind, + "login_tasks", + "ix_login_tasks_account_attempt_status", + ["account_id", "attempt_status"], + ) + + # 4. 回填 + 自校验 + _backfill_login_state(bind) + _backfill_lifecycle_status(bind) + _verify_backfill(bind) + + +def downgrade() -> None: + bind = op.get_bind() + + for name in ( + "ix_login_tasks_account_attempt_status", + "ix_account_events_ledger", + "ix_accounts_status", + "ix_accounts_status_relogin_next", + "ix_accounts_ck_checked_at", + ): + table = ( + "login_tasks" + if name == "ix_login_tasks_account_attempt_status" + else ( + "account_events" if name == "ix_account_events_ledger" else "accounts" + ) + ) + if table in _tables(bind) and name in _indexes(bind, table): + op.drop_index(name, table_name=table) + + if "account_events" in _tables(bind): + existing = _columns(bind, "account_events") + for name, _type, _nullable in reversed(_EVENT_COLUMNS): + if name in existing: + op.drop_column("account_events", name) + + existing = _columns(bind, "accounts") + for name, _type, _nullable in reversed(_ACCOUNT_COLUMNS): + if name in existing: + op.drop_column("accounts", name) diff --git a/web/backend/models.py b/web/backend/models.py index d55001b..847e4ec 100644 --- a/web/backend/models.py +++ b/web/backend/models.py @@ -127,6 +127,66 @@ class Account(Base): UTCDateTime, default=_utcnow, onupdate=_utcnow ) + # ── 登录态:CK 本体(账号自己的凭证,不再寄存在 login_tasks 上)────────── + # 历史问题:CK 曾只存在于 login_tasks.cookie,删一条登录历史就等于删掉账号的 + # 登录态(2026-10-03 误删事故的根因)。CK 是**账号级凭证**,必须挂在账号上。 + # 注意:**不能给这列加 server_default** —— MySQL 不允许 TEXT/BLOB/JSON 列有默认值 + # (错误 1101)。已有行由 ALTER TABLE 隐式填 '',新建行由 Python 侧 default 兜。 + cookie: Mapped[str] = mapped_column(EncryptedText(), nullable=False, default="") + cookie_updated_at: Mapped[datetime | None] = mapped_column( + UTCDateTime, nullable=True + ) + # 这份 CK 是怎么来的:web / app / sms / import(空 = 历史回填,来源未知) + cookie_source: Mapped[str] = mapped_column( + String(16), nullable=False, default="", server_default="" + ) + + # ── 登录态:CK 健康(巡检/人工检测的当前结论)────────────────────────── + ck_check_status: Mapped[str] = mapped_column( + String(16), nullable=False, default="", server_default="" + ) # valid / invalid / error / 空 = 未检测 + ck_check_result: Mapped[dict | None] = mapped_column(JSON, nullable=True) + ck_checked_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True) + + # ── 登录态:自动重登的退避进度 ───────────────────────────────────────── + relogin_attempts: Mapped[int] = mapped_column( + Integer, nullable=False, default=0, server_default="0" + ) + relogin_next_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True) + relogin_last_error: Mapped[str] = mapped_column( + String(512), nullable=False, default="", server_default="" + ) + # 失败分型:captcha / credential 转人工,email_code / email_quota / + # interrupted / network 自动重试。空 = 没失败过。 + relogin_failure_kind: Mapped[str] = mapped_column( + String(16), nullable=False, default="", server_default="" + ) + + # ── 生命周期状态 ─────────────────────────────────────────────────────── + # 取值域见 docs/设计-账号全生命周期.md §3.4。**写入必须收口到 + # services/account_lifecycle.py**,禁止各处直接赋值 —— accounts.bind_status + # 就是「谁都能写、语义没人管」的反面教材(36 种取值互相覆盖)。 + status: Mapped[str] = mapped_column( + String(16), nullable=False, default="imported", server_default="imported" + ) + status_reason: Mapped[str] = mapped_column( + String(64), nullable=False, default="", server_default="" + ) + status_changed_at: Mapped[datetime | None] = mapped_column( + UTCDateTime, nullable=True + ) + disabled_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True) + archived_at: Mapped[datetime | None] = mapped_column(UTCDateTime, nullable=True) + + __table_args__ = ( + # 巡检按「最久没查的优先」选行 + Index("ix_accounts_ck_checked_at", "ck_checked_at"), + # 重登按状态 + 退避时间选行 + Index("ix_accounts_status_relogin_next", "status", "relogin_next_at"), + # 列表/统计按生命周期状态筛选 + Index("ix_accounts_status", "status"), + ) + assigned_user: Mapped[User | None] = relationship( "User", back_populates="assigned_accounts", foreign_keys=[assigned_to] ) @@ -729,6 +789,10 @@ class AccountEvent(Base): - `amount` 有符号(充值为正、兑换消耗为负),便于页面直接算累计; - `source_ref` 是幂等键,可空且唯一:收集器重跑、事件重复发射都靠它兜住; - `detail` 只存精简字段,不放原始大 JSON(那是 douyu_tasks 的包袱)。 + + 除了「时间线」,它还承担**业务台账**(开通 / 积分 / 兑换 / 充值记录)—— + 查询靠 `(account_id, event_type, subject_key, occurred_at)` 复合索引, + 所以**有流程的东西才另建表**(如兑换工单),「事实」一律进这里。 """ __tablename__ = "account_events" @@ -738,18 +802,48 @@ class AccountEvent(Base): # 前缀已覆盖按账号查,再单建一个纯属冗余。 account_id: Mapped[int] = mapped_column(ForeignKey("accounts.id"), nullable=False) event_type: Mapped[str] = mapped_column(String(32), nullable=False, index=True) + # 粗分类:lifecycle(状态变迁)/ auth(登录与 CK)/ business(业务)/ ops(人工) + category: Mapped[str] = mapped_column( + String(16), nullable=False, default="", server_default="" + ) occurred_at: Mapped[datetime] = mapped_column(UTCDateTime, nullable=False) title: Mapped[str] = mapped_column(String(256), default="", server_default="") amount: Mapped[int | None] = mapped_column(Integer, nullable=True) balance_after: Mapped[int | None] = mapped_column(Integer, nullable=True) + # 台账检索键:subject_type = commodity / skin / order / role, + # subject_key = 斗鱼 commodity_id / 归一后的 skin_key / 订单号。 + # 只放**需要检索**的键,其余细节进 detail —— 别把这里变成第二个 detail。 + subject_type: Mapped[str] = mapped_column( + String(16), nullable=False, default="", server_default="" + ) + subject_key: Mapped[str] = mapped_column( + String(96), nullable=False, default="", server_default="" + ) detail: Mapped[dict | None] = mapped_column(JSON, nullable=True) source: Mapped[str] = mapped_column(String(24), default="", server_default="") + # 谁触发的:user / system / public(收集链接) + actor_type: Mapped[str] = mapped_column( + String(16), nullable=False, default="", server_default="" + ) + actor_id: Mapped[int | None] = mapped_column(Integer, nullable=True) # 幂等键:可空且唯一(MySQL 允许多个 NULL,便于「没有幂等键」的事件共存) source_ref: Mapped[str | None] = mapped_column( String(96), nullable=True, index=True, unique=True ) created_at: Mapped[datetime] = mapped_column(UTCDateTime, default=_utcnow) + __table_args__ = ( + # 台账查询:「该账号该事件类型该客体最近一次」——365 天同皮肤规则靠它。 + # 跨账号反查(「这件皮肤被谁兑过」)吃 event_type + subject_key 前缀。 + Index( + "ix_account_events_ledger", + "account_id", + "event_type", + "subject_key", + "occurred_at", + ), + ) + class AutomationJob(Base): """自动化作业定义。状态落库,进程重启不丢调度进度。"""