dachui/live-hub-py
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(douyu): 加固高价值三态迁移与状态口径

Browse Source 
修复迁移中断重跑、旧快照回填覆盖、降级将 unknown 误写成可兑换，以及迁移前 CK 已失效账号未作废旧结论的问题。统一 CK error 的展示与筛选口径，补齐旧快照迁移和兑换兜底反向用例，并同步高价值结论时间语义。同时纳入 .gitignore 更新。
This commit is contained in:
yml2213
2026-11-06 13:25:45 +08:00
parent c9fbbdb828
commit 578e259eea
13 changed files with 286 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_account_overview.py
+36 -4
View File
@@ -343,8 +343,8 @@ class TestOverviewAggregation:
def test_high_value_goods_are_flagged(self):
r = self._run().marked["result"]
goods = {g["name"]: g for g in r["high_value_goods"]}
assert goods["套装-兔兔白日梦"]["owned"] is True # 兑换记录里有
assert goods["套装-狂战萌兔"]["owned"] is False # 记录里没有
assert goods["套装-兔兔白日梦"]["owned"] is True # 官方 status=1
assert goods["套装-狂战萌兔"]["owned"] is False # 官方 status=0
assert "兔兔白日梦" in r["high_value_goods"][0]["name"]
def test_single_item_failure_does_not_fail_the_batch(self):
@@ -1920,7 +1920,7 @@ class TestHighValueAccountFilters:
assert "acc_no_mall" not in names
def test_high_value_never_filter(self):
"""「从未读到官方商城状态」这一档必须**单独可筛**。
"""「从未形成高价值结论」这一档必须**单独可筛**。
之前这批账号（商城接口失败 / 没配 manualID）在界面上无处可寻：
既不满足 ``unqueried``（资料查过），也不满足 ``high_value_unknown``
@@ -2222,6 +2222,37 @@ class TestHighValueTriState:
assert payload["state"] == "stale"
assert payload["stale"] is True
def test_ck_error_keeps_cached_conclusion_in_sync_with_filters(self):
"""CK 检测 error 只表示这次没查成，不改变账号结论或筛选口径。"""
from web.backend.services.account_value import (
high_value_unknown_expr,
suit_available_expr,
)
from web.backend.services.high_value_service import high_value_payload
acc = self._account(
high_value_bairimeng_status="available",
high_value_mengtu_status="available",
high_value_checked_at=datetime.now(UTC),
ck_check_status="error",
)
payload = high_value_payload(acc)
assert payload["state"] == "complete"
assert payload["stale"] is False
assert (
self.session.query(Account)
.filter(Account.id == acc.id, suit_available_expr("bairimeng"))
.count()
== 1
)
assert (
self.session.query(Account)
.filter(Account.id == acc.id, high_value_unknown_expr())
.count()
== 0
)
def test_stale_account_is_not_counted_as_available(self):
"""过期的账号不能进「双套均可兑换」筛选。"""
from web.backend.routers.accounts import accounts_summary
@@ -2397,9 +2428,10 @@ class TestHighValueAfterExchange:
self.account,
manual_id="m",
rid="r",
commodity_name="套装-狂兔萌兔",
commodity_name="套装-狂战萌兔",
)
assert self.account.high_value_bairimeng_status == "unknown"
assert self.account.high_value_mengtu_status == "owned"
assert any(level == "warning" for level, _ in runner.logs)
def test_unrelated_goods_name_is_ignored(self):
tests/test_migration_high_value_roundtrip.py
+138 -14
View File
@@ -1,6 +1,8 @@
"""0065 高价值三态迁移的 upgrade/downgrade 往返验证（真实 MySQL）。"""
import json
import secrets
from datetime import UTC, datetime
from urllib.parse import quote_plus
from alembic import command
@@ -31,9 +33,67 @@ def test_0065_upgrade_downgrade_roundtrip():
cfg = Config("alembic.ini")
cfg.set_main_option("script_location", "web/backend/migrations")
cfg.set_main_option("sqlalchemy.url", mig_url)
# 先在旧版本落真实数据，验证快照回填，而不是只在空库上验 DDL。
command.upgrade(cfg, "20261021_0064")
engine = create_engine(mig_url)
queried_at = datetime(2026, 11, 6, 12, 0, 0, tzinfo=UTC).replace(tzinfo=None)
snapshots = {
"snapshot-status": (
{
"queried_at": "2026-11-06T12:00:00+00:00",
"high_value_goods": [
{"name": "套装-兔兔白日梦", "status": 1},
{"name": "套装-狂战萌兔", "status": 0},
],
},
"valid",
),
"snapshot-no-status": (
{
"queried_at": "2026-11-05T10:00:00+00:00",
"high_value_goods": [
{"name": "套装-兔兔白日梦", "owned": True},
],
},
"valid",
),
"snapshot-invalid": (
{
"queried_at": "2026-11-04T09:00:00+00:00",
"high_value_goods": [
{"name": "套装-兔兔白日梦", "status": 1},
{"name": "套装-狂战萌兔", "status": 0},
],
},
"invalid",
),
"snapshot-missing": (None, None),
}
with engine.begin() as conn:
for username, (snapshot, ck_status) in snapshots.items():
conn.execute(
text(
"INSERT INTO accounts (username, password, email, email_password,"
" cookie, ck_check_status, overview_snapshot, overview_queried_at,"
" high_value_owned)"
" VALUES (:u, 'p', 'e@x.com', 'p', '', :ck, :snapshot, :queried,"
" :owned)"
),
{
"u": username,
"ck": ck_status or "",
"snapshot": json.dumps(snapshot)
if snapshot is not None
else None,
"queried": queried_at if snapshot is not None else None,
"owned": json.dumps([]) if snapshot is not None else None,
},
)
command.upgrade(cfg, "head")
insp = inspect(create_engine(mig_url))
insp = inspect(engine)
cols = {c["name"] for c in insp.get_columns("accounts")}
assert "high_value_bairimeng_status" in cols
assert "high_value_mengtu_status" in cols
@@ -43,21 +103,78 @@ def test_0065_upgrade_downgrade_roundtrip():
i["name"] for i in insp.get_indexes("accounts")
}
# 塞两行数据：owned/owned 与 owned/unknown，验证降级数组干净
engine = create_engine(mig_url)
with engine.connect() as conn:
rows = conn.execute(
text(
"SELECT username, high_value_bairimeng_status,"
" high_value_mengtu_status, high_value_checked_at"
" FROM accounts ORDER BY username"
)
).fetchall()
by_name = {row[0]: row for row in rows}
assert by_name["snapshot-status"][1:3] == ("owned", "available")
assert by_name["snapshot-status"][3] == queried_at
assert by_name["snapshot-no-status"][1:3] == ("unknown", "unknown")
assert by_name["snapshot-no-status"][3] == datetime(
2026, 11, 5, 10, 0, 0, tzinfo=UTC
).replace(tzinfo=None)
assert by_name["snapshot-invalid"][1:3] == ("unknown", "unknown")
assert by_name["snapshot-invalid"][3] == datetime(
2026, 11, 4, 9, 0, 0, tzinfo=UTC
).replace(tzinfo=None)
assert by_name["snapshot-missing"][1:3] == (None, None)
assert by_name["snapshot-missing"][3] is None
# 模拟“回填和索引已完成，但删旧列前进程中断”：版本仍停在 0064，
# 旧列还在。重跑必须能成功，且不能覆盖已经形成的新结论。
with engine.begin() as conn:
for i, (b, m) in enumerate([("owned", "owned"), ("owned", "unknown")]):
conn.execute(
text("ALTER TABLE accounts ADD COLUMN high_value_owned JSON NULL")
)
conn.execute(
text(
"UPDATE accounts SET high_value_bairimeng_status='available',"
" high_value_mengtu_status='unknown', high_value_checked_at=:now"
" WHERE username='snapshot-status'"
),
{
"now": datetime(2026, 11, 6, 13, 0, 0, tzinfo=UTC).replace(
tzinfo=None
)
},
)
conn.execute(text("UPDATE alembic_version SET version_num='20261021_0064'"))
command.upgrade(cfg, "head")
with engine.connect() as conn:
resumed = conn.execute(
text(
"SELECT high_value_bairimeng_status, high_value_mengtu_status"
" FROM accounts WHERE username='snapshot-status'"
)
).one()
assert resumed == ("available", "unknown")
# 降级无法表达 unknown；必须留 NULL，不能生成 [] 让旧版本误判为可兑换。
with engine.begin() as conn:
for index, (b, m) in enumerate(
[
("available", "available"),
("owned", "available"),
("owned", "owned"),
]
):
conn.execute(
text(
"INSERT INTO accounts (username, password, email, email_password,"
" cookie, high_value_bairimeng_status, high_value_mengtu_status)"
" VALUES (:u, 'p', 'e@x.com', 'p', '', :b, :m)"
" cookie, overview_queried_at, high_value_bairimeng_status,"
" high_value_mengtu_status)"
" VALUES (:u, 'p', 'e@x.com', 'p', '', :q, :b, :m)"
),
{"u": f"rt-{i}", "b": b, "m": m},
{"u": f"downgrade-{index}", "q": queried_at, "b": b, "m": m},
)
command.downgrade(cfg, "20261021_0064")
insp = inspect(create_engine(mig_url))
insp = inspect(engine)
cols = {c["name"] for c in insp.get_columns("accounts")}
assert "high_value_owned" in cols
assert "high_value_bairimeng_status" not in cols
@@ -65,19 +182,26 @@ def test_0065_upgrade_downgrade_roundtrip():
i["name"] for i in insp.get_indexes("accounts")
}
with engine.begin() as conn:
with engine.connect() as conn:
rows = conn.execute(
text(
"SELECT username, high_value_owned FROM accounts ORDER BY username"
)
).fetchall()
# JSON 列读出来是字符串；重点是里面**不能有 null**（旧实现会写出 [null,...]）
assert rows[0][1] == '["套装-兔兔白日梦", "套装-狂战萌兔"]', rows[0]
assert rows[1][1] == '["套装-兔兔白日梦"]', rows[1]
assert "null" not in rows[0][1] and "null" not in rows[1][1]
old_values = {
name: json.loads(value) if isinstance(value, str) else value
for name, value in rows
}
assert old_values["snapshot-status"] is None
assert old_values["snapshot-no-status"] is None
assert old_values["snapshot-invalid"] is None
assert old_values["snapshot-missing"] is None
assert old_values["downgrade-0"] == []
assert old_values["downgrade-1"] == ["套装-兔兔白日梦"]
assert old_values["downgrade-2"] == ["套装-兔兔白日梦", "套装-狂战萌兔"]
command.upgrade(cfg, "head")
insp = inspect(create_engine(mig_url))
insp = inspect(engine)
cols = {c["name"] for c in insp.get_columns("accounts")}
assert "high_value_owned" not in cols
assert "high_value_checked_at" in cols