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

完善虎牙外部单展示、状态恢复与礼包配置

Browse Source
This commit is contained in:
yml2213
2026-11-11 12:40:06 +08:00
parent 3a48d9c624
commit b18e46d72d
18 changed files with 1216 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_huya_collab_orders.py
+390 -7
View File
@@ -30,7 +30,15 @@ from web.backend.models import (
)
from web.backend.permissions import ROLE_PERMISSIONS
from web.backend.routers.huya import collab as collab_api
from web.backend.schemas import HuyaExternalOrderListOut, HuyaExternalOrderUpdate
from web.backend.schemas import (
HuyaCollabCkPreviewOut,
HuyaConfigOut,
HuyaExternalOrderCreate,
HuyaExternalOrderListOut,
HuyaExternalOrderOut,
HuyaExternalOrderStatusBatchRequest,
HuyaExternalOrderUpdate,
)
from web.backend.services import huya_collab_service as collab
from web.backend.services.huya_collab_events import collab_events
@@ -437,6 +445,64 @@ class TestCollabStatusMachine(TestCollabBase):
collab.set_status(self.db, self.user, order, "incomplete", rollback=True)
assert excinfo.value.code == "status_from_void"
def test_super_admin_can_force_recover_void_order(self):
"""作废默认不可逆；超级管理员可凭原因强制恢复，并保留审计证据。"""
order = self._bind_ready(self._create(order_no="VOID-FORCE-1"))
collab.set_status(self.db, self.user, order, "void", reason="客户退款")
admin = User(username="force-admin", password_hash="h", role="super_admin")
self.db.add(admin)
self.db.commit()
collab.set_status(
self.db,
admin,
order,
collab.STATUS_TO_RECHARGE,
reason="误作废，客户要求继续处理",
force=True,
)
assert order.status == collab.STATUS_TO_RECHARGE
assert order.void_reason == ""
audit = [
row for row in self._audits(order) if row.action == "huya:collab:status"
][-1]
detail = json.loads(audit.detail)
assert detail["force"] is True
assert detail["force_reason"] == "误作废，客户要求继续处理"
assert detail["previous_void_reason"] == "客户退款"
def test_force_recover_void_requires_super_admin_and_reason(self):
order = self._bind_ready(self._create(order_no="VOID-FORCE-2"))
collab.set_status(self.db, self.user, order, "void", reason="客户退款")
with pytest.raises(collab.CollabError) as excinfo:
collab.set_status(
self.db,
self.user,
order,
collab.STATUS_TO_RECHARGE,
reason="普通运营尝试恢复",
force=True,
)
assert excinfo.value.code == "force_admin_required"
assert excinfo.value.status_code == 403
admin = User(username="force-admin-2", password_hash="h", role="super_admin")
self.db.add(admin)
self.db.commit()
with pytest.raises(collab.CollabError) as excinfo:
collab.set_status(
self.db,
admin,
order,
collab.STATUS_TO_RECHARGE,
force=True,
)
assert excinfo.value.code == "field_required"
assert "强制恢复原因" in excinfo.value.detail
assert order.status == "void"
def test_same_status_is_noop(self):
order = self._bind_ready(self._create())
before = len(self._audits(order))
@@ -493,8 +559,10 @@ class TestCollabCkLifecycle(TestCollabBase):
def test_pasting_new_ck_resets_verification(self):
order = self._create(ck_text=fake_ck("1001"))
collab.set_ck_status(self.db, self.user, order, "valid")
order.nickname = "旧虎牙昵称"
collab.set_ck(self.db, self.user, order, fake_ck("1002"))
assert order.uid == "1002"
assert order.nickname == ""
assert order.ck_status == "unverified"
assert order.ck_checked_at is None
@@ -804,14 +872,20 @@ def points_center_blob(
stock: int = 39,
remaining: int | None = None,
category: str = "礼包",
category_id: int = 431,
extra_tabs: list[dict] | None = None,
) -> bytes:
"""构造 queryAllocation 的 ``tag5`` 最小商品列表（TAF LIST）。"""
os = TafOutputStream()
os.write_head(0, TafType.LIST)
os.write_int32(0, 1)
extra_tabs = list(extra_tabs or [])
os.write_int32(0, 1 + len(extra_tabs))
# 分类
os.write_head(0, TafType.STRUCT_BEGIN)
os.write_int32(0, category_id)
os.write_string(1, category)
os.write_int32(3, 2)
os.write_int32(5, 0)
os.write_head(6, TafType.LIST)
os.write_int32(0, 1)
# 商品
@@ -841,7 +915,16 @@ def points_center_blob(
os.write_struct_end() # tag0
os.write_struct_end() # tag1
os.write_struct_end() # 商品
os.write_int32(7, 1)
os.write_struct_end() # 分类
for tab in extra_tabs:
os.write_head(0, TafType.STRUCT_BEGIN)
os.write_int32(0, int(tab.get("tab_id") or 0))
os.write_string(1, str(tab.get("name") or ""))
os.write_int32(3, 2)
os.write_int32(5, 0)
os.write_int32(7, int(tab.get("count") or 0))
os.write_struct_end()
return os.get_bytes()
@@ -881,14 +964,22 @@ class FakeProbeClient:
"""离线探针客户端：按脚本依次给出 ``None`` / 异常 / 响应，并记录调用参数。"""
def __init__(
self, *outcomes, gifts: list | None = None, points_gifts: list | None = None
self,
*outcomes,
gifts: list | None = None,
points_gifts: list | None = None,
profile_nick: str | None = None,
profile_outcome: object | None = None,
):
self.outcomes = list(outcomes)
self.gift_outcomes = list(gifts or [])
self.point_outcomes = list(points_gifts or [])
self.profile_nick = profile_nick
self.profile_outcome = profile_outcome
self.calls: list[dict] = []
self.gift_calls: list[dict] = []
self.points_calls: list[dict] = []
self.profile_calls: list[dict] = []
self._current_bind_outcome: object | None = None
def check_user_bind_game_account(self, **kwargs):
@@ -926,6 +1017,19 @@ class FakeProbeClient:
raise outcome
return outcome
def get_user_profile_batch(self, **kwargs):
self.profile_calls.append(kwargs)
if self.profile_outcome is not None:
if isinstance(self.profile_outcome, Exception):
raise self.profile_outcome
return self.profile_outcome
if self.profile_nick is None:
return None
target_uid = int((kwargs.get("target_uids") or [0])[0] or 0)
return SimpleNamespace(
profiles=[SimpleNamespace(uid=target_uid, nick=self.profile_nick)]
)
class TestCollabCkProbe(TestCollabBase):
"""S5：只读探针。判据（真机对照 2026-10-16）= 响应非 None 且 status == 200。"""
@@ -942,7 +1046,8 @@ class TestCollabCkProbe(TestCollabBase):
role_id="1234567",
plat="手Q",
area="苹果(iOS)",
)
),
profile_nick="暖阳冰笑月",
)
result = collab.verify_ck(self.db, self.user, order, client=client)
@@ -952,9 +1057,31 @@ class TestCollabCkProbe(TestCollabBase):
assert order.ck_checked_at is not None
assert "随遇而安" in order.ck_check_message
assert "吃鸡199706" in order.ck_check_message
assert order.nickname == "暖阳冰笑月"
assert result["nickname"] == "暖阳冰笑月"
assert "暖阳冰笑月" in order.ck_check_message
assert client.profile_calls[0]["target_uids"] == [1199647239697]
assert order.version == 1
assert "huya:collab:ck_status" in [row.action for row in self._audits(order)]
def test_profile_failure_does_not_invalidate_cookie(self):
"""虎牙昵称是展示辅助：资料接口失败不能把已经通过的 CK 主结论判失效。"""
order = self._row_with_ck()
client = FakeProbeClient(
FakeBindResp(nick="7", role="小凛饼干_"),
profile_outcome=RuntimeError("profile boom"),
)
result = collab.verify_ck(self.db, self.user, order, client=client)
assert result["ok"] is True
assert order.ck_status == "valid"
assert order.bind_game_account == "7"
assert order.bind_game_role == "小凛饼干_"
assert order.nickname == ""
assert "虎牙昵称未读到" in order.ck_check_message
assert "RuntimeError" in order.ck_check_message
def test_verify_uses_this_rows_ck_and_the_default_activity(self):
"""探针必须拿**本行的** CK 去问默认活动，不能借用别的行/别的活动。"""
order = self._row_with_ck(uid="1199647239697")
@@ -1081,7 +1208,7 @@ class TestCollabCkProbe(TestCollabBase):
"bind_game_role": "吃鸡199706",
"bind_game_channel": "手Q · 苹果(iOS)",
}
# 人工填的列一个都不碰；nickname 是「虎牙号昵称」，这个探针给不出
# 人工填的列一个都不碰；这台假客户端没给虎牙资料，昵称保持空
assert order.bind_douyu_account == ""
assert order.nickname == ""
assert "huya:collab:probe_snapshot" in [
@@ -1332,7 +1459,8 @@ class TestCollabCkPreview(TestCollabBase):
client = FakeProbeClient(
FakeBindResp(
nick="随遇而安", role="吃鸡199706", plat="手Q", area="苹果(iOS)"
)
),
profile_nick="暖阳冰笑月",
)
preview = collab.preview_ck(fake_ck("1199647239697"), client=client)
@@ -1341,7 +1469,11 @@ class TestCollabCkPreview(TestCollabBase):
assert preview["has_role"] is True
assert preview["uid"] == "1199647239697"
assert preview["hy_username"] == "hy_1199647239697"
assert preview["nickname"] == "暖阳冰笑月"
assert preview["bind_game_role"] == "吃鸡199706"
# 接口响应模型必须真的序列化昵称，避免 Pydantic 静默丢字段。
dumped = HuyaCollabCkPreviewOut.model_validate(preview).model_dump()
assert dumped["nickname"] == "暖阳冰笑月"
assert preview["gifts"] == [
{
"product_id": "hy-1001",
@@ -1455,6 +1587,100 @@ class TestCollabGifts(TestCollabBase):
]
assert message == "已拉取 1 个礼包（积分中心）"
def test_points_center_gift_fetch_loads_all_discovered_tabs(self):
"""基础响应只给第一个页签；其余页签按 pageId + tabId 自动补拉。"""
base_payload = {
"tag2": [
{
"tag0": 343,
"tag5": points_center_blob(
name="基础礼包",
product_id="hy-1001",
points=200,
category="基础礼包",
extra_tabs=[{"tab_id": 433, "name": "限量返场", "count": 1}],
),
}
]
}
tab_payload = {
"tag2": [
{
"tag0": 343,
"tag5": points_center_blob(
name="套装-甜橙豚豚",
product_id="hy-9400194",
points=988,
category="限量返场",
category_id=433,
),
}
]
}
client = FakeProbeClient(
points_gifts=[
FakePointsCenterResp(payload=base_payload),
FakePointsCenterResp(payload=tab_payload),
]
)
gifts, message = collab.fetch_points_center_gifts(
1199647239697,
fake_ck("1199647239697"),
client=client,
page_code="custom_page",
source_id="custom_source",
scene=8,
)
assert [gift["name"] for gift in gifts] == ["基础礼包", "套装-甜橙豚豚"]
assert client.points_calls[0]["page_code"] == "custom_page"
assert client.points_calls[0]["source_id"] == "custom_source"
assert client.points_calls[0]["scene"] == 8
assert "waterfall_tab_id" not in client.points_calls[0]
assert client.points_calls[1]["page_id"] == 343
assert client.points_calls[1]["waterfall_tab_id"] == 433
assert "2 个页签" in message
def test_points_center_tab_override_uses_configured_ids(self):
base_payload = {
"tag2": [
{
"tag0": 343,
"tag5": points_center_blob(name="基础礼包", product_id="hy-1001"),
}
]
}
tab_payload = {
"tag2": [
{
"tag0": 343,
"tag5": points_center_blob(
name="手工页签礼包", product_id="hy-2002"
),
}
]
}
client = FakeProbeClient(
points_gifts=[
FakePointsCenterResp(payload=base_payload),
FakePointsCenterResp(payload=tab_payload),
]
)
gifts, message = collab.fetch_points_center_gifts(
1199647239697,
fake_ck("1199647239697"),
client=client,
page_id=343,
tab_ids=[999],
)
assert [gift["name"] for gift in gifts] == ["基础礼包", "手工页签礼包"]
assert client.points_calls[1]["page_id"] == 343
assert client.points_calls[1]["waterfall_tab_id"] == 999
assert "2 个页签" in message
def test_points_center_gift_fetch_reports_failure(self):
gifts, message = collab.fetch_points_center_gifts(
1199647239697,
@@ -1531,6 +1757,41 @@ class TestCollabGifts(TestCollabBase):
self.db.commit()
assert collab.resolve_gifts_sid(self.db) == collab.GIFTS_SID
def test_resolve_points_center_config_reads_manual_overrides(self):
self.db.add(
HuyaConfig(
points_page_code="custom_page",
points_source_id="custom_source",
points_scene="8",
points_page_id="343",
points_tab_ids="431, 433，999",
)
)
self.db.commit()
assert collab.resolve_points_center_config(self.db) == {
"page_code": "custom_page",
"source_id": "custom_source",
"scene": 8,
"page_id": 343,
"tab_ids": [431, 433, 999],
}
def test_config_out_serializes_points_center_fields(self):
dumped = HuyaConfigOut(
points_page_code="custom_page",
points_source_id="custom_source",
points_scene="8",
points_page_id="343",
points_tab_ids="431,433",
).model_dump()
assert dumped["points_page_code"] == "custom_page"
assert dumped["points_source_id"] == "custom_source"
assert dumped["points_scene"] == "8"
assert dumped["points_page_id"] == "343"
assert dumped["points_tab_ids"] == "431,433"
def test_preview_ignores_legacy_sid_and_uses_points_center(self):
self.db.add(HuyaConfig(sid="2277"))
self.db.commit()
@@ -1670,7 +1931,8 @@ class TestCollabCreateStrict(TestCollabBase):
client=FakeProbeClient(
FakeBindResp(
nick="随遇而安", role="吃鸡199706", plat="手Q", area="苹果(iOS)"
)
),
profile_nick="暖阳冰笑月",
),
)
@@ -1679,6 +1941,7 @@ class TestCollabCreateStrict(TestCollabBase):
assert order.bind_game_account == "随遇而安"
assert order.bind_game_role == "吃鸡199706"
assert order.bind_game_channel == "手Q · 苹果(iOS)"
assert order.nickname == "暖阳冰笑月"
assert order.bind_snapshot_at is not None
assert order.gift_name == "不朽伊人大礼包"
assert order.home_images == [self.HOME_IMAGE, "collab/2026/10/16/home02.png"]
@@ -1718,6 +1981,79 @@ class TestCollabCreateStrict(TestCollabBase):
assert '"draft"' in detail and "true" in detail
assert collab.summary(self.db)["by_status"]["incomplete"] == 1
def test_draft_with_ck_auto_fills_snapshot_and_nickname(self):
"""草稿单带了 CK 也自动验证，不用建完再手动点一次「验证 CK」。"""
order = collab.create_order(
self.db,
self.user,
order_no="DRAFT-AUTO-1",
shop="佳佳",
ck_text=fake_ck("1199661291272"),
draft=True,
auto_verify_ck=True,
client=FakeProbeClient(
FakeBindResp(nick="7", role="小凛饼干_", plat="苹果(iOS)", area="手Q"),
profile_nick="暖阳冰笑月",
),
)
assert order.status == collab.STATUS_INCOMPLETE
assert order.ck_status == "valid"
assert order.ck_checked_at is not None
assert order.bind_game_account == "7"
assert order.bind_game_role == "小凛饼干_"
assert order.bind_game_channel == "苹果(iOS) · 手Q"
assert order.nickname == "暖阳冰笑月"
assert order.bind_snapshot_at is not None
def test_draft_with_bad_ck_still_creates_and_exposes_failure(self):
"""自动探针失败不阻止草稿建单，但必须落成「已失效」并写明原因。"""
order = collab.create_order(
self.db,
self.user,
order_no="DRAFT-AUTO-BAD",
shop="佳佳",
ck_text=fake_ck("1199661291272"),
draft=True,
auto_verify_ck=True,
client=FakeProbeClient(FakeBindResp(status=401, msg="登录态失效")),
)
assert order.status == collab.STATUS_INCOMPLETE
assert order.ck_status == "invalid"
assert order.ck_checked_at is not None
assert "401" in order.ck_check_message
assert order.bind_game_account == ""
assert order.bind_game_role == ""
assert order.nickname == ""
def test_create_route_auto_verifies_draft_ck(self, monkeypatch):
"""新建接口是用户实际入口：草稿单带 CK 也必须自动补全并返回新字段。"""
client = FakeProbeClient(
FakeBindResp(nick="7", role="小凛饼干_", plat="苹果(iOS)", area="手Q"),
profile_nick="暖阳冰笑月",
)
monkeypatch.setattr(collab, "_build_probe_client", lambda: client)
out = collab_api.create_external_order(
HuyaExternalOrderCreate(
order_no="DRAFT-ROUTE-AUTO",
shop="佳佳",
ck_text=fake_ck("1199661291272"),
draft=True,
),
db=self.db,
current=self.user,
)
assert out["ck_status"] == "valid"
assert out["bind_game_account"] == "7"
assert out["bind_game_role"] == "小凛饼干_"
assert out["nickname"] == "暖阳冰笑月"
dumped = HuyaExternalOrderOut.model_validate(out).model_dump()
assert dumped["nickname"] == "暖阳冰笑月"
assert dumped["bind_game_role"] == "小凛饼干_"
def test_incomplete_cannot_advance_until_customer_info_complete(self):
order = collab.create_order(
self.db, self.user, order_no="DRAFT-2", shop="佳佳", draft=True
@@ -2426,6 +2762,53 @@ class TestCollabStatusBatch(TestCollabBase):
assert accepted["updated"] == 1
assert order.status == collab.STATUS_TO_RECHARGE
def test_batch_force_recovers_void_order_for_super_admin(self):
order = self._bind_ready(self._create(order_no="BATCH-VOID-FORCE"))
collab.set_status(self.db, self.user, order, "void", reason="误操作")
admin = User(
username="batch-force-admin", password_hash="h", role="super_admin"
)
self.db.add(admin)
self.db.commit()
result = collab.set_status_batch(
self.db,
admin,
[order],
collab.STATUS_TO_RECHARGE,
reason="批量恢复误作废订单",
force=True,
)
assert result["updated"] == 1
assert result["failed"] == 0
assert order.status == collab.STATUS_TO_RECHARGE
def test_batch_status_route_serializes_force_and_recovers_void(self):
"""批量接口必须真正带上 force，不能只在服务层测试里生效。"""
order = self._bind_ready(self._create(order_no="BATCH-ROUTE-FORCE"))
collab.set_status(self.db, self.user, order, "void", reason="误操作")
admin = User(
username="batch-route-admin", password_hash="h", role="super_admin"
)
self.db.add(admin)
self.db.commit()
req = HuyaExternalOrderStatusBatchRequest(
ids=[order.id],
status=collab.STATUS_TO_RECHARGE,
reason="接口层强制恢复",
force=True,
)
assert req.model_dump()["force"] is True
out = collab_api.batch_update_external_order_status(
req, db=self.db, current=admin
)
assert out["updated"] == 1
assert out["failed"] == 0
assert order.status == collab.STATUS_TO_RECHARGE
def test_batch_rejects_empty_and_too_many(self):
with pytest.raises(collab.CollabError) as excinfo:
collab.set_status_batch(self.db, self.user, [], "scripted")
tests/test_huya_points_shop_protocol.py
+15
View File
@@ -307,6 +307,21 @@ def test_new_points_rpcs_match_capture():
assert t_req["tag1"]["tag6"] == "yellowcarlist"
assert t_req["tag2"] == "heping_baodian_credit_center_page"
await client.play_mall_query_allocation(
uid=UID,
guid=GUID,
cookie=COOKIE,
scene=8,
page_id=343,
waterfall_tab_id=433,
)
decoded = client.ws.last_request()
t_req = decoded["tReq"]
assert t_req["tag1"]["tag5"] == 8
assert t_req["tag7"] == '{"waterfallTabId":"433"}'
assert t_req["tag9"] == 343
assert t_req["tag11"] == []
await client.play_mall_get_my_promotion(
uid=UID,
guid=GUID,
tests/test_migrations.py
+1 -1
View File
@@ -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 = "20261109_0075"
HEAD_REVISION = "20261111_0076"
# 迁移新增、但不声明在模型里的复合查询索引。
EXTRA_INDEXES = (