diff --git a/云翼云盘.py b/云翼云盘.py new file mode 100644 index 0000000..de02476 --- /dev/null +++ b/云翼云盘.py @@ -0,0 +1,1067 @@ + + +#!/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 +import json +from typing import Any, Optional +import base64 +import random +import time +from io import BytesIO +from datetime import datetime +from pathlib import Path + +import numpy as np +from PIL import Image, ImageDraw + +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 _to_gray(rgb: np.ndarray) -> np.ndarray: + return ( + rgb[:, :, 0] * 0.299 + rgb[:, :, 1] * 0.587 + rgb[:, :, 2] * 0.114 + ).astype(np.float32) + + +def _grad_x(gray: np.ndarray) -> np.ndarray: + pad = np.pad(gray, ((0, 0), (1, 1)), mode="edge") + return np.abs(pad[:, 2:] - pad[:, :-2]) * 0.5 + + +def calc_drag_distance_from_bytes( + bg_bytes: bytes, piece_bytes: bytes, alpha_threshold: int = 12 +) -> dict: + """计算拼图目标位移，输出多个候选位移提高滑块命中率。""" + bg = _to_rgb_array(bg_bytes).astype(np.float32) + piece_rgba = _to_rgba_array(piece_bytes).astype(np.float32) + + 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] + alpha_crop = alpha[y0:y1, x0:x1] + mask = alpha_crop > alpha_threshold + ys, xs = np.where(mask) + if len(xs) < 10: + raise ValueError("拼图有效像素过少，无法稳定匹配") + + piece_gray = _to_gray(piece_crop) + bg_gray = _to_gray(bg) + piece_grad = _grad_x(piece_gray) + bg_grad = _grad_x(bg_gray) + + piece_gray_pixels = piece_gray[ys, xs] + piece_grad_pixels = piece_grad[ys, xs] + weights = np.clip(alpha_crop[ys, xs].astype(np.float32), 1.0, 255.0) + weights = weights / float(weights.sum()) + + patch_h, patch_w = piece_crop.shape[:2] + if patch_w > bw or patch_h > bh: + raise ValueError("拼图块裁剪尺寸超过背景图") + + max_x = bw - patch_w + scores: list[tuple[float, int, float, float]] = [] + + for x in range(max_x + 1): + patch_gray_pixels = bg_gray[y0 + ys, x + xs] + patch_grad_pixels = bg_grad[y0 + ys, x + xs] + color_score = float((np.abs(patch_gray_pixels - piece_gray_pixels) * weights).sum()) + grad_score = float((np.abs(patch_grad_pixels - piece_grad_pixels) * weights).sum()) + # 颜色 + 边缘加权，较纯 RGB 差值在压缩噪声下更稳定 + score = color_score * 0.72 + grad_score * 0.28 + scores.append((score, x, color_score, grad_score)) + + scores.sort(key=lambda item: item[0]) + best_score, best_x, best_color_score, best_grad_score = scores[0] + second_best = scores[1][0] if len(scores) > 1 else float("inf") + + drag_distance = best_x - x0 + confidence_ratio = (second_best / best_score) if best_score > 0 else float("inf") + + candidate_xs: list[int] = [] + for _score, x, _color, _grad in scores: + if any(abs(x - picked) <= 1 for picked in candidate_xs): + continue + candidate_xs.append(int(x)) + if len(candidate_xs) >= 5: + break + + candidate_distances = [int(x - x0) for x in candidate_xs] + + 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, + "best_color_score": best_color_score, + "best_grad_score": best_grad_score, + "second_best": second_best, + "confidence_ratio": confidence_ratio, + "candidate_target_xs": candidate_xs, + "candidate_drag_distances": candidate_distances, + } + + +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 _is_element_visible(page, selector: str) -> bool: + try: + visible = page.run_js( + """ + const el = document.querySelector(arguments[0]); + if (!el) return false; + const st = window.getComputedStyle(el); + if (st.display === 'none' || st.visibility === 'hidden' || st.opacity === '0') return false; + const r = el.getBoundingClientRect(); + return r.width > 0 && r.height > 0; + """, + selector, + ) + return bool(visible) + except Exception: + return False + + +def _read_slider_toast_text(page) -> str: + try: + text = page.run_js( + """ + const node = document.querySelector('.van-toast--text .van-toast__text') + || document.querySelector('.van-toast__text'); + if (!node) return ''; + const popup = node.closest('.van-toast') || node; + const st = window.getComputedStyle(popup); + const visible = st.display !== 'none' && st.visibility !== 'hidden' && st.opacity !== '0'; + if (!visible) return ''; + return (node.innerText || node.textContent || '').trim(); + """ + ) + return (text or "").strip() + except Exception: + return "" + + +def _get_slider_layout_metrics(page, bg_img, slider_ele) -> dict: + try: + metrics = page.run_js( + """ + const bg = arguments[0]; + const slider = arguments[1]; + const bar = document.querySelector('.verify-bar-area'); + const sub = slider ? slider.querySelector('.verify-sub-block') : null; + const bgR = bg ? bg.getBoundingClientRect() : {width: 0}; + const sliderR = slider ? slider.getBoundingClientRect() : {width: 0}; + const barR = bar ? bar.getBoundingClientRect() : {width: 0}; + const subR = sub ? sub.getBoundingClientRect() : {width: 0}; + return { + bg_display_w: Number(bgR.width || 0), + slider_w: Number(sliderR.width || 0), + bar_w: Number(barR.width || 0), + sub_w: Number(subR.width || 0), + }; + """, + bg_img, + slider_ele, + ) + except Exception: + metrics = None + + if not isinstance(metrics, dict): + metrics = {} + + bar_w = float(metrics.get("bar_w", 0) or 0) + slider_w = float(metrics.get("slider_w", 0) or 0) + track_max = int(round(max(0.0, bar_w - slider_w))) if bar_w > 0 and slider_w > 0 else -1 + + return { + "bg_display_w": float(metrics.get("bg_display_w", 0) or 0), + "slider_w": slider_w, + "bar_w": bar_w, + "sub_w": float(metrics.get("sub_w", 0) or 0), + "track_max": track_max, + } + + +def _build_move_distance_candidates( + match: dict, + scale: float, + distance_adjust: int, + track_max: int, +) -> list[int]: + base_distances = match.get("candidate_drag_distances") or [int(match["drag_distance"])] + micro_offsets = (0, -1, 1, -2, 2, -3, 3) + + candidates: list[int] = [] + seen: set[int] = set() + + for base in base_distances[:4]: + scaled = int(round(int(base) * scale)) + int(distance_adjust) + for offset in micro_offsets: + val = scaled + offset + if track_max >= 0: + val = max(0, min(track_max, val)) + if val not in seen: + seen.add(val) + candidates.append(val) + + if not candidates: + fallback = int(round(int(match["drag_distance"]) * scale)) + int(distance_adjust) + if track_max >= 0: + fallback = max(0, min(track_max, fallback)) + candidates = [fallback] + + return candidates + + +def _normalize_listen_packet(packet): + if packet is False: + return None + if isinstance(packet, list): + return packet[0] if packet else None + return packet + + +def _coerce_json_body(body: Any) -> Any: + if not isinstance(body, str): + return body + raw = body.strip() + if not raw: + return body + if (raw.startswith("{") and raw.endswith("}")) or (raw.startswith("[") and raw.endswith("]")): + try: + return json.loads(raw) + except Exception: + return body + return body + + +def _code_indicates_success(value: Any) -> Optional[bool]: + if value is None: + return None + if isinstance(value, bool): + return value + if isinstance(value, (int, float)): + return int(value) in (0, 1, 200) + + text = str(value).strip() + if not text: + return None + lower = text.lower() + if lower in {"0", "00", "000", "0000", "1", "200", "ok", "success", "true"}: + return True + if lower in {"false", "fail", "failed", "error", "err"}: + return False + if lower.lstrip("-").isdigit(): + return int(lower) in (0, 1, 200) + return False + + +def _to_int_or_none(value: Any) -> Optional[int]: + try: + if value is None: + return None + if isinstance(value, bool): + return int(value) + if isinstance(value, (int, float)): + return int(value) + text = str(value).strip() + if not text: + return None + if text.lstrip("-").isdigit(): + return int(text) + return None + except Exception: + return None + + +def _extract_interface_message(data: dict) -> str: + for key in ("msg", "message", "error", "errorMsg", "detail", "retMsg"): + value = data.get(key) + if isinstance(value, str) and value.strip(): + return value.strip() + return "" + + +def _is_sms_already_sent_response(data: dict, message: str) -> bool: + code_keys = ("code", "retCode", "resultCode", "statusCode", "errno") + code_val = None + for key in code_keys: + if key in data: + code_val = _to_int_or_none(data.get(key)) + if code_val is not None: + break + if code_val != 1001: + return False + + text = (message or "").strip() + if not text: + return False + # 这类文案通常表示验证码已下发或短信频控，不应判定为滑块失败。 + hints = ("验证码已发送", "短信验证码已发送", "请稍后重试", "请稍等") + return any(h in text for h in hints) + + +def _assert_interface_success(body: Any) -> None: + data = _coerce_json_body(body) + if not isinstance(data, dict): + return + + fail_reasons: list[str] = [] + message = _extract_interface_message(data) + is_soft_success = _is_sms_already_sent_response(data, message) + + for key in ("success", "ok"): + if key in data and not bool(data[key]): + fail_reasons.append(f"{key}={data[key]}") + + for key in ("code", "retCode", "resultCode", "statusCode", "errno"): + if key not in data: + continue + code_ok = _code_indicates_success(data[key]) + if code_ok is False: + if is_soft_success: + continue + fail_reasons.append(f"{key}={data[key]}") + + if message: + lowered = message.lower() + fail_words = ("失败", "错误", "无效", "fail", "error", "invalid", "请重新", "未通过") + if any(word in lowered for word in fail_words): + if is_soft_success: + return + fail_reasons.append(f"msg={message}") + + if fail_reasons: + preview = str(data) + if len(preview) > 300: + preview = preview[:300] + "..." + raise RuntimeError(f"接口返回失败: {', '.join(fail_reasons)}; body={preview}") + + +def _wait_packet_or_feedback(page, timeout: float) -> tuple[Any, str]: + deadline = time.time() + timeout + last_toast = "" + while time.time() < deadline: + wait_span = min(0.35, max(0.05, deadline - time.time())) + packet = _normalize_listen_packet(page.listen.wait(timeout=wait_span, fit_count=False)) + if packet is not None: + return packet, last_toast + + toast = _read_slider_toast_text(page) + if toast: + last_toast = toast + + return None, last_toast + + +def _to_jsonable(value: Any) -> Any: + if value is None or isinstance(value, (str, int, float, bool)): + return value + if isinstance(value, dict): + return {str(k): _to_jsonable(v) for k, v in value.items()} + if isinstance(value, (list, tuple, set)): + return [_to_jsonable(v) for v in value] + return str(value) + + +def _analyze_slider_failure(match: dict, layout: dict, attempts: list[dict], last_reason: str) -> dict: + hypotheses: list[str] = [] + confidence_ratio = float(match.get("confidence_ratio", 0) or 0) + + if confidence_ratio < 1.03: + hypotheses.append("拼图匹配置信度偏低，背景与缺口特征区分不明显，易出现位移误差。") + + toast_texts = [str(a.get("toast") or "") for a in attempts if a.get("toast")] + if any("验证码错误" in t for t in toast_texts): + hypotheses.append("前端返回“验证码错误”，通常是最终落点偏差或拖动轨迹被风控识别。") + + interface_errors = [str(a.get("interface_error") or "") for a in attempts if a.get("interface_error")] + if interface_errors: + hypotheses.append("接口已返回失败状态，说明滑块校验请求已发出但服务端未判定通过。") + + packet_count = sum(1 for a in attempts if a.get("packet_received")) + if attempts and packet_count == 0: + hypotheses.append("多次拖动均未捕获到目标接口，请检查接口监听地址是否变化。") + + if attempts and bool(attempts[-1].get("verify_bar_visible_after")): + hypotheses.append("验证码弹窗仍可见，当前 challenge 未完成。") + + boundary_hits = sum(1 for a in attempts if a.get("boundary_hit")) + if boundary_hits > 0: + hypotheses.append("拖动距离命中轨道边界，存在位移被截断风险。") + + if not hypotheses and last_reason: + hypotheses.append(last_reason) + if not hypotheses: + hypotheses.append("未识别出单一原因，建议查看保存的标注图和报告。") + + return { + "summary": hypotheses[0], + "hypotheses": hypotheses, + "metrics": { + "confidence_ratio": confidence_ratio, + "best_score": match.get("best_score"), + "second_best": match.get("second_best"), + "attempt_count": len(attempts), + "packet_count": packet_count, + "track_max": layout.get("track_max"), + }, + } + + +def _save_failure_artifacts( + page, + bg_bytes: bytes, + piece_bytes: bytes, + match: dict, + layout: dict, + move_distances: list[int], + attempts: list[dict], + last_reason: str, + analysis: dict, +) -> Path: + root = Path(__file__).resolve().parent / "captcha_failures" + case_dir = root / datetime.now().strftime("%Y%m%d_%H%M%S_%f") + case_dir.mkdir(parents=True, exist_ok=True) + + (case_dir / "bg.png").write_bytes(bg_bytes) + (case_dir / "piece.png").write_bytes(piece_bytes) + + try: + page.get_screenshot(path=str(case_dir / "page.png")) + except Exception: + pass + + try: + bg_img = Image.open(BytesIO(bg_bytes)).convert("RGB") + draw = ImageDraw.Draw(bg_img) + y0 = int(match.get("piece_bbox_y0", 0) or 0) + h = int(match.get("piece_bbox_h", 0) or 0) + w = int(match.get("piece_bbox_w", 0) or 0) + candidate_xs = match.get("candidate_target_xs") or [match.get("target_x", 0)] + colors = [(255, 0, 0), (255, 140, 0), (0, 128, 255), (0, 180, 80), (160, 80, 255)] + for idx, x in enumerate(candidate_xs[:5], start=1): + color = colors[(idx - 1) % len(colors)] + x = int(x) + draw.rectangle([x, y0, x + w, y0 + h], outline=color, width=2 if idx == 1 else 1) + draw.text((x + 2, max(0, y0 - 14)), f"#{idx}", fill=color) + bg_img.save(case_dir / "bg_overlay.png") + except Exception: + pass + + report = { + "created_at": datetime.now().isoformat(timespec="seconds"), + "failure_reason": last_reason, + "analysis": analysis, + "match": _to_jsonable(match), + "layout": _to_jsonable(layout), + "move_distances": move_distances, + "attempts": _to_jsonable(attempts), + } + (case_dir / "report.json").write_text( + json.dumps(report, ensure_ascii=False, indent=2), + encoding="utf-8", + ) + return case_dir + + +# ---------- 滑块拖动（仿人轨迹） ---------- + +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("验证码关键元素缺失（背景图/拼图块/滑块）") + + # 6. 每次重试前重新抓取验证码图片并重算位移（不复用旧距离） + deadline = time.time() + 15.0 + last_reason = "" + attempts: list[dict[str, Any]] = [] + max_attempts = 6 + last_bg_bytes = b"" + last_piece_bytes = b"" + last_match: dict[str, Any] = {} + last_layout: dict[str, Any] = {} + last_move_distances: list[int] = [] + + for idx in range(1, max_attempts + 1): + if time.time() >= deadline: + break + + bg_img = find_first(page, ["css:.verify-img-panel img"], timeout=2) + piece_img = find_first(page, ["css:.verify-sub-block img"], timeout=2) + slider = find_first(page, ["css:.verify-move-block"], timeout=2) + + attempt: dict[str, Any] = { + "index": idx, + "move_distance": None, + "boundary_hit": False, + "packet_received": False, + "toast": "", + "interface_error": "", + "verify_bar_visible_after": False, + } + + if not bg_img or not piece_img or not slider: + last_reason = "重试时未找到验证码关键元素（背景图/拼图块/滑块）" + attempt["interface_error"] = last_reason + attempts.append(attempt) + break + + try: + bg_src = wait_for_data_src(bg_img, timeout=4) + piece_src = wait_for_data_src(piece_img, timeout=4) + bg_bytes = parse_data_url(bg_src) + piece_bytes = parse_data_url(piece_src) + except Exception as e: + last_reason = f"重试时读取验证码图片失败: {e}" + attempt["interface_error"] = last_reason + attempts.append(attempt) + continue + + if len(bg_bytes) < 100 or len(piece_bytes) < 100: + last_reason = f"验证码图片数据异常: bg={len(bg_bytes)}B piece={len(piece_bytes)}B" + attempt["interface_error"] = last_reason + attempts.append(attempt) + continue + + try: + match = calc_drag_distance_from_bytes( + bg_bytes, piece_bytes, alpha_threshold=alpha_threshold + ) + except Exception as e: + last_reason = f"重试时位移计算失败: {e}" + attempt["interface_error"] = last_reason + attempts.append(attempt) + continue + + layout = _get_slider_layout_metrics(page, bg_img, slider) + bg_display_w = layout["bg_display_w"] if layout["bg_display_w"] > 0 else match["bg_width"] + scale = float(bg_display_w) / max(1, match["bg_width"]) + track_max = int(layout["track_max"]) + move_distances = _build_move_distance_candidates( + match=match, + scale=scale, + distance_adjust=int(distance_adjust), + track_max=track_max, + ) + pick_index = min(max(0, idx - 1), len(move_distances) - 1) + move_distance = int(move_distances[pick_index]) + + attempt["move_distance"] = move_distance + attempt["boundary_hit"] = bool(track_max >= 0 and move_distance in (0, track_max)) + attempt["confidence_ratio"] = float(match.get("confidence_ratio", 0) or 0) + attempt["candidate_count"] = len(move_distances) + + # 记录最后一次有效计算，用于失败样本保存与分析 + last_bg_bytes = bg_bytes + last_piece_bytes = piece_bytes + last_match = match + last_layout = layout + last_move_distances = move_distances + + drag_slider(page, slider, move_distance) + time.sleep(0.2) + + remaining = max(0.1, deadline - time.time()) + packet, toast_text = _wait_packet_or_feedback(page, timeout=min(4.0, remaining)) + attempt["toast"] = toast_text or "" + if packet is not None: + attempt["packet_received"] = True + response = getattr(packet, "response", None) + if response is None: + attempt["interface_error"] = f"捕获到 {GET_YAN_ZHEN_MA_URL} 数据包但无 response 字段" + last_reason = attempt["interface_error"] + attempt["verify_bar_visible_after"] = _is_element_visible(page, ".verify-bar-area") + attempts.append(attempt) + if not attempt["verify_bar_visible_after"]: + break + continue + + body = _coerce_json_body(response.body) + attempt["response_body"] = _to_jsonable(body) + try: + _assert_interface_success(body) + return body + except Exception as e: + attempt["interface_error"] = str(e) + last_reason = attempt["interface_error"] + attempt["verify_bar_visible_after"] = _is_element_visible(page, ".verify-bar-area") + attempts.append(attempt) + if not attempt["verify_bar_visible_after"]: + break + continue + + if toast_text: + last_reason = f"滑块验证失败：{toast_text}" + elif _is_element_visible(page, ".verify-bar-area"): + last_reason = f"第{idx}次拖动未触发 {GET_YAN_ZHEN_MA_URL}" + else: + last_reason = "验证码弹窗已关闭但未捕获接口响应" + attempt["verify_bar_visible_after"] = _is_element_visible(page, ".verify-bar-area") + attempt["interface_error"] = last_reason if not attempt["interface_error"] else attempt["interface_error"] + attempts.append(attempt) + + analysis = _analyze_slider_failure( + match=last_match, + layout=last_layout, + attempts=attempts, + last_reason=last_reason, + ) + save_info = "" + try: + case_dir = _save_failure_artifacts( + page=page, + bg_bytes=last_bg_bytes, + piece_bytes=last_piece_bytes, + match=last_match, + layout=last_layout, + move_distances=last_move_distances, + attempts=attempts, + last_reason=last_reason, + analysis=analysis, + ) + save_info = f"; 失败样本目录: {case_dir}" + except Exception as e: + save_info = f"; 失败样本保存失败: {e}" + + raise RuntimeError( + f"滑块未通过或接口未返回成功结果: {last_reason or '未知原因'}" + f"; 分析: {analysis.get('summary', '无')}" + f"{save_info}" + ) + + +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,clear=True) + time.sleep(0.5) + + page.listen.start("hn_userEquitys/common/order") + page.ele('x://*[@id="app"]/div/img').click(by_js=True) + time.sleep(0.5) + page.ele('x://*[@id="app"]/div/div[7]/div/div[3]/button[2]').click(by_js=True) + res = page.listen.wait() + print(res.response.body) + +if __name__ == "__main__": + main()