190 lines
6.7 KiB
Python
190 lines
6.7 KiB
Python
"""虎牙验证码数据采集与分析脚本。
|
||
|
||
功能：
|
||
1. 【URL 采集模式】直接传入 safe_auth URL（dx_auth 点选 / pt_auth 滑块），
|
||
拉取验证码图片、提示语与元数据并落盘保存；
|
||
2. 【登录探测模式】输入账号密码触发风控，捕获验证码全套图片与提示语。
|
||
|
||
用法示例：
|
||
# 模式一：直接解析已有风控 URL
|
||
python scripts/capture_huya_captcha.py url "https://aq.huya.com/p/safe_auth/dx_auth.html?param=..."
|
||
|
||
# 模式二：仅拉取图片和提示语，不向服务端提交答案
|
||
python scripts/capture_huya_captcha.py url "https://aq.huya.com/p/safe_auth/dx_auth.html?param=..." --dump-only
|
||
|
||
# 模式三：登录触发捕获
|
||
python scripts/capture_huya_captcha.py login <账号> <密码> --channel app
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
from urllib.parse import parse_qs, urlparse
|
||
|
||
from loguru import logger
|
||
|
||
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
||
if str(PROJECT_ROOT) not in sys.path:
|
||
sys.path.insert(0, str(PROJECT_ROOT))
|
||
|
||
from core.huya.verification.capture import DEFAULT_CAPTURE_DIR
|
||
from core.huya.verification.solver import (
|
||
HuyaQrAuthRequiredError,
|
||
HuyaSmsAuthRequiredError,
|
||
HuyaVerificationSolver,
|
||
)
|
||
|
||
|
||
def capture_from_url(
|
||
url: str,
|
||
*,
|
||
dump_only: bool = False,
|
||
account: str = "",
|
||
proxies: dict | None = None,
|
||
) -> None:
|
||
"""从 safe_auth URL 抓取并保存验证码数据。"""
|
||
logger.info(f"开始从 URL 采集验证码: {url[:100]}...")
|
||
q = parse_qs(urlparse(url).query)
|
||
app_id = q.get("appId", ["5002"])[0]
|
||
|
||
solver = HuyaVerificationSolver(
|
||
app_id=app_id,
|
||
page_url=url,
|
||
account=account,
|
||
proxies=proxies,
|
||
enable_capture=True,
|
||
)
|
||
solver._parse_strategy(url)
|
||
|
||
if dump_only:
|
||
# 只拉取 config3 和 gurl，不提交 vurl
|
||
config_payload = solver._build_common_payload({"cTuIndex": 0})
|
||
config = solver._request_json(
|
||
"post",
|
||
"https://udbrtt.huya.com/auth/client/config3",
|
||
json=config_payload,
|
||
)
|
||
data = config.get("data") or {}
|
||
gurl = str(data.get("gurl") or "")
|
||
dx_js = str(data.get("js") or "")
|
||
if not gurl:
|
||
logger.error(f"config3 未下发 gurl: {config}")
|
||
return
|
||
if dx_js:
|
||
solver._compile_js(dx_js)
|
||
point_payload = solver._build_common_payload({"ver": 5})
|
||
point = solver._get_track_point(gurl, point_payload)
|
||
logger.info(f"【仅采集模式】验证码解析完成，识别结果: {point}")
|
||
if solver.current_capture and solver.current_capture.target_dir:
|
||
logger.info(f"验证码数据已成功保存至: {solver.current_capture.target_dir}")
|
||
return
|
||
|
||
# 完整求解并记录提交结果
|
||
try:
|
||
result = solver.solve(url)
|
||
logger.info(f"验证码提交完成: authId={result.get('authId')}")
|
||
except HuyaQrAuthRequiredError as exc:
|
||
logger.warning(f"该 URL 属于扫码验证: {exc}")
|
||
except HuyaSmsAuthRequiredError as exc:
|
||
logger.warning(f"该 URL 触发短信二次验证: {exc}")
|
||
except Exception as exc: # noqa: BLE001
|
||
logger.error(f"验证码处理异常: {exc}")
|
||
|
||
|
||
def capture_from_login(
|
||
account: str,
|
||
password: str,
|
||
channel: str = "app",
|
||
proxies: dict | None = None,
|
||
) -> None:
|
||
"""通过尝试登录触发风控并自动落盘验证码。"""
|
||
logger.info(f"尝试登录账号 {account} (渠道: {channel}) 探测验证码...")
|
||
if channel == "app":
|
||
from core.huya.app_login import login_huya_app_password
|
||
|
||
res = login_huya_app_password(account, password, proxies=proxies)
|
||
logger.info(
|
||
f"登录结果: success={res.success}, code={res.code}, msg={res.message}"
|
||
)
|
||
else:
|
||
from core.huya.login import login_huya_password
|
||
|
||
res = login_huya_password(account, password, proxies=proxies)
|
||
logger.info(
|
||
f"登录结果: success={res.success}, code={res.code}, msg={res.message}"
|
||
)
|
||
|
||
|
||
def list_recent_captures() -> None:
|
||
"""列出最近采集到的验证码。"""
|
||
if not DEFAULT_CAPTURE_DIR.exists():
|
||
print("暂无捕获数据目录: ", DEFAULT_CAPTURE_DIR)
|
||
return
|
||
folders = sorted(
|
||
DEFAULT_CAPTURE_DIR.glob("*"), key=lambda p: p.stat().st_mtime, reverse=True
|
||
)
|
||
if not folders:
|
||
print("当前尚未捕获任何验证码数据。")
|
||
return
|
||
print(f"\n找到 {len(folders)} 条捕获记录 (位于 {DEFAULT_CAPTURE_DIR}):\n")
|
||
for f in folders[:10]:
|
||
meta_file = f / "meta.json"
|
||
if meta_file.exists():
|
||
try:
|
||
meta = json.loads(meta_file.read_text(encoding="utf-8"))
|
||
auth_name = meta.get("auth_name", "未知")
|
||
prompts = meta.get("prompts", {})
|
||
question = (
|
||
prompts.get("question") or prompts.get("default_instruction") or ""
|
||
)
|
||
files = [p.name for p in f.glob("*") if p.name != "meta.json"]
|
||
print(f"📁 {f.name}")
|
||
print(f" 类型: {auth_name} | 提示语: {question}")
|
||
print(f" 文件: {', '.join(files)}")
|
||
except Exception: # noqa: BLE001
|
||
print(f"📁 {f.name}")
|
||
else:
|
||
print(f"📁 {f.name}")
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="虎牙验证码采集与测试工具")
|
||
subparsers = parser.add_subparsers(dest="cmd")
|
||
|
||
url_parser = subparsers.add_parser("url", help="从 safe_auth URL 抓取并落盘验证码")
|
||
url_parser.add_argument("url", help="safe_auth 完整 URL")
|
||
url_parser.add_argument("--account", default="", help="关联账号名 (可选)")
|
||
url_parser.add_argument(
|
||
"--dump-only",
|
||
action="store_true",
|
||
help="仅下载图片与提示语，不向服务器提交答案",
|
||
)
|
||
|
||
login_parser = subparsers.add_parser("login", help="通过密码登录触发验证码")
|
||
login_parser.add_argument("account", help="虎牙账号或手机号")
|
||
login_parser.add_argument("password", help="密码")
|
||
login_parser.add_argument(
|
||
"--channel", choices=["app", "web"], default="app", help="登录渠道 (默认 app)"
|
||
)
|
||
|
||
subparsers.add_parser("list", help="查看最近采集到的验证码记录")
|
||
|
||
args = parser.parse_args()
|
||
if args.cmd == "url":
|
||
capture_from_url(args.url, dump_only=args.dump_only, account=args.account)
|
||
elif args.cmd == "login":
|
||
capture_from_login(args.account, args.password, channel=args.channel)
|
||
elif args.cmd == "list":
|
||
list_recent_captures()
|
||
else:
|
||
parser.print_help()
|
||
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|