From aab84e0fb5f7393abb637a1cb34ff181e48f4a47 Mon Sep 17 00:00:00 2001 From: yml2213 Date: Tue, 1 Sep 2026 12:20:58 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=87=E6=8D=A2=E8=99=8E=E7=89=99=E5=AE=9D?= =?UTF-8?q?=E5=85=B8=E5=BC=80=E9=80=9A=E4=BB=BB=E5=8A=A1=E5=88=B0WSS?= =?UTF-8?q?=E4=BC=9A=E8=AF=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- core/huya/__init__.py | 2 + core/huya/elite_session.py | 136 +++++++++++++++++++ web/backend/services/huya_runner.py | 1 + web/backend/services/huya_runner_core.py | 23 ++++ web/backend/services/huya_runner_recharge.py | 17 ++- 5 files changed, 175 insertions(+), 4 deletions(-) create mode 100644 core/huya/elite_session.py diff --git a/core/huya/__init__.py b/core/huya/__init__.py index 4742139..940a1db 100644 --- a/core/huya/__init__.py +++ b/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", diff --git a/core/huya/elite_session.py b/core/huya/elite_session.py new file mode 100644 index 0000000..257817e --- /dev/null +++ b/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"] diff --git a/web/backend/services/huya_runner.py b/web/backend/services/huya_runner.py index 6173e6d..b470bec 100644 --- a/web/backend/services/huya_runner.py +++ b/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): diff --git a/web/backend/services/huya_runner_core.py b/web/backend/services/huya_runner_core.py index fb5e83d..3c7dc15 100644 --- a/web/backend/services/huya_runner_core.py +++ b/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: diff --git a/web/backend/services/huya_runner_recharge.py b/web/backend/services/huya_runner_recharge.py index 657acaf..0405e23 100644 --- a/web/backend/services/huya_runner_recharge.py +++ b/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,9 +394,14 @@ class RechargeMixin: ) unit_price = int(snapshot.price or 0) if snapshot else 0 - client: Any = HuyaHttpClient( - logger=lambda msg: self._push_log("info", f"[{uid}] {msg}") - ) + 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( uid=uid, guid="", @@ -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: