"""虎牙登录风控验证闭环。""" from __future__ import annotations import base64 import io import json import random import time from collections.abc import Mapping from http.cookies import SimpleCookie from urllib.parse import parse_qs, quote, urlparse import cv2 import execjs import numpy as np import requests from PIL import Image from loguru import logger 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): """虎牙风控验证失败。""" 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 后的滑块或点选验证。""" 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), ): 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.huyapk = "" self.scene_type = "" self.code = "" self.url_param_map: dict[str, str] = {} self.strategy_url = "" self.js_ctx = None self.behavior_events: list[dict] = [] @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: raise HuyaVerificationError("验证码图片为空") 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) 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) 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]) 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) -> dict: payload = { "urlParamMap": self.url_param_map, "wupData": "", "page": self._auth_page_url(), "behavior": self._encode_behavior(), "info": self.sdid, } payload.update(extra) return {"appId": "5002", "data": payload} 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) -> 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", "")), ) 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") if segment: logger.info("虎牙验证类型: 滑块") bg_bytes = self._decode_image(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) return int((opencv_x + yolo_x) / 2) logger.debug("虎牙滑块识别距离: yolo={}, opencv={}, 使用yolo", yolo_x, opencv_x) return yolo_x logger.info("虎牙验证类型: 点选") bg_bytes = self._decode_image(data.get("dTuStr")) tip_bytes = self._decode_image(data.get("hZiStr")) merged = self._vertical_concat_align_left(bg_bytes, tip_bytes) points = self.ocr.ocr_click_points(merged) return [[int(x), int(y) - 44] for x, y in points] 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": "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": "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")), ) return data def _solve_current_strategy(self) -> dict: """提交当前 strategy_url 对应的验证。""" 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) data = config.get("data") if isinstance(config.get("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 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: raise HuyaVerificationError( "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)}" ) self._compile_js(dx_js) point_payload = self._build_common_payload({"ver": 5}) last_result: dict = {} for index in range(3): if index: time.sleep(0.4) point_payload = self._build_common_payload({"ver": 5}) logger.info("虎牙验证失败后重试: {}/3", index + 1) point = self._get_track_point(gurl, point_payload) logger.debug(f"虎牙验证识别结果: {point}") if 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: """提交虎牙风控验证，返回 vurl 响应中的 data。""" self._parse_strategy(challenge_payload) result = self._solve_current_strategy() next_auth_url = result.get("nextAuthUrl") if isinstance(result, dict) else None if next_auth_url: logger.info("虎牙滑块后继续点选验证") self._parse_strategy(next_auth_url) result = self._solve_current_strategy() 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, ) -> dict: """函数式入口，便于登录流程直接调用。""" solver = HuyaVerificationSolver( cookie=cookie, ua=ua, sdid=sdid, session=session, proxies=proxies, ) return solver.solve(challenge_payload)