"""虎牙登录风控验证闭环。""" from __future__ import annotations import base64 import io import json import random import time import uuid from collections.abc import Mapping from datetime import UTC, datetime from http.cookies import SimpleCookie from pathlib import Path from typing import Any, cast from urllib.parse import parse_qs, quote, urlparse import re import cv2 import execjs import numpy as np import requests from loguru import logger from PIL import Image, ImageDraw, ImageFont from .capture import CaptchaCaptureRecord from .ocr import HuyaCaptchaOcr, default_ocr from .track import format_track, generate_slide_track DEFAULT_UA = ( "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " "AppleWebKit/537.36 (KHTML, like Gecko) " "Chrome/120.0.0.0 Safari/537.36" ) class HuyaVerificationError(RuntimeError): """虎牙风控验证失败。""" class HuyaNoSessionError(HuyaVerificationError): """config3 未下发图形验证会话（策略不匹配或无需滑块）。""" class HuyaQrAuthRequiredError(HuyaVerificationError): """风控策略为扫码验证（qr_auth），需已登录设备确认，无法自动化。""" class HuyaSmsAuthRequiredError(HuyaVerificationError): """风控策略为短信验证（sms_auth/sms_auth2），需接码/收码后提交。""" def __init__( self, challenge_payload: dict | str, message: str = "", mobile: str = "" ): self.challenge_payload = challenge_payload self.mobile = mobile super().__init__(message or f"虎牙风控要求短信验证，手机号: {mobile or '?'}") def _cookie_mapping(cookie: Mapping[str, str] | str | None) -> dict[str, str]: """兼容 dict 和 Cookie 字符串。""" if not cookie: return {} if isinstance(cookie, str): parsed = SimpleCookie() parsed.load(cookie) return {key: morsel.value for key, morsel in parsed.items()} return {str(key): str(value) for key, value in cookie.items()} class HuyaVerificationSolver: """处理虎牙 returnCode=10030/10039 后的滑块或点选验证。""" _cloze_corpus: set[str] | None = None def __init__( self, cookie: Mapping[str, str] | str | None = None, ua: str = DEFAULT_UA, sdid: str = "", session: requests.Session | None = None, ocr: HuyaCaptchaOcr | None = None, proxies: Mapping[str, str] | None = None, timeout: tuple[float, float] = (8, 20), app_id: str = "5002", page_url: str | None = None, use_touch_events: bool = False, account: str = "", capture_dir: str | Path | None = None, enable_capture: bool = True, sms_url: str = "", ): self.cookie = _cookie_mapping(cookie) self.ua = ua or DEFAULT_UA self.sdid = sdid or self.cookie.get("sdid", "") self.session = session or requests.Session() self.session.trust_env = False if self.cookie: self.session.cookies.update(self.cookie) if proxies: self.session.proxies = dict(proxies) self.session.headers.update( { "Accept-Language": "zh-CN,zh;q=0.9", "Cache-Control": "no-cache", "Connection": "keep-alive", "Origin": "https://aq.huya.com", "Pragma": "no-cache", "Referer": "https://aq.huya.com/", "User-Agent": self.ua, "accept": "application/json, text/plain, */*", "content-type": "application/json", } ) self.timeout = timeout self.ocr = ocr or default_ocr() self.account = account self.capture_dir = capture_dir self.enable_capture = enable_capture self.sms_url = sms_url self.current_capture: CaptchaCaptureRecord | None = None self.auth_type = "" self.huyapk = "" self.scene_type = "" self.code = "" self.url_param_map: dict[str, str] = {} self.strategy_url = "" self.js_ctx: Any = None self.behavior_events: list[dict] = [] self.app_id = app_id self.page_url = page_url or self._auth_page_url() self.use_touch_events = use_touch_events @staticmethod def _auth_page_url() -> str: """生成 auth 页面地址，形式参考易语言实现。""" return f"https://aq.huya.com/p/udb_login.html?v={str(int(time.time() * 1000))[:8]}00000" @staticmethod def _decode_image(base64_string: str | None) -> bytes: if not base64_string: return b"" if ";base64," in base64_string: base64_string = base64_string.split(";base64,")[-1] return base64.b64decode(base64_string) @staticmethod def _vertical_concat_align_left( bg_bytes: bytes, tip_bytes: bytes, image_format: str = "JPEG" ) -> bytes: bg = cv2.imdecode(np.frombuffer(bg_bytes, np.uint8), cv2.IMREAD_COLOR) tip = cv2.imdecode(np.frombuffer(tip_bytes, np.uint8), cv2.IMREAD_COLOR) if bg is None or tip is None: raise HuyaVerificationError("验证码图片解码失败") max_width = max(bg.shape[1], tip.shape[1]) def pad_right(img, target_width): _height, width = img.shape[:2] if width >= target_width: return img return cv2.copyMakeBorder( img, 0, 0, 0, target_width - width, cv2.BORDER_CONSTANT, value=(255, 255, 255), ) concat_img = np.vstack([pad_right(tip, max_width), pad_right(bg, max_width)]) success, buffer = cv2.imencode(f".{image_format.lower()}", concat_img) if not success: raise HuyaVerificationError("点选验证码图片编码失败") return buffer.tobytes() @staticmethod def _remove_black_borders(image: Image.Image) -> tuple[Image.Image, int]: if image.mode != "RGBA": image = image.convert("RGBA") bbox = image.getbbox() if not bbox: return image, 0 cropped_height = bbox[3] - bbox[1] black_border_height = image.height - cropped_height return image.crop(bbox), black_border_height @classmethod def _composite_slider_images( cls, bg_bytes: bytes, tip_bytes: bytes, y_position: int ) -> bytes: bg = Image.open(io.BytesIO(bg_bytes)).convert("RGB") tip = Image.open(io.BytesIO(tip_bytes)).convert("RGBA") tip, black_border_height = cls._remove_black_borders(tip) adjusted_y = y_position + int(black_border_height / 2) + 1 bg.paste(tip, (10, adjusted_y), tip) output = io.BytesIO() bg.save(output, format="PNG") return output.getvalue() @staticmethod def _opencv_slider_x(target_bytes: bytes, background_bytes: bytes) -> int: target = cv2.imdecode( np.frombuffer(target_bytes, np.uint8), cv2.IMREAD_ANYCOLOR ) background = cv2.imdecode( np.frombuffer(background_bytes, np.uint8), cv2.IMREAD_ANYCOLOR ) if target is None or background is None: raise HuyaVerificationError("滑块图片解码失败") target = cv2.cvtColor(cv2.Canny(target, 100, 200), cv2.COLOR_GRAY2RGB) background = cv2.cvtColor(cv2.Canny(background, 100, 200), cv2.COLOR_GRAY2RGB) result = cv2.matchTemplate(background, target, cv2.TM_CCOEFF_NORMED) _, _, _, max_loc = cv2.minMaxLoc(result) return int(max_loc[0]) @staticmethod def _opencv_rotate_angle( background_bytes: bytes, puzzle_bytes: bytes, default_cx: float = 195.0, default_cy: float = 90.0, ) -> int: """通过圆周边界颜色与边缘连续性计算旋转验证码的最佳旋转角度 (0~359度)。""" bg = cv2.imdecode(np.frombuffer(background_bytes, np.uint8), cv2.IMREAD_COLOR) puzzle_raw = cv2.imdecode( np.frombuffer(puzzle_bytes, np.uint8), cv2.IMREAD_UNCHANGED ) if bg is None or puzzle_raw is None: raise HuyaVerificationError("旋转验证码图片解码失败") bg_gray = cv2.cvtColor(bg, cv2.COLOR_BGR2GRAY).astype(np.float32) p_gray = cv2.cvtColor(puzzle_raw[:, :, :3], cv2.COLOR_BGR2GRAY).astype( np.float32 ) gray_u8 = cv2.cvtColor(bg, cv2.COLOR_BGR2GRAY) black_mask = (gray_u8 < 25).astype(np.uint8) * 255 contours, _ = cv2.findContours( black_mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE ) cx, cy, r_hole = default_cx, default_cy, 58.0 for c in contours: if cv2.contourArea(c) > 3000: (tcx, tcy), tr = cv2.minEnclosingCircle(c) if abs(tcx - default_cx) < 35 and abs(tcy - default_cy) < 25: cx, cy, r_hole = tcx, tcy, tr break ph, pw = puzzle_raw.shape[:2] pcx, pcy = pw / 2.0, ph / 2.0 angles = np.linspace(0, 2 * np.pi, 360, endpoint=False) r_out_list = [r_hole + 2.0, r_hole + 4.0] r_in_list = [r_hole - 5.0, r_hole - 7.0] bg_outer_vals_list = [] for ro in r_out_list: bx = cx + ro * np.cos(angles) by = cy + ro * np.sin(angles) vals = cv2.remap( bg_gray, bx.astype(np.float32), by.astype(np.float32), cv2.INTER_LINEAR, ).flatten() bg_outer_vals_list.append(vals) bg_outer_vals = np.mean(bg_outer_vals_list, axis=0) best_deg = 0 best_corr = -1.0 for deg in range(360): rad = np.radians(deg) p_inner_vals_list = [] for ri in r_in_list: px = pcx + ri * np.cos(angles - rad) py = pcy + ri * np.sin(angles - rad) pvals = cv2.remap( p_gray, px.astype(np.float32), py.astype(np.float32), cv2.INTER_LINEAR, ).flatten() p_inner_vals_list.append(pvals) p_inner_vals = np.mean(p_inner_vals_list, axis=0) corr = np.corrcoef(bg_outer_vals, p_inner_vals)[0, 1] if corr > best_corr: best_corr = corr best_deg = deg return best_deg @classmethod def _get_cloze_corpus(cls) -> set[str]: if cls._cloze_corpus is None: corpus: set[str] = set() base_dir = Path(__file__).resolve().parent for filename in ("idioms.txt", "poems.txt"): file_path = base_dir / filename if file_path.is_file(): try: corpus.update( file_path.read_text(encoding="utf-8").splitlines() ) except Exception as exc: # noqa: BLE001 logger.debug(f"[solver] 加载语料 {filename} 失败: {exc}") cls._cloze_corpus = corpus return cls._cloze_corpus @classmethod def _find_chinese_font( cls, size: int = 30 ) -> ImageFont.FreeTypeFont | ImageFont.ImageFont: font_candidates = [ "/System/Library/Fonts/PingFang.ttc", "/System/Library/Fonts/STHeiti Medium.ttc", "/System/Library/Fonts/STHeiti Light.ttc", "/System/Library/Fonts/Hiragino Sans GB.ttc", "/System/Library/Fonts/Supplemental/Songti.ttc", "/usr/share/fonts/truetype/wqy/wqy-microhei.ttc", "/usr/share/fonts/truetype/wqy/wqy-zenhei.ttc", "/usr/share/fonts/opentype/noto/NotoSansCJK-Regular.ttc", "/usr/share/fonts/truetype/droid/DroidSansFallbackFull.ttf", "C:/Windows/Fonts/msyh.ttc", "C:/Windows/Fonts/simhei.ttf", ] for font_path in font_candidates: if Path(font_path).is_file(): try: return ImageFont.truetype(font_path, size) except Exception: continue return ImageFont.load_default() @classmethod def _synthesize_text_tip(cls, chars: list[str]) -> bytes: """为没有提示小图的文本点选题目动态合成 44px 高度的提示图像。""" if not chars: return b"" font = cls._find_chinese_font(30) char_box_size = 44 total_width = len(chars) * char_box_size image = Image.new("RGB", (total_width, char_box_size), (255, 255, 255)) draw = ImageDraw.Draw(image) for i, ch in enumerate(chars): offset_x = i * char_box_size + 7 offset_y = 5 draw.text((offset_x, offset_y), ch, fill=(0, 0, 0), font=font) buf = io.BytesIO() image.save(buf, format="PNG") return buf.getvalue() @classmethod def _extract_target_chars_from_question(cls, question: str) -> list[str]: """从点选题目中提取需要点击的目标汉字（支持完整成语/菜名及古诗/成语填空题）。""" if not question: return [] m = re.search(r'["“]([^"”]+)["”]', question) if not m: return [] text = m.group(1).strip() if "_" in text: regex = "^" + text.replace("_", "(.)") + "$" compiled = re.compile(regex) for phrase in cls._get_cloze_corpus(): matched = compiled.match(phrase) if matched: return list(matched.groups()) return [] return [ch for ch in text if "\u4e00" <= ch <= "\u9fa5"] def _request_json(self, method: str, url: str, **kwargs) -> dict: response = self.session.request(method, url, timeout=self.timeout, **kwargs) response.raise_for_status() try: return response.json() except json.JSONDecodeError as exc: raise HuyaVerificationError( f"虎牙验证接口返回不是 JSON: {response.text[:200]}" ) from exc def _compile_js(self, dx_js: str) -> None: js_data = """ window = globalThis; $$self_js$$ function get_regoin_arr(segment, huyapk) { return JSON.parse(atob(window.UdbCipher.decrypt(segment, huyapk))).ord } function get_act(act_data, huyapk) { act_data = encodeURI(JSON.stringify(act_data)) return window.UdbCipher.encrypt(btoa(act_data), huyapk) } """ self.js_ctx = execjs.compile(js_data.replace("$$self_js$$", dx_js)) def _build_common_payload( self, extra: dict, info_override: str | None = None ) -> dict: payload = { "urlParamMap": self.url_param_map, "wupData": "", "page": self.page_url, "behavior": self._encode_behavior(), "info": info_override if info_override is not None else self.sdid, } payload.update(extra) return {"appId": self.app_id, "data": payload} def _csid(self) -> str: """App WebView 场景 get3 使用 csid_ 前缀会话 id。""" return f"csid_{uuid.uuid4().hex}" def _pointer_action(self, action: str) -> str: """web 用 mouse 事件，App WebView 场景用 touch 事件。""" if not self.use_touch_events: return action return { "mousedown": "touchstart", "mousemove": "touchmove", "mouseup": "touchend", }.get(action, action) def _encode_behavior(self) -> str: """编码浏览器行为，失败重试时会携带前一次点击记录。""" return quote( json.dumps(self.behavior_events, separators=(",", ":")), safe="~()*!.'" ) @staticmethod def _extract_strategy_url(challenge_payload: dict | str) -> str: """兼容完整响应、data 子对象、data URL 字符串三种输入。""" if isinstance(challenge_payload, str): return challenge_payload data = challenge_payload.get("data") if isinstance(data, str) and data.startswith("http"): return data if isinstance(data, str): try: data = json.loads(data) except json.JSONDecodeError: data = {} strategy_root = data if isinstance(data, dict) else challenge_payload try: return strategy_root["strategys"][0]["data"] except (KeyError, IndexError, TypeError) as exc: raise HuyaVerificationError("虎牙响应中没有 strategys[0].data") from exc def _parse_strategy(self, challenge_payload: dict | str, step: int = 1) -> None: strategy_url = self._extract_strategy_url(challenge_payload) query_params = parse_qs(urlparse(strategy_url).query) self.url_param_map = {key: value[0] for key, value in query_params.items()} self.scene_type = self.url_param_map.get("sceneType", "") self.strategy_url = strategy_url if not self.url_param_map.get("param"): raise HuyaVerificationError("虎牙验证链接缺少 param") logger.debug( "虎牙验证参数: sceneType={}, mobile={}, param长度={}", self.scene_type, self.url_param_map.get("mobile", ""), len(self.url_param_map.get("param", "")), ) strategies = [] if isinstance(challenge_payload, dict): strategies = (challenge_payload.get("data") or {}).get("strategys") or [] if strategies: try: logger.warning( "虎牙登录风控strategys完整结构: {}", json.dumps(strategies, ensure_ascii=False)[:1200], ) except Exception as exc: # noqa: BLE001 logger.debug(f"风控策略日志序列化失败: {exc}") strategy_url_lower = strategy_url.lower() # 判定依据是 URL 路径，不是 strategy 数值。 # 实测(2026-08-25): strategy=64 时 pt_auth.html 是滑块、qr_auth.html 才是扫码， # 旧的 `strategy==64 → 扫码` 硬编码会把可自动化的滑块误杀成死路。 is_qr_auth = "qr_auth" in strategy_url_lower account_id = self.account or self.url_param_map.get("mobile", "") if is_qr_auth: if self.enable_capture: cap = CaptchaCaptureRecord( auth_type="qr_auth", account=account_id, page_url=self.strategy_url, scene_type=self.scene_type, output_base_dir=Path(self.capture_dir) if self.capture_dir else None, ) cap.capture_qr_auth_data( mobile=self.url_param_map.get("mobile", "?"), page_url=self.strategy_url, scene_type=self.scene_type, step=step, raw_params=self.url_param_map, ) logger.warning( "[solver] 虎牙风控在第 {} 步要求扫码验证 (qr_auth)，需 App 确认，账号: {}", step, self.url_param_map.get("mobile", "?"), ) raise HuyaQrAuthRequiredError( "虎牙风控要求扫码验证(qr_auth)，需已登录App设备确认，" f"账号: {self.url_param_map.get('mobile', '?')}" ) if "sms_auth" in strategy_url_lower: mobile = self.url_param_map.get("mobile", "?") if self.enable_capture: cap = CaptchaCaptureRecord( auth_type="sms_auth", account=account_id, page_url=self.strategy_url, scene_type=self.scene_type, output_base_dir=Path(self.capture_dir) if self.capture_dir else None, ) cap.capture_sms_auth_data( mobile=mobile, page_url=self.strategy_url, scene_type=self.scene_type, step=step, raw_params=self.url_param_map, ) if self.sms_url: logger.info( "[solver] 虎牙风控在第 {} 步要求短信验证 (sms_auth)，已配置接码链接，准备自动发码并接码过验", step, ) self.auth_type = "sms_auth" return logger.warning( "[solver] 虎牙风控在第 {} 步要求短信验证 (sms_auth)，手机号: {}", step, mobile, ) raise HuyaSmsAuthRequiredError( challenge_payload, f"虎牙风控要求短信验证(sms_auth)，手机号: {mobile}", mobile=mobile, ) if "rotate_auth" in strategy_url_lower or "/auth/xz/" in strategy_url_lower: self.auth_type = "rotate_auth" elif "dx_auth" in strategy_url_lower or "/auth/dx/" in strategy_url_lower: self.auth_type = "dx_auth" else: self.auth_type = "pt_auth" if self.enable_capture: self.current_capture = CaptchaCaptureRecord( auth_type=self.auth_type, account=account_id, page_url=self.strategy_url, scene_type=self.scene_type, output_base_dir=Path(self.capture_dir) if self.capture_dir else None, ) self.behavior_events = [] def _restore_slider_background( self, bg_bytes: bytes, segment: str, huyapk: str ) -> bytes: recover_list = self.js_ctx.call("get_regoin_arr", segment, huyapk) image = Image.open(io.BytesIO(bg_bytes)).convert("RGB") width, height = image.size captcha = Image.new("RGB", (width, height)) for index, region_index in enumerate(recover_list): region = image.crop((region_index * 26, 0, region_index * 26 + 26, 180)) captcha.paste(region, (index * 26, 0, index * 26 + 26, 180)) output = io.BytesIO() captcha.save(output, format="JPEG") return output.getvalue() def _get_track_point(self, url: str, payload: dict): logger.debug("虎牙验证获取图片: {}", url) result = self._request_json("post", url, json=payload) if result.get("returnCode") != 0 or not result.get("data", {}).get("code"): raise HuyaVerificationError(f"获取验证码图片失败: {result}") data = result["data"] self.code = data["code"] self.huyapk = data["huyapk"] segment = data.get("segment") is_rotate = ( self.auth_type == "rotate_auth" or "/auth/xz/" in url or "rotate" in self.strategy_url ) if is_rotate: logger.info("虎牙验证类型: 旋转验证码 (rotate_auth / xz)") bg_bytes = self._decode_image(data.get("dtuStr") or data.get("dTuStr")) tip_bytes = self._decode_image(data.get("puzzleStr")) restored_bg = ( self._restore_slider_background(bg_bytes, segment, self.huyapk) if segment else bg_bytes ) angle = self._opencv_rotate_angle(restored_bg, tip_bytes) logger.info(f"虎牙旋转验证码识别角度: {angle}°") if self.current_capture: try: self.current_capture.capture_rotate_data( code=self.code, dtu_bytes=bg_bytes, puzzle_bytes=tip_bytes, restored_bg_bytes=restored_bg, angle=angle, raw_data=data, ) except Exception as exc: # noqa: BLE001 logger.debug(f"[solver] 记录旋转验证码捕获数据异常: {exc}") return angle if segment: logger.info("虎牙验证类型: 滑块") bg_bytes = self._decode_image(data.get("dtuStr") or data.get("dTuStr")) tip_bytes = self._decode_image(data.get("puzzleStr")) restored_bg = self._restore_slider_background( bg_bytes, segment, self.huyapk ) composite = self._composite_slider_images( restored_bg, tip_bytes, int(data.get("location_y") or 0) ) slider_box = self.ocr.ocr_slider_box(composite) yolo_x = int(slider_box[0] - 13) opencv_x = self._opencv_slider_x(tip_bytes, restored_bg) if abs(opencv_x - yolo_x) < 10: logger.debug("虎牙滑块识别距离: yolo={}, opencv={}", yolo_x, opencv_x) final_x = int((opencv_x + yolo_x) / 2) else: logger.debug( "虎牙滑块识别距离: yolo={}, opencv={}, 使用yolo", yolo_x, opencv_x ) final_x = yolo_x if self.current_capture: try: self.current_capture.capture_slider_data( code=self.code, dtu_bytes=bg_bytes, puzzle_bytes=tip_bytes, restored_bg_bytes=restored_bg, composite_bytes=composite, point_x=final_x, raw_data=data, ) except Exception as exc: # noqa: BLE001 logger.debug(f"[solver] 记录滑块捕获数据异常: {exc}") return final_x logger.info("虎牙验证类型: 点选 (dx_auth)") bg_raw = data.get("dTuStr") or data.get("dtuStr") or data.get("bgStr") bg_bytes = self._decode_image(bg_raw) if not bg_bytes: raise HuyaVerificationError( f"点选验证码背景图为空: keys={list(data.keys())}" ) tip_raw = data.get("hZiStr") or data.get("hziStr") or data.get("puzzleStr") tip_bytes = self._decode_image(tip_raw) if tip_raw else b"" if tip_bytes: merged = self._vertical_concat_align_left(bg_bytes, tip_bytes) points = self.ocr.ocr_click_points(merged) final_points = [[int(x), int(y) - 44] for x, y in points] else: question = str(data.get("question") or "") target_chars = self._extract_target_chars_from_question(question) if target_chars: logger.info(f"虎牙点选从题目解析出目标字: {target_chars}") synth_tip = self._synthesize_text_tip(target_chars) merged = self._vertical_concat_align_left(bg_bytes, synth_tip) points = self.ocr.ocr_click_points(merged) final_points = [[int(x), int(y) - 44] for x, y in points] tip_bytes = synth_tip else: logger.info(f"虎牙点选题目未能提取明确目标字: {question}") final_points = [] if self.current_capture: try: self.current_capture.capture_click_data( code=self.code, dtu_bytes=bg_bytes, hzi_bytes=tip_bytes, points=final_points, raw_data=data, ) except Exception as exc: # noqa: BLE001 logger.debug(f"[solver] 记录点选捕获数据异常: {exc}") return final_points def _send_rotate_verify(self, url: str, angle: float) -> dict: drag_distance = max(5, int((angle / 360.0) * 260)) track_points = generate_slide_track(drag_distance) act_data = { "point": int(round(angle)), "travel": format_track(track_points), "code": self.code, "sceneType": self.scene_type, "endTime": int(track_points[-1][-1]) - int(track_points[0][-1]), } act = self.js_ctx.call("get_act", act_data, self.huyapk) self.behavior_events.append( { "action": self._pointer_action("mousedown"), "id": "", "txt": "", "clientX": track_points[0][0], "clientY": track_points[0][1], "clientWidth": 350, "clientHeight": 268, "isTrusted": True, "timeStamp": random.randint(1800, 5500), "nowTime": int(time.time() * 1000), } ) payload = self._build_common_payload({"act": act, "ver": 5}) return self._submit_verify(url, payload) def _send_slider_verify(self, url: str, point: int) -> dict: track_points = generate_slide_track(point) act_data = { "point": int(point), "travel": format_track(track_points), "code": self.code, "sceneType": self.scene_type, "endTime": int(track_points[-1][-1]) - int(track_points[0][-1]), } act = self.js_ctx.call("get_act", act_data, self.huyapk) self.behavior_events.append( { "action": self._pointer_action("mousedown"), "id": "", "txt": "", "clientX": track_points[0][0], "clientY": track_points[0][1], "clientWidth": 350, "clientHeight": 268, "isTrusted": True, "timeStamp": random.randint(1800, 5500), "nowTime": int(time.time() * 1000), } ) payload = self._build_common_payload({"act": act, "ver": 5}) return self._submit_verify(url, payload) def _send_click_verify(self, url: str, points: list[list[int]]) -> dict: if not points: raise HuyaVerificationError("点选坐标识别为空") start_time = int(time.time() * 1000) - random.randint(5000, 10000) click_data = [] current_time = start_time for point in points: current_time += random.randint(1000, 2000) click_data.append([int(point[0]), int(point[1]), current_time]) interval_time_list = [item[-1] for item in click_data] act_data = { "clickData": ",".join(f"{x}_{y}" for x, y in points), "intervalTime": "_".join(map(str, interval_time_list)), "endTime": interval_time_list[-1] - interval_time_list[0], "sceneType": self.scene_type, "code": self.code, } act = self.js_ctx.call("get_act", act_data, self.huyapk) txt_list = ["", "1", "1\\n2"] behavior_list = [] for index, item in enumerate(click_data): behavior_list.append( { "action": self._pointer_action("mousedown"), "id": "", "txt": txt_list[index] if index < len(txt_list) else "", "clientX": item[0], "clientY": item[1], "clientWidth": 350, "clientHeight": 268, "isTrusted": True, "timeStamp": random.randint(2000, 5000), "nowTime": item[2], } ) self.behavior_events.extend(behavior_list) payload = self._build_common_payload({"act": act, "ver": 5}) return self._submit_verify(url, payload) def _submit_verify(self, url: str, payload: dict) -> dict: logger.debug("虎牙验证提交: {}", url) body = json.dumps(payload, separators=(",", ":")) result = self._request_json("post", url, params={"lock": "true"}, data=body) data = result.get("data") or result if isinstance(data, dict): logger.debug( "虎牙验证提交结果: authId={}, nextAuthUrl={}", bool(data.get("authId")), bool(data.get("nextAuthUrl")), ) if self.current_capture: try: self.current_capture.record_submit_result(data) except Exception as exc: # noqa: BLE001 logger.debug(f"[solver] 记录提交结果异常: {exc}") return data def _solve_sms_strategy(self, wait_seconds: float = 90.0) -> dict: """自动通过 sms_url 接码并提交虎牙 sms_auth 验证。""" from core.sms_provider import SmsLine, SmsProviderClient, detect_provider mobile = self.url_param_map.get("mobile", "?") logger.info( f"[solver] 开始自动接码流程 (手机: {mobile}, 接码URL: {self.sms_url[:40]}...)" ) # 1. 触发虎牙官方发送短信验证码 send_url = "https://udbrtt.huya.com/auth/sms/send-sms-code" send_payload = { "data": { "urlParamMap": self.url_param_map, "ver": 5, } } sent_at = datetime.now(UTC) try: body = json.dumps(send_payload, separators=(",", ":")) send_res = self._request_json("post", send_url, data=body) logger.info(f"[solver] 虎牙短信验证码发送结果: {send_res}") except Exception as exc: logger.warning(f"[solver] 虎牙安全验证发送短信失败: {exc}") raise HuyaVerificationError(f"发送虎牙短信验证码失败: {exc}") from exc # 2. 轮询接码链接获取验证码 client = SmsProviderClient(timeout=(8, 15)) item = SmsLine( phone=mobile, url=self.sms_url, provider=detect_provider(self.sms_url), raw=f"{mobile}----{self.sms_url}", ) deadline = time.monotonic() + wait_seconds poll_interval = 2.5 sms_code = "" while time.monotonic() < deadline: poll_result = client.poll(item, after_time=sent_at) if poll_result.has_code: sms_code = poll_result.code logger.info(f"[solver] 成功从接码链接收到验证码: {sms_code}") break time.sleep(poll_interval) if not sms_code: raise HuyaSmsAuthRequiredError( self.strategy_url, f"接码超时（{int(wait_seconds)}秒内未从接码链接获取到新验证码）", mobile=mobile, ) # 3. 提交验证码给虎牙官方 verify_url = "https://udbrtt.huya.com/auth/sms/verify-sms-code" sms_code_hex = sms_code.encode("utf-8").hex() verify_payload = { "encrypt": 1, "data": { "urlParamMap": self.url_param_map, "behavior": self._encode_behavior(), "info": self.sdid, "page": self.page_url, "smsCode": sms_code_hex, "wupData": "", "ver": 5, }, } logger.debug(f"[solver] 提交虎牙短信验证码: {sms_code} (hex={sms_code_hex})") return self._submit_verify(verify_url, verify_payload) def _solve_current_strategy(self) -> dict: """提交当前 strategy_url 对应的验证。""" if self.auth_type == "sms_auth": return self._solve_sms_strategy() config_payload = self._build_common_payload({"cTuIndex": 0}) config = {} gurl = "" vurl = "" dx_js = "" for index in range(2): logger.debug("虎牙验证获取config3 ({}/2)", index + 1) config = self._request_json( "post", "https://udbrtt.huya.com/auth/client/config3", json=config_payload, ) raw_data = config.get("data") data = cast(dict[str, Any], raw_data) if isinstance(raw_data, dict) else {} gurl = str(data.get("gurl") or "") vurl = str(data.get("vurl") or "") dx_js = str(data.get("js") or "") logger.debug( "虎牙config3返回: returnCode={}, desc={}, gurl={}, vurl={}, js长度={}", config.get("returnCode"), config.get("description") or config.get("message") or "", bool(gurl), bool(vurl), len(dx_js), ) if not (gurl and vurl): logger.warning( "虎牙config3未下发验证会话, 完整响应: {}", json.dumps(config, ensure_ascii=False)[:1500], ) if gurl and vurl and dx_js: break if index == 0 and self.url_param_map.get("sceneType") != "4": self.url_param_map["sceneType"] = "4" self.scene_type = "4" config_payload = self._build_common_payload({"cTuIndex": 0}) logger.debug("虎牙config3取参不完整，改用旧流程sceneType=4重试") time.sleep(0.3) if not gurl or not vurl or not dx_js: data_summary = { k: str(v)[:80] for k, v in (data or {}).items() if k != "js" } raise HuyaNoSessionError( "config3未下发验证会话: " f"returnCode={config.get('returnCode')}, " f"description={config.get('description') or config.get('message') or ''}, " f"gurl={bool(gurl)}, vurl={bool(vurl)}, js={bool(dx_js)}, " f"data={json.dumps(data_summary, ensure_ascii=False)[:600]}" ) self._compile_js(dx_js) point_payload = self._build_common_payload( {"ver": 5}, info_override=self._csid() if self.use_touch_events else None ) last_result: dict = {} for index in range(3): if index: time.sleep(0.4) point_payload = self._build_common_payload( {"ver": 5}, info_override=self._csid() if self.use_touch_events else None, ) logger.info("虎牙验证失败后重试: {}/3", index + 1) point = self._get_track_point(gurl, point_payload) logger.debug(f"虎牙验证识别结果: {point}") if ( self.auth_type == "rotate_auth" or "/auth/xz/" in vurl or "rotate" in self.strategy_url ): last_result = self._send_rotate_verify(vurl, float(point)) elif isinstance(point, list): last_result = self._send_click_verify(vurl, point) else: last_result = self._send_slider_verify(vurl, int(point)) if isinstance(last_result, dict) and ( last_result.get("authId") or last_result.get("nextAuthUrl") ): return last_result logger.info( "虎牙验证提交未通过: returnCode={}, message={} ({}/3)", last_result.get("returnCode") if isinstance(last_result, dict) else "", last_result.get("message") if isinstance(last_result, dict) else "", index + 1, ) return last_result def solve(self, challenge_payload: dict | str) -> dict: """提交虎牙风控验证，支持多级风控流转 (如 pt_auth -> dx_auth -> ...)。""" current_challenge: dict | str = challenge_payload result: dict = {} for step in range(5): self._parse_strategy(current_challenge, step=step + 1) result = self._solve_current_strategy() if not isinstance(result, dict): break if result.get("authId"): logger.info(f"虎牙验证在第 {step + 1} 步成功获取 authId") break next_auth_url = result.get("nextAuthUrl") if not next_auth_url: break logger.info( f"虎牙验证第 {step + 1} 步完成，进入下一阶段验证: {next_auth_url[:80]}..." ) current_challenge = next_auth_url return result def solve_huya_verification( challenge_payload: dict | str, cookie: Mapping[str, str] | str | None = None, ua: str = DEFAULT_UA, sdid: str = "", session: requests.Session | None = None, proxies: Mapping[str, str] | None = None, app_id: str = "5002", page_url: str | None = None, use_touch_events: bool = False, account: str = "", capture_dir: str | Path | None = None, enable_capture: bool = True, sms_url: str = "", ) -> dict: """函数式入口，便于登录流程直接调用。""" solver = HuyaVerificationSolver( cookie=cookie, ua=ua, sdid=sdid, session=session, proxies=proxies, app_id=app_id, page_url=page_url, use_touch_events=use_touch_events, account=account, capture_dir=capture_dir, enable_capture=enable_capture, sms_url=sms_url, ) return solver.solve(challenge_payload)