迁移虎牙验证码核心逻辑
This commit is contained in:
yml2213
@@ -8,6 +8,9 @@ __all__ = [
|
||||
"HuyaWssClient",
|
||||
"GetUserScoreReq",
|
||||
"GetUserScoreResp",
|
||||
"HuyaVerificationError",
|
||||
"HuyaVerificationSolver",
|
||||
"solve_huya_verification",
|
||||
]
|
||||
|
||||
|
||||
@@ -21,4 +24,14 @@ def __getattr__(name: str):
|
||||
}
|
||||
globals().update(values)
|
||||
return values[name]
|
||||
if name in {"HuyaVerificationError", "HuyaVerificationSolver", "solve_huya_verification"}:
|
||||
from .verification import HuyaVerificationError, HuyaVerificationSolver, solve_huya_verification
|
||||
|
||||
values = {
|
||||
"HuyaVerificationError": HuyaVerificationError,
|
||||
"HuyaVerificationSolver": HuyaVerificationSolver,
|
||||
"solve_huya_verification": solve_huya_verification,
|
||||
}
|
||||
globals().update(values)
|
||||
return values[name]
|
||||
raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
"""虎牙风控验证求解器。"""
|
||||
|
||||
from .solver import HuyaVerificationError, HuyaVerificationSolver, solve_huya_verification
|
||||
|
||||
__all__ = [
|
||||
"HuyaVerificationError",
|
||||
"HuyaVerificationSolver",
|
||||
"solve_huya_verification",
|
||||
]
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -0,0 +1,248 @@
|
||||
"""虎牙滑块/点选验证码 OCR。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from functools import lru_cache
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
import cv2
|
||||
import numpy as np
|
||||
import onnxruntime as ort
|
||||
from PIL import Image
|
||||
from scipy.optimize import linear_sum_assignment
|
||||
|
||||
|
||||
MODEL_DIR = Path(__file__).resolve().parent / "models"
|
||||
|
||||
|
||||
class YoloOnnx:
|
||||
"""轻量 YOLO ONNX 推理封装。"""
|
||||
|
||||
def __init__(self, model_path: str | Path, classes: list[str], providers: list[str] | None = None):
|
||||
providers = providers or ["CPUExecutionProvider"]
|
||||
ort.set_default_logger_severity(3)
|
||||
self.session = ort.InferenceSession(str(model_path), providers=providers)
|
||||
self.input_name = self.session.get_inputs()[0].name
|
||||
self.output_name = self.session.get_outputs()[0].name
|
||||
self.input_shape = self.session.get_inputs()[0].shape[2:]
|
||||
self.names = classes
|
||||
self.scale = 1.0
|
||||
self.pad_top = 0
|
||||
self.pad_left = 0
|
||||
|
||||
@staticmethod
|
||||
def xywh2xyxy(value: np.ndarray) -> np.ndarray:
|
||||
result = np.copy(value)
|
||||
result[..., 0] = value[..., 0] - value[..., 2] / 2
|
||||
result[..., 1] = value[..., 1] - value[..., 3] / 2
|
||||
result[..., 2] = value[..., 0] + value[..., 2] / 2
|
||||
result[..., 3] = value[..., 1] + value[..., 3] / 2
|
||||
return result
|
||||
|
||||
def preprocess(self, image: Image.Image) -> np.ndarray:
|
||||
img = cv2.cvtColor(np.array(image), cv2.COLOR_RGB2BGR)
|
||||
height, width = img.shape[:2]
|
||||
|
||||
self.scale = min(self.input_shape[0] / height, self.input_shape[1] / width)
|
||||
new_size = (int(width * self.scale), int(height * self.scale))
|
||||
resized = cv2.resize(img, new_size, interpolation=cv2.INTER_LINEAR)
|
||||
|
||||
dh = self.input_shape[0] - new_size[1]
|
||||
dw = self.input_shape[1] - new_size[0]
|
||||
self.pad_top = dh // 2
|
||||
pad_bottom = dh - self.pad_top
|
||||
self.pad_left = dw // 2
|
||||
pad_right = dw - self.pad_left
|
||||
|
||||
padded = cv2.copyMakeBorder(
|
||||
resized,
|
||||
self.pad_top,
|
||||
pad_bottom,
|
||||
self.pad_left,
|
||||
pad_right,
|
||||
cv2.BORDER_CONSTANT,
|
||||
value=(114, 114, 114),
|
||||
)
|
||||
input_tensor = padded.transpose(2, 0, 1).astype(np.float32) / 255.0
|
||||
return np.expand_dims(input_tensor, axis=0)
|
||||
|
||||
@staticmethod
|
||||
def nms(boxes: np.ndarray, scores: np.ndarray, iou_threshold: float = 0.3) -> list[int]:
|
||||
order = scores.argsort()[::-1]
|
||||
keep = []
|
||||
|
||||
while order.size > 0:
|
||||
index = order[0]
|
||||
keep.append(index)
|
||||
|
||||
xx1 = np.maximum(boxes[index, 0], boxes[order[1:], 0])
|
||||
yy1 = np.maximum(boxes[index, 1], boxes[order[1:], 1])
|
||||
xx2 = np.minimum(boxes[index, 2], boxes[order[1:], 2])
|
||||
yy2 = np.minimum(boxes[index, 3], boxes[order[1:], 3])
|
||||
|
||||
width = np.maximum(0.0, xx2 - xx1)
|
||||
height = np.maximum(0.0, yy2 - yy1)
|
||||
intersection = width * height
|
||||
area_i = (boxes[index, 2] - boxes[index, 0]) * (boxes[index, 3] - boxes[index, 1])
|
||||
area_j = (boxes[order[1:], 2] - boxes[order[1:], 0]) * (
|
||||
boxes[order[1:], 3] - boxes[order[1:], 1]
|
||||
)
|
||||
iou = intersection / (area_i + area_j - intersection + 1e-7)
|
||||
|
||||
inds = np.where(iou <= iou_threshold)[0]
|
||||
order = order[inds + 1]
|
||||
|
||||
return keep
|
||||
|
||||
def postprocess(
|
||||
self,
|
||||
outputs: list[np.ndarray],
|
||||
conf_threshold: float = 0.1,
|
||||
iou_threshold: float = 0.3,
|
||||
) -> list[dict]:
|
||||
predictions = np.squeeze(outputs[0]).T
|
||||
class_scores = predictions[:, 4:]
|
||||
scores = np.max(class_scores, axis=1)
|
||||
class_ids = np.argmax(class_scores, axis=1)
|
||||
|
||||
valid_indices = scores > conf_threshold
|
||||
predictions = predictions[valid_indices]
|
||||
scores = scores[valid_indices]
|
||||
class_ids = class_ids[valid_indices]
|
||||
if predictions.size == 0:
|
||||
return []
|
||||
|
||||
boxes = predictions[:, :4].copy()
|
||||
boxes[:, 0] = (boxes[:, 0] - self.pad_left) / self.scale
|
||||
boxes[:, 1] = (boxes[:, 1] - self.pad_top) / self.scale
|
||||
boxes[:, 2] /= self.scale
|
||||
boxes[:, 3] /= self.scale
|
||||
xyxy_boxes = self.xywh2xyxy(boxes)
|
||||
|
||||
keep_indices = []
|
||||
for class_id in np.unique(class_ids):
|
||||
class_indices = np.where(class_ids == class_id)[0]
|
||||
class_keep = self.nms(
|
||||
xyxy_boxes[class_indices],
|
||||
scores[class_indices],
|
||||
iou_threshold,
|
||||
)
|
||||
keep_indices.extend(class_indices[class_keep])
|
||||
|
||||
results = []
|
||||
for index in keep_indices:
|
||||
class_id = int(class_ids[index])
|
||||
x1, y1, x2, y2 = map(int, xyxy_boxes[index].tolist())
|
||||
results.append({
|
||||
"label_id": class_id,
|
||||
"label_name": self.names[class_id],
|
||||
"confidence": float(scores[index]),
|
||||
"box_mid_xy": [(x1 + x2) // 2, (y1 + y2) // 2],
|
||||
"xyxy": [x1, y1, x2, y2],
|
||||
})
|
||||
|
||||
results.sort(key=lambda item: item["confidence"], reverse=True)
|
||||
return results
|
||||
|
||||
def detect(self, image: Image.Image) -> list[dict]:
|
||||
input_tensor = self.preprocess(image)
|
||||
outputs = self.session.run([self.output_name], {self.input_name: input_tensor})
|
||||
return self.postprocess(outputs)
|
||||
|
||||
|
||||
class SimilarityOnnx:
|
||||
"""点选验证码中目标图和候选字的相似度模型。"""
|
||||
|
||||
def __init__(self, model_path: str | Path, providers: list[str] | None = None):
|
||||
providers = providers or ["CPUExecutionProvider"]
|
||||
ort.set_default_logger_severity(3)
|
||||
self.session = ort.InferenceSession(str(model_path), providers=providers)
|
||||
self.input_shape = [64, 64]
|
||||
|
||||
@staticmethod
|
||||
def sigmoid(value: np.ndarray) -> np.ndarray:
|
||||
return 1 / (1 + np.exp(-value))
|
||||
|
||||
@staticmethod
|
||||
def _to_image(value) -> Image.Image:
|
||||
if isinstance(value, np.ndarray):
|
||||
return Image.fromarray(value)
|
||||
if isinstance(value, bytes):
|
||||
return Image.open(BytesIO(value))
|
||||
if isinstance(value, Image.Image):
|
||||
return value
|
||||
return Image.open(value)
|
||||
|
||||
def _tensor(self, value) -> np.ndarray:
|
||||
image = self._to_image(value).convert("RGB").resize(tuple(reversed(self.input_shape)), 1)
|
||||
array = np.array(image).astype(np.float32) / 255.0
|
||||
return np.expand_dims(np.transpose(array, (2, 0, 1)), 0)
|
||||
|
||||
def score(self, image_1, image_2) -> int:
|
||||
out = self.session.run(None, {"x1": self._tensor(image_1), "x2": self._tensor(image_2)})
|
||||
similarity = self.sigmoid(out[0])[0][0]
|
||||
return int(round(similarity.item(), 2) * 100)
|
||||
|
||||
|
||||
class HuyaCaptchaOcr:
|
||||
"""封装虎牙滑块和点选识别。"""
|
||||
|
||||
def __init__(self, model_dir: str | Path = MODEL_DIR, providers: list[str] | None = None):
|
||||
model_dir = Path(model_dir)
|
||||
self.similarity = SimilarityOnnx(model_dir / "weights.onnx", providers=providers)
|
||||
self.click_model = YoloOnnx(model_dir / "best.onnx", classes=["target", "char"], providers=providers)
|
||||
self.slider_model = YoloOnnx(model_dir / "slider_2.onnx", classes=["slider"], providers=providers)
|
||||
|
||||
@staticmethod
|
||||
def _open_image(image_bytes: bytes) -> Image.Image:
|
||||
return Image.open(BytesIO(image_bytes)).convert("RGB")
|
||||
|
||||
def ocr_slider_box(self, image_bytes: bytes) -> list[int]:
|
||||
"""识别滑块缺口框，返回 [x1, y1, x2, y2]。"""
|
||||
image = self._open_image(image_bytes)
|
||||
detections = self.slider_model.detect(image)
|
||||
boxes = [item["xyxy"] for item in detections]
|
||||
return boxes[0] if boxes else [100, 100, 100, 100]
|
||||
|
||||
def ocr_click_points(self, image_bytes: bytes) -> list[list[int]]:
|
||||
"""识别点选坐标，返回按目标顺序排列的点击点。"""
|
||||
image = self._open_image(image_bytes)
|
||||
detections = self.click_model.detect(image)
|
||||
results = [{**item, "cropped_image": image.crop(tuple(item["xyxy"]))} for item in detections]
|
||||
|
||||
char_list = [item for item in results if item.get("label_name") == "char"]
|
||||
target_list = [item for item in results if item.get("label_name") == "target"]
|
||||
target_list = [item for item in target_list if item["xyxy"][2] - item["xyxy"][0] > 10]
|
||||
char_list.sort(key=lambda item: item["xyxy"][0])
|
||||
target_list.sort(key=lambda item: item["xyxy"][0])
|
||||
|
||||
if not char_list or not target_list:
|
||||
return []
|
||||
|
||||
score_matrix = np.zeros((len(target_list), len(char_list)))
|
||||
for target_index, target in enumerate(target_list):
|
||||
for char_index, char in enumerate(char_list):
|
||||
try:
|
||||
score_matrix[target_index][char_index] = -self.similarity.score(
|
||||
target["cropped_image"],
|
||||
char["cropped_image"],
|
||||
)
|
||||
except Exception:
|
||||
score_matrix[target_index][char_index] = 1e6
|
||||
|
||||
row_ind, col_ind = linear_sum_assignment(score_matrix)
|
||||
matched = []
|
||||
for target_index, char_index in zip(row_ind, col_ind):
|
||||
if score_matrix[target_index][char_index] == 1e6:
|
||||
continue
|
||||
char = char_list[char_index]
|
||||
matched.append({**char, "index": target_index + 1})
|
||||
|
||||
return [item["box_mid_xy"] for item in matched]
|
||||
|
||||
|
||||
@lru_cache(maxsize=1)
|
||||
def default_ocr() -> HuyaCaptchaOcr:
|
||||
"""懒加载默认 OCR，避免导入 core.huya 时立即加载 150MB 模型。"""
|
||||
return HuyaCaptchaOcr()
|
||||
@@ -0,0 +1,388 @@
|
||||
"""虎牙登录风控验证闭环。"""
|
||||
|
||||
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
|
||||
|
||||
@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": "%5B%5D",
|
||||
"info": self.sdid,
|
||||
}
|
||||
payload.update(extra)
|
||||
return {"appId": "5002", "data": payload}
|
||||
|
||||
@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")
|
||||
|
||||
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):
|
||||
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:
|
||||
return int((opencv_x + yolo_x) / 2)
|
||||
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)
|
||||
behavior = quote(
|
||||
json.dumps([{
|
||||
"action": "mousedown",
|
||||
"id": "",
|
||||
"txt": "",
|
||||
"clientX": track_points[0][0],
|
||||
"clientY": track_points[0][1],
|
||||
"clientWidth": 350,
|
||||
"clientHeight": 268,
|
||||
"isTrusted": True,
|
||||
"timeStamp": 1829,
|
||||
"nowTime": int(time.time() * 1000),
|
||||
}], separators=(",", ":")),
|
||||
safe="~()*!.'",
|
||||
)
|
||||
payload = self._build_common_payload({"behavior": behavior, "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],
|
||||
})
|
||||
|
||||
behavior = quote(json.dumps(behavior_list, separators=(",", ":")), safe="~()*!.'")
|
||||
payload = self._build_common_payload({"behavior": behavior, "act": act, "ver": 5})
|
||||
return self._submit_verify(url, payload)
|
||||
|
||||
def _submit_verify(self, url: str, payload: dict) -> dict:
|
||||
body = json.dumps(payload, separators=(",", ":"))
|
||||
result = self._request_json("post", url, params={"lock": "true"}, data=body)
|
||||
return result.get("data") or result
|
||||
|
||||
def _solve_current_strategy(self) -> dict:
|
||||
"""提交当前 strategy_url 对应的验证。"""
|
||||
config_payload = self._build_common_payload({"cTuIndex": 0})
|
||||
config = self._request_json("post", "https://udbrtt.huya.com/auth/client/config3", json=config_payload)
|
||||
try:
|
||||
gurl = config["data"]["gurl"]
|
||||
vurl = config["data"]["vurl"]
|
||||
dx_js = config["data"]["js"]
|
||||
except KeyError as exc:
|
||||
raise HuyaVerificationError(f"config3 取参失败: {config}") from exc
|
||||
|
||||
self._compile_js(dx_js)
|
||||
point = self._get_track_point(gurl, config_payload)
|
||||
logger.debug(f"虎牙验证识别结果: {point}")
|
||||
if isinstance(point, list):
|
||||
return self._send_click_verify(vurl, point)
|
||||
return self._send_slider_verify(vurl, int(point))
|
||||
|
||||
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)
|
||||
@@ -0,0 +1,45 @@
|
||||
"""虎牙滑块轨迹生成工具。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
import time
|
||||
from collections.abc import Iterable
|
||||
|
||||
|
||||
def ease_out_expo(sep: float) -> float:
|
||||
"""缓出曲线，让轨迹更接近人工拖动。"""
|
||||
if sep == 1:
|
||||
return 1
|
||||
return 1 - pow(2, -10 * sep)
|
||||
|
||||
|
||||
def generate_slide_track(distance: int) -> list[list[int]]:
|
||||
"""生成虎牙滑块验证所需 travel 轨迹。"""
|
||||
if not isinstance(distance, int) or distance < 0:
|
||||
raise ValueError(f"distance 必须是大于等于 0 的整数: {distance!r}")
|
||||
|
||||
distance = distance + random.randint(10, 40)
|
||||
current_time = int(time.time() * 1000) - random.randint(1500, 2000)
|
||||
slide_x = random.randint(56, 63)
|
||||
slide_y = random.randint(210, 213)
|
||||
track = [[slide_x, slide_y, current_time]]
|
||||
|
||||
count = random.randint(40, 60)
|
||||
last_x = 0
|
||||
for index in range(count):
|
||||
x = round(ease_out_expo(index / count) * distance)
|
||||
current_time += random.randint(10, 20)
|
||||
if x == last_x:
|
||||
continue
|
||||
y = random.randint(-1, 1)
|
||||
track.append([x + slide_x, y + slide_y, current_time])
|
||||
last_x = x
|
||||
|
||||
track.append(track[-1])
|
||||
return track
|
||||
|
||||
|
||||
def format_track(points: Iterable[Iterable[int]]) -> str:
|
||||
"""把轨迹点格式化成虎牙 actData.travel 字符串。"""
|
||||
return ";".join(f"{x},{y},{t}" for x, y, t in points)
|
||||
Reference in New Issue
Block a user