534 lines
16 KiB
Python
534 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
天翼云盘订购页 http://yscnb.com/tyyp/ 自动化脚本（全部代码自包含）
|
||
|
||
自动化流程：
|
||
1. 打开页面
|
||
2. 勾选协议
|
||
3. 点击「立即订购」
|
||
4. 输入手机号
|
||
5. 点击「获取验证码」
|
||
6. 解析拼图验证码并执行滑块拖动
|
||
7. 监听 hn_userEquitys/getYanZhenMa/v2 响应并输出
|
||
|
||
用法：python test1.py [--phone 手机号]
|
||
调试时可在代码中修改 DEFAULT_PHONE
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
from typing import Any
|
||
import base64
|
||
import random
|
||
import time
|
||
from io import BytesIO
|
||
|
||
import numpy as np
|
||
from PIL import Image
|
||
|
||
from DrissionPage import ChromiumPage, ChromiumOptions
|
||
|
||
# ========== 调试用：在代码中直接指定手机号 ==========
|
||
DEFAULT_PHONE = "17375712810"
|
||
|
||
# 需要监听的 URL 特征（滑块通过后前端会请求此接口）
|
||
GET_YAN_ZHEN_MA_URL = "hn_userEquitys/getYanZhenMa/v2"
|
||
|
||
|
||
# ---------- 拼图验证码计算 ----------
|
||
|
||
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 click_safe(ele) -> None:
|
||
try:
|
||
ele.click()
|
||
return
|
||
except Exception:
|
||
pass
|
||
ele.click(by_js=True)
|
||
|
||
|
||
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。"""
|
||
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 _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]:
|
||
"""生成仿人轨迹列表：加速-匀速-减速-过冲-回弹。"""
|
||
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)
|
||
|
||
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:
|
||
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)
|
||
|
||
_dispatch_mouse(page, "mouseMoved", cx, cy)
|
||
time.sleep(random.uniform(0.03, 0.08))
|
||
|
||
_dispatch_mouse(page, "mousePressed", cx, cy)
|
||
time.sleep(random.uniform(0.02, 0.06))
|
||
|
||
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"])
|
||
|
||
time.sleep(random.uniform(0.02, 0.06))
|
||
_dispatch_mouse(page, "mouseReleased", cur_x, cur_y)
|
||
|
||
|
||
# ---------- 核心自动化流程 ----------
|
||
|
||
def submit_phone(
|
||
page,
|
||
phone: str,
|
||
url: str = "http://yscnb.com/tyyp/",
|
||
alpha_threshold: int = 12,
|
||
distance_adjust: int = 0,
|
||
wait_page: float = 0.3,
|
||
) -> Any:
|
||
"""
|
||
填写手机号、点击获取验证码、执行滑块，返回 getYanZhenMa/v2 接口响应体。
|
||
"""
|
||
page.get(url)
|
||
time.sleep(wait_page)
|
||
|
||
# 1. 勾选协议
|
||
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)
|
||
time.sleep(0.4)
|
||
|
||
# 2. 立即订购
|
||
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: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)
|
||
time.sleep(0.4)
|
||
else:
|
||
try:
|
||
page.run_js("""
|
||
var nodes = document.querySelectorAll('button, a, span, div');
|
||
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;
|
||
""")
|
||
except Exception:
|
||
pass
|
||
time.sleep(0.4)
|
||
|
||
# 3. 输入手机号
|
||
phone_input = find_first(
|
||
page,
|
||
[
|
||
'x://*[@placeholder="请输入手机号码"]',
|
||
"css:input.inp-txt",
|
||
],
|
||
timeout=8,
|
||
)
|
||
if not phone_input:
|
||
raise RuntimeError("未找到手机号输入框")
|
||
|
||
phone_input.input(phone, clear=True)
|
||
|
||
agree = find_first(
|
||
page,
|
||
[
|
||
"css:i.ico-checkbox",
|
||
'x://i[contains(@class,"ico-checkbox")]',
|
||
],
|
||
timeout=2,
|
||
)
|
||
if agree:
|
||
try:
|
||
click_safe(agree)
|
||
except Exception:
|
||
pass
|
||
|
||
# 4. 启动监听（必须在点击获取验证码之前）
|
||
page.listen.start(GET_YAN_ZHEN_MA_URL)
|
||
|
||
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)
|
||
|
||
# 5. 等待滑块弹窗
|
||
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)
|
||
|
||
if not bg_img or not piece_img or not slider:
|
||
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("验证码图片数据异常")
|
||
|
||
match = calc_drag_distance_from_bytes(
|
||
bg_bytes, piece_bytes, alpha_threshold=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(distance_adjust)
|
||
|
||
# 6. 执行滑块并等待 getYanZhenMa 响应
|
||
drag_slider(page, slider, move_distance)
|
||
time.sleep(0.5)
|
||
|
||
packet = page.listen.wait(timeout=15, fit_count=False)
|
||
if packet is False:
|
||
raise RuntimeError(f"滑块拖动后 15s 内未收到 {GET_YAN_ZHEN_MA_URL} 响应")
|
||
if isinstance(packet, list):
|
||
packet = packet[0] if packet else None
|
||
if packet is None:
|
||
raise RuntimeError(f"未捕获到 {GET_YAN_ZHEN_MA_URL} 数据包")
|
||
|
||
return packet.response.body
|
||
|
||
|
||
def submit_code(page, code: str) -> dict:
|
||
"""填写短信验证码，点击确认订购按钮。"""
|
||
code_input = find_first(
|
||
page,
|
||
[
|
||
'x://input[@placeholder*="验证码"]',
|
||
'x://input[@placeholder*="短信"]',
|
||
"css:input.inp-txt[type='text']",
|
||
"css:input[type='tel']",
|
||
"css:.code-input",
|
||
"css:input.verify-input",
|
||
],
|
||
timeout=8,
|
||
)
|
||
if not code_input:
|
||
raise RuntimeError("未找到验证码输入框")
|
||
|
||
code_input.input(code, clear=True)
|
||
time.sleep(0.2)
|
||
|
||
confirm_btn = find_first(
|
||
page,
|
||
[
|
||
"css:img.btn-buy",
|
||
"css:.btn-buy",
|
||
'x://img[contains(@class,"btn-buy")]',
|
||
'x://*[contains(@class,"btn-buy")]',
|
||
],
|
||
timeout=5,
|
||
)
|
||
if not confirm_btn:
|
||
raise RuntimeError("未找到确认订购按钮（img.btn-buy）")
|
||
|
||
try:
|
||
confirm_btn.run_js("this.scrollIntoView({block:'center'})")
|
||
time.sleep(0.05)
|
||
except Exception:
|
||
pass
|
||
|
||
click_safe(confirm_btn)
|
||
return {"success": True, "message": "已点击确认订购"}
|
||
|
||
|
||
# ---------- 入口 ----------
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(description="天翼云盘订购页自动化")
|
||
parser.add_argument(
|
||
"--phone", default=DEFAULT_PHONE, help="手机号码（默认用代码中的 DEFAULT_PHONE）"
|
||
)
|
||
parser.add_argument("--url", default="http://yscnb.com/tyyp/", help="目标页面 URL")
|
||
parser.add_argument("--port", type=int, default=0, help="连接已有浏览器端口，0 表示新建")
|
||
args = parser.parse_args()
|
||
|
||
if args.port:
|
||
co = ChromiumOptions().set_local_port(port=args.port)
|
||
page = ChromiumPage(addr_or_opts=co)
|
||
else:
|
||
page = ChromiumPage()
|
||
|
||
print(f"打开页面: {args.url}，手机号: {args.phone}")
|
||
body = submit_phone(
|
||
page=page,
|
||
phone=args.phone,
|
||
url=args.url,
|
||
)
|
||
print("hn_userEquitys/getYanZhenMa/v2 响应:")
|
||
print(body)
|
||
|
||
input_code(page=page, code=123123)
|
||
|
||
|
||
def input_code(page, code):
|
||
page.ele('x://input[@placeholder="请输入验证码"]').input(code)
|
||
page.ele('x://*[@id="app"]/div/img').click(by_js=True)
|
||
page.ele('x://*[@id="app"]/div/div[7]/div/div[3]/button[2]').click(by_js=True)
|
||
|
||
if __name__ == "__main__":
|
||
main()
|