diff --git a/core/huya/login.py b/core/huya/login.py index fc1150d..e8541ff 100644 --- a/core/huya/login.py +++ b/core/huya/login.py @@ -28,6 +28,12 @@ APP_SIGN = "1ce3bf682483d03f146f58232ec10635" # appKey 来自 APK 资源 R.string.cgs。 APP_ID_MOBILE = "5008" 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" PASSWORD_LOGIN_URI = "30001" DF_TOKEN_URL = "https://df.huya.com/web/df/token" @@ -131,14 +137,24 @@ class HuyaPasswordLogin: username: str, password: str, cookie: Mapping[str, str] | str | None = None, - ua: str = DEFAULT_UA, + ua: str | None = None, session: requests.Session | None = None, proxies: Mapping[str, str] | None = None, timeout: tuple[float, float] | None = None, + app_id: str = APP_ID, + proxy_fetcher=None, ): self.username = username 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.request_id = generate_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.sdid = "" self.sdid_source = "" + # 代理源(鸭子类型,需提供 fetch_new_proxy() -> Optional[str])。 + # 用于扫码/短信风控时换 IP 重试,尝试把风控等级降回滑块。 + self.proxy_fetcher = proxy_fetcher self.session = session or requests.Session() self.session.trust_env = False @@ -250,8 +269,8 @@ class HuyaPasswordLogin: "uri": PASSWORD_LOGIN_URI, "version": APP_VERSION, "context": self.context, - "appId": APP_ID, - "appSign": APP_SIGN, + "appId": self.app_id, + "appSign": self.app_sign, "authId": auth_id, "sdid": self.sdid, "lcid": LCID, @@ -310,8 +329,56 @@ class HuyaPasswordLogin: data = payload.get("data") return data if isinstance(data, dict) else {} - def login(self) -> HuyaLoginResult: - """执行密码登录,成功返回 Cookie。""" + def _swap_proxy(self) -> bool: + """换一个代理 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}") self.prepare_device() self.init_udb_middle() @@ -380,11 +447,21 @@ def login_huya_password( username: str, password: str, cookie: Mapping[str, str] | str | None = None, - ua: str = DEFAULT_UA, + ua: str | None = None, proxies: Mapping[str, str] | None = None, timeout: tuple[float, float] | None = None, + app_id: str = APP_ID, + proxy_fetcher=None, + max_ip_retries: int = 3, ) -> 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( username=username, password=password, @@ -392,4 +469,6 @@ def login_huya_password( ua=ua, proxies=proxies, timeout=timeout, - ).login() + app_id=app_id, + proxy_fetcher=proxy_fetcher, + ).login(max_ip_retries=max_ip_retries)