切换虎牙宝典开通任务到WSS会话

core/huya/__init__.py

@@ -2,6 +2,7 @@
 
 from typing import TYPE_CHECKING
 
+from .elite_session import HuyaEliteWssSession
 from .http_client import HuyaHttpClient
 from .wss_client import HuyaWssClient
 
@@ -39,6 +40,7 @@ __all__ = [
     "HuyaAppPasswordLogin",
     "HuyaAppQrAuthRequiredError",
     "HuyaCredentialError",
+    "HuyaEliteWssSession",
     "HuyaHttpClient",
     "HuyaLoginError",
     "HuyaLoginResult",

core/huya/elite_session.py

@@ -0,0 +1,136 @@
+"""虎牙精英宝典的同步 WSS 会话封装。
+
+任务执行器运行在线程中，使用本类把异步 WSS 生命周期限制在一个账号任务内，
+避免在每次 RPC 时重复建连，也不把 asyncio 对象跨线程传递。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import urllib.parse
+from collections.abc import Callable
+from typing import Literal
+
+from .http_client import HuyaHttpClient, generate_http_baseinfo
+from .wss_client import ACTIVITY_WS_HOST, SHOP_BASEINFO, SHOP_WS_HOST, HuyaWssClient
+
+
+class HuyaEliteWssSession:
+    """一个账号、一个通道、一个可复用的同步 WSS 会话。"""
+
+    def __init__(
+        self,
+        uid: int,
+        cookie: str,
+        kind: Literal["activity", "shop"],
+        logger: Callable[[str], None] | None = None,
+        host: str | None = None,
+    ):
+        self.uid = int(uid or 0)
+        self.cookie = cookie or ""
+        self.kind = kind
+        self.logger = logger or (lambda _message: None)
+        self.loop = asyncio.new_event_loop()
+        self.client: HuyaWssClient | None = None
+        self._closed = False
+
+        if kind == "activity":
+            guid = HuyaHttpClient._resolve_cookie_guid(self.cookie)
+            baseinfo = urllib.parse.unquote(
+                generate_http_baseinfo(self.uid, guid, "")
+            )
+            connect_host = host or ACTIVITY_WS_HOST
+            origin = "https://zt.huya.com"
+        else:
+            baseinfo = SHOP_BASEINFO
+            connect_host = host or SHOP_WS_HOST
+            origin = "https://m-shop.yaoguo.com"
+
+        self.client = HuyaWssClient(baseinfo=baseinfo, logger=self.logger)
+        try:
+            self.loop.run_until_complete(
+                self.client.connect(host=connect_host, origin=origin, cookie=self.cookie)
+            )
+            initialized = self.loop.run_until_complete(
+                self.client.initialize_activity(self.uid, "", self.cookie)
+                if kind == "activity"
+                else self.client.initialize(self.uid, "", self.cookie)
+            )
+            if not initialized:
+                raise RuntimeError(f"虎牙 {kind} WSS 初始化失败")
+        except Exception:
+            self.close()
+            raise
+
+    def _run(self, coroutine):
+        if self._closed or self.client is None:
+            raise RuntimeError("虎牙 WSS 会话已关闭")
+        return self.loop.run_until_complete(coroutine)
+
+    def close(self):
+        if self._closed:
+            return
+        self._closed = True
+        if self.client is not None and self.client.ws is not None:
+            try:
+                self.loop.run_until_complete(self.client.disconnect())
+            except Exception:  # noqa: BLE001 - cleanup must not mask task result
+                self.logger("[WSS] 会话清理异常")
+        self.loop.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, _exc_type, _exc, _tb):
+        self.close()
+
+    def get_act_info(self, act_id: int):
+        return self._run(self.client.get_act_info(act_id))
+
+    def get_act_task_detail(self, uid: int, cookie: str, act_id: int):
+        return self._run(self.client.get_act_task_detail(uid, cookie, act_id))
+
+    def get_act_user_task_detail(self, uid: int, cookie: str, act_id: int):
+        return self._run(self.client.get_act_user_task_detail(uid, cookie, act_id))
+
+    def get_user_score(self, uid: int, cookie: str, sid: int):
+        return self._run(self.client.get_user_score(uid, cookie, sid))
+
+    def get_act_prize_list(self, uid: int, cookie: str, sid: int):
+        return self._run(self.client.get_act_prize_list(uid, cookie, sid))
+
+    def get_act_prize_detail(self, uid: int, cookie: str, sid: int, pid: int):
+        return self._run(self.client.get_act_prize_detail(uid, cookie, sid, pid))
+
+    def score_exchange_prize(self, uid: int, cookie: str, sid: int, pid: int):
+        return self._run(self.client.score_exchange_prize(uid, cookie, sid, pid))
+
+    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 get_goods_info(self, **kwargs):
+        return self._run(self.client.get_goods_info(**kwargs))
+
+    def list_pay_channels(self, **kwargs):
+        return self._run(self.client.list_pay_channels(**kwargs))
+
+    def check_hy_protocol(self, **kwargs):
+        return self._run(self.client.check_hy_protocol(**kwargs))
+
+    def check_user_buy_auth(self, **kwargs):
+        return self._run(self.client.check_user_buy_auth(**kwargs))
+
+    def create_order(self, **kwargs):
+        return self._run(self.client.create_order(**kwargs))
+
+    def pay_order_submit(self, **kwargs):
+        return self._run(self.client.pay_order_submit(**kwargs))
+
+    def order_detail(self, **kwargs):
+        return self._run(self.client.order_detail(**kwargs))
+
+    def query_user_order_list(self, **kwargs):
+        return self._run(self.client.query_user_order_list(**kwargs))
+
+
+__all__ = ["HuyaEliteWssSession"]

web/backend/services/huya_runner.py

@@ -127,6 +127,7 @@ class HuyaBatchRunner(
                 self._mark_task(worker_db, task, "error", f"执行异常: {exc}")
                 self._push_log("error", f"[{current}] {name} 执行异常: {exc}")
         finally:
+            self._close_wss_sessions()
             worker_db.close()
 
     def run(self):

web/backend/services/huya_runner_core.py

@@ -7,6 +7,7 @@ import copy
 import threading
 import time
 from datetime import UTC, datetime
+from typing import Any
 
 from loguru import logger
 from sqlalchemy.orm import Session
@@ -45,10 +46,32 @@ class HuyaBatchRunnerCore:
         self._stop = threading.Event()
         self._counter_lock = threading.Lock()
         self._started = 0
+        self._wss_sessions: list[Any] = []
 
     def stop(self):
         self._stop.set()
 
+    def _open_wss_session(self, uid: int, cookie: str, kind: str):
+        """在当前任务线程创建并登记 WSS 会话，统一由 finally 回收。"""
+        from core.huya.elite_session import HuyaEliteWssSession
+
+        session = HuyaEliteWssSession(
+            uid=uid,
+            cookie=cookie,
+            kind=kind,
+            logger=lambda message: self._push_log("info", f"[{uid}] {message}"),
+        )
+        self._wss_sessions.append(session)
+        return session
+
+    def _close_wss_sessions(self):
+        for session in reversed(self._wss_sessions):
+            try:
+                session.close()
+            except Exception:  # noqa: BLE001 - cleanup must not mask task status
+                logger.debug("[huya] WSS 会话清理异常")
+        self._wss_sessions.clear()
+
     def _push_log(self, level: str, message: str):
         if level == "result":
             try:

web/backend/services/huya_runner_recharge.py

@@ -42,6 +42,7 @@ class RechargeMixin:
         def _push_log(self, level: str, message: str) -> None: ...
         def _mark_task(self, *args: Any, **kwargs: Any) -> None: ...
         def _update_task_progress(self, *args: Any, **kwargs: Any) -> None: ...
+        def _open_wss_session(self, uid: int, cookie: str, kind: str): ...
 
         @staticmethod
         def _format_local_time(timestamp: int) -> str: ...
@@ -393,7 +394,12 @@ class RechargeMixin:
         )
         unit_price = int(snapshot.price or 0) if snapshot else 0
 
-        client: Any = HuyaHttpClient(
+        try:
+            client: Any = self._open_wss_session(uid, cookie, "shop")
+            self._push_log("info", f"[{uid}] 商城链路使用 WSS 会话")
+        except Exception as exc:  # noqa: BLE001 - HTTP fallback is explicit
+            self._push_log("warning", f"[{uid}] 商城 WSS 不可用，回退 HTTP: {type(exc).__name__}")
+            client = HuyaHttpClient(
                 logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")
             )
         detail_resp = client.get_goods_info(
@@ -564,7 +570,10 @@ class RechargeMixin:
         account.updated_at = datetime.now(UTC)
         if payment_status == "paid":
             account.status = "recharge_paid"
-            post_score = client.query_user_score(
+            score_client = client if hasattr(client, "query_user_score") else HuyaHttpClient(
+                logger=lambda msg: self._push_log("info", f"[{uid}] {msg}")
+            )
+            post_score = score_client.query_user_score(
                 uid=uid, cookie=cookie, sid=self._to_int(config_info.get("sid")) or 2203
             )
             if post_score is not None: