Files
live-hub-py/tests/test_huya_elite_query.py
T
yml2213 f61284e808 feat(huya): 开通支付码支持多选批量 + 支付后自动确认开通
多选账号点「生成开通支付码」以前只有弹窗里最后一个码能被看到(每个账号一个
任务,前端只留一个 qrTask),实际只能一个账号一个账号地开。

- 前端:支付码弹窗改成列表 —— 一次列出该批次所有账号的二维码(账号/UID/
  商品金额/状态),未支付的排前面,可逐个扫也可一起扫;行点击也会打开对应批次
- 支付成功后自动确认开通:新增 EliteMixin._refresh_elite_open_after_payment,
  交付有几秒延迟,tag5 还没翻成 1 时重试(默认 4 次 × 3 秒),顺带取最新积分,
  并把 elite_opened/elite_open_time/elite_expire_time 写回账号
- 已支付或已开通的账号二维码自动从弹窗收起,只留一行「已开通」标签
- 开通时间优先用支付回调里的支付时间(比再去翻订单更准,也少一次请求)
- 补 3 条测试:重试到已开通、重试到上限、支付成功后任务结果与账号字段齐全

全量 746 passed / 6 skipped,前端 tsc 通过
2026-10-15 17:14:20 +08:00

533 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""虎牙精英宝典开通状态与「一键查询」(2026-10-15 真机实测)。
判定来源与真机结论:
* 宝典开通状态 = 积分中心 ``shopMiddleUI.getRedeemNeedGoods`` 的 ``tag5``。
下面固化的两段响应字节分别来自已开通账号(hy_302016799)与未开通账号
(hy_260034554),与网页显示的「已开通精英宝典 / 获取积分需先开通精英宝典」
一致。
* 开通时间 = ``revenueWebUI.queryUserOrderList`` 里已支付的宝典订单。
"""
import json
from types import SimpleNamespace
from core.huya.frame_decoder import decode_struct_bytes
from core.huya.http_client import HuyaHttpClient
from core.huya.shop_structs import ShopRawRsp
from web.backend.models import HuyaAccount
from web.backend.services.huya_runner import HuyaBatchRunner
UID = 1199668595203
COOKIE = f"yyuid={UID}; udb_uid={UID}"
# 真机 getRedeemNeedGoods 响应:已开通(tag5=1、tag6=1、tag7 为空)
REDEEM_OPENED_HEX = "0a0100c8160ce69fa5e8afa2e68890e58a9f21011830144c50016001790c0b"
# 真机 getRedeemNeedGoods 响应:未开通(tag5=0、tag6=0、tag7 列出开通套餐)
REDEEM_NOT_OPENED_HEX = (
"0a0100c8160ce69fa5e8afa2e68890e58a9f2c301440145c6c7900040a02007f134f1200"
"765c0b260ce7b2bee88bb1e5ae9de585b830054101fb51012c610bb87c9001a90cb90cca010"
"bb8110bb82c310bb8490c5c610bb8790c8c9cacbcccdc0bf00f010b0a02007ede691200766"
"17f2616e5bc80e9809ae5ae9de585b82be7b2bee88bb1e5b88130054101fc51012d610bc2"
"70018900020a02007ede6912007f134f2200765c0b310bb841012c5001660ce7b2bee88bb1"
"e5ae9de585b87c0b0a02007ede6912007f10bc220076662f300a400150016609e7b2bee88b"
"b1e5b88170010b9ca90cb90cca010bc2110bc22c310bc2490c5c610bc2790c8c9cacbcccdc0b"
"fc0f0b"
)
# 真机订单列表里已支付的宝典订单(pay_time 与 spu 均来自真机响应)
PAID_ELITE_ORDER = {
"item_name": "精英宝典",
"order_status": 60,
"pay_time": 1791895396000,
"goods_detail": {"spu_id": "hy-8328015", "sku_id": 7756811},
}
UNPAID_ELITE_ORDER = {
"item_name": "精英宝典",
"order_status": 10,
"pay_time": 0,
"goods_detail": {"spu_id": "hy-8328015", "sku_id": 7756811},
}
OTHER_ORDER = {
"item_name": "电竞徽章",
"order_status": 60,
"pay_time": 1788000000000,
"goods_detail": {"spu_id": "hy-7000000", "sku_id": 1},
}
# 真机 queryAllocation 响应里的赛季节点(精英宝典福利季丨Q4:2026-10-09 ~ 2026-12-31)
ALLOCATION_SEASON_PAYLOAD = {
"tag0": {"tag0": 200, "tag1": "成功", "tag2": ""},
"tag1": {
"tag0": "和平精英宝典积分中心",
"tag1": "heping_baodian_credit_center_page",
},
"tag2": [
{"tag0": 343, "tag4": 6},
{
"tag0": 1,
"tag2_2": {
"tag0": {
"tag0": 15,
"tag1": "精英宝典福利季",
"tag2": "精英宝典福利季丨Q4",
"tag3": 3203,
"tag5": 17096,
"tag6": 9815,
"tag10": 1791475200000,
"tag11": 1798732799000,
},
"tag4": {"tag0": 300},
},
},
],
}
def _redeem(hex_text: str) -> ShopRawRsp:
return ShopRawRsp.from_payload(decode_struct_bytes(bytes.fromhex(hex_text)))
class _FakeOrder:
def __init__(self, payload: dict):
self._payload = payload
def to_dict(self) -> dict:
return dict(self._payload)
class _FakeOrderList:
def __init__(self, orders: list[dict], total_count: int | None = None):
self.orders = [_FakeOrder(item) for item in orders]
self.totalCount = len(orders) if total_count is None else total_count
class _FakeBindStatus:
"""只喂给 EliteMixin 用到的那几个字段(角色解析由 BindMixin 自己的用例覆盖)。"""
def __init__(self, status: int = 200, msg: str = ""):
self.status = status
self.msg = msg
def to_dict(self) -> dict:
return {"status": self.status, "msg": self.msg}
class _StubClient:
"""积分中心通道的假客户端:可返回宝典状态/订单列表/赛季配置,并记录调用。"""
def __init__(self, redeem=None, order_pages=None, season: bool = False):
self.redeem = redeem
self.order_pages = order_pages or []
self.season_payload = ALLOCATION_SEASON_PAYLOAD if season else None
self.calls: list[tuple[str, dict]] = []
self.redeem_calls = 0
def get_redeem_need_goods(self, **kwargs):
self.calls.append(("get_redeem_need_goods", kwargs))
self.redeem_calls += 1
return self.redeem
def play_mall_query_allocation(self, **kwargs):
self.calls.append(("play_mall_query_allocation", kwargs))
if self.season_payload is None:
return None
return ShopRawRsp.from_payload(self.season_payload)
def query_user_order_list(self, **kwargs):
self.calls.append(("query_user_order_list", kwargs))
offset = int(kwargs.get("offset") or 0)
index = offset // 20
if index >= len(self.order_pages):
return None
page = self.order_pages[index]
return _FakeOrderList(page, total_count=sum(len(p) for p in self.order_pages))
def _runner(monkeypatch, shop_client, activity_client=None):
runner = HuyaBatchRunner(db=None, batch_id="test-batch", task_type="query_all")
marks: list[tuple[str, str, dict | None]] = []
logs: list[tuple[str, str]] = []
monkeypatch.setattr(
runner, "_shop_client", lambda uid, cookie: shop_client, raising=False
)
if activity_client is not None:
monkeypatch.setattr(
runner,
"_activity_client",
lambda uid, cookie: activity_client,
raising=False,
)
monkeypatch.setattr(
runner,
"_mark_task",
lambda _db, _task, status, message, result=None: marks.append(
(status, message, result)
),
raising=False,
)
monkeypatch.setattr(
runner,
"_push_log",
lambda level, message: logs.append((level, message)),
raising=False,
)
return runner, marks, logs
def test_real_redeem_bytes_report_elite_opened():
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
assert runner._elite_opened_from_redeem(_redeem(REDEEM_OPENED_HEX)) is True
assert runner._elite_opened_from_redeem(_redeem(REDEEM_NOT_OPENED_HEX)) is False
# 不是 200 或没有 tag5 时不下结论(界面显示「未查询」而不是猜一个值)
assert runner._elite_opened_from_redeem(None) is None
assert (
runner._elite_opened_from_redeem(ShopRawRsp.from_payload({"tag0": 500})) is None
)
def test_elite_open_time_uses_latest_paid_handbook_order():
client = _StubClient(
order_pages=[[OTHER_ORDER, UNPAID_ELITE_ORDER, PAID_ELITE_ORDER]]
)
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
assert runner._resolve_elite_open_time(client, UID, COOKIE) == 1791895396000
def test_elite_open_time_ignores_unpaid_handbook_order():
"""未支付的宝典订单不算开通时间(真机:order_status=10、pay_time=0)。"""
client = _StubClient(order_pages=[[UNPAID_ELITE_ORDER]])
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
assert runner._resolve_elite_open_time(client, UID, COOKIE) is None
def test_elite_open_time_scans_following_pages():
later_order = {**PAID_ELITE_ORDER, "pay_time": 1792000000000}
client = _StubClient(order_pages=[[OTHER_ORDER] * 20, [later_order]])
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
assert runner._resolve_elite_open_time(client, UID, COOKIE) == 1792000000000
offsets = [
kwargs.get("offset")
for name, kwargs in client.calls
if name == "query_user_order_list"
]
assert offsets == [0, 20]
def test_extract_elite_season_from_allocation_payload():
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
season = runner._extract_elite_season(ALLOCATION_SEASON_PAYLOAD)
assert season == {
"name": "精英宝典福利季丨Q4",
"start_ms": 1791475200000,
"end_ms": 1798732799000,
"config_id": 15,
}
# 认不出赛季结构时不下结论(界面就不显示有效期)
assert runner._extract_elite_season({"tag0": {"tag1": "x"}}) is None
assert runner._extract_elite_season(None) is None
def test_query_all_fills_expire_time_from_season(monkeypatch):
shop_client = _StubClient(
redeem=_redeem(REDEEM_OPENED_HEX),
order_pages=[[PAID_ELITE_ORDER]],
season=True,
)
runner, marks, _logs = _runner(monkeypatch, shop_client, activity_client=object())
monkeypatch.setattr(
runner, "_resolve_bind_status", lambda **kwargs: (None, {}), raising=False
)
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
runner._execute_query_all(
None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271"}
)
status, message, result = marks[-1]
assert status == "success"
assert result["elite_expire_time"] == 1798732799
assert result["elite_season"]["name"] == "精英宝典福利季丨Q4"
assert account.elite_expire_time == 1798732799
assert "有效期至 2026-12-31" in message
def test_refresh_elite_open_after_payment_retries_until_opened():
"""刚支付完 tag5 可能还是 0(交付延迟)→ 重试到已开通,并写回账号。"""
class _FlakyClient(_StubClient):
def get_redeem_need_goods(self, **kwargs):
result = super().get_redeem_need_goods(**kwargs)
if self.redeem_calls > 1:
return _redeem(REDEEM_OPENED_HEX)
return result
client = _FlakyClient(season=True)
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="create_recharge_order")
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
state = runner._refresh_elite_open_after_payment(
client,
account,
UID,
COOKIE,
goods_num=8315982,
pay_time_ms=1792053457000,
attempts=3,
interval=0,
)
assert client.redeem_calls == 2 # 第一次未开通 → 重试一次即成功
assert state["elite_opened"] is True
assert state["elite_open_time"] == 1792053457
assert state["elite_expire_time"] == 1798732799
assert state["redeem_points"] == 280
assert account.elite_opened is True
assert account.elite_open_time == 1792053457
assert account.elite_expire_time == 1798732799
def test_refresh_elite_open_after_payment_gives_up_after_attempts():
client = _StubClient(redeem=_redeem(REDEEM_NOT_OPENED_HEX))
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="create_recharge_order")
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
state = runner._refresh_elite_open_after_payment(
client,
account,
UID,
COOKIE,
goods_num=8315982,
pay_time_ms=1792053457000,
attempts=2,
interval=0,
)
assert client.redeem_calls == 2 # 重试到上限
assert state["elite_opened"] is False
assert account.elite_opened is False
assert account.elite_open_time is None
def test_create_recharge_order_confirms_elite_open_after_payment(db, monkeypatch):
"""支付成功后自动确认开通状态:任务结果与账号字段都带上宝典状态。"""
outcome = SimpleNamespace(
ok=True,
message="",
detail={"spu_id": "hy-8328015", "sku_id": 7756811, "name": "精英宝典"},
sku_id=7756811,
product_name="精英宝典",
unit_price=3000,
order_raw={"order_id": 11399999},
pay_raw={"pay_order_id": "pay-1"},
amount=3000,
order_id=11399999,
app_order_id="app-1",
pay_order_id="pay-1",
pay_url="https://example.com/pay",
supplier_name="和平精英官方小店",
promotion={},
promotion_info="本单无返利",
)
shop_client = _StubClient(redeem=_redeem(REDEEM_OPENED_HEX), season=True)
runner = HuyaBatchRunner(
db=None,
batch_id="t",
task_type="create_recharge_order",
payload={"spu_id": "hy-8328015", "sku_id": 7756811, "count": 1},
)
marks: list[tuple[str, str, dict | None]] = []
monkeypatch.setattr(
runner, "_shop_client", lambda uid, cookie: shop_client, raising=False
)
monkeypatch.setattr(
runner,
"_mark_task",
lambda _db, _task, status, message, result=None: marks.append(
(status, message, result)
),
raising=False,
)
monkeypatch.setattr(
runner, "_update_task_progress", lambda *args, **kwargs: None, raising=False
)
monkeypatch.setattr(
runner, "_push_log", lambda *args, **kwargs: None, raising=False
)
def fake_wait_payment(**kwargs):
# 真实现会边轮询边写 result,这里照抄「已支付」那几个字段
target = kwargs.get("result")
if isinstance(target, dict):
target.update(
{
"payment_status": "paid",
"payment_status_label": "已支付",
"payment_paid": True,
}
)
return "paid", {"pay_time": 1792053457000}
monkeypatch.setattr(
runner, "_wait_recharge_payment", fake_wait_payment, raising=False
)
monkeypatch.setattr(
"web.backend.services.huya_runner_recharge.run_open_flow",
lambda *args, **kwargs: outcome,
)
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
db.add(account)
db.commit()
runner._execute_create_recharge_order(
db,
None,
account,
{"uid": UID, "cookie": COOKIE},
{"room_pid": "1199650619883", "pay_channel": "Zfb", "sid": "2203"},
)
status, _message, result = marks[-1]
assert status == "success"
assert result["payment_paid"] is True
assert result["elite_opened"] is True
assert result["elite_open_time"] == 1792053457
assert result["elite_expire_time"] == 1798732799
assert result["post_payment_points"] == 280
assert account.elite_opened is True
assert account.elite_expire_time == 1798732799
assert account.points == 280
assert account.status == "recharge_paid"
def test_query_all_combines_points_elite_and_role(monkeypatch):
shop_client = _StubClient(
redeem=_redeem(REDEEM_OPENED_HEX), order_pages=[[PAID_ELITE_ORDER]]
)
runner, marks, _logs = _runner(monkeypatch, shop_client, activity_client=object())
monkeypatch.setattr(
runner,
"_resolve_bind_status",
lambda **kwargs: (_FakeBindStatus(), {"bind_status_source": "outer"}),
raising=False,
)
monkeypatch.setattr(
runner,
"_bind_role_result",
staticmethod(
lambda _status: {"role_name": "软踏日潺湲", "game_title": "和平精英"}
),
raising=False,
)
monkeypatch.setattr(
runner,
"_bind_change_state",
staticmethod(lambda _status: {"is_bound": True}),
raising=False,
)
monkeypatch.setattr(
runner,
"_role_channel",
staticmethod(lambda _status: "苹果(iOS) / 手Q"),
raising=False,
)
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
runner._execute_query_all(
None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271"}
)
status, message, result = marks[-1]
assert status == "success"
assert result["points"] == 280
assert result["elite_opened"] is True
assert result["elite_open_time"] == 1791895396
assert result["role_name"] == "软踏日潺湲"
assert result["failures"] == []
assert account.points == 280
assert account.elite_opened is True
assert account.elite_open_time == 1791895396
assert account.game_name == "软踏日潺湲"
assert account.game_channel == "苹果(iOS) / 手Q"
assert account.status == "all_queried"
assert (
"积分 280" in message
and "已开通宝典" in message
and "角色 软踏日潺湲" in message
)
def test_query_all_reports_not_opened_without_order_lookup(monkeypatch):
shop_client = _StubClient(redeem=_redeem(REDEEM_NOT_OPENED_HEX))
runner, marks, _logs = _runner(monkeypatch, shop_client, activity_client=object())
monkeypatch.setattr(
runner, "_resolve_bind_status", lambda **kwargs: (None, {}), raising=False
)
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
runner._execute_query_all(
None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271"}
)
status, message, result = marks[-1]
# 角色接口挂了不影响宝典/积分结论
assert status == "success"
assert result["elite_opened"] is False
assert account.elite_opened is False
assert account.elite_open_time is None
assert "未开通宝典" in message
assert "角色信息接口无响应" in message
# 未开通不查订单(省一次请求)
assert [name for name, _kwargs in shop_client.calls] == ["get_redeem_need_goods"]
def test_query_all_fails_when_points_center_unreachable(monkeypatch):
shop_client = _StubClient(redeem=None)
runner, marks, _logs = _runner(monkeypatch, shop_client, activity_client=object())
monkeypatch.setattr(
runner, "_resolve_bind_status", lambda **kwargs: (None, {}), raising=False
)
account = HuyaAccount(uid=str(UID), cookie=COOKIE)
runner._execute_query_all(
None, None, account, {"uid": UID, "cookie": COOKIE}, {"bind_act_id": "9271"}
)
status, message, _result = marks[-1]
assert status == "failed"
assert "积分中心接口无响应" in message
def test_shop_client_falls_back_to_http_when_wss_init_fails(monkeypatch):
"""设备态齐全但积分中心端点不回包时(wsLaunch 超时)回退 HTTP,而不是判整批失败。"""
cookie = (
"udb_cred=cred; udb_biztoken=biz; udb_uid=1199659425268; yyuid=1199659425268; "
"sdid=sdid; udb_deviceid=w_1234567890123456789; udb_guiddata=guiddata; "
"udb_anobiztoken=anobiz"
)
logs: list[tuple[str, str]] = []
runner = HuyaBatchRunner(db=None, batch_id="t", task_type="query_all")
def boom(uid: int, cookie_text: str, kind: str):
raise RuntimeError(f"虎牙 {kind} WSS 初始化失败")
monkeypatch.setattr(runner, "_open_wss_session", boom, raising=False)
monkeypatch.setattr(
runner,
"_push_log",
lambda level, message: logs.append((level, message)),
raising=False,
)
client = runner._shop_client(1199659425268, cookie)
assert isinstance(client, HuyaHttpClient)
assert logs and logs[-1][0] == "warning"
assert "WSS 会话不可用" in logs[-1][1]
assert "sdid" not in json.dumps(logs[-1][1]) # 日志不带 Cookie 值