diff --git a/core/huya/http_client.py b/core/huya/http_client.py index 3c74e0c..c2f7586 100644 --- a/core/huya/http_client.py +++ b/core/huya/http_client.py @@ -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 diff --git a/tests/test_huya_exchange_proxy.py b/tests/test_huya_exchange_proxy.py new file mode 100644 index 0000000..368a727 --- /dev/null +++ b/tests/test_huya_exchange_proxy.py @@ -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) diff --git a/tests/test_huya_points_exchange.py b/tests/test_huya_points_exchange.py index ac84d12..c8cd3d4 100644 --- a/tests/test_huya_points_exchange.py +++ b/tests/test_huya_points_exchange.py @@ -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", diff --git a/web/backend/services/huya_runner_core.py b/web/backend/services/huya_runner_core.py index 50b0766..f465ff3 100644 --- a/web/backend/services/huya_runner_core.py +++ b/web/backend/services/huya_runner_core.py @@ -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, ...]: diff --git a/web/backend/services/huya_runner_goods.py b/web/backend/services/huya_runner_goods.py index feb36c2..f7822f8 100644 --- a/web/backend/services/huya_runner_goods.py +++ b/web/backend/services/huya_runner_goods.py @@ -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 = {