diff --git a/core/huya/http_client.py b/core/huya/http_client.py index 6e611e1..83c82e1 100644 --- a/core/huya/http_client.py +++ b/core/huya/http_client.py @@ -1522,6 +1522,87 @@ class HuyaHttpClient: return self._raw_from_wup(resp) if isinstance(resp, WupResponse) else None +def _bind_game_account_has_role(resp: Any) -> bool: + """响应是否真的带出了角色名（角色名是判断「最新角色」的唯一可靠信号）。""" + account_data = getattr(resp, "accountData", None) + game_role = getattr(account_data, "gameRole", None) + return bool(getattr(game_role, "roleName", "")) + + +def query_bind_game_account( + client: Any, + *, + uid: int, + cookie: str, + b_act_id: int, + gid: int = 0, + outer_act_id: str = "", + is_use_outer_act_id: int = 1, + scene: str = "", + timeout: float | None = None, +) -> tuple[Any, dict[str, Any]]: + """查询绑定游戏账号（官方活动页两条查询，返回最新角色）。 + + 2026-11-09 真机定论：新版活动页先查 ``gid + outerActId``，再查 + ``gid + outerActId=空``；**角色名以第二路为准**（第一路会返回旧的缓存角色）。 + 旧协议（gid=0）第二路会被服务端判 501，保持外层优先、缺角色时补查内层。 + + 返回 ``(status, detail)``： + - ``status``：最终选中的 ``CheckUserBindGameAccountResp``（``None`` 表示接口无响应）； + - ``detail``：两路响应与来源标记，供任务结果透传（可 JSON 序列化）。 + """ + + def _call(outer_id: str, use_outer: int): + kwargs: dict[str, Any] = { + "uid": uid, + "cookie": cookie, + "b_act_id": b_act_id, + "gid": gid, + "outer_act_id": outer_id, + "is_use_outer_act_id": use_outer, + "scene": scene, + } + if timeout is not None: + kwargs["timeout"] = timeout + return client.check_user_bind_game_account(**kwargs) + + outer = _call(outer_act_id, is_use_outer_act_id) + if outer is None: + return None, {} + detail: dict[str, Any] = { + "bind_status_source": "outer", + "outer_bind_status": outer.to_dict(), + } + if outer.status != 200: + return outer, detail + + chosen = outer + if gid: + # 第二路（gid + outerActId=空）是最新角色源，带出角色就优先采用。 + inner = _call("", 0) + if inner is not None: + detail["inner_bind_status"] = inner.to_dict() + if inner.status == 200 and _bind_game_account_has_role(inner): + chosen = inner + detail["bind_status_source"] = "inner" + else: + account_data = getattr(outer, "accountData", None) + if not _bind_game_account_has_role(outer) and bool( + getattr(account_data, "isNeedActCheck", 0) + or not getattr(account_data, "isBindAcount", 0) + or not getattr(account_data, "isBindRole", 0) + ): + inner = _call("", 0) + if inner is not None: + detail["inner_bind_status"] = inner.to_dict() + if inner.status == 200 and _bind_game_account_has_role(inner): + chosen = inner + detail["bind_status_source"] = "inner" + + detail["bind_status"] = chosen.to_dict() + return chosen, detail + + def _self_check(): uid = 1199647239697 guid = "0a8973013ad3476a53029ddb9edb071f" diff --git a/tests/test_huya_collab_orders.py b/tests/test_huya_collab_orders.py index e0c3c09..65a5597 100644 --- a/tests/test_huya_collab_orders.py +++ b/tests/test_huya_collab_orders.py @@ -18,8 +18,7 @@ from core.huya.taf_protocol import TafOutputStream, TafType from tests.support.db import get_test_engine, make_test_session from web.backend.deps import require_permission from web.backend.huya_defaults import ( - HUYA_DEFAULT_BIND_ACT_ID, - HUYA_DEFAULT_OUTER_ACT_ID, + HUYA_DEFAULT_BIND_GID, ) from web.backend.models import ( AuditLog, @@ -763,6 +762,13 @@ class FakeBindResp: ), ) + def to_dict(self) -> dict: + return { + "status": self.status, + "msg": self.msg, + "game_name": self.gameName, + } + class FakeGiftResp: """合成一份 ``getActPrizeList`` 响应（礼包列表）。""" @@ -883,12 +889,21 @@ class FakeProbeClient: self.calls: list[dict] = [] self.gift_calls: list[dict] = [] self.points_calls: list[dict] = [] + self._current_bind_outcome: object | None = None def check_user_bind_game_account(self, **kwargs): self.calls.append(kwargs) - if not self.outcomes: - raise AssertionError("探针被调用的次数超出用例预期") - outcome = self.outcomes.pop(0) + if kwargs.get("outer_act_id"): + # 外层（第一路）：按脚本弹一个响应。 + if not self.outcomes: + raise AssertionError("探针被调用的次数超出用例预期") + outcome = self.outcomes.pop(0) + self._current_bind_outcome = outcome + else: + # 内层（第二路）：复用当前外层响应；需要内外层不同时用自定义 client。 + outcome = self._current_bind_outcome + if outcome is None: + raise AssertionError("内层查询在没有外层结果时被调用") if isinstance(outcome, Exception): raise outcome return outcome @@ -947,21 +962,49 @@ class TestCollabCkProbe(TestCollabBase): collab.verify_ck(self.db, self.user, order, client=client) - call = client.calls[0] - assert call["uid"] == 1199647239697 - assert isinstance(call["uid"], int) - assert call["cookie"] == order.cookie - assert ( - call["b_act_id"] - == collab.PROBE_BIND_ACT_ID - == int(HUYA_DEFAULT_BIND_ACT_ID) - ) - assert call["is_use_outer_act_id"] == 1 - assert ( - call["outer_act_id"] - == collab.PROBE_OUTER_ACT_ID - == HUYA_DEFAULT_OUTER_ACT_ID - ) + # 通用查询两路：外层 gid+outerActId，内层 gid+outerActId=空。 + outer, inner = client.calls[0], client.calls[1] + assert outer["uid"] == 1199647239697 + assert isinstance(outer["uid"], int) + assert outer["cookie"] == order.cookie + assert outer["b_act_id"] == 0 + assert outer["gid"] == collab.PROBE_GID == int(HUYA_DEFAULT_BIND_GID) + assert outer["outer_act_id"] == collab.PROBE_OUTER_ACT_ID + assert outer["is_use_outer_act_id"] == 0 + assert outer["scene"] == collab.PROBE_SCENE == "undefined" + assert inner["gid"] == collab.PROBE_GID == 3203 + assert inner["outer_act_id"] == "" + assert inner["is_use_outer_act_id"] == 0 + assert inner["scene"] == "undefined" + + def test_verify_queries_second_path_for_latest_role(self): + """外层返回旧角色时，必须取第二路（outerActId=空）的最新角色。""" + order = self._row_with_ck(uid="1199647239697") + + class StaleThenLatestClient: + def __init__(self): + self.calls: list[dict] = [] + + def check_user_bind_game_account(self, **kwargs): + self.calls.append(kwargs) + if kwargs["outer_act_id"]: + return FakeBindResp(nick="冰点", role="软芙乎") + return FakeBindResp(nick="冰点", role="唉哟酪酪") + + client = StaleThenLatestClient() + + collab.verify_ck(self.db, self.user, order, client=client) + + outer, inner = client.calls[0], client.calls[1] + assert outer["outer_act_id"] == collab.PROBE_OUTER_ACT_ID + assert inner["outer_act_id"] == "" + assert inner["gid"] == collab.PROBE_GID == 3203 + assert inner["scene"] == "undefined" + # 快照落的是第二路的最新角色，而不是外层旧角色 + assert order.bind_game_role == "唉哟酪酪" + assert order.bind_game_account == "冰点" + assert "唉哟酪酪" in order.ck_check_message + assert "软芙乎" not in order.ck_check_message def test_verify_marks_invalid_when_status_is_not_200(self): order = self._row_with_ck() @@ -1241,7 +1284,7 @@ class TestCollabCkProbe(TestCollabBase): # 没 CK 的行保持「未验证」，不能谎报成「已失效」 assert without_ck.ck_status == "unverified" assert "CK" in result["results"][1]["message"] - assert len(client.calls) == 1 # 没 CK 的行不发请求 + assert len(client.calls) == 2 # 有 CK 的行两路（外层+内层），没 CK 的行不发请求 def test_batch_keeps_interval_between_rows(self, monkeypatch): """人工触发也要串行 + 间隔，别把账号送进风控名单。""" @@ -2134,8 +2177,8 @@ class TestCollabRecharge(TestCollabBase): assert ( order.points_check_detail["group_key"] == f"{self.BIND} / {self.ROLE}" ) - # 同一 uid 只验一次 CK、只查一次积分 - assert len(client.calls) == 1 + # 同一 uid 只验一次 CK（两路查询）、只查一次积分 + assert len(client.calls) == 2 assert len(client.score_calls) == 1 assert "huya:collab:recharge_check" in [ row.action for row in self._audits(first) @@ -3136,7 +3179,7 @@ class TestProbeCache(TestCollabBase): collab.probe_ck(order) collab.probe_ck(order) - assert len(calls) == 1 + assert len(calls) == 2 # 第一次探针两路（外层+内层），第二次命中缓存 collab._probe_cache.clear() def test_different_ck_is_probed_separately(self, monkeypatch): @@ -3155,7 +3198,7 @@ class TestProbeCache(TestCollabBase): collab.probe_ck(first) collab.probe_ck(second) - assert len(calls) == 2 + assert len(calls) == 4 # 两个 CK 各自两路（外层+内层） collab._probe_cache.clear() diff --git a/tests/test_huya_elite_query.py b/tests/test_huya_elite_query.py index f18af34..57362e6 100644 --- a/tests/test_huya_elite_query.py +++ b/tests/test_huya_elite_query.py @@ -1050,14 +1050,20 @@ def test_confirm_bind_uses_latest_har_protocol(monkeypatch): "outer_act_id": "17096", } ] - assert all( - call["b_act_id"] == 0 - and call["gid"] == 3203 - and call["outer_act_id"] == "17096" - and call["is_use_outer_act_id"] == 0 - and call["scene"] == "undefined" - for call in client.check_calls - ) + # 通用查询会连打两路：外层 gid+outerActId=17096，内层 gid+outerActId=空。 + assert len(client.check_calls) == 4 # 绑定前查一次 + 确认后刷新查一次 + for call in client.check_calls[0::2]: + assert call["b_act_id"] == 0 + assert call["gid"] == 3203 + assert call["outer_act_id"] == "17096" + assert call["is_use_outer_act_id"] == 0 + assert call["scene"] == "undefined" + for call in client.check_calls[1::2]: + assert call["b_act_id"] == 0 + assert call["gid"] == 3203 + assert call["outer_act_id"] == "" + assert call["is_use_outer_act_id"] == 0 + assert call["scene"] == "undefined" def test_get_bind_qr_uses_latest_har_protocol(monkeypatch): diff --git a/web/backend/services/huya_collab_service.py b/web/backend/services/huya_collab_service.py index a214ef0..12a856d 100644 --- a/web/backend/services/huya_collab_service.py +++ b/web/backend/services/huya_collab_service.py @@ -30,8 +30,10 @@ from loguru import logger from sqlalchemy import func, or_, tuple_ from sqlalchemy.orm import Session +from core.huya.http_client import query_bind_game_account + from ..huya_defaults import ( - HUYA_DEFAULT_BIND_ACT_ID, + HUYA_DEFAULT_BIND_GID, HUYA_DEFAULT_OUTER_ACT_ID, HUYA_DEFAULT_SID, ) @@ -1412,9 +1414,12 @@ def set_images( # CK 探针（S5，只读验证） # -------------------------------------------------------------------------- -# 探针参数与 web/backend/huya_defaults.py 的默认活动一致（2026-10-16 真机实测）。 -PROBE_BIND_ACT_ID = int(HUYA_DEFAULT_BIND_ACT_ID) +# 探针参数与 web/backend/huya_defaults.py 的默认活动一致（2026-11-09 真机实测）。 +# 新版活动页绑定走 gid=3203、bActId=0；最终角色由通用查询的第二路 +# （gid + outerActId=空）返回，这里只给出外层活动参数。 +PROBE_GID = int(HUYA_DEFAULT_BIND_GID) PROBE_OUTER_ACT_ID = HUYA_DEFAULT_OUTER_ACT_ID +PROBE_SCENE = "undefined" # 与 runner 绑定协议保持一致 PROBE_TIMEOUT_SECONDS = 15.0 # 探针独占的绑定快照列（每次验证**覆盖写入**，人工不编辑）： @@ -1572,12 +1577,15 @@ def probe_ck(order: HuyaExternalOrder, *, client: Any | None = None) -> CkProbeR try: probe_client = client or _build_probe_client() - resp = probe_client.check_user_bind_game_account( + resp, _detail = query_bind_game_account( + probe_client, uid=uid, cookie=cookie, - b_act_id=PROBE_BIND_ACT_ID, - is_use_outer_act_id=1, + b_act_id=0, + gid=PROBE_GID, outer_act_id=PROBE_OUTER_ACT_ID, + is_use_outer_act_id=0, + scene=PROBE_SCENE, timeout=PROBE_TIMEOUT_SECONDS, ) except Exception as exc: # noqa: BLE001 探针异常=验证不通过，不能让接口 500 diff --git a/web/backend/services/huya_runner_bind.py b/web/backend/services/huya_runner_bind.py index c27f756..463122d 100644 --- a/web/backend/services/huya_runner_bind.py +++ b/web/backend/services/huya_runner_bind.py @@ -8,6 +8,8 @@ from typing import TYPE_CHECKING, Any from sqlalchemy.orm import Session +from core.huya.http_client import query_bind_game_account + from ..huya_defaults import HUYA_DEFAULT_OUTER_ACT_ID from ..models import HuyaAccount, HuyaTask from .huya_runner_core import HUYA_RECHARGE_SOURCE_ID, HuyaBatchRunnerCore @@ -48,10 +50,6 @@ class BindMixin: account_data = bind_status.accountData return account_data.gameRole.roleName or "" - @staticmethod - def _has_bind_role(bind_status) -> bool: - return bool(bind_status and BindMixin._role_name(bind_status)) - @staticmethod def _bind_role_result(bind_status) -> dict: account_data = bind_status.accountData @@ -149,56 +147,18 @@ class BindMixin: is_use_outer_act_id: int = 1, scene: str = "", ): - """按活动页逻辑解析绑定状态，优先返回含角色的状态。""" - probe_kwargs: dict[str, Any] = { - "uid": uid, - "cookie": cookie, - "b_act_id": b_act_id_int, - "gid": gid, - "outer_act_id": outer_act_id, - "is_use_outer_act_id": is_use_outer_act_id, - "scene": scene, - } - if timeout is not None: - probe_kwargs["timeout"] = timeout - outer_status = client.check_user_bind_game_account(**probe_kwargs) - if outer_status is None: - return None, {} - - query_result = { - "bind_status_source": "outer", - "outer_bind_status": outer_status.to_dict(), - } - if outer_status.status != 200: - return outer_status, query_result - - chosen_status = outer_status - account_data = outer_status.accountData - # 官方页面会同时查两路：gid + outerActId，以及 gid + outerActId=空。 - # 角色名通常由第二路返回；第二路必须保留 gid，不能传成 gid=0。 - should_check_inner = not self._has_bind_role(outer_status) and bool( - account_data.isNeedActCheck - or not account_data.isBindAcount - or not account_data.isBindRole + """按活动页逻辑解析绑定状态（统一委托 core.huya.http_client 的通用查询）。""" + return query_bind_game_account( + client, + uid=uid, + cookie=cookie, + b_act_id=b_act_id_int, + gid=gid, + outer_act_id=outer_act_id, + is_use_outer_act_id=is_use_outer_act_id, + scene=scene, + timeout=timeout, ) - if should_check_inner: - inner_status = client.check_user_bind_game_account( - uid=uid, - cookie=cookie, - b_act_id=b_act_id_int, - is_use_outer_act_id=0, - gid=gid, - outer_act_id="", - scene=scene, - ) - if inner_status is not None: - query_result["inner_bind_status"] = inner_status.to_dict() - if inner_status.status == 200 and self._has_bind_role(inner_status): - chosen_status = inner_status - query_result["bind_status_source"] = "inner" - - query_result["bind_status"] = chosen_status.to_dict() - return chosen_status, query_result def _probe_bind_status( self,