378 lines
15 KiB
Python
378 lines
15 KiB
Python
"""虎牙验证码全要素数据捕获与落盘模块。
|
|
|
|
每次触发验证码（滑块 pt_auth / 点选 dx_auth）时，将图片、提示语、识别结果、
|
|
请求元数据及提交响应完整落盘保存到 data/captcha_captures/，便于分析与排查。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime as dt
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import cv2
|
|
import numpy as np
|
|
from loguru import logger
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[3]
|
|
DEFAULT_CAPTURE_DIR = PROJECT_ROOT / "data" / "captcha_captures"
|
|
|
|
|
|
class CaptchaCaptureRecord:
|
|
"""单次验证码捕获会话。"""
|
|
|
|
def __init__(
|
|
self,
|
|
auth_type: str,
|
|
account: str = "",
|
|
page_url: str = "",
|
|
scene_type: str = "",
|
|
output_base_dir: Path | None = None,
|
|
):
|
|
self.auth_type = auth_type
|
|
self.account = account
|
|
self.page_url = page_url
|
|
self.scene_type = scene_type
|
|
self.output_base_dir = Path(output_base_dir or DEFAULT_CAPTURE_DIR)
|
|
self.created_at = dt.datetime.now(dt.timezone(dt.timedelta(hours=8)))
|
|
|
|
self.session_id = self.created_at.strftime("%Y%m%d_%H%M%S")
|
|
self.target_dir: Path | None = None
|
|
|
|
self.meta: dict[str, Any] = {
|
|
"timestamp": self.created_at.isoformat(),
|
|
"account": account,
|
|
"auth_type": auth_type,
|
|
"auth_name": (
|
|
"点选验证 (dx_auth)"
|
|
if "dx" in auth_type
|
|
else (
|
|
"短信二次验证 (sms_auth)"
|
|
if "sms" in auth_type
|
|
else (
|
|
"扫码验证 (qr_auth)"
|
|
if "qr" in auth_type
|
|
else "滑块验证 (pt_auth)"
|
|
)
|
|
)
|
|
),
|
|
"page_url": page_url,
|
|
"scene_type": scene_type,
|
|
"prompts": {},
|
|
"images": {},
|
|
"solution": {},
|
|
"submit_result": {},
|
|
}
|
|
|
|
def _ensure_target_dir(self, code: str = "") -> Path:
|
|
if self.target_dir is None:
|
|
short_code = code[:8] if code else "sample"
|
|
dir_name = f"{self.session_id}_{self.auth_type}_{short_code}"
|
|
self.target_dir = self.output_base_dir / dir_name
|
|
self.target_dir.mkdir(parents=True, exist_ok=True)
|
|
return self.target_dir
|
|
|
|
def capture_slider_data(
|
|
self,
|
|
*,
|
|
code: str,
|
|
dtu_bytes: bytes,
|
|
puzzle_bytes: bytes,
|
|
restored_bg_bytes: bytes | None = None,
|
|
composite_bytes: bytes | None = None,
|
|
point_x: int | None = None,
|
|
raw_data: dict[str, Any] | None = None,
|
|
) -> Path:
|
|
"""捕获滑块验证码的所有图片和元数据。"""
|
|
folder = self._ensure_target_dir(code)
|
|
|
|
# 1. 提示语与服务端描述
|
|
raw = raw_data or {}
|
|
self.meta["code"] = code
|
|
self.meta["prompts"] = {
|
|
"title": "安全验证 (滑块拼图)",
|
|
"default_instruction": "拖动下方滑块完成拼图",
|
|
"location_y": raw.get("location_y"),
|
|
"description": raw.get("description") or raw.get("message") or "",
|
|
"tip": raw.get("tip") or "",
|
|
}
|
|
|
|
# 2. 保存原始与合成图片
|
|
images_info = {}
|
|
if dtu_bytes:
|
|
(folder / "bg_scrambled.jpg").write_bytes(dtu_bytes)
|
|
images_info["bg_scrambled"] = "bg_scrambled.jpg"
|
|
|
|
if puzzle_bytes:
|
|
(folder / "puzzle.png").write_bytes(puzzle_bytes)
|
|
images_info["puzzle"] = "puzzle.png"
|
|
|
|
if restored_bg_bytes:
|
|
(folder / "bg_restored.jpg").write_bytes(restored_bg_bytes)
|
|
images_info["bg_restored"] = "bg_restored.jpg"
|
|
|
|
if composite_bytes:
|
|
(folder / "composite.png").write_bytes(composite_bytes)
|
|
images_info["composite"] = "composite.png"
|
|
|
|
# 3. 绘制识别标注图（在还原后的背景图上标注缺口 X 坐标与 Y 坐标）
|
|
if restored_bg_bytes and point_x is not None:
|
|
try:
|
|
bg_arr = cv2.imdecode(
|
|
np.frombuffer(restored_bg_bytes, np.uint8), cv2.IMREAD_COLOR
|
|
)
|
|
if bg_arr is not None:
|
|
loc_y = int(raw.get("location_y") or 0)
|
|
cv2.rectangle(
|
|
bg_arr,
|
|
(point_x, loc_y),
|
|
(point_x + 50, loc_y + 50),
|
|
(0, 0, 255),
|
|
2,
|
|
)
|
|
cv2.putText(
|
|
bg_arr,
|
|
f"X={point_x}",
|
|
(max(0, point_x - 10), max(20, loc_y - 10)),
|
|
cv2.FONT_HERSHEY_SIMPLEX,
|
|
0.6,
|
|
(0, 0, 255),
|
|
2,
|
|
)
|
|
cv2.imwrite(str(folder / "annotated.jpg"), bg_arr)
|
|
images_info["annotated"] = "annotated.jpg"
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.debug(f"[captcha_capture] 标注滑块图异常: {exc}")
|
|
|
|
self.meta["images"] = images_info
|
|
if point_x is not None:
|
|
self.meta["solution"] = {"type": "slider", "distance_x": point_x}
|
|
|
|
if raw_data:
|
|
clean_raw = {
|
|
k: v for k, v in raw_data.items() if not str(v).startswith("data:image")
|
|
}
|
|
(folder / "raw_data.json").write_text(
|
|
json.dumps(clean_raw, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
|
|
self._flush_meta()
|
|
self._log_summary()
|
|
return folder
|
|
|
|
def capture_click_data(
|
|
self,
|
|
*,
|
|
code: str,
|
|
dtu_bytes: bytes,
|
|
hzi_bytes: bytes,
|
|
points: list[list[int]] | None = None,
|
|
raw_data: dict[str, Any] | None = None,
|
|
) -> Path:
|
|
"""捕获文字/图标点选验证码的所有图片和元数据。"""
|
|
folder = self._ensure_target_dir(code)
|
|
|
|
raw = raw_data or {}
|
|
self.meta["code"] = code
|
|
question_text = str(raw.get("question") or "")
|
|
if not question_text:
|
|
question_text = "按顺序点击图中对应内容"
|
|
word_size = raw.get("wordSize")
|
|
self.meta["prompts"] = {
|
|
"title": str(raw.get("title") or "安全验证 (点选验证 dx_auth)"),
|
|
"question": question_text,
|
|
"word_size": word_size,
|
|
"tu_ming": raw.get("tuMing"),
|
|
"description": raw.get("description") or raw.get("message") or "",
|
|
}
|
|
|
|
images_info = {}
|
|
if dtu_bytes:
|
|
(folder / "bg_main.jpg").write_bytes(dtu_bytes)
|
|
images_info["bg_main"] = "bg_main.jpg"
|
|
|
|
if hzi_bytes:
|
|
(folder / "tip_text.png").write_bytes(hzi_bytes)
|
|
images_info["tip_text"] = "tip_text.png"
|
|
|
|
# 绘制点选标注预览图
|
|
if dtu_bytes and points:
|
|
try:
|
|
bg_arr = cv2.imdecode(
|
|
np.frombuffer(dtu_bytes, np.uint8), cv2.IMREAD_COLOR
|
|
)
|
|
if bg_arr is not None:
|
|
for idx, (px, py) in enumerate(points):
|
|
# 绘制点击圆圈
|
|
cv2.circle(bg_arr, (int(px), int(py)), 14, (0, 0, 255), 2)
|
|
cv2.circle(bg_arr, (int(px), int(py)), 4, (0, 255, 0), -1)
|
|
# 标注点击序号 1, 2, 3...
|
|
cv2.putText(
|
|
bg_arr,
|
|
str(idx + 1),
|
|
(int(px) - 5, int(py) - 18),
|
|
cv2.FONT_HERSHEY_SIMPLEX,
|
|
0.7,
|
|
(0, 0, 255),
|
|
2,
|
|
)
|
|
cv2.imwrite(str(folder / "annotated.jpg"), bg_arr)
|
|
images_info["annotated"] = "annotated.jpg"
|
|
except Exception as exc: # noqa: BLE001
|
|
logger.debug(f"[captcha_capture] 标注点选图异常: {exc}")
|
|
|
|
self.meta["images"] = images_info
|
|
if points is not None:
|
|
self.meta["solution"] = {"type": "click", "points": points}
|
|
|
|
if raw_data:
|
|
clean_raw = {
|
|
k: v for k, v in raw_data.items() if not str(v).startswith("data:image")
|
|
}
|
|
(folder / "raw_data.json").write_text(
|
|
json.dumps(clean_raw, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
|
|
self._flush_meta()
|
|
self._log_summary()
|
|
return folder
|
|
|
|
def capture_sms_auth_data(
|
|
self,
|
|
*,
|
|
mobile: str = "",
|
|
page_url: str = "",
|
|
scene_type: str = "",
|
|
step: int = 1,
|
|
raw_params: dict[str, Any] | None = None,
|
|
) -> Path:
|
|
"""捕获短信风控验证拦截数据并落盘。"""
|
|
folder = self._ensure_target_dir(code="sms_auth")
|
|
self.meta["auth_type"] = "sms_auth"
|
|
self.meta["auth_name"] = "短信二次验证 (sms_auth)"
|
|
self.meta["page_url"] = page_url
|
|
self.meta["scene_type"] = scene_type
|
|
self.meta["mobile"] = mobile
|
|
self.meta["step"] = step
|
|
self.meta["prompts"] = {
|
|
"title": "安全验证 (短信二次验证 sms_auth)",
|
|
"instruction": f"前置图形验证码已通过，虎牙触发第 {step} 级短信安全验证，需向手机 {mobile} 发送验证码",
|
|
"mobile": mobile,
|
|
}
|
|
self.meta["submit_result"] = {
|
|
"status": "sms_auth_required",
|
|
"mobile": mobile,
|
|
"nextAuthUrl": page_url,
|
|
}
|
|
if raw_params:
|
|
(folder / "params.json").write_text(
|
|
json.dumps(raw_params, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
self._flush_meta()
|
|
|
|
logger.warning(
|
|
f"\n┌── [虎牙风控拦截捕获] ──────────────────────────────────────────\n"
|
|
f"│ 类型: 短信二次验证 (sms_auth) (sceneType: {self.scene_type or scene_type or '4'})\n"
|
|
f"│ 账号: {self.account or '未指定'}\n"
|
|
f"│ 阶段: 第 {step} 阶段（前置滑块/点选已成功通过）\n"
|
|
f"│ 手机: {mobile or '未知'}\n"
|
|
f"│ 提示: 虎牙官方下发短信接码页面，协议自动化无法跳过，需人工接码\n"
|
|
f"│ 页面: {page_url[:80]}...\n"
|
|
f"│ 存储目录: {self.target_dir}\n"
|
|
f"└─────────────────────────────────────────────────────────────"
|
|
)
|
|
return folder
|
|
|
|
def capture_qr_auth_data(
|
|
self,
|
|
*,
|
|
mobile: str = "",
|
|
page_url: str = "",
|
|
scene_type: str = "",
|
|
step: int = 1,
|
|
raw_params: dict[str, Any] | None = None,
|
|
) -> Path:
|
|
"""捕获扫码风控验证拦截数据并落盘。"""
|
|
folder = self._ensure_target_dir(code="qr_auth")
|
|
self.meta["auth_type"] = "qr_auth"
|
|
self.meta["auth_name"] = "扫码二次验证 (qr_auth)"
|
|
self.meta["page_url"] = page_url
|
|
self.meta["scene_type"] = scene_type
|
|
self.meta["mobile"] = mobile
|
|
self.meta["step"] = step
|
|
self.meta["prompts"] = {
|
|
"title": "安全验证 (扫码二次验证 qr_auth)",
|
|
"instruction": f"虎牙触发第 {step} 级扫码安全验证，需在手机 App 上确认",
|
|
"mobile": mobile,
|
|
}
|
|
self.meta["submit_result"] = {
|
|
"status": "qr_auth_required",
|
|
"nextAuthUrl": page_url,
|
|
}
|
|
if raw_params:
|
|
(folder / "params.json").write_text(
|
|
json.dumps(raw_params, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
self._flush_meta()
|
|
|
|
logger.warning(
|
|
f"\n┌── [虎牙风控拦截捕获] ──────────────────────────────────────────\n"
|
|
f"│ 类型: 扫码二次验证 (qr_auth) (sceneType: {self.scene_type or scene_type or '4'})\n"
|
|
f"│ 账号: {self.account or '未指定'}\n"
|
|
f"│ 阶段: 第 {step} 阶段\n"
|
|
f"│ 提示: 虎牙要求在手机 App 上扫码确认或使用常用设备登录\n"
|
|
f"│ 页面: {page_url[:80]}...\n"
|
|
f"│ 存储目录: {self.target_dir}\n"
|
|
f"└─────────────────────────────────────────────────────────────"
|
|
)
|
|
return folder
|
|
|
|
def record_submit_result(self, result: dict[str, Any]) -> None:
|
|
"""记录验证码提交验证结果（authId、是否通过、是否流转下一阶段等）。"""
|
|
has_auth = bool(result.get("authId"))
|
|
has_next = bool(result.get("nextAuthUrl"))
|
|
status = (
|
|
"auth_success"
|
|
if has_auth
|
|
else ("passed_to_next_auth" if has_next else "failed")
|
|
)
|
|
self.meta["submit_result"] = {
|
|
"status": status,
|
|
"success": has_auth or has_next,
|
|
"authId": result.get("authId"),
|
|
"nextAuthUrl": result.get("nextAuthUrl"),
|
|
"returnCode": result.get("returnCode"),
|
|
"message": result.get("message") or result.get("description") or "",
|
|
}
|
|
self._flush_meta()
|
|
|
|
def _flush_meta(self) -> None:
|
|
if self.target_dir:
|
|
(self.target_dir / "meta.json").write_text(
|
|
json.dumps(self.meta, ensure_ascii=False, indent=2), encoding="utf-8"
|
|
)
|
|
|
|
def _log_summary(self) -> None:
|
|
prompts = self.meta.get("prompts") or {}
|
|
title = prompts.get("title") or "安全验证"
|
|
question = (
|
|
prompts.get("question")
|
|
or prompts.get("default_instruction")
|
|
or "按顺序点击"
|
|
)
|
|
word_size = prompts.get("word_size")
|
|
word_info = f", 目标字数: {word_size}" if word_size else ""
|
|
img_list = ", ".join(self.meta.get("images", {}).values())
|
|
|
|
logger.info(
|
|
f"\n┌── [虎牙验证码捕获] ──────────────────────────────────────────\n"
|
|
f"│ 类型: {self.meta.get('auth_name')} (sceneType: {self.scene_type or '默认'})\n"
|
|
f"│ 账号: {self.account or '未指定'}\n"
|
|
f"│ 标题: {title}\n"
|
|
f"│ 提示语: {question}{word_info}\n"
|
|
f"│ 存储目录: {self.target_dir}\n"
|
|
f"│ 保存图片: {img_list}\n"
|
|
f"└─────────────────────────────────────────────────────────────"
|
|
)
|