yml2213
f101414f5e
- 新增登录失败异常分型统计：支持「必须绑定手机号」、「极验滑块」、「凭证错误」、「邮箱取码失败」、「邮箱配额已满」、「网络异常」分型识别、生命周期联动（待人工）与一键筛选； - 新增账号名单与明细导出功能：支持勾选多选或按当前筛选条件全选，可导出纯用户名名单 (TXT) 或 UTF-8 BOM 结构化明细报表 (CSV)，并支持一键复制到剪贴板； - 优化筛选条件记忆机制：支持页面切换/返回后自动保留所有筛选与排序状态，新增顶部「重置全部筛选」快捷按钮； - 模块化拆分 AccountsPage：将 2100+ 行的大文件拆解至 pages/accounts/ 子目录（包含状态卡片、筛选面板、工具栏、操作弹窗与表格构建器），提升可维护性； - 验证方式：新增 5 个后端自动化单测 (test_account_failure_and_export.py)，全量 694 个测试用例通过，前端 TypeScript 编译与 pre-commit 代码规范检查全部通过。
1034 lines
41 KiB
Python
1034 lines
41 KiB
Python
"""虎牙登录风控验证闭环。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import io
|
|
import json
|
|
import random
|
|
import re
|
|
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 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)
|