feat(huya): 密码登录appId透传修正+风控换IP重试

- _password_payload 改用 self.app_id/self.app_sign(原先硬编码5002,
  导致5008实例静默发错appId)
- app_id=5008 时自动切换 Android WebView UA
- login() 增加换IP重试循环: 扫码/短信风控时经 proxy_fetcher 换代理重试,
  上限 max_ip_retries, 耗尽后返回 code=risk
This commit is contained in:
yml2213
2026-08-25 11:50:48 +08:00
parent 0a060bb777
commit 705c51bf48
+88 -9
View File
@@ -28,6 +28,12 @@ APP_SIGN = "1ce3bf682483d03f146f58232ec10635"
# appKey 来自 APK 资源 R.string.cgs。 # appKey 来自 APK 资源 R.string.cgs。
APP_ID_MOBILE = "5008" APP_ID_MOBILE = "5008"
APP_SIGN_MOBILE = "60576904" APP_SIGN_MOBILE = "60576904"
# App 场景 UA(与 core/huya/fingerprint/env.js 的 Android WebView 仿真一致)
APP_UA_MOBILE = (
"Mozilla/5.0 (Linux; Android 11; M2102J2SC Build/RKQ1.200826.002; wv) "
"AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 "
"Chrome/149.0.7827.159 Mobile Safari/537.36 huya adr/13.4.22/xiaomi/30"
)
LCID = "2052" LCID = "2052"
PASSWORD_LOGIN_URI = "30001" PASSWORD_LOGIN_URI = "30001"
DF_TOKEN_URL = "https://df.huya.com/web/df/token" DF_TOKEN_URL = "https://df.huya.com/web/df/token"
@@ -131,14 +137,24 @@ class HuyaPasswordLogin:
username: str, username: str,
password: str, password: str,
cookie: Mapping[str, str] | str | None = None, cookie: Mapping[str, str] | str | None = None,
ua: str = DEFAULT_UA, ua: str | None = None,
session: requests.Session | None = None, session: requests.Session | None = None,
proxies: Mapping[str, str] | None = None, proxies: Mapping[str, str] | None = None,
timeout: tuple[float, float] | None = None, timeout: tuple[float, float] | None = None,
app_id: str = APP_ID,
proxy_fetcher=None,
): ):
self.username = username self.username = username
self.password = password self.password = password
self.ua = ua or DEFAULT_UA self.app_id = app_id if app_id in (APP_ID, APP_ID_MOBILE) else APP_ID
self.app_sign = APP_SIGN_MOBILE if self.app_id == APP_ID_MOBILE else APP_SIGN
# ua 显式传入优先;否则按 app_id 自动选择(App 场景用 Android WebView UA
if ua:
self.ua = ua
elif self.app_id == APP_ID_MOBILE:
self.ua = APP_UA_MOBILE
else:
self.ua = DEFAULT_UA
self.timeout = timeout or self.timeout self.timeout = timeout or self.timeout
self.request_id = generate_request_id() self.request_id = generate_request_id()
self.exchange = self.request_id self.exchange = self.request_id
@@ -148,6 +164,9 @@ class HuyaPasswordLogin:
self.middle_url = f"https://udblgn.huya.com/web/middle/{APP_VERSION}/{self.exchange}/https/{self.device_id}" self.middle_url = f"https://udblgn.huya.com/web/middle/{APP_VERSION}/{self.exchange}/https/{self.device_id}"
self.sdid = "" self.sdid = ""
self.sdid_source = "" self.sdid_source = ""
# 代理源(鸭子类型,需提供 fetch_new_proxy() -> Optional[str])。
# 用于扫码/短信风控时换 IP 重试,尝试把风控等级降回滑块。
self.proxy_fetcher = proxy_fetcher
self.session = session or requests.Session() self.session = session or requests.Session()
self.session.trust_env = False self.session.trust_env = False
@@ -250,8 +269,8 @@ class HuyaPasswordLogin:
"uri": PASSWORD_LOGIN_URI, "uri": PASSWORD_LOGIN_URI,
"version": APP_VERSION, "version": APP_VERSION,
"context": self.context, "context": self.context,
"appId": APP_ID, "appId": self.app_id,
"appSign": APP_SIGN, "appSign": self.app_sign,
"authId": auth_id, "authId": auth_id,
"sdid": self.sdid, "sdid": self.sdid,
"lcid": LCID, "lcid": LCID,
@@ -310,8 +329,56 @@ class HuyaPasswordLogin:
data = payload.get("data") data = payload.get("data")
return data if isinstance(data, dict) else {} return data if isinstance(data, dict) else {}
def login(self) -> HuyaLoginResult: def _swap_proxy(self) -> bool:
"""执行密码登录,成功返回 Cookie。""" """换一个代理 IP,成功返回 True。"""
if not self.proxy_fetcher:
return False
try:
proxy = self.proxy_fetcher.fetch_new_proxy()
except Exception as exc: # noqa: BLE001
logger.warning(f"获取新代理失败: {exc}")
return False
if not proxy:
return False
self.session.proxies = {"http": proxy, "https": proxy}
logger.info(f"虎牙登录切换代理 IP: {proxy}")
return True
def login(self, max_ip_retries: int = 3) -> HuyaLoginResult:
"""执行密码登录,成功返回 Cookie。
遇到扫码/短信风控(无法自动化)时,若有代理源则换 IP 重试,
尝试把风控等级降回可自动化的滑块。
"""
from .verification import HuyaQrAuthRequiredError, HuyaSmsAuthRequiredError
last_exc: Exception | None = None
for attempt in range(max_ip_retries + 1):
if attempt > 0:
if not self._swap_proxy():
logger.warning("无可用代理可切换,停止换 IP 重试")
break
try:
return self._login_once()
except (HuyaQrAuthRequiredError, HuyaSmsAuthRequiredError) as exc:
last_exc = exc
kind = "扫码" if isinstance(exc, HuyaQrAuthRequiredError) else "短信"
logger.warning(
f"虎牙风控要求{kind}验证({exc}),换 IP 重试 ({attempt + 1}/{max_ip_retries})"
)
continue
return HuyaLoginResult(
success=False,
message=f"换 IP 后仍触发风控: {last_exc}",
code="risk",
sdid=self.sdid,
context=self.context,
request_id=self.request_id,
)
def _login_once(self) -> HuyaLoginResult:
"""单次登录流程(不换 IP),成功返回 Cookie。"""
logger.info(f"开始虎牙密码登录: {self.username}") logger.info(f"开始虎牙密码登录: {self.username}")
self.prepare_device() self.prepare_device()
self.init_udb_middle() self.init_udb_middle()
@@ -380,11 +447,21 @@ def login_huya_password(
username: str, username: str,
password: str, password: str,
cookie: Mapping[str, str] | str | None = None, cookie: Mapping[str, str] | str | None = None,
ua: str = DEFAULT_UA, ua: str | None = None,
proxies: Mapping[str, str] | None = None, proxies: Mapping[str, str] | None = None,
timeout: tuple[float, float] | None = None, timeout: tuple[float, float] | None = None,
app_id: str = APP_ID,
proxy_fetcher=None,
max_ip_retries: int = 3,
) -> HuyaLoginResult: ) -> HuyaLoginResult:
"""函数式入口,便于服务层直接调用。""" """函数式入口,便于服务层直接调用。
app_id 可选 APP_ID(5002, web) 或 APP_ID_MOBILE(5008, App 场景)
传 5008 且未显式传 ua 时,内部自动切换 appSign 与 Android WebView UA。
proxy_fetcher: 代理源(需提供 fetch_new_proxy()),扫码/短信风控时换 IP 重试。
max_ip_retries: 最多换 IP 次数。
"""
return HuyaPasswordLogin( return HuyaPasswordLogin(
username=username, username=username,
password=password, password=password,
@@ -392,4 +469,6 @@ def login_huya_password(
ua=ua, ua=ua,
proxies=proxies, proxies=proxies,
timeout=timeout, timeout=timeout,
).login() app_id=app_id,
proxy_fetcher=proxy_fetcher,
).login(max_ip_retries=max_ip_retries)