dachui/live-hub-py
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(huya): 为精英宝典兑换接入代理

Browse Source
This commit is contained in:
yml
2026-11-13 19:17:13 +08:00
parent 711be0e2b6
commit 563b32b7b7
5 changed files with 374 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/huya/http_client.py
+21 -4
View File
@@ -15,7 +15,7 @@ import re
import time
import urllib.parse
import urllib.request
from collections.abc import Callable
from collections.abc import Callable, Mapping
from typing import Any
from .cookie_utils import cookie_pairs, normalize_cookie_pairs, normalize_huya_cookie
@@ -164,8 +164,25 @@ def generate_ws_baseinfo(
class HuyaHttpClient:
"""虎牙 HTTP POST RPC 客户端"""
def __init__(self, logger: Callable[[str], None] | None = None):
def __init__(
self,
logger: Callable[[str], None] | None = None,
proxies: Mapping[str, str] | None = None,
):
self.logger = logger or print
self.proxies = dict(proxies) if proxies else {}
self._opener = (
urllib.request.build_opener(urllib.request.ProxyHandler(self.proxies))
if self.proxies
else None
)
if self.proxies:
self.logger("[HTTP] 本客户端已启用代理")
def _urlopen(self, request: urllib.request.Request, timeout: float):
if self._opener is not None:
return self._opener.open(request, timeout=timeout)
return urllib.request.urlopen(request, timeout=timeout)
@staticmethod
def _cookie_first_value(cookie: str, key: str) -> str:
@@ -346,7 +363,7 @@ class HuyaHttpClient:
},
)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
with self._urlopen(req, timeout=timeout) as resp:
resp_data = resp.read()
# 检测 gzip 压缩 (响应 hex 1f8b 开头)
if resp_data[:2] == b"\x1f\x8b":
@@ -1279,7 +1296,7 @@ class HuyaHttpClient:
},
)
try:
with urllib.request.urlopen(req, timeout=15) as resp:
with self._urlopen(req, timeout=15) as resp:
resp_data = resp.read()
if resp_data[:2] == b"\x1f\x8b":
import gzip
tests/test_huya_exchange_proxy.py
+259
View File
@@ -0,0 +1,259 @@
"""Huya elite exchange proxy scoping and transport tests."""
from __future__ import annotations
import urllib.request
from types import SimpleNamespace
from unittest.mock import Mock
from core.huya.http_client import HuyaHttpClient
from web.backend.models import HuyaAccount, ProxyConfig
from web.backend.services import huya_points_exchange as exchange
from web.backend.services.huya_runner import HuyaBatchRunner
def _cfg(**overrides) -> ProxyConfig:
values = {
"enabled": False,
"api_url": "",
"http": "",
"https": "",
"whitelist_enabled": False,
"whitelist_platform": "xiequ",
"whitelist_credentials": None,
"whitelist_uid": "",
"whitelist_ukey": "",
}
values.update(overrides)
cfg = ProxyConfig()
for key, value in values.items():
setattr(cfg, key, value)
return cfg
def _runner(cfg: ProxyConfig, task_type: str = "exchange_goods"):
db = Mock()
db.query.return_value.first.return_value = cfg
return HuyaBatchRunner(
db=db,
batch_id="proxy-test",
task_type=task_type,
payload={"product_id": "hy-8315982"},
)
def _direct_client():
client = Mock()
client.get_redeem_need_goods.return_value = SimpleNamespace(
code=200,
message="",
value=lambda *path: {
("tag2",): 20,
("tag3",): 20,
("tag4",): 0,
("tag5",): 1,
("tag6",): 1,
("tag7",): [],
}.get(path),
to_dict=dict,
)
client.get_goods_info.return_value = SimpleNamespace(
code=200,
message="",
to_dict=lambda: {
"sku_id": 7747527,
"name": "gift",
"price": 20,
"goods_type": 5,
"virtual_type": 507,
},
)
client.create_order_v6.return_value = SimpleNamespace(
code=200,
message="",
to_dict=lambda: {"order_id": 101},
)
return client
def test_non_exchange_task_does_not_read_proxy_config():
db = Mock()
runner = HuyaBatchRunner(
db=db,
batch_id="query-test",
task_type="query_points",
)
db.query.assert_not_called()
assert runner._proxy_cfg is None
assert runner._http_client().proxies == {}
def test_static_proxy_is_used_for_entire_exchange_task(monkeypatch):
runner = _runner(
_cfg(
enabled=True,
http="http://10.0.0.1:8080",
https="http://10.0.0.1:8080",
)
)
client = _direct_client()
calls = []
marks = []
monkeypatch.setattr(
runner,
"_http_client",
lambda **kwargs: calls.append(kwargs) or client,
)
monkeypatch.setattr(
runner,
"_mark_task",
lambda _db, _task, status, message, result=None: marks.append(
(status, message, result)
),
)
runner._execute_exchange_goods(
None,
SimpleNamespace(id=1),
HuyaAccount(uid="100", cookie="yyuid=100"),
{"uid": 100, "cookie": "yyuid=100"},
{"room_pid": "123", "pay_channel": "Zfb"},
)
assert calls == [
{
"proxies": {
"http": "http://10.0.0.1:8080",
"https": "http://10.0.0.1:8080",
}
}
]
assert marks[-1][0] == "success"
assert marks[-1][2]["proxy_used"] is True
assert marks[-1][2]["proxy_source"] == "static"
def test_api_proxy_is_fetched_once_per_exchange_task(monkeypatch):
runner = _runner(_cfg(enabled=True, api_url="https://proxy.example/fetch"))
fetcher = Mock()
fetcher.fetch_new_proxy.return_value = "http://10.0.0.2:3128"
runner._proxy_fetcher = fetcher
client = _direct_client()
calls = []
marks = []
monkeypatch.setattr(
runner,
"_http_client",
lambda **kwargs: calls.append(kwargs) or client,
)
monkeypatch.setattr(
runner,
"_mark_task",
lambda _db, _task, status, message, result=None: marks.append(
(status, message, result)
),
)
runner._execute_exchange_goods(
None,
SimpleNamespace(id=2),
HuyaAccount(uid="100", cookie="yyuid=100"),
{"uid": 100, "cookie": "yyuid=100"},
{"room_pid": "123", "pay_channel": "Zfb"},
)
fetcher.fetch_new_proxy.assert_called_once_with(max_attempts=3)
assert calls[0]["proxies"] == {
"http": "http://10.0.0.2:3128",
"https": "http://10.0.0.2:3128",
}
assert marks[-1][2]["proxy_source"] == "api"
def test_enabled_proxy_without_source_fails_exchange_before_request(monkeypatch):
runner = _runner(_cfg(enabled=True))
marks = []
monkeypatch.setattr(
exchange,
"prepare_exchange",
lambda *_args, **_kwargs: (_ for _ in ()).throw(
AssertionError("exchange must not run without a proxy")
),
)
monkeypatch.setattr(
runner,
"_mark_task",
lambda _db, _task, status, message, result=None: marks.append(
(status, message, result)
),
)
runner._execute_exchange_goods(
None,
SimpleNamespace(id=3),
HuyaAccount(uid="100", cookie="yyuid=100"),
{"uid": 100, "cookie": "yyuid=100"},
{"room_pid": "123"},
)
assert marks == [
(
"failed",
"兑换代理不可用：已启用代理，但未配置静态代理或代理 API",
{"proxy_used": False, "proxy_source": "error"},
)
]
def test_huya_http_client_routes_methods_through_configured_opener(monkeypatch):
opener = Mock()
opener.open.return_value = SimpleNamespace()
monkeypatch.setattr(urllib.request, "build_opener", lambda _handler: opener)
client = HuyaHttpClient(
logger=lambda _message: None,
proxies={"http": "http://10.0.0.3:8080"},
)
request = urllib.request.Request("https://cdnws.api.huya.com/")
assert client._urlopen(request, timeout=5) is opener.open.return_value
opener.open.assert_called_once_with(request, timeout=5)
def test_exchange_rpc_and_payment_both_use_proxied_transport(monkeypatch):
client = HuyaHttpClient(
logger=lambda _message: None,
proxies={"http": "http://10.0.0.4:8080"},
)
calls = []
def stop_request(request, timeout):
calls.append((request.full_url, timeout))
raise RuntimeError("stop before network")
monkeypatch.setattr(client, "_urlopen", stop_request)
assert (
client.get_redeem_need_goods(
uid=100,
guid="",
cookie="yyuid=100",
)
is None
)
assert (
client.pay_order_submit(
uid=100,
guid="",
cookie="yyuid=100",
order_id=101,
)
is None
)
assert len(calls) == 2
assert all(url.startswith("https://cdnws.api.huya.com/") for url, _ in calls)
tests/test_huya_points_exchange.py
+2 -2
View File
@@ -216,7 +216,7 @@ def test_runner_uses_points_exchange_service_for_direct_redeem(monkeypatch):
payload={"product_id": "hy-8315982"},
)
marks = []
monkeypatch.setattr(runner, "_http_client", lambda: client)
monkeypatch.setattr(runner, "_http_client", lambda **_kwargs: client)
monkeypatch.setattr(
runner,
"_mark_task",
@@ -269,7 +269,7 @@ def test_runner_waits_points_payment_then_redeems(monkeypatch):
)
marks = []
progress = []
monkeypatch.setattr(runner, "_http_client", lambda: client)
monkeypatch.setattr(runner, "_http_client", lambda **_kwargs: client)
monkeypatch.setattr(
runner,
"_mark_task",
web/backend/services/huya_runner_core.py
+73 -2
View File
@@ -12,7 +12,10 @@ from typing import Any
from loguru import logger
from sqlalchemy.orm import Session
from core.douyu.proxy_fetcher import ProxyFetcher
from ..models import HuyaTask
from ..models import ProxyConfig as ProxyConfigModel
from .huya_service import cookie_value
# 绑定跳转与充值下单共用（bind 与 recharge 两个域都引用了 sourceId）
@@ -47,6 +50,15 @@ class HuyaBatchRunnerCore:
self._counter_lock = threading.Lock()
self._started = 0
self._wss_sessions = threading.local()
self._proxy_cfg = None
self._proxy_fetcher = None
self._static_proxies = None
if self.task_type == "exchange_goods":
self._proxy_cfg = (
db.query(ProxyConfigModel).first() if db is not None else None
)
self._proxy_fetcher = self._create_proxy_fetcher()
self._static_proxies = self._resolve_static_proxies()
def stop(self):
self._stop.set()
@@ -130,11 +142,70 @@ class HuyaBatchRunnerCore:
)
return None
def _http_client(self):
def _create_proxy_fetcher(self) -> ProxyFetcher | None:
"""兑换任务使用现有全局代理 API，每个账号任务取一个新出口。"""
cfg = self._proxy_cfg
if not cfg or not cfg.enabled or not cfg.api_url:
return None
platform = getattr(cfg, "whitelist_platform", None) or "xiequ"
credentials = getattr(cfg, "whitelist_credentials", None)
if (
not credentials
and getattr(cfg, "whitelist_uid", "")
and getattr(cfg, "whitelist_ukey", "")
):
credentials = {
"uid": cfg.whitelist_uid,
"ukey": cfg.whitelist_ukey,
}
return ProxyFetcher(
api_url=cfg.api_url,
whitelist_platform=platform,
whitelist_credentials=(
credentials if getattr(cfg, "whitelist_enabled", False) else None
),
stop_event=self._stop,
)
def _resolve_static_proxies(self) -> dict[str, str] | None:
cfg = self._proxy_cfg
if not cfg or not cfg.enabled:
return None
http = str(getattr(cfg, "http", "") or "").strip()
https = str(getattr(cfg, "https", "") or "").strip()
if not http and not https:
return None
return {"http": http or https, "https": https or http}
def _exchange_proxies(self) -> tuple[dict[str, str] | None, str, str]:
"""返回兑换任务的代理、来源和错误；其他任务不会调用这里。"""
cfg = self._proxy_cfg
if not cfg or not cfg.enabled:
return None, "disabled", ""
if self._static_proxies:
return self._static_proxies, "static", ""
if self._proxy_fetcher:
try:
proxy_url = self._proxy_fetcher.fetch_new_proxy(max_attempts=3)
except Exception as exc: # noqa: BLE001 外部代理接口异常必须收敛到任务
return None, "error", f"获取代理失败: {exc}"
if proxy_url:
return (
{"http": proxy_url, "https": proxy_url},
"api",
"",
)
return None, "error", "代理 API 未返回可用代理"
return None, "error", "已启用代理，但未配置静态代理或代理 API"
def _http_client(self, proxies: dict[str, str] | None = None):
"""HTTP cdnws 通道客户端（无需网页设备态，方法签名与 WSS 会话保持一致）。"""
from core.huya.http_client import HuyaHttpClient
return HuyaHttpClient(logger=lambda message: self._push_log("info", message))
return HuyaHttpClient(
logger=lambda message: self._push_log("info", message),
proxies=proxies,
)
@staticmethod
def _web_wss_cookie_missing(cookie: str) -> tuple[str, ...]:
web/backend/services/huya_runner_goods.py
+19 -2
View File
@@ -40,7 +40,10 @@ class GoodsMixin:
def _push_log(self, level: str, message: str) -> None: ...
def _activity_client(self, uid: int, cookie: str): ...
def _shop_client(self, uid: int, cookie: str): ...
def _http_client(self): ...
def _http_client(self, proxies: dict[str, str] | None = None): ...
def _exchange_proxies(
self,
) -> tuple[dict[str, str] | None, str, str]: ...
def _probe_bind_status(
self, uid: int, cookie: str, b_act_id_int: int, outer_act_id: str = ""
): ...
@@ -444,7 +447,19 @@ class GoodsMixin:
pay_channel = self._normalize_pay_channel(
self.payload.get("pay_channel") or config_info.get("pay_channel")
)
client: Any = self._http_client()
proxies, proxy_source, proxy_error = self._exchange_proxies()
if proxy_error:
self._mark_task(
worker_db,
task,
"failed",
f"兑换代理不可用：{proxy_error}",
{"proxy_used": False, "proxy_source": "error"},
)
return
if proxies:
self._push_log("info", f"[{uid}] 兑换任务使用代理: {proxy_source}")
client: Any = self._http_client(proxies=proxies)
prepared = prepare_exchange(
client,
uid=uid,
@@ -464,6 +479,8 @@ class GoodsMixin:
"redeem_state": prepared.state.to_dict() if prepared.state else None,
"target": prepared.target,
"exchange_plan": prepared.plan,
"proxy_used": bool(proxies),
"proxy_source": proxy_source,
}
if prepared.status == EXCHANGE_STATUS_EXCHANGED:
result = {