完成虎牙宝典WSS绑定与快照隔离重构

core/huya/elite_session.py

@@ -31,6 +31,7 @@ class HuyaEliteWssSession:
         self.guid = HuyaHttpClient._resolve_cookie_guid(self.cookie)
         self.kind = kind
         self.logger = logger or (lambda _message: None)
+        self.http = HuyaHttpClient(logger=self.logger)
         self.loop = asyncio.new_event_loop()
         self.client: HuyaWssClient | None = None
         self._closed = False
@@ -111,6 +112,27 @@ class HuyaEliteWssSession:
     def get_user_prize_records(self, uid: int, cookie: str, sid: int):
         return self._run(self.client.get_user_prize_records(uid, cookie, sid))
 
+    def check_user_bind_game_account(self, **kwargs):
+        return self._run(self.client.check_user_bind_game_account(**kwargs))
+
+    def confirm_bind_act_account(self, **kwargs):
+        return self._run(self.client.confirm_bind_act_account(**kwargs))
+
+    def get_live_link_param(self, **kwargs):
+        return self._run(self.client.get_live_link_param(**kwargs))
+
+    def get_user_profile_batch(self, **kwargs):
+        return self._run(self.client.get_user_profile_batch(**kwargs))
+
+    def build_bind_urls(self, *args, **kwargs):
+        return self.http.build_bind_urls(*args, **kwargs)
+
+    def get_livelink_mini_qrcode(self, *args, **kwargs):
+        return self.http.get_livelink_mini_qrcode(*args, **kwargs)
+
+    def get_livelink_qrcode_status(self, *args, **kwargs):
+        return self.http.get_livelink_qrcode_status(*args, **kwargs)
+
     def get_goods_info(self, **kwargs):
         return self._run(self.client.get_goods_info(**kwargs))
 

core/huya/wss_client.py

@@ -800,6 +800,69 @@ class HuyaWssClient:
         req.sid = int(sid or 0)
         return await self.call_rpc("webActUI", "getUserPrizeRecords", req, GetUserPrizeRecordsResp)
 
+    async def check_user_bind_game_account(
+        self, uid: int, cookie: str, b_act_id: int, is_use_outer_act_id: int = 1,
+        gid: int = 0, outer_act_id: str = "", scene: str = "",
+    ):
+        from .activity_structs import (
+            CheckUserBindGameAccountReq,
+            CheckUserBindGameAccountResp,
+        )
+
+        req = CheckUserBindGameAccountReq()
+        req.userId = self._build_activity_user(uid, cookie)
+        req.gid = int(gid or 0)
+        req.outerActId = str(outer_act_id or "")
+        req.scene = scene or ""
+        req.bActId = int(b_act_id or 0)
+        req.isUseOuterActId = int(is_use_outer_act_id or 0)
+        return await self.call_rpc(
+            "webActUI", "checkUserBindGameAccount", req, CheckUserBindGameAccountResp
+        )
+
+    async def confirm_bind_act_account(
+        self, uid: int, cookie: str, b_act_id: int, gid: int = 0, outer_act_id: str = ""
+    ):
+        from .activity_structs import (
+            ConfirmBindActAccountReq,
+            ConfirmBindActAccountResp,
+        )
+
+        req = ConfirmBindActAccountReq()
+        req.userId = self._build_activity_user(uid, cookie)
+        req.gid = int(gid or 0)
+        req.outerActId = str(outer_act_id or "")
+        req.bActId = int(b_act_id or 0)
+        return await self.call_rpc(
+            "webActUI", "confirmBindActAccount", req, ConfirmBindActAccountResp
+        )
+
+    async def get_live_link_param(
+        self, uid: int, cookie: str, b_act_id: int, gid: int = 0,
+        outer_act_id: int = 0, game_auth_scene: str = "",
+    ):
+        from .activity_structs import GetLiveLinkParamReq, GetLiveLinkParamResp
+
+        req = GetLiveLinkParamReq()
+        req.userId = self._build_activity_user(uid, cookie)
+        req.gid = int(gid or 0)
+        req.outerActId = int(outer_act_id or 0)
+        req.gameAuthScene = game_auth_scene or ""
+        req.bActId = int(b_act_id or 0)
+        return await self.call_rpc("webActUI", "getLiveLinkParam", req, GetLiveLinkParamResp)
+
+    async def get_user_profile_batch(self, uid: int, cookie: str, target_uids: list[int]):
+        from .activity_structs import GetUserProfileBatchReq, GetUserProfileBatchResp
+
+        req = GetUserProfileBatchReq()
+        req.userId = self._build_activity_user(0, "")
+        req.uidList = [int(item) for item in target_uids if int(item or 0)]
+        if not req.uidList:
+            return None
+        return await self.call_rpc(
+            "huyauserui", "getUserProfileBatch", req, GetUserProfileBatchResp
+        )
+
     async def list_pay_channels(
         self, uid: int, guid: str, cookie: str, spu_id: str, sku_id: int = 0,
         supplier_id: int = 87401,