feat(accounts): 增加高价值资产一键筛选与高级自定义筛选功能

1. 后端新增高价值套装状态筛选(双套均可换/可换白日梦/可换萌兔/已拥有/高价值待救/未查资料),兼容 MySQL 原生 JSON 与 SQLite unicode 转义存储;
2. 后端新增多维资产与状态区间筛选(积分范围、鱼翅范围、宝典开通、电竞开通、角色绑定状态),并支持与批量打标签/删除联动;
3. 统计接口 accounts_summary 新增 high_value 聚合数据,毫秒级返回各类高价值资产数量;
4. 前端账号管理页面新增「高价值一键筛选」胶囊区(带数量徽标与激活状态,即点即滤);
5. 前端新增可折叠「高级自定义筛选」面板,支持套装状态下拉、积分快捷区间(≥300、≥600)、鱼翅快捷区间(>0、≥6)、宝典与角色绑定状态多条件组合;
6. 补充 6 个高价值筛选与资产区间单元测试,全量用例通过。
This commit is contained in:
yml2213
2026-10-05 11:23:48 +08:00
parent 19f1cbceba
commit 512ea3db06
7 changed files with 967 additions and 18 deletions
+177
View File
@@ -1610,3 +1610,180 @@ class TestPeaceScopeCanBeTurnedOff:
marked, called = self._run(peace_enabled=True)
assert "xpd_records" in called
assert "exchange_records_peace" in marked["result"].get("errors", {})
class TestHighValueAccountFilters:
"""高价值账号筛选与自定义范围筛选测试。"""
def setup_method(self):
self.engine = get_test_engine()
self.session = make_test_session(self.engine)
self.admin = User(username="hv-admin", password_hash="h", role="super_admin")
self.session.add(self.admin)
self.session.commit()
# 构造不同特征的账号
now = datetime.now(UTC)
# 1. 优质皇冠号:双可换 + 积分 500 + 鱼翅 10 + 宝典开通 + CK valid
self.acc_crown = Account(
username="acc_crown",
password="p",
email="e1@test.com",
email_password="p",
overview_queried_at=now,
high_value_owned=[],
points=500,
gold_balance=10,
elite_opened=True,
ck_check_status="valid",
status="normal",
)
# 2. 换过白日梦(可换萌兔)+ 积分 100 + 鱼翅 0 + 宝典未开通
self.acc_bairimeng_owned = Account(
username="acc_bairimeng_owned",
password="p",
email="e2@test.com",
email_password="p",
overview_queried_at=now,
high_value_owned=["套装-兔兔白日梦"],
points=100,
gold_balance=0,
elite_opened=False,
ck_check_status="valid",
status="normal",
)
# 3. 换过萌兔(可换白日梦)+ 积分 50 + 鱼翅 20 + 宝典开通
self.acc_mengtu_owned = Account(
username="acc_mengtu_owned",
password="p",
email="e3@test.com",
email_password="p",
overview_queried_at=now,
high_value_owned=["套装-狂战萌兔"],
points=50,
gold_balance=20,
elite_opened=True,
ck_check_status="valid",
status="normal",
)
# 4. 双已换 + 高价值已空
self.acc_both_owned = Account(
username="acc_both_owned",
password="p",
email="e4@test.com",
email_password="p",
overview_queried_at=now,
high_value_owned=["套装-兔兔白日梦", "套装-狂战萌兔"],
points=0,
gold_balance=0,
elite_opened=False,
ck_check_status="valid",
status="normal",
)
# 5. 未查询资料的账号
self.acc_unqueried = Account(
username="acc_unqueried",
password="p",
email="e5@test.com",
email_password="p",
overview_queried_at=None,
high_value_owned=None,
points=None,
gold_balance=None,
elite_opened=None,
ck_check_status="valid",
status="normal",
)
# 6. 高价值但 CK 坏掉(待抢救)
self.acc_broken = Account(
username="acc_broken",
password="p",
email="e6@test.com",
email_password="p",
overview_queried_at=now,
high_value_owned=[],
points=800,
gold_balance=0,
elite_opened=True,
ck_check_status="invalid",
status="normal",
)
self.session.add_all(
[
self.acc_crown,
self.acc_bairimeng_owned,
self.acc_mengtu_owned,
self.acc_both_owned,
self.acc_unqueried,
self.acc_broken,
]
)
self.session.commit()
def _query(self, **kwargs):
from web.backend.routers.accounts import list_accounts
return list_accounts(
assigned_only=False,
has_cookie=False,
search="acc_",
page=1,
page_size=20,
db=self.session,
current=self.admin,
**kwargs,
)["items"]
def test_both_available_filter(self):
items = self._query(high_value_filter="both_available")
names = [it.username for it in items]
assert "acc_crown" in names
assert "acc_broken" in names
assert "acc_bairimeng_owned" not in names
assert "acc_mengtu_owned" not in names
assert "acc_both_owned" not in names
assert "acc_unqueried" not in names
def test_can_bairimeng_filter(self):
items = self._query(high_value_filter="can_bairimeng")
names = [it.username for it in items]
assert "acc_crown" in names
assert "acc_mengtu_owned" in names # 拥有萌兔但没拥有白日梦,所以可以换白日梦
assert "acc_bairimeng_owned" not in names
assert "acc_both_owned" not in names
def test_can_mengtu_filter(self):
items = self._query(high_value_filter="can_mengtu")
names = [it.username for it in items]
assert "acc_crown" in names
assert "acc_bairimeng_owned" in names # 拥有白日梦但没拥有萌兔,所以可以换萌兔
assert "acc_mengtu_owned" not in names
assert "acc_both_owned" not in names
def test_numeric_ranges_and_opened_filter(self):
# 积分 >= 300 且 鱼翅 >= 5
items = self._query(min_points=300, min_gold=5)
names = [it.username for it in items]
assert names == ["acc_crown"]
# 宝典已开通
items = self._query(elite_opened=True)
names = [it.username for it in items]
assert set(names) == {"acc_crown", "acc_mengtu_owned", "acc_broken"}
def test_high_value_broken_filter(self):
# 高价值但 CK 异常待抢救
items = self._query(high_value_filter="high_value_broken")
names = [it.username for it in items]
assert names == ["acc_broken"]
def test_summary_returns_high_value_breakdown(self):
from web.backend.routers.accounts import accounts_summary
res = accounts_summary(db=self.session, current=self.admin)
hv = res.get("high_value")
assert hv is not None
assert hv["both_available"] == 2 # acc_crown + acc_broken
assert hv["unqueried"] == 1 # acc_unqueried
assert hv["high_points"] == 2 # 500, 800
assert hv["high_value_broken"] == 1 # acc_broken
@@ -9,13 +9,11 @@ from web.backend.services.douyu_service import create_douyu_planned_tasks
def test_douyu_task_assigned_to_flow():
from tests.support.db import get_test_engine
init_test_schema()
DouyuTask.__table__.drop(get_test_engine(), checkfirst=True)
DouyuTask.__table__.create(get_test_engine(), checkfirst=True)
db: Session = make_test_session()
try:
db.query(DouyuTask).delete()
db.commit()
# 创建测试客服用户与账号
user_a = User(
username="test_kefu_a",