hahza
This commit is contained in:
77
.gitignore
vendored
Normal file
77
.gitignore
vendored
Normal file
@@ -0,0 +1,77 @@
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
|
||||
# C extensions
|
||||
*.so
|
||||
|
||||
# Distribution / packaging
|
||||
.Python
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
downloads/
|
||||
eggs/
|
||||
.eggs/
|
||||
lib/
|
||||
lib64/
|
||||
parts/
|
||||
sdist/
|
||||
var/
|
||||
wheels/
|
||||
*.egg-info/
|
||||
.installed.cfg
|
||||
*.egg
|
||||
|
||||
# PyInstaller
|
||||
*.manifest
|
||||
*.spec
|
||||
|
||||
# Installer logs
|
||||
pip-log.txt
|
||||
pip-delete-this-directory.txt
|
||||
|
||||
# Unit test / coverage reports
|
||||
htmlcov/
|
||||
.tox/
|
||||
.nox/
|
||||
.coverage
|
||||
.coverage.*
|
||||
.cache
|
||||
nosetests.xml
|
||||
coverage.xml
|
||||
*.cover
|
||||
*.py,cover
|
||||
.hypothesis/
|
||||
.pytest_cache/
|
||||
|
||||
# Translations
|
||||
*.mo
|
||||
*.pot
|
||||
|
||||
# Environments
|
||||
.env
|
||||
.venv
|
||||
env/
|
||||
venv/
|
||||
ENV/
|
||||
env.bak/
|
||||
venv.bak/
|
||||
|
||||
# IDE
|
||||
.idea/
|
||||
.vscode/
|
||||
*.swp
|
||||
*.swo
|
||||
*~
|
||||
|
||||
# 本项目调试/临时输出
|
||||
captcha_debug/
|
||||
slider_match_debug.png
|
||||
下载.png
|
||||
下载 (1).png
|
||||
|
||||
# OS
|
||||
.DS_Store
|
||||
Thumbs.db
|
||||
279
dp_slider_verify.py
Normal file
279
dp_slider_verify.py
Normal file
@@ -0,0 +1,279 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
根据背景图 + 拼图块图计算滑块目标偏移,并可选使用 DrissionPage 执行拖动。
|
||||
|
||||
默认读取当前项目根目录:
|
||||
- 下载 (1).png (背景图)
|
||||
- 下载.png (拼图块,RGBA 透明图)
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import random
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image, ImageDraw
|
||||
|
||||
|
||||
def _load_images(bg_path: Path, piece_path: Path) -> tuple[np.ndarray, np.ndarray]:
|
||||
bg = np.array(Image.open(bg_path).convert("RGB"), dtype=np.int16)
|
||||
piece_rgba = np.array(Image.open(piece_path).convert("RGBA"), dtype=np.int16)
|
||||
return bg, piece_rgba
|
||||
|
||||
|
||||
def _piece_bbox(alpha: np.ndarray, threshold: int = 12) -> tuple[int, int, int, int, np.ndarray]:
|
||||
mask = alpha > threshold
|
||||
if not mask.any():
|
||||
raise ValueError("拼图块图片 alpha 全透明,无法匹配")
|
||||
|
||||
ys, xs = np.where(mask)
|
||||
y0, y1 = int(ys.min()), int(ys.max()) + 1
|
||||
x0, x1 = int(xs.min()), int(xs.max()) + 1
|
||||
return x0, y0, x1, y1, mask
|
||||
|
||||
|
||||
def calc_drag_distance(
|
||||
bg_path: Path,
|
||||
piece_path: Path,
|
||||
alpha_threshold: int = 12,
|
||||
) -> dict:
|
||||
"""返回最佳匹配位置与建议拖动距离。"""
|
||||
bg, piece_rgba = _load_images(bg_path, piece_path)
|
||||
|
||||
bh, bw = bg.shape[:2]
|
||||
ph, pw = piece_rgba.shape[:2]
|
||||
if ph != bh:
|
||||
raise ValueError(f"图片高度不一致:背景={bh}, 拼图块={ph}")
|
||||
|
||||
alpha = piece_rgba[:, :, 3]
|
||||
x0, y0, x1, y1, _ = _piece_bbox(alpha, threshold=alpha_threshold)
|
||||
|
||||
piece_crop = piece_rgba[y0:y1, x0:x1, :3]
|
||||
mask = alpha[y0:y1, x0:x1] > alpha_threshold
|
||||
ys, xs = np.where(mask)
|
||||
piece_pixels = piece_crop[ys, xs]
|
||||
|
||||
patch_h, patch_w = piece_crop.shape[:2]
|
||||
if patch_w > bw or patch_h > bh:
|
||||
raise ValueError("拼图块裁剪后尺寸超过背景图")
|
||||
|
||||
max_x = bw - patch_w
|
||||
best_x = 0
|
||||
best_score = float("inf")
|
||||
second_best = float("inf")
|
||||
|
||||
# 固定 y,按 x 扫描。评分越小越好。
|
||||
for x in range(max_x + 1):
|
||||
patch_pixels = bg[y0 + ys, x + xs]
|
||||
score = float(np.abs(patch_pixels - piece_pixels).mean())
|
||||
if score < best_score:
|
||||
second_best = best_score
|
||||
best_score = score
|
||||
best_x = x
|
||||
elif score < second_best:
|
||||
second_best = score
|
||||
|
||||
# 如果拼图初始从最左侧出发,建议拖动距离可用:best_x - x0
|
||||
drag_distance = best_x - x0
|
||||
confidence_ratio = (second_best / best_score) if best_score > 0 else float("inf")
|
||||
|
||||
return {
|
||||
"target_x": int(best_x),
|
||||
"piece_bbox_x0": int(x0),
|
||||
"piece_bbox_y0": int(y0),
|
||||
"piece_bbox_w": int(patch_w),
|
||||
"piece_bbox_h": int(patch_h),
|
||||
"drag_distance": int(drag_distance),
|
||||
"best_score": best_score,
|
||||
"second_best": second_best,
|
||||
"confidence_ratio": confidence_ratio,
|
||||
"bg_width": int(bw),
|
||||
"bg_height": int(bh),
|
||||
}
|
||||
|
||||
|
||||
def save_debug_overlay(bg_path: Path, out_path: Path, match: dict) -> None:
|
||||
img = Image.open(bg_path).convert("RGB")
|
||||
draw = ImageDraw.Draw(img)
|
||||
x = int(match["target_x"])
|
||||
y = int(match["piece_bbox_y0"])
|
||||
w = int(match["piece_bbox_w"])
|
||||
h = int(match["piece_bbox_h"])
|
||||
draw.rectangle([x, y, x + w, y + h], outline=(255, 0, 0), width=2)
|
||||
img.save(out_path)
|
||||
|
||||
|
||||
def build_human_track(distance: int) -> list[int]:
|
||||
"""生成拟人拖动轨迹(整数步长列表)。"""
|
||||
if distance == 0:
|
||||
return []
|
||||
|
||||
direction = 1 if distance > 0 else -1
|
||||
remaining = abs(int(round(distance)))
|
||||
|
||||
track: list[int] = []
|
||||
moved = 0
|
||||
velocity = 0.0
|
||||
interval = 0.02
|
||||
accelerate_until = remaining * 0.7
|
||||
|
||||
while moved < remaining:
|
||||
if moved < accelerate_until:
|
||||
acc = random.uniform(2.0, 4.0)
|
||||
else:
|
||||
acc = -random.uniform(3.0, 6.0)
|
||||
|
||||
step = max(1, int(velocity * interval + 0.5 * acc * interval * interval * 100))
|
||||
moved += step
|
||||
velocity = max(0.0, velocity + acc * interval)
|
||||
track.append(step)
|
||||
|
||||
overflow = moved - remaining
|
||||
if overflow > 0 and track:
|
||||
track[-1] -= overflow
|
||||
if track[-1] <= 0:
|
||||
track.pop()
|
||||
|
||||
# 小幅回拉,增加拟人性
|
||||
if remaining >= 30:
|
||||
back = random.randint(1, 3)
|
||||
track.extend([-back, back])
|
||||
|
||||
return [direction * s for s in track if s != 0]
|
||||
|
||||
|
||||
def drag_slider_with_dp(page, slider_selector: str, distance: int) -> None:
|
||||
slider = page.ele(slider_selector, timeout=8)
|
||||
if not slider:
|
||||
raise RuntimeError(f"未找到滑块元素:{slider_selector}")
|
||||
|
||||
# 若元素支持直接 drag,优先用内置实现
|
||||
if hasattr(slider, "drag"):
|
||||
try:
|
||||
slider.drag(offset_x=distance, offset_y=0, duration=0.8)
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
actions = page.actions
|
||||
actions.move_to(slider, duration=0.2)
|
||||
|
||||
hold_ok = False
|
||||
for hold_call in (
|
||||
lambda: actions.hold(slider),
|
||||
lambda: actions.hold(),
|
||||
):
|
||||
try:
|
||||
hold_call()
|
||||
hold_ok = True
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if not hold_ok:
|
||||
raise RuntimeError("无法按下滑块,请确认 DrissionPage 版本与页面状态")
|
||||
|
||||
for step in build_human_track(distance):
|
||||
dy = random.randint(-1, 1)
|
||||
moved = False
|
||||
for move_call in (
|
||||
lambda: actions.move(step, dy, duration=random.uniform(0.01, 0.04)),
|
||||
lambda: actions.move(offset_x=step, offset_y=dy, duration=random.uniform(0.01, 0.04)),
|
||||
lambda: actions.move(step, dy),
|
||||
lambda: actions.move(offset_x=step, offset_y=dy),
|
||||
):
|
||||
try:
|
||||
move_call()
|
||||
moved = True
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
if not moved:
|
||||
raise RuntimeError("动作链 move 调用失败,请按当前 DrissionPage 版本调整 move 参数")
|
||||
|
||||
released = False
|
||||
for release_call in (
|
||||
lambda: actions.release(),
|
||||
lambda: actions.release(on_ele=None),
|
||||
):
|
||||
try:
|
||||
release_call()
|
||||
released = True
|
||||
break
|
||||
except Exception:
|
||||
continue
|
||||
|
||||
if not released:
|
||||
raise RuntimeError("动作链 release 调用失败")
|
||||
|
||||
|
||||
def maybe_run_dp_drag(args: argparse.Namespace, distance: int) -> None:
|
||||
if not args.url or not args.slider:
|
||||
return
|
||||
|
||||
from DrissionPage import ChromiumOptions, ChromiumPage
|
||||
|
||||
if args.port:
|
||||
co = ChromiumOptions().set_local_port(port=args.port)
|
||||
page = ChromiumPage(addr_or_opts=co)
|
||||
else:
|
||||
page = ChromiumPage()
|
||||
|
||||
page.get(args.url)
|
||||
if args.wait_after_open > 0:
|
||||
time.sleep(args.wait_after_open)
|
||||
|
||||
drag_slider_with_dp(page, args.slider, distance)
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
root = Path(__file__).resolve().parent
|
||||
|
||||
parser = argparse.ArgumentParser(description="DP 滑块验证辅助脚本")
|
||||
parser.add_argument("--bg", type=Path, default=root / "下载 (1).png", help="背景图路径")
|
||||
parser.add_argument("--piece", type=Path, default=root / "下载.png", help="拼图块图路径")
|
||||
parser.add_argument("--alpha-threshold", type=int, default=12, help="透明阈值(0-255)")
|
||||
parser.add_argument("--debug-out", type=Path, default=root / "slider_match_debug.png", help="匹配结果标注图")
|
||||
|
||||
# 以下参数用于真实页面拖动(可选)
|
||||
parser.add_argument("--url", type=str, default="", help="验证码页面 URL")
|
||||
parser.add_argument("--slider", type=str, default="", help="滑块元素选择器")
|
||||
parser.add_argument("--port", type=int, default=0, help="连接已有浏览器的本地端口")
|
||||
parser.add_argument("--distance-adjust", type=int, default=0, help="对计算结果追加修正像素")
|
||||
parser.add_argument("--wait-after-open", type=float, default=1.0, help="打开页面后等待秒数")
|
||||
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = parse_args()
|
||||
|
||||
if not args.bg.exists():
|
||||
raise FileNotFoundError(f"背景图不存在:{args.bg}")
|
||||
if not args.piece.exists():
|
||||
raise FileNotFoundError(f"拼图块图不存在:{args.piece}")
|
||||
|
||||
match = calc_drag_distance(args.bg, args.piece, alpha_threshold=args.alpha_threshold)
|
||||
save_debug_overlay(args.bg, args.debug_out, match)
|
||||
|
||||
distance = int(match["drag_distance"]) + int(args.distance_adjust)
|
||||
|
||||
print("匹配结果:")
|
||||
print(f" target_x={match['target_x']}")
|
||||
print(f" piece_bbox_x0={match['piece_bbox_x0']}")
|
||||
print(f" 建议拖动距离 drag_distance={match['drag_distance']}")
|
||||
print(f" 调整后拖动距离 distance={distance}")
|
||||
print(f" best_score={match['best_score']:.4f}, second_best={match['second_best']:.4f}")
|
||||
print(f" confidence_ratio={match['confidence_ratio']:.4f} (越大越好)")
|
||||
print(f" 标注图已输出:{args.debug_out}")
|
||||
|
||||
if args.url and args.slider:
|
||||
maybe_run_dp_drag(args, distance)
|
||||
print("DP 拖动已执行。")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
16
main.py
Normal file
16
main.py
Normal file
@@ -0,0 +1,16 @@
|
||||
# 这是一个示例 Python 脚本。
|
||||
|
||||
# 按 Shift+F10 执行或将其替换为您的代码。
|
||||
# 按 双击 Shift 在所有地方搜索类、文件、工具窗口、操作和设置。
|
||||
|
||||
|
||||
def print_hi(name):
|
||||
# 在下面的代码行中使用断点来调试脚本。
|
||||
print(f'Hi, {name}') # 按 Ctrl+F8 切换断点。
|
||||
|
||||
|
||||
# 按装订区域中的绿色按钮以运行脚本。
|
||||
if __name__ == '__main__':
|
||||
print_hi('PyCharm')
|
||||
|
||||
# 访问 https://www.jetbrains.com/help/pycharm/ 获取 PyCharm 帮助
|
||||
4
requirements.txt
Normal file
4
requirements.txt
Normal file
@@ -0,0 +1,4 @@
|
||||
# 项目依赖
|
||||
DrissionPage>=4.0.0
|
||||
Pillow>=10.0.0
|
||||
numpy>=1.24.0
|
||||
549
tyyp_1html_dp.py
Normal file
549
tyyp_1html_dp.py
Normal file
@@ -0,0 +1,549 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
站点流程自动化(DrissionPage):
|
||||
1) 打开 http://yscnb.com/tyyp/1.html
|
||||
2) 输入手机号
|
||||
3) 点击“获取验证码”
|
||||
4) 解析弹窗中的拼图验证码并执行拖动
|
||||
|
||||
说明:仅实现到“滑块拖动”步骤,不会自动填写短信验证码。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import base64
|
||||
import random
|
||||
import re
|
||||
import time
|
||||
from io import BytesIO
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
from PIL import Image
|
||||
|
||||
|
||||
def _to_rgba_array(image_bytes: bytes) -> np.ndarray:
|
||||
return np.array(Image.open(BytesIO(image_bytes)).convert("RGBA"), dtype=np.int16)
|
||||
|
||||
|
||||
def _to_rgb_array(image_bytes: bytes) -> np.ndarray:
|
||||
return np.array(Image.open(BytesIO(image_bytes)).convert("RGB"), dtype=np.int16)
|
||||
|
||||
|
||||
def _piece_bbox(alpha: np.ndarray, threshold: int = 12) -> tuple[int, int, int, int]:
|
||||
mask = alpha > threshold
|
||||
if not mask.any():
|
||||
raise ValueError("拼图块 alpha 全透明,无法匹配")
|
||||
ys, xs = np.where(mask)
|
||||
return int(xs.min()), int(ys.min()), int(xs.max()) + 1, int(ys.max()) + 1
|
||||
|
||||
|
||||
def calc_drag_distance_from_bytes(bg_bytes: bytes, piece_bytes: bytes, alpha_threshold: int = 12) -> dict:
|
||||
"""计算拼图目标位移(基于背景图 + 拼图块图)。"""
|
||||
bg = _to_rgb_array(bg_bytes)
|
||||
piece_rgba = _to_rgba_array(piece_bytes)
|
||||
|
||||
bh, bw = bg.shape[:2]
|
||||
ph, _ = piece_rgba.shape[:2]
|
||||
if bh != ph:
|
||||
raise ValueError(f"背景与拼图块高度不一致: {bh} != {ph}")
|
||||
|
||||
alpha = piece_rgba[:, :, 3]
|
||||
x0, y0, x1, y1 = _piece_bbox(alpha, threshold=alpha_threshold)
|
||||
|
||||
piece_crop = piece_rgba[y0:y1, x0:x1, :3]
|
||||
mask = alpha[y0:y1, x0:x1] > alpha_threshold
|
||||
ys, xs = np.where(mask)
|
||||
piece_pixels = piece_crop[ys, xs]
|
||||
|
||||
patch_h, patch_w = piece_crop.shape[:2]
|
||||
if patch_w > bw or patch_h > bh:
|
||||
raise ValueError("拼图块裁剪尺寸超过背景图")
|
||||
|
||||
best_x = 0
|
||||
best_score = float("inf")
|
||||
second_best = float("inf")
|
||||
max_x = bw - patch_w
|
||||
|
||||
for x in range(max_x + 1):
|
||||
patch_pixels = bg[y0 + ys, x + xs]
|
||||
score = float(np.abs(patch_pixels - piece_pixels).mean())
|
||||
if score < best_score:
|
||||
second_best = best_score
|
||||
best_score = score
|
||||
best_x = x
|
||||
elif score < second_best:
|
||||
second_best = score
|
||||
|
||||
drag_distance = best_x - x0
|
||||
confidence_ratio = (second_best / best_score) if best_score > 0 else float("inf")
|
||||
|
||||
return {
|
||||
"target_x": int(best_x),
|
||||
"piece_bbox_x0": int(x0),
|
||||
"piece_bbox_y0": int(y0),
|
||||
"piece_bbox_w": int(patch_w),
|
||||
"piece_bbox_h": int(patch_h),
|
||||
"bg_width": int(bw),
|
||||
"bg_height": int(bh),
|
||||
"drag_distance": int(drag_distance),
|
||||
"best_score": best_score,
|
||||
"second_best": second_best,
|
||||
"confidence_ratio": confidence_ratio,
|
||||
}
|
||||
|
||||
|
||||
def parse_data_url(data_url: str) -> bytes:
|
||||
if not data_url.startswith("data:image"):
|
||||
raise ValueError("图片不是 data:image URL")
|
||||
_, data = data_url.split(",", 1)
|
||||
return base64.b64decode(data)
|
||||
|
||||
|
||||
def style_px(style_text: str, key: str, default: float) -> float:
|
||||
if not style_text:
|
||||
return default
|
||||
m = re.search(rf"{re.escape(key)}\s*:\s*([0-9.]+)px", style_text)
|
||||
if not m:
|
||||
return default
|
||||
return float(m.group(1))
|
||||
|
||||
|
||||
def click_safe(ele) -> None:
|
||||
try:
|
||||
ele.click()
|
||||
return
|
||||
except Exception:
|
||||
pass
|
||||
ele.click(by_js=True)
|
||||
|
||||
|
||||
def _ease_out_quad(t: float) -> float:
|
||||
return t * (2 - t)
|
||||
|
||||
|
||||
def _ease_out_cubic(t: float) -> float:
|
||||
return 1 - (1 - t) ** 3
|
||||
|
||||
|
||||
def _ease_out_bounce(t: float) -> float:
|
||||
if t < 1 / 2.75:
|
||||
return 7.5625 * t * t
|
||||
elif t < 2 / 2.75:
|
||||
t -= 1.5 / 2.75
|
||||
return 7.5625 * t * t + 0.75
|
||||
elif t < 2.5 / 2.75:
|
||||
t -= 2.25 / 2.75
|
||||
return 7.5625 * t * t + 0.9375
|
||||
else:
|
||||
t -= 2.625 / 2.75
|
||||
return 7.5625 * t * t + 0.984375
|
||||
|
||||
|
||||
def build_human_track(distance: int, num_steps: int = 0) -> list[dict]:
|
||||
"""生成仿人轨迹列表,每项 {'dx': int, 'dy': int, 'dt': float(秒)}。
|
||||
|
||||
包含:加速-匀速-减速-过冲-回弹 五阶段。
|
||||
"""
|
||||
if distance == 0:
|
||||
return []
|
||||
|
||||
dist = abs(distance)
|
||||
sign = 1 if distance > 0 else -1
|
||||
|
||||
if num_steps <= 0:
|
||||
num_steps = max(12, int(dist * random.uniform(0.25, 0.4)))
|
||||
|
||||
overshoot = random.randint(max(2, int(dist * 0.03)), max(3, int(dist * 0.08)))
|
||||
total = dist + overshoot # 先超过,再回弹
|
||||
|
||||
easing = random.choice([_ease_out_quad, _ease_out_cubic])
|
||||
|
||||
# 正向阶段
|
||||
raw_positions: list[float] = []
|
||||
for i in range(1, num_steps + 1):
|
||||
t = i / num_steps
|
||||
raw_positions.append(easing(t) * total)
|
||||
|
||||
# 回弹阶段 (2~4 步)
|
||||
bounce_steps = random.randint(2, 4)
|
||||
for j in range(1, bounce_steps + 1):
|
||||
t = j / bounce_steps
|
||||
raw_positions.append(total - _ease_out_bounce(t) * overshoot)
|
||||
|
||||
track: list[dict] = []
|
||||
prev_x = 0.0
|
||||
for pos in raw_positions:
|
||||
dx = round(pos - prev_x)
|
||||
if dx == 0 and random.random() < 0.3:
|
||||
continue
|
||||
prev_x += dx
|
||||
dy = random.choice([-1, 0, 0, 0, 1])
|
||||
# 前半段快、后半段慢(加快整体速度)
|
||||
dt = random.uniform(0.005, 0.012) if prev_x < dist * 0.6 else random.uniform(0.008, 0.025)
|
||||
if random.random() < 0.03:
|
||||
dt += random.uniform(0.02, 0.06)
|
||||
track.append({"dx": sign * dx, "dy": dy, "dt": dt})
|
||||
|
||||
# 最终位置校正
|
||||
actual = sum(s["dx"] for s in track)
|
||||
diff = distance - actual
|
||||
if diff != 0:
|
||||
track.append({"dx": diff, "dy": 0, "dt": random.uniform(0.01, 0.03)})
|
||||
|
||||
return track
|
||||
|
||||
|
||||
def _dispatch_mouse(page, event_type: str, x: int, y: int, button: str = "left") -> None:
|
||||
"""通过 CDP Input.dispatchMouseEvent 发送鼠标事件。"""
|
||||
page.run_cdp(
|
||||
"Input.dispatchMouseEvent",
|
||||
type=event_type,
|
||||
x=x,
|
||||
y=y,
|
||||
button=button,
|
||||
clickCount=1 if event_type == "mousePressed" else 0,
|
||||
)
|
||||
|
||||
|
||||
def _get_element_center(page, ele) -> tuple[int, int]:
|
||||
"""获取元素在视口中的中心坐标。"""
|
||||
rect = page.run_js(
|
||||
"""const r = arguments[0].getBoundingClientRect();
|
||||
return {x: Math.round(r.x + r.width/2), y: Math.round(r.y + r.height/2)}""",
|
||||
ele,
|
||||
)
|
||||
if rect and isinstance(rect, dict):
|
||||
return int(rect["x"]), int(rect["y"])
|
||||
# 回退:通过元素属性
|
||||
loc = ele.rect.midpoint
|
||||
return int(loc[0]), int(loc[1])
|
||||
|
||||
|
||||
def drag_slider(page, slider_ele, distance: int) -> None:
|
||||
"""用 CDP 级鼠标事件完成拖拽,模拟真人操作轨迹。"""
|
||||
cx, cy = _get_element_center(page, slider_ele)
|
||||
|
||||
# 1. 鼠标移到滑块中心
|
||||
_dispatch_mouse(page, "mouseMoved", cx, cy)
|
||||
time.sleep(random.uniform(0.03, 0.08))
|
||||
|
||||
# 2. 按下
|
||||
_dispatch_mouse(page, "mousePressed", cx, cy)
|
||||
time.sleep(random.uniform(0.02, 0.06))
|
||||
|
||||
# 3. 沿轨迹移动
|
||||
cur_x, cur_y = cx, cy
|
||||
track = build_human_track(distance)
|
||||
for step in track:
|
||||
cur_x += step["dx"]
|
||||
cur_y += step["dy"]
|
||||
_dispatch_mouse(page, "mouseMoved", cur_x, cur_y)
|
||||
time.sleep(step["dt"])
|
||||
|
||||
# 4. 到达终点后短暂停留
|
||||
time.sleep(random.uniform(0.02, 0.06))
|
||||
|
||||
# 5. 释放
|
||||
_dispatch_mouse(page, "mouseReleased", cur_x, cur_y)
|
||||
|
||||
|
||||
def find_first(page, selectors: list[str], timeout: float = 5):
|
||||
for sel in selectors:
|
||||
try:
|
||||
ele = page.ele(sel, timeout=timeout)
|
||||
if ele:
|
||||
return ele
|
||||
except Exception:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
def wait_for_data_src(img_ele, timeout: float = 6, interval: float = 0.12) -> str:
|
||||
"""轮询等待 img 元素的 src 变为有效 data:image URL(含非空 base64 数据)。"""
|
||||
deadline = time.time() + timeout
|
||||
while time.time() < deadline:
|
||||
src = img_ele.attr("src") or ""
|
||||
if src.startswith("data:image"):
|
||||
_prefix, _, b64 = src.partition(",")
|
||||
if b64.strip():
|
||||
return src
|
||||
time.sleep(interval)
|
||||
raise RuntimeError(f"等待 data:image src 超时({timeout}s),当前 src 前80字符: {(img_ele.attr('src') or '')[:80]}")
|
||||
|
||||
|
||||
def save_debug(debug_dir: Path, bg_bytes: bytes, piece_bytes: bytes) -> tuple[Path, Path]:
|
||||
debug_dir.mkdir(parents=True, exist_ok=True)
|
||||
bg_path = debug_dir / "captcha_bg.png"
|
||||
piece_path = debug_dir / "captcha_piece.png"
|
||||
bg_path.write_bytes(bg_bytes)
|
||||
piece_path.write_bytes(piece_bytes)
|
||||
return bg_path, piece_path
|
||||
|
||||
|
||||
def run(args: argparse.Namespace) -> None:
|
||||
from DrissionPage import ChromiumOptions, ChromiumPage
|
||||
|
||||
t_start = time.perf_counter()
|
||||
|
||||
if args.port:
|
||||
co = ChromiumOptions().set_local_port(port=args.port)
|
||||
page = ChromiumPage(addr_or_opts=co)
|
||||
else:
|
||||
page = ChromiumPage()
|
||||
|
||||
page.get(args.url)
|
||||
time.sleep(args.wait_page)
|
||||
|
||||
# 首页先勾选协议 checkbox(id=color-input-red 的复选框),再点「立即订购」
|
||||
agree_checkbox = find_first(page, [
|
||||
"css:#color-input-red",
|
||||
"css:input[name='color-input-red']",
|
||||
'x://input[@id="color-input-red"]',
|
||||
"css:input.right-box[type='checkbox']",
|
||||
], timeout=5)
|
||||
if agree_checkbox:
|
||||
click_safe(agree_checkbox)
|
||||
print("已勾选协议复选框")
|
||||
time.sleep(0.4) # 勾选后等待
|
||||
|
||||
# 立即订购:点击 div.paybg 即为立即订购
|
||||
order_btn = None
|
||||
for attempt in range(4):
|
||||
order_btn = find_first(page, [
|
||||
"css:div.paybg",
|
||||
"css:.paybg",
|
||||
'x://button[contains(.,"立即订购")]',
|
||||
'x://a[contains(.,"立即订购")]',
|
||||
'x://span[contains(.,"立即订购")]',
|
||||
'x://div[contains(.,"立即订购")]',
|
||||
'x://*[contains(text(),"立即订购")]',
|
||||
'x://*[contains(.,"立即订购")]',
|
||||
"css:.btn-order",
|
||||
"css:.order-btn",
|
||||
"css:button.btn-primary",
|
||||
"css:button.btn",
|
||||
"css:a.btn",
|
||||
], timeout=1)
|
||||
if order_btn:
|
||||
break
|
||||
time.sleep(0.25)
|
||||
if order_btn:
|
||||
try:
|
||||
order_btn.run_js("this.scrollIntoView({block:'center'})")
|
||||
time.sleep(0.05)
|
||||
except Exception:
|
||||
pass
|
||||
click_safe(order_btn)
|
||||
print("已点击立即订购")
|
||||
time.sleep(0.4)
|
||||
else:
|
||||
# 兜底1:若页面有 jQuery,用 :contains 查找
|
||||
jq_clicked = False
|
||||
try:
|
||||
jq_clicked = page.run_js("""
|
||||
if (typeof $ !== 'undefined') {
|
||||
var el = $('button, a, span, div').filter(function(){ return $(this).text().indexOf('立即订购')>=0; }).first();
|
||||
if (el.length) { el[0].scrollIntoView({block:'center'}); el[0].click(); return true; }
|
||||
el = $('*').filter(function(){ return $(this).text().trim()==='立即订购'; }).first();
|
||||
if (el.length) { el[0].scrollIntoView({block:'center'}); el[0].click(); return true; }
|
||||
}
|
||||
return false;
|
||||
""")
|
||||
if jq_clicked:
|
||||
print("已通过 jQuery 点击立即订购")
|
||||
time.sleep(0.4)
|
||||
except Exception:
|
||||
pass
|
||||
if not jq_clicked:
|
||||
clicked = page.run_js("""
|
||||
var nodes = document.querySelectorAll('button, a, span, div, input[type=button], input[type=submit]');
|
||||
for (var i = 0; i < nodes.length; i++) {
|
||||
var t = (nodes[i].innerText || nodes[i].textContent || '').trim();
|
||||
if (t === '立即订购' || (t.indexOf('立即订购') >= 0 && t.length < 20)) {
|
||||
var el = nodes[i];
|
||||
if (el.offsetParent !== null || el.tagName === 'BODY') {
|
||||
el.scrollIntoView({block: 'center'});
|
||||
el.dispatchEvent(new MouseEvent('click', {bubbles: true, cancelable: true, view: window}));
|
||||
return true;
|
||||
}
|
||||
}
|
||||
}
|
||||
for (var i = 0; i < nodes.length; i++) {
|
||||
var t = (nodes[i].innerText || nodes[i].textContent || '').trim();
|
||||
if (t.indexOf('立即订购') >= 0) {
|
||||
nodes[i].scrollIntoView({block: 'center'});
|
||||
nodes[i].dispatchEvent(new MouseEvent('click', {bubbles: true, cancelable: true, view: window}));
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
""")
|
||||
if clicked:
|
||||
print("已通过 JS 点击立即订购")
|
||||
time.sleep(0.4)
|
||||
elif not clicked:
|
||||
# 尝试在 iframe 内点击
|
||||
in_iframe = page.run_js("""
|
||||
var iframes = document.querySelectorAll('iframe');
|
||||
for (var i = 0; i < iframes.length; i++) {
|
||||
try {
|
||||
var doc = iframes[i].contentDocument || iframes[i].contentWindow.document;
|
||||
var all = doc.querySelectorAll('*');
|
||||
for (var j = 0; j < all.length; j++) {
|
||||
var t = (all[j].innerText || all[j].textContent || '').trim();
|
||||
if (t.indexOf('立即订购') >= 0) {
|
||||
all[j].scrollIntoView({block:'center'});
|
||||
all[j].dispatchEvent(new MouseEvent('click',{bubbles:true,cancelable:true,view:window}));
|
||||
return true;
|
||||
}
|
||||
}
|
||||
} catch(e) {}
|
||||
}
|
||||
return false;
|
||||
""")
|
||||
if in_iframe:
|
||||
print("已在 iframe 内点击立即订购")
|
||||
time.sleep(0.4)
|
||||
else:
|
||||
# 调试:输出包含「立即订购」的元素信息
|
||||
try:
|
||||
info = page.run_js("""
|
||||
var out=[], all=document.querySelectorAll('*');
|
||||
for(var i=0;i<all.length;i++){
|
||||
var t=(all[i].innerText||all[i].textContent||'').trim();
|
||||
if(t.indexOf('立即订购')>=0)
|
||||
out.push(all[i].tagName+(all[i].id?'#'+all[i].id:'')+(all[i].className?'.'+all[i].className.split(' ')[0]:''));
|
||||
}
|
||||
return out.slice(0,5).join(', ') || '无';
|
||||
""")
|
||||
print(f"调试: 包含「立即订购」的元素(前5个): {info}")
|
||||
except Exception:
|
||||
pass
|
||||
print("警告: 未找到「立即订购」按钮,尝试继续...")
|
||||
|
||||
phone_input = find_first(page, [
|
||||
'x://input[@placeholder="请输入手机号码"]',
|
||||
"css:input.inp-txt",
|
||||
], timeout=8)
|
||||
if not phone_input:
|
||||
raise RuntimeError("未找到手机号输入框")
|
||||
|
||||
phone_input.input(args.phone, clear=True)
|
||||
print(f"已输入手机号: {args.phone}")
|
||||
|
||||
if not args.skip_agree:
|
||||
agree = find_first(page, [
|
||||
"css:i.ico-checkbox",
|
||||
'x://i[contains(@class,"ico-checkbox")]',
|
||||
], timeout=2)
|
||||
if agree:
|
||||
try:
|
||||
click_safe(agree)
|
||||
print("已点击同意勾选")
|
||||
except Exception:
|
||||
print("同意勾选点击失败,继续执行")
|
||||
|
||||
send_btn = find_first(page, [
|
||||
"css:button.btn-code",
|
||||
'x://button[contains(text(),"获取验证码")]',
|
||||
], timeout=8)
|
||||
if not send_btn:
|
||||
raise RuntimeError("未找到“获取验证码”按钮")
|
||||
|
||||
click_safe(send_btn)
|
||||
print("已点击获取验证码,等待滑块弹窗")
|
||||
|
||||
# 等待验证码弹窗出现
|
||||
verify_box = find_first(page, [
|
||||
"css:.verifybox",
|
||||
"css:.verify-bar-area",
|
||||
], timeout=6)
|
||||
if not verify_box:
|
||||
raise RuntimeError("未检测到滑块验证码弹窗")
|
||||
|
||||
bg_img = find_first(page, ["css:.verify-img-panel img"], timeout=5)
|
||||
piece_img = find_first(page, ["css:.verify-sub-block img"], timeout=5)
|
||||
slider = find_first(page, ["css:.verify-move-block"], timeout=5)
|
||||
bar = find_first(page, ["css:.verify-bar-area"], timeout=5)
|
||||
|
||||
if not bg_img or not piece_img or not slider or not bar:
|
||||
raise RuntimeError("验证码关键元素缺失(背景图/拼图块/滑块)")
|
||||
|
||||
bg_src = wait_for_data_src(bg_img, timeout=10)
|
||||
piece_src = wait_for_data_src(piece_img, timeout=10)
|
||||
|
||||
bg_bytes = parse_data_url(bg_src)
|
||||
piece_bytes = parse_data_url(piece_src)
|
||||
|
||||
if len(bg_bytes) < 100 or len(piece_bytes) < 100:
|
||||
raise RuntimeError(f"验证码图片数据异常: bg={len(bg_bytes)}B, piece={len(piece_bytes)}B")
|
||||
|
||||
if args.debug_dir:
|
||||
bg_path, piece_path = save_debug(Path(args.debug_dir), bg_bytes, piece_bytes)
|
||||
print(f"已保存验证码图片: {bg_path} | {piece_path}")
|
||||
|
||||
match = calc_drag_distance_from_bytes(bg_bytes, piece_bytes, alpha_threshold=args.alpha_threshold)
|
||||
|
||||
# 用渲染宽度做像素→屏幕映射
|
||||
bg_display_w = page.run_js(
|
||||
"""const el = arguments[0]; const r = el.getBoundingClientRect(); return r.width;""",
|
||||
bg_img,
|
||||
)
|
||||
if not bg_display_w or bg_display_w <= 0:
|
||||
bg_display_w = match["bg_width"]
|
||||
scale = float(bg_display_w) / max(1, match["bg_width"])
|
||||
move_distance = int(round(match["drag_distance"] * scale)) + int(args.distance_adjust)
|
||||
|
||||
print(
|
||||
"滑块匹配结果: "
|
||||
f"bg_width={match['bg_width']}, bg_display_w={bg_display_w}, "
|
||||
f"target_x={match['target_x']}, drag_distance={match['drag_distance']}, "
|
||||
f"scale={scale:.4f}, move_distance={move_distance}, "
|
||||
f"confidence={match['confidence_ratio']:.4f}"
|
||||
)
|
||||
|
||||
drag_slider(page, slider, move_distance)
|
||||
time.sleep(args.wait_result)
|
||||
|
||||
elapsed = time.perf_counter() - t_start
|
||||
print(f"总耗时: {elapsed:.2f} 秒(程序开始 → 滑块滑动完成)")
|
||||
|
||||
# 尝试判断是否通过:遮罩是否消失
|
||||
still_visible = page.run_js(
|
||||
"""
|
||||
const m = document.querySelector('.mask');
|
||||
if (!m) return false;
|
||||
const s = window.getComputedStyle(m);
|
||||
return s.display !== 'none' && s.visibility !== 'hidden' && s.opacity !== '0';
|
||||
"""
|
||||
)
|
||||
if still_visible:
|
||||
print("拖动已执行,但验证码弹窗仍在,可能需要微调 --distance-adjust")
|
||||
else:
|
||||
print("滑块拖动完成,验证码弹窗已关闭(疑似验证通过)")
|
||||
|
||||
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
root = Path(__file__).resolve().parent
|
||||
p = argparse.ArgumentParser(description="天翼订购页 1.html 自动滑块脚本")
|
||||
p.add_argument("--url", default="http://yscnb.com/tyyp/1.html", help="目标页面 URL")
|
||||
p.add_argument("--phone", required=True, help="手机号")
|
||||
p.add_argument("--port", type=int, default=0, help="连接已有浏览器端口(可选)")
|
||||
p.add_argument("--skip-agree", action="store_true", help="跳过勾选“同意办理”")
|
||||
p.add_argument("--alpha-threshold", type=int, default=12, help="拼图 alpha 透明阈值")
|
||||
p.add_argument("--distance-adjust", type=int, default=0, help="拖动距离微调像素")
|
||||
p.add_argument("--wait-page", type=float, default=0.3, help="打开页面后等待秒数")
|
||||
p.add_argument("--wait-result", type=float, default=0.5, help="拖动后等待结果秒数")
|
||||
p.add_argument("--debug-dir", default=str(root / "captcha_debug"), help="验证码图片输出目录")
|
||||
return p
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = build_parser().parse_args()
|
||||
run(args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user