"""test6 — 方案一：连续轨迹、分阶段定时、TouchPipe 与到位轮询。""" import sys, os, json, time, random, math, threading import uiautomator2 as u2 from uiautomator2.core import AdbHTTPConnection import cv2, numpy as np, base64, requests from rapidocr_onnxruntime import RapidOCR os.environ["PYTHONIOENCODING"] = "utf-8" sys.path.insert(0, r"D:\drug\sg") from detect_slider_button import detect_slider_button from detect_captcha_edge import detect_captcha_left_edge ocr_eng = RapidOCR() TOKEN = "1nDVocTE2mJ0yLEYb2sZJ5uUY2VIEoGTkIpW44X7Kgk" JFBYM_URL = "http://api.jfbym.com/api/YmServer/customApi" BASE = os.path.dirname(os.path.abspath(__file__)) IMG_ROOT = os.path.join(BASE, "image", "test6_image") OUT = os.path.join(IMG_ROOT, "d") ALIGN_DIR = os.path.join(IMG_ROOT, "align") # 按天分类：success/failure 下按日期建子目录（如 success/2026-08-19/xxx.png） _TODAY = time.strftime("%Y-%m-%d") SUCCESS_DIR = os.path.join(IMG_ROOT, "success", _TODAY) FAILURE_DIR = os.path.join(IMG_ROOT, "failure", _TODAY) os.makedirs(OUT, exist_ok=True) os.makedirs(ALIGN_DIR, exist_ok=True) os.makedirs(SUCCESS_DIR, exist_ok=True) os.makedirs(FAILURE_DIR, exist_ok=True) OFFSET_COMPENSATE = 0 MICRO_MODE = "none" # 真人速度模板的慢速尾段原本约占最后 10% 路程。运行时把这段压缩到 # 最后 3%~5%，让前段继续快速靠近，离缺口很近后才明显减速。 RETURN_BRAKE_SOURCE_START = 0.90 RETURN_BRAKE_START_RANGE = (0.95, 0.97) # 折回主段使用固定的快速节奏，不再直接照搬某一条可能很慢的真人模板。 # 采样间隔保持在触摸事件可稳定消费的范围，距离越长只增加点数，不拉长点间隔。 RETURN_SAMPLE_INTERVAL_RANGE = (0.0042, 0.0052) RETURN_BRAKE_TIME_RANGE = (0.58, 0.66) RETURN_CURVE_LIMIT_1220 = 42 RETURN_Y_CORRIDOR_1220 = 46 RETURN_END_DRIFT_LIMIT_1220 = 28 # 方案一参数：按真人样本的量级修正点密度与阶段时长。 RIGHT_SAMPLE_INTERVAL_RANGE = (0.0038, 0.0052) RIGHT_POINT_LIMITS = (60, 110) # 右滑移动本身按 test5 的真人节奏控制；到最右端后的幕布展开等待 # 由 RIGHT_SETTLE_RANGE 单独负责，不计入右滑阶段。 RIGHT_DURATION_RANGE = (0.25, 0.40) TRACK_SKIP_LATE_S = 0.010 # 真人样本的右端折返点集中在约 1023~1109 px；不要每次都固定在 1190。 # 下限略高于滑块可确认的右端，避免随机到太靠左导致幕布未完全展开。 RIGHT_END_RANGE_1220 = (1060, 1105) # 真人右滑大多是轻微向上收尾（Y 减小），按水平距离归一化。 RIGHT_Y_DRIFT_RATIO_RANGE = (-0.055, -0.018) # TouchPipe 异步发送后，至少给幕布动画和设备事件队列留出稳定时间。 RIGHT_SETTLE_RANGE = (0.55, 0.85) RIGHT_READY_TIMEOUT_S = 3.50 RETURN_READY_TIMEOUT_S = 1.80 SLIDER_POSITION_TOLERANCE = 12 # 真人轨迹模板目录。模板只使用归一化后的形状，不直接复用原始触摸坐标， # 因此不会把旧设备的绝对坐标带到当前屏幕。 GUIJI_ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "guiji") _GUIJI_TEMPLATES = None def _clamp(v, lo, hi): return max(lo, min(v, hi)) def _detect_slider_center(image): """从 OpenCV 截图中检测橙色滑块中心。""" if image is None or getattr(image, "ndim", 0) != 3: return None rgb = cv2.cvtColor(image, cv2.COLOR_BGR2RGB) hsv = cv2.cvtColor(rgb, cv2.COLOR_RGB2HSV) mask = cv2.inRange(hsv, np.array([8, 230, 230]), np.array([22, 255, 255])) h, _ = mask.shape[:2] mask[:int(h * 0.55), :] = 0 kernel = np.ones((5, 5), np.uint8) mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel) mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel) contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) candidates = [] for cnt in contours: x, y, bw, bh = cv2.boundingRect(cnt) area = bw * bh if area < 300 or bw < 30 or bh < 30: continue aspect = min(bw, bh) / max(bw, bh) if aspect < 0.5: continue candidates.append((area * aspect, x, bw)) if not candidates: return None _, x, bw = max(candidates, key=lambda item: item[0]) return x + bw / 2.0 def _wait_for_slider_position(pipe, device, x, y, expected_center=None, min_center=None, timeout_s=3.0, label="", track_pts=None, t0=None, phase="confirm", stable_required=2): """重复发送终点并轮询真实滑块位置，避免截图/松手早于异步事件落地。""" started = time.perf_counter() stable = 0 last_center = None last_frame = None while time.perf_counter() - started < timeout_s: send_ms = 0.0 try: send_ms = pipe.move(int(x), int(y), 0) except Exception: pass if track_pts is not None: track_pts.append({ "x": int(x), "y": int(y), "pressure": 0, "phase": phase, "confirm": True, "scheduled_ms": None, "lag_ms": None, "send_ms": send_ms, "rel_ms": ((time.perf_counter() - t0) * 1000 if t0 is not None else None), }) time.sleep(0.06) try: frame = device.screenshot(format="opencv") except Exception: frame = None if frame is not None: last_frame = frame center = _detect_slider_center(frame) if center is not None: last_center = center if expected_center is not None: ok = abs(center - expected_center) <= SLIDER_POSITION_TOLERANCE elif min_center is not None: ok = center >= min_center else: ok = True stable = stable + 1 if ok else 0 if stable >= stable_required: elapsed = (time.perf_counter() - started) * 1000 return last_frame, last_center, elapsed, True time.sleep(0.06) elapsed = (time.perf_counter() - started) * 1000 if label: actual = "未检测到" if last_center is None else f"{last_center:.0f}" target = (f"{expected_center:.0f}" if expected_center is not None else f">={min_center:.0f}" if min_center is not None else "任意") print(f" {label}确认超时: actual={actual}, target={target}") return last_frame, last_center, elapsed, False def _dedupe_track(points): """移除取整后相邻的重复坐标，避免末端连续发送同一个点。""" result = [] for x, y in points: point = (int(x), int(y)) if not result or point != result[-1]: result.append(point) return result def _pressure_curve(i, n, phase): # uiautomator2 injectInputEvent 的第四参数是 metaState，不是压力；保持为 0。 return 0 def _choose_right_end_x(start_x, screen_w): """返回接近真人分布、且仍能完全揭开幕布的右侧折返点。""" scale = screen_w / 1220.0 min_travel = 800 * scale lo = int(round(RIGHT_END_RANGE_1220[0] * scale)) hi = int(round(RIGHT_END_RANGE_1220[1] * scale)) sampled = random.uniform(lo, hi) return int(round(_clamp(sampled, start_x + min_travel, screen_w - 12))) def _return_duration(distance, scale): """生成快速折回总时长；接口等待和右端展开等待不计入移动时间。""" # 先用较高的巡航速度完成大部分距离，再为最后3%~5%保留短暂收尾。 # 速度按屏幕比例归一化，避免长距离被旧模板拖到1.5秒以上。 speed = random.uniform(1450.0, 1750.0) * max(scale, 0.8) if distance > 600 * scale: tail = random.uniform(0.18, 0.24) lo, hi = 0.62, 0.88 elif distance > 450 * scale: tail = random.uniform(0.16, 0.22) lo, hi = 0.48, 0.70 else: tail = random.uniform(0.14, 0.20) lo, hi = 0.34, 0.52 return _clamp(distance / speed + tail, lo, hi) def _timing_schedule(count, duration_s, phase): """生成总时长固定、点间隔轻微相关的发送时间表。""" if count <= 1: return [0.0] phase_shift = random.uniform(0, math.tau) weights = [] for i in range(count - 1): t = i / max(1, count - 2) correlated = 0.10 * math.sin(math.tau * (1.2 * t) + phase_shift) jitter = random.uniform(-0.05, 0.05) if phase == 'return' and t > 0.75: correlated += 0.08 * (t - 0.75) / 0.25 weights.append(max(0.70, 1.0 + correlated + jitter)) total = sum(weights) elapsed = 0.0 schedule = [0.0] for weight in weights: elapsed += duration_s * weight / total schedule.append(elapsed) schedule[-1] = duration_s return schedule def _emit_track(pipe, track, phase, track_pts, t0, duration_s, schedule_override=None): """按绝对时间表发送；落后过多时丢弃过期中间点，终点始终发送。""" if not track: return 0.0, { "planned_points": 0, "sent_points": 0, "skipped_points": 0, "target_ms": duration_s * 1000, "elapsed_ms": 0.0, "lag_p95_ms": 0.0, "lag_max_ms": 0.0, "send_p95_ms": 0.0, "send_max_ms": 0.0, } started = time.perf_counter() if schedule_override is not None and len(schedule_override) == len(track): schedule = list(schedule_override) schedule[0] = 0.0 schedule[-1] = duration_s else: schedule = _timing_schedule(len(track), duration_s, phase) lag_samples = [] send_samples = [] skipped = 0 for i, (x, y) in enumerate(track): deadline = started + schedule[i] wait_s = deadline - time.perf_counter() if wait_s > 0: time.sleep(wait_s) lag_s = max(0.0, time.perf_counter() - deadline) # 中间点已经过期时继续补发只会制造事件突发；最后一点不能跳过。 if i < len(track) - 1 and lag_s > TRACK_SKIP_LATE_S: skipped += 1 continue pressure = _pressure_curve(i, len(track), phase) send_ms = pipe.move(x, y, pressure) lag_ms = max(0.0, (time.perf_counter() - deadline) * 1000) lag_samples.append(lag_ms) send_samples.append(send_ms) track_pts.append({"x": x, "y": y, "pressure": pressure, "phase": phase, "scheduled_ms": schedule[i] * 1000, "lag_ms": lag_ms, "send_ms": send_ms, "rel_ms": (time.perf_counter()-t0)*1000}) elapsed_s = time.perf_counter() - started def percentile(values, ratio): if not values: return 0.0 ordered = sorted(values) return ordered[int(round((len(ordered) - 1) * ratio))] stats = { "planned_points": len(track), "sent_points": len(track) - skipped, "skipped_points": skipped, "target_ms": duration_s * 1000, "elapsed_ms": elapsed_s * 1000, "lag_p95_ms": percentile(lag_samples, 0.95), "lag_max_ms": max(lag_samples, default=0.0), "send_p95_ms": percentile(send_samples, 0.95), "send_max_ms": max(send_samples, default=0.0), } return elapsed_s, stats class TouchPipe: def __init__(self, dev): self._dev = dev; self._conn = None; self._sock = None self._stop = threading.Event(); self._drainer = None self._lock = threading.Lock(); self._fallback = False self._open_ok = False; self._open_error = None self._fallback_reason = None; self._last_error = None self._send_errors = 0; self._move_errors = 0 self._pipe_move_ms = []; self._fallback_move_ms = [] def open(self): try: self._conn = AdbHTTPConnection(self._dev.adb_device, port=9008) self._conn.timeout = 15; self._conn.connect() self._sock = self._conn.sock; self._sock.settimeout(0.5) self._drainer = threading.Thread(target=self._drain, daemon=True) self._drainer.start() self._open_ok = True except Exception as exc: self._fallback = True self._fallback_reason = "open_error" self._open_error = repr(exc) self._last_error = repr(exc) return self def _drain(self): while not self._stop.is_set(): try: if not self._sock.recv(65536): break except Exception: continue def move(self, x, y, pressure=0): if self._fallback: started = time.perf_counter() try: self._dev.touch.move(int(x), int(y)) except Exception as exc: self._move_errors += 1 self._last_error = repr(exc) elapsed_ms = (time.perf_counter() - started) * 1000 self._fallback_move_ms.append(elapsed_ms) return elapsed_ms started = time.perf_counter() try: with self._lock: self._sock.sendall(self._req(x, y, pressure)) elapsed_ms = (time.perf_counter() - started) * 1000 self._pipe_move_ms.append(elapsed_ms) return elapsed_ms except Exception as exc: self._send_errors += 1 self._last_error = repr(exc) self._fallback = True self._fallback_reason = "send_error" return self.move(x, y, pressure) def _req(self, x, y, pressure): body = json.dumps({"jsonrpc":"2.0","id":1,"method":"injectInputEvent", "params":[2,int(x),int(y),0]}).encode() return (f"POST /jsonrpc/0 HTTP/1.1\r\nHost: localhost\r\n" f"User-Agent: u2\r\nAccept-Encoding: \r\n" f"Content-Type: application/json\r\nContent-Length: {len(body)}\r\n" f"Connection: keep-alive\r\n\r\n").encode() + body def close(self): self._stop.set() if self._drainer: self._drainer.join(timeout=1) try: if self._sock: self._sock.close() except Exception: pass def diagnostics(self): def percentile(values, ratio): if not values: return 0.0 ordered = sorted(values) return ordered[int(round((len(ordered) - 1) * ratio))] return { "open_ok": self._open_ok, "fallback_used": self._fallback, "fallback_reason": self._fallback_reason, "open_error": self._open_error, "last_error": self._last_error, "send_errors": self._send_errors, "move_errors": self._move_errors, "pipe_moves": len(self._pipe_move_ms), "fallback_moves": len(self._fallback_move_ms), "pipe_send_p50_ms": percentile(self._pipe_move_ms, 0.50), "pipe_send_p95_ms": percentile(self._pipe_move_ms, 0.95), "pipe_send_max_ms": max(self._pipe_move_ms, default=0.0), "fallback_move_p95_ms": percentile(self._fallback_move_ms, 0.95), "fallback_move_max_ms": max(self._fallback_move_ms, default=0.0), } def _print_touchpipe_diagnostics(pipe): diag = pipe.diagnostics() print(f" TouchPipe: open={'OK' if diag['open_ok'] else 'FAIL'} " f"fallback={diag['fallback_used']} errors={diag['send_errors']} " f"send95={diag['pipe_send_p95_ms']:.2f}ms " f"sendMax={diag['pipe_send_max_ms']:.2f}ms") if diag["last_error"]: print(f" TouchPipe最后错误: {diag['last_error']}") return diag def _normalize_phase_template(points, phase): """Convert one recorded guiji phase into a device-independent curve. u is horizontal progress (0..1); residual is the signed distance from the straight endpoint chord divided by the horizontal span. This keeps the shape reusable on a different screen size and avoids copying raw touch coordinates from the recording device. """ if not points or len(points) < 12: return None xy = [(float(p.get("x", 0)), float(p.get("y", 0))) for p in points] if phase == "right": split = max(range(len(xy)), key=lambda i: xy[i][0]) xy = xy[:split + 1] x0, x1 = xy[0][0], xy[-1][0] span = x1 - x0 if span < 1000: return None raw_u = [(x - x0) / span for x, _ in xy] else: split = max(range(len(xy)), key=lambda i: xy[i][0]) xy = xy[split:] if len(xy) < 12: return None x0, x1 = xy[0][0], xy[-1][0] span = x0 - x1 if span < 800: return None raw_u = [(x0 - x) / span for x, _ in xy] # End-of-phase recordings can contain a few backtracking pixels. Preserve # the broad human curve but make the interpolation coordinate monotonic. raw_u = np.maximum.accumulate(np.asarray(raw_u, dtype=np.float64)) raw_u = np.clip(raw_u, 0.0, 1.0) unique_u, unique_idx = np.unique(raw_u, return_index=True) if len(unique_u) < 8: return None yy = np.asarray([xy[i][1] for i in unique_idx], dtype=np.float64) sample_u = np.linspace(0.0, 1.0, 81) sample_y = np.interp(sample_u, unique_u, yy) chord_y = sample_y[0] + (sample_y[-1] - sample_y[0]) * sample_u residual = (sample_y - chord_y) / span residual[0] = 0.0 residual[-1] = 0.0 # A few recordings contain unusually large endpoint excursions. They are # valid data, but clipping keeps one outlier from producing an unsafe path. residual = np.clip(residual, -0.18, 0.18) return { "u": sample_u.tolist(), "residual": residual.tolist(), "curvature": float(np.max(np.abs(residual))), "raw_points": len(xy), } def _normalize_return_velocity_template(points): """Extract horizontal progress as a function of elapsed return time.""" if not points or len(points) < 20: return None turn_idx = max(range(len(points)), key=lambda i: float(points[i].get("x", 0))) turn_x = float(points[turn_idx].get("x", 0)) end_x = float(points[-1].get("x", 0)) full_span = turn_x - end_x if full_span < 500: return None threshold = max(10.0, full_span * 0.01) start_idx = None for i in range(turn_idx + 1, len(points)): if turn_x - float(points[i].get("x", 0)) >= threshold: start_idx = i break if start_idx is None or len(points) - start_idx < 12: return None phase = points[start_idx:] x0 = float(phase[0].get("x", 0)) x1 = float(phase[-1].get("x", 0)) span = x0 - x1 t0 = float(phase[0].get("rel_ms", 0.0)) t1 = float(phase[-1].get("rel_ms", 0.0)) duration_ms = t1 - t0 if span < 300 or duration_ms < 100: return None raw_t = np.asarray([ (float(p.get("rel_ms", 0.0)) - t0) / duration_ms for p in phase ], dtype=np.float64) raw_u = np.asarray([ (x0 - float(p.get("x", 0))) / span for p in phase ], dtype=np.float64) raw_t = np.clip(raw_t, 0.0, 1.0) raw_u = np.maximum.accumulate(np.clip(raw_u, 0.0, 1.0)) unique_t, unique_idx = np.unique(raw_t, return_index=True) if len(unique_t) < 8: return None unique_u = raw_u[unique_idx] # The recorder often keeps sending the final coordinate after the slider # has already reached the target. That endpoint plateau is a hold phase, # not return motion, so trim it from the velocity template. end_idx = next((i for i, value in enumerate(unique_u) if value >= 0.999), len(unique_u) - 1) motion_end_t = max(float(unique_t[end_idx]), 1e-6) event_t = unique_t[:end_idx + 1] / motion_end_t event_progress = unique_u[:end_idx + 1] event_t[0] = 0.0 event_t[-1] = 1.0 event_progress[0] = 0.0 event_progress[-1] = 1.0 sample_t = np.linspace(0.0, 1.0, 101) sample_u = np.interp(sample_t, event_t, event_progress) sample_u[0] = 0.0 sample_u[-1] = 1.0 return { "t": sample_t.tolist(), "progress": sample_u.tolist(), "event_t": event_t.tolist(), "event_progress": event_progress.tolist(), "duration_ms": duration_ms * motion_end_t, "plateau_ms": duration_ms * (1.0 - motion_end_t), "raw_points": len(event_t), } def _load_guiji_templates(): global _GUIJI_TEMPLATES if _GUIJI_TEMPLATES is not None: return _GUIJI_TEMPLATES templates = { "right": [], "short": [], "medium": [], "long": [], "velocity": {"short": [], "medium": [], "long": []}, } try: names = sorted(n for n in os.listdir(GUIJI_ROOT) if n.lower().endswith(".json")) except OSError: names = [] for name in names: try: with open(os.path.join(GUIJI_ROOT, name), "r", encoding="utf-8") as fh: record = json.load(fh) points = record.get("points") or [] right = _normalize_phase_template(points, "right") ret = _normalize_phase_template(points, "return") velocity = _normalize_return_velocity_template(points) if right: templates["right"].append(right) if ret: span = max(float(points[i]["x"]) for i in range(len(points))) - float(points[-1]["x"]) if span < 6300: key = "short" elif span < 8400: key = "medium" else: key = "long" templates[key].append(ret) if velocity: templates["velocity"][key].append(velocity) except (OSError, ValueError, KeyError, TypeError): continue # The medium bucket has few observations, so use all return templates as a # safe fallback instead of inventing an unrelated synthetic curve. all_returns = templates["short"] + templates["medium"] + templates["long"] for key in ("short", "medium", "long"): if not templates[key]: templates[key] = all_returns[:] all_velocity = (templates["velocity"]["short"] + templates["velocity"]["medium"] + templates["velocity"]["long"]) for key in ("short", "medium", "long"): if not templates["velocity"][key]: templates["velocity"][key] = all_velocity[:] _GUIJI_TEMPLATES = templates print(" 真人模板: right={} short={} medium={} long={} velocity={}/{}/{}".format( len(templates["right"]), len(templates["short"]), len(templates["medium"]), len(templates["long"]), len(templates["velocity"]["short"]), len(templates["velocity"]["medium"]), len(templates["velocity"]["long"]))) return templates def _template_residual(template, t): if not template: return 0.0 return float(np.interp(float(t), template["u"], template["residual"])) def _template_progress(template, t): if not template: # Smooth fallback with a short acceleration section and a long braking # tail. Real templates are available in normal operation. t = _clamp(float(t), 0.0, 1.0) knots_t = np.asarray([0.0, 0.08, 0.20, 0.35, 0.55, 0.75, 0.90, 1.0]) knots_u = np.asarray([0.0, 0.07, 0.23, 0.43, 0.63, 0.80, 0.93, 1.0]) return float(np.interp(t, knots_t, knots_u)) return float(np.interp(float(t), template["t"], template["progress"])) def _compress_return_braking_tail(progress, brake_start): """把模板最后 10% 的慢速路程压缩到最后 3%~5%，保持时间顺序不变。""" values = np.asarray(progress, dtype=np.float64).copy() source = RETURN_BRAKE_SOURCE_START target = float(_clamp(brake_start, source + 0.01, 0.99)) before = values <= source values[before] *= target / source values[~before] = ( target + (values[~before] - source) * (1.0 - target) / (1.0 - source) ) values = np.maximum.accumulate(np.clip(values, 0.0, 1.0)) values[0] = 0.0 values[-1] = 1.0 return values def _build_fast_return_progress(count, brake_start, brake_time): """前段快速匀速推进，最后一小段才进入明显的减速/微调。""" if count <= 1: return np.asarray([0.0]), np.asarray([0.0]) event_t = np.linspace(0.0, 1.0, int(count), dtype=np.float64) event_u = np.empty_like(event_t) main = event_t <= brake_time main_t = np.clip(event_t[main] / max(brake_time, 1e-6), 0.0, 1.0) # 轻微的自然起步，不制造旧模板那种长时间慢爬。 event_u[main] = brake_start * np.power(main_t, 0.96) tail_t = np.clip( (event_t[~main] - brake_time) / max(1.0 - brake_time, 1e-6), 0.0, 1.0, ) # 刚进入最后一段时仍有少量位移，随后逐步减小到目标。 event_u[~main] = brake_start + (1.0 - brake_start) * ( 1.0 - np.power(1.0 - tail_t, 2.2) ) event_u[0] = 0.0 event_u[-1] = 1.0 return event_t, np.maximum.accumulate(np.clip(event_u, 0.0, 1.0)) def _fallback_bow(t, category, amplitude=None): """Low-frequency fallback with the same distance-dependent curvature.""" ratios = {"short": (0.020, 0.055), "medium": (0.035, 0.075), "long": (0.050, 0.115)} lo, hi = ratios[category] amp = amplitude if amp is None: amp = random.uniform(lo, hi) * random.choice((-1.0, 1.0)) # One broad asymmetric bow; no high-frequency wobble. return amp * math.sin(math.pi * t) * (0.88 + 0.24 * t) def _build_right_track(start_x, start_y, end_x, point_count): """右滑轨迹：连续低频起伏，避免逐点随机造成锯齿。""" dist = end_x - start_x if dist <= 0: return [(int(start_x), int(start_y))] # Use a normalized guiji curve for the broad motion. The old branch below # is retained as unreachable reference code while the new generator is # validated against saved images. steps = max(1, int(point_count) - 1) templates = _load_guiji_templates().get("right", []) template = random.choice(templates) if templates else None template_sign = random.choice((-1.0, 1.0)) template_gain = random.uniform(0.82, 1.08) fallback_amp = random.uniform(0.020, 0.055) * random.choice((-1.0, 1.0)) # 真人右滑通常是轻微向上（Y 减小），不再生成明显向下的漂移。 drift = random.uniform(*RIGHT_Y_DRIFT_RATIO_RANGE) * dist ease_exp = random.uniform(1.65, 2.05) points = [(int(start_x), int(start_y))] for i in range(1, steps + 1): t = i / steps u = 1.0 - (1.0 - t) ** ease_exp x = start_x + dist * u residual = (_template_residual(template, u) * template_sign * template_gain if template else _fallback_bow(u, "short", fallback_amp)) residual = _clamp(residual, -0.06, 0.06) y = start_y + drift * u + residual * dist points.append((int(round(x)), int(round(y)))) points[-1] = (int(end_x), points[-1][1]) return _dedupe_track(points) # Legacy procedural branch kept below for easy rollback during validation. steps = max(1, int(point_count) - 1) form = random.choices( ['flat', 'arch', 'decline', 'slope'], weights=[0.25, 0.35, 0.20, 0.20], k=1, )[0] drift = random.triangular(-70, 95, 22) if abs(drift) < 18: drift = 18 * random.choice([-1, 1]) arch_h = random.triangular(18, 70, 38) * random.choice([-1, 1]) decline_dy = random.triangular(20, 90, 45) slope_dy = random.triangular(-75, 100, 25) ease_exp = random.uniform(1.65, 2.15) wobble_amp = random.uniform(1.5, 5.0) wobble_cycles = random.uniform(1.0, 2.2) wobble_phase = random.uniform(0, math.tau) points = [(int(start_x), int(start_y))] for i in range(1, steps + 1): t = i / steps x = start_x + dist * (1.0 - (1.0 - t) ** ease_exp) if form == 'flat': y = start_y + drift * t elif form == 'arch': y = start_y + drift * t + arch_h * math.sin(math.pi * t) elif form == 'decline': y = start_y + drift * t + abs(arch_h) * math.sin(math.pi * t) + decline_dy * (t ** 2) else: # slope y = start_y + slope_dy * t y += (wobble_amp * math.sin(math.tau * wobble_cycles * t + wobble_phase) * math.sin(math.pi * t)) points.append((int(round(x)), int(round(y)))) points[-1] = (int(end_x), points[-1][1]) return _dedupe_track(points) def _build_human_return_track(start_x, start_y, target_x, duration_s=None, return_schedule=False): """折返轨迹：按距离调整点密度，并保留轻微的平滑回修。""" distance = start_x - target_x if distance <= 0: result = [(int(target_x), int(start_y))] return (result, [0.0]) if return_schedule else result scale = W / 1220.0 if distance > 600 * scale: cat = "long" end_ratio = random.triangular(-0.045, 0.12, 0.035) elif distance > 450 * scale: cat = "medium" end_ratio = random.triangular(-0.035, 0.10, 0.025) else: cat = "short" end_ratio = random.triangular(-0.025, 0.08, 0.018) if duration_s is None: duration_s = _return_duration(distance, scale) templates = _load_guiji_templates() geometry_templates = templates.get(cat, []) geometry = random.choice(geometry_templates) if geometry_templates else None end_dy_limit = int(round(RETURN_END_DRIFT_LIMIT_1220 * scale)) end_dy = int(_clamp(round(end_ratio * distance), -end_dy_limit, end_dy_limit)) target_y = start_y + end_dy shape_gain = random.uniform(0.82, 1.08) shape_sign = random.choice((-1.0, 1.0)) curve_limit_px = { "short": 30.0, "medium": 36.0, "long": float(RETURN_CURVE_LIMIT_1220), }[cat] * scale residual_limit = min( {"short": 0.06, "medium": 0.075, "long": 0.115}[cat], curve_limit_px / max(distance, 1.0), ) fallback_lo = min({"short": 0.020, "medium": 0.035, "long": 0.050}[cat], residual_limit) fallback_hi = min({"short": 0.055, "medium": 0.075, "long": 0.115}[cat], residual_limit) fallback_amp = random.uniform(fallback_lo, max(fallback_lo, fallback_hi)) \ * random.choice((-1.0, 1.0)) interval = random.uniform(*RETURN_SAMPLE_INTERVAL_RANGE) steps = int(_clamp(round(duration_s / interval) + 1, {"short": 60, "medium": 90, "long": 120}[cat], {"short": 180, "medium": 230, "long": 300}[cat])) brake_start = random.uniform(*RETURN_BRAKE_START_RANGE) brake_time = random.uniform(*RETURN_BRAKE_TIME_RANGE) event_t, event_u = _build_fast_return_progress( steps, brake_start, brake_time ) track = [] max_residual = 0.0 max_y_deviation = 0.0 y_corridor = RETURN_Y_CORRIDOR_1220 * scale for t, u in zip(event_t, event_u): residual = (_template_residual(geometry, u) * shape_sign * shape_gain if geometry else _fallback_bow(float(u), cat, fallback_amp)) residual = _clamp(residual, -residual_limit, residual_limit) max_residual = max(max_residual, abs(residual)) x = start_x - distance * float(u) y = start_y + end_dy * float(u) + residual * distance y = _clamp(y, start_y - y_corridor, start_y + y_corridor) max_y_deviation = max(max_y_deviation, abs(y - start_y)) track.append((int(round(x)), int(round(y)))) if track: track[0] = (int(start_x), int(start_y)) track[-1] = (int(target_x), int(target_y)) # 取整后相同的坐标不再重复发送；下一次不同坐标的时间仍保留， # 这样末端是短暂停顿而不是高频轰炸同一个像素。 compact_track = [] compact_t = [] for point, t in zip(track, event_t): if not compact_track or point != compact_track[-1]: compact_track.append(point) compact_t.append(float(t)) track = compact_track event_t = np.asarray(compact_t, dtype=np.float64) if len(event_t) > 1: event_t[0] = 0.0 event_t[-1] = 1.0 schedule = (event_t * float(duration_s)).tolist() print(f" 折回: {cat} distance={distance:.0f}px {len(track)}点 " f"Y漂={end_dy:+d} 弯曲={max_residual * distance:.0f}px " f"比例={max_residual * 100:.1f}% Y范围={max_y_deviation:.0f}px " f"速度=快速分段 时长={duration_s * 1000:.0f}ms " f"快速段={brake_time * duration_s * 1000:.0f}ms " f"减速起点={brake_start * 100:.1f}% " f"尾段={(1.0 - brake_start) * 100:.1f}%") return (track, schedule) if return_schedule else track # Legacy procedural branch kept below for rollback during validation. if distance > 600 * scale: cat = "long" steps = int(_clamp(distance / random.uniform(2.8, 4.1), 180, 380)) end_ratio = random.triangular(-0.045, 0.12, 0.035) elif distance > 450 * scale: cat = "medium" steps = int(_clamp(distance / random.uniform(2.7, 4.2), 130, 250)) end_ratio = random.triangular(-0.035, 0.10, 0.025) else: cat = "short" steps = int(_clamp(distance / random.uniform(2.0, 3.8), 75, 210)) end_ratio = random.triangular(-0.025, 0.08, 0.018) end_dy = int(round(end_ratio * distance)) templates = _load_guiji_templates().get(cat, []) template = random.choice(templates) if templates else None shape_gain = random.uniform(0.80, 1.12) shape_sign = random.choice((-1.0, 1.0)) fallback_amp = random.uniform( {"short": 0.020, "medium": 0.035, "long": 0.050}[cat], {"short": 0.055, "medium": 0.075, "long": 0.115}[cat], ) * random.choice((-1.0, 1.0)) x_exp = random.uniform(1.55, 2.05) linear_tail = random.uniform(0.16, 0.24) target_y = start_y + end_dy track = [] max_residual = 0.0 for i in range(steps): t = i / max(1, steps - 1) # Horizontal movement eases into the target. The residual is a # single broad bow learned from guiji, rather than random wobble. ease = 1.0 - (1.0 - t) ** x_exp u = ease * (1.0 - linear_tail) + t * linear_tail x = start_x - distance * u residual = (_template_residual(template, u) * shape_sign * shape_gain if template else _fallback_bow(u, cat, fallback_amp)) residual_limit = {"short": 0.06, "medium": 0.075, "long": 0.115}[cat] residual = _clamp(residual, -residual_limit, residual_limit) max_residual = max(max_residual, abs(residual)) chord_y = start_y + end_dy * u y = chord_y + residual * distance track.append((int(round(x)), int(round(y)))) if track: track[-1] = (int(target_x), int(target_y)) print(f" 折回: {cat} distance={distance:.0f}px {len(track)}点 " f"Y漂={end_dy:+d} 弯曲={max_residual * distance:.0f}px " f"比例={max_residual * 100:.1f}%") return _dedupe_track(track) # Legacy endpoint-easing branch kept below for rollback during validation. scale = W / 1220.0 if distance > 600 * scale: # Keep the long return close to the slider rail. A 300+ px Y drift # puts the final events outside the control and the device stops # applying the horizontal motion reliably. steps = int(_clamp(distance / random.uniform(4.0, 6.0), 150, 210)) end_dy = int(round(random.triangular(60, 180, 110) * scale)) cat = '长折' elif distance > 450 * scale: steps = int(_clamp(distance / random.uniform(3.0, 5.0), 110, 180)) end_dy = int(round(random.triangular(-25, 45, 10) * scale)) cat = '中折' else: steps = int(_clamp(distance / random.uniform(1.2, 2.2), 80, 240)) end_dy = int(round(random.triangular(0, 135, 40) * scale)) cat = '短折' x_exp = random.uniform(1.55, 2.10) y_exp = random.uniform(1.15, 1.85) print(f' 折回: {cat} distance={distance:.0f}px {steps}点 Y漂=+{end_dy} ' f'X弧度={x_exp:.2f} Y弧度={y_exp:.2f}') target_y = start_y + end_dy track = [] for i in range(1, steps + 1): t = i / steps # 末端保留 12% 线性分量，避免取整后长时间停在同一坐标。 ease = 1 - (1 - t) ** x_exp x = start_x - distance * (ease * 0.88 + t * 0.12) y = start_y + (target_y - start_y) * (1 - (1 - t) ** y_exp) track.append((int(round(x)), int(round(y)))) if track: track[-1] = (int(target_x), int(target_y)) track = _dedupe_track(track) return _dedupe_track(track) def save_track_image(pts, filepath): if len(pts) < 2: return xs=[p[0] for p in pts]; ys=[p[1] for p in pts]; m=50 w=max(xs)-min(xs)+m*2; h=max(ys)-min(ys)+m*2 w,h=max(w,200),max(h,100) c=np.ones((h,w,3),dtype=np.uint8)*255 for i in range(1,len(pts)): r=i/len(pts); g=200 if r<0.5 else int(200*(1-r)*2); b=int(200*r*2) if r<0.5 else 200 cv2.line(c,(pts[i-1][0]-min(xs)+m,pts[i-1][1]-min(ys)+m), (pts[i][0]-min(xs)+m,pts[i][1]-min(ys)+m),(0,g,b),1) cv2.circle(c,(pts[0][0]-min(xs)+m,pts[0][1]-min(ys)+m),4,(0,200,0),-1) cv2.circle(c,(pts[-1][0]-min(xs)+m,pts[-1][1]-min(ys)+m),4,(0,0,200),-1) cv2.imwrite(filepath,c) def solve_slider(driver, sx=None): global d, W, H d = driver; W, H = d.window_size(); d.screen_on() if sx is None: sx = 163 if W > 1000 else 87 print(f" 屏幕: {W}x{H} -> sx={sx}") screen = d.screenshot(format="opencv") slider_y = y_top = slider_bottom = None try: cv2.imwrite(os.path.join(OUT, "_tmp_slider.png"), screen) info = detect_slider_button(os.path.join(OUT, "_tmp_slider.png")) coords = info[0] if isinstance(info, tuple) else info if isinstance(coords, dict): sx = int(round((coords["top_left"][0] + coords["bottom_right"][0]) / 2)) except: pass initial_button_center = _detect_slider_center(screen) handle_center_offset = ((initial_button_center - sx) if initial_button_center is not None else 0.0) ocr_r = ocr_eng(screen) if ocr_r and ocr_r[0]: for item in ocr_r[0]: t = item[1]; cy = int((item[0][0][1]+item[0][2][1])/2) if "请按照说明拖动滑块" in t: slider_y = slider_y or cy; slider_bottom = int(item[0][2][1]) if "松开" in t: y_top = int(item[0][0][1]) if slider_y is None: return False sy = slider_y # 右滑 — TouchPipe；injectInputEvent 第四参数固定为 0 right_end_x = _choose_right_end_x(sx, W) print(f" 右滑: 按住({sx},{sy}) -> 折返点({right_end_x})") touch_started = time.perf_counter() d.touch.down(sx, sy) down_hold_s = random.uniform(0.09, 0.15) time.sleep(down_hold_s) pipe = TouchPipe(d).open() t0 = time.perf_counter(); track_pts = [] right_target_s = random.uniform(*RIGHT_DURATION_RANGE) right_sample_interval_s = random.uniform(*RIGHT_SAMPLE_INTERVAL_RANGE) right_point_count = int(round(_clamp( right_target_s / right_sample_interval_s + 1, RIGHT_POINT_LIMITS[0], RIGHT_POINT_LIMITS[1], ))) right_track = _build_right_track( sx, sy, right_end_x, right_point_count ) right_elapsed_s, right_emit = _emit_track( pipe, right_track, 'right', track_pts, t0, right_target_s ) right_emit["requested_points"] = right_point_count right_emit["sample_interval_ms"] = right_sample_interval_s * 1000 right_finished = time.perf_counter() print(f" 右滑: 计划{len(right_track)}点/实发{right_emit['sent_points']}点 " f"跳过{right_emit['skipped_points']}点 {right_elapsed_s*1000:.0f}ms " f"lag95={right_emit['lag_p95_ms']:.1f}ms") settle_s = random.uniform(*RIGHT_SETTLE_RANGE) time.sleep(settle_s) # TouchPipe 的请求可能仍在设备端排队。重复发送右端点并以真实滑块 # 中心确认到右侧后，才截取给接口的图片，避免幕布尚未展开。 # 手指可以发送到 1190，但滑块按钮受滑轨右边界限制：1220 宽屏上 # 滑轨约止于 1130，154px 宽的按钮中心最大约为 1053。 # 因此右端确认必须按 UI 的物理极限判断，不能按手指折返点判断。 scale = W / 1220.0 right_min_center = max(0, int(round(W - 180 * scale))) unfolded, right_center, right_ready_ms, right_ready = _wait_for_slider_position( pipe, d, right_end_x, right_track[-1][1], track_pts=track_pts, t0=t0, phase="right_confirm", min_center=right_min_center, timeout_s=RIGHT_READY_TIMEOUT_S, label="右端到位" ) print(f" 右端确认: center=" f"{right_center if right_center is not None else '未检测到'} " f"目标>={right_min_center} wait={right_ready_ms:.0f}ms " f"{'OK' if right_ready else 'TIMEOUT'}") if not right_ready: print(f" 右端到位未确认，仍使用最后截图 center=" f"{right_center if right_center is not None else '未检测到'}") if unfolded is None: unfolded = d.screenshot(format="opencv") if not right_ready: cv2.imwrite(os.path.join(OUT, "_unfold_not_ready.png"), unfolded) print(" 幕布未确认完全展开，本次不调用接口，避免使用错误坐标") _print_touchpipe_diagnostics(pipe) pipe.close() d.touch.up(int(right_end_x), int(right_track[-1][1])) return False # JFBYM # 先做一次渲染/事件队列冲刷，再截取真正提交给识别接口的画面。 time.sleep(random.uniform(0.06, 0.12)) crop = d.screenshot(format="opencv") if y_top and slider_bottom: crop = crop[y_top:slider_bottom, :] _, buf = cv2.imencode(".png", crop) gap = None api_started = time.perf_counter() for a in range(3): try: r = requests.post(JFBYM_URL, json={"token":TOKEN,"type":"20226","image":base64.b64encode(buf).decode()}, timeout=35).json() if r.get("data") and r["data"].get("data"): gap = int(r["data"]["data"]) elif r.get("data") and isinstance(r["data"],(int,float)): gap = int(r["data"]) if gap is not None: break time.sleep(2) except: time.sleep(2) api_elapsed_s = time.perf_counter() - api_started if gap is None: _print_touchpipe_diagnostics(pipe) pipe.close() d.touch.up(*right_track[-1]) return False print(f" gap={gap}") # 折回 — 按距离确定点数和阶段时长 turn_x, turn_y = right_track[-1] raw_target_x = gap + OFFSET_COMPENSATE target_x = int(_clamp(raw_target_x, 10, turn_x - 1)) if target_x != raw_target_x: print(f" 目标X超出折返范围: {raw_target_x} -> {target_x}") return_target_s = _return_duration(turn_x - target_x, W / 1220.0) return_track, return_schedule = _build_human_return_track( turn_x, turn_y, target_x, return_target_s, return_schedule=True ) return_target_s = float(return_schedule[-1]) if return_schedule else return_target_s hold_ms = (time.perf_counter() - right_finished) * 1000 return_elapsed_s, return_emit = _emit_track( pipe, return_track, 'return', track_pts, t0, return_target_s, schedule_override=return_schedule ) return_emit["schedule_mode"] = "fast_then_brake" print(f" 折回: 计划{len(return_track)}点/实发{return_emit['sent_points']}点 " f"跳过{return_emit['skipped_points']}点 {return_elapsed_s*1000:.0f}ms " f"lag95={return_emit['lag_p95_ms']:.1f}ms 远端停顿={hold_ms:.0f}ms") # 回滑同样不能把“最后一个已发送点”当成“设备已经到位”。 # 轮询真实滑块中心，并重复发送目标点，直到旧事件队列被消化。 target_center = target_x + handle_center_offset return_ready_frame, return_center, return_ready_ms, return_ready = ( _wait_for_slider_position( pipe, d, target_x, return_track[-1][1], track_pts=track_pts, t0=t0, phase="return_confirm", expected_center=target_center, timeout_s=RETURN_READY_TIMEOUT_S, stable_required=1, label="回滑到位" ) ) if not return_ready: print(f" 回滑到位未确认，释放前实际 center=" f"{return_center if return_center is not None else '未检测到'}") phase_timing = { "down_hold_ms": down_hold_s * 1000, "right_ms": right_elapsed_s * 1000, "right_emit": right_emit, "settle_ms": settle_s * 1000, "right_ready_ms": right_ready_ms, "right_ready": right_ready, "right_center": right_center, "api_ms": api_elapsed_s * 1000, "hold_ms": hold_ms, "return_ms": return_elapsed_s * 1000, "return_emit": return_emit, "return_ready_ms": return_ready_ms, "return_ready": return_ready, "return_center": return_center, } # 到位处理：以滑块按钮的真实中心确认，不再把幕布橙色左边缘 # 直接和 API gap 相减。两者不是同一个物理坐标点。 alignment_started = time.perf_counter() aligned = return_ready last_diff = None correction_count = 0 cur_x, cur_y = return_track[-1] print(f" 回滑确认: 滑块中心=" f"{return_center if return_center is not None else '未检测到'} " f"目标中心={target_center:.0f} wait={return_ready_ms:.0f}ms " f"{'OK' if return_ready else 'TIMEOUT'}") aligned_img = return_ready_frame # 微调（默认关闭；如果启用 fixed，仍以 API 目标点为释放点） if MICRO_MODE == "fixed": target_cur_x = int(_clamp(cur_x + OFFSET_COMPENSATE, 10, W-10)) steps = abs(target_cur_x - cur_x) sign = 1 if target_cur_x > cur_x else -1 for s in range(steps): cur_x += sign pipe.move(cur_x, cur_y, 0) time.sleep(random.uniform(0.005, 0.010)) print(f" 固定偏移: {OFFSET_COMPENSATE:+.0f}px 分{steps}步 -> x={cur_x}") elif MICRO_MODE == "visual": fallback_x = cur_x; prev_pl = None for attempt in range(3): time.sleep(0.3) check = d.screenshot(format="opencv") cv2.imwrite(os.path.join(ALIGN_DIR, f"align_{attempt}.png"), check) try: pl, _ = detect_captcha_left_edge(check) except Exception: break diff = pl - gap marked = check.copy() cv2.line(marked, (pl, 0), (pl, marked.shape[0]), (0, 255, 0), 3) cv2.line(marked, (gap, 0), (gap, marked.shape[0]), (0, 0, 255), 3) cv2.imwrite(os.path.join(ALIGN_DIR, f"align_{attempt}_marked.png"), marked) if abs(diff) <= 5: break if prev_pl is not None and pl == prev_pl: cur_x = int(_clamp(fallback_x,10,W-10)); pipe.move(cur_x,cur_y,0); break prev_pl = pl cur_x = int(_clamp(cur_x-diff,10,W-10)); pipe.move(cur_x,cur_y,0); time.sleep(0.05) # MICRO_MODE=none has already obtained a fresh confirmation frame above; # avoid another synchronous screenshot before releasing the finger. if MICRO_MODE != "none" or aligned_img is None: aligned_img = d.screenshot(format="opencv") phase_timing["alignment_ms"] = (time.perf_counter() - alignment_started) * 1000 phase_timing["alignment_corrections"] = correction_count pipe_diag = _print_touchpipe_diagnostics(pipe) phase_timing["touchpipe"] = pipe_diag pipe.close(); d.touch.up(int(cur_x), int(cur_y)) phase_timing["touch_total_ms"] = (time.perf_counter() - touch_started) * 1000 time.sleep(2) # 验证 final = d.screenshot(format="opencv"); check_r = ocr_eng(final) passed = True if check_r and check_r[0]: if any("拖动滑块" in it[1] or "请按住滑块" in it[1] or "安全验证" in it[1] for it in check_r[0]): passed = False if not passed: time.sleep(2); final = d.screenshot(format="opencv"); check_r = ocr_eng(final) passed = True if check_r and check_r[0]: if any("拖动滑块" in it[1] or "请按住滑块" in it[1] or "安全验证" in it[1] for it in check_r[0]): passed = False print(f" 结果: {'OK' if passed else 'FAIL'}") # 图片和 JSON 使用实际发送点；被调度器跳过的计划点不再画进轨迹。 all_track = [(item["x"], item["y"]) for item in track_pts] rd = SUCCESS_DIR if passed else FAILURE_DIR now = time.localtime() dev_id = getattr(d, 'serial', getattr(d, '_serial', 'unknown')) prefix = f"{dev_id}_{now.tm_year:04d}{now.tm_mon:02d}{now.tm_mday:02d}_{now.tm_hour:02d}{now.tm_min:02d}{now.tm_sec:02d}" save_track_image(all_track, os.path.join(rd, f"{prefix}_track.png")) if aligned_img is not None: b_img = aligned_img.copy() for i in range(1, len(all_track)): cv2.line(b_img, all_track[i-1], all_track[i], (0, 200, 200), 2) cv2.circle(b_img, all_track[0], 6, (0, 255, 0), -1) cv2.circle(b_img, all_track[-1], 6, (0, 0, 255), -1) cv2.imwrite(os.path.join(rd, f"{prefix}_b.png"), b_img) tj = {"device": dev_id, "screen": {"w": W, "h": H}, "time": time.strftime("%Y-%m-%d %H:%M:%S", now), "total_points": len(track_pts), "duration_ms": track_pts[-1]["rel_ms"] if track_pts else 0, "phase_timing": phase_timing, "start": {"x": track_pts[0]["x"], "y": track_pts[0]["y"]}, "turn": {"x": turn_x, "y": turn_y}, "end": {"x": track_pts[-1]["x"], "y": track_pts[-1]["y"]}, "dx": track_pts[-1]["x"]-track_pts[0]["x"], "dy": track_pts[-1]["y"]-track_pts[0]["y"], "passed": passed, "gap": gap, "points": track_pts} with open(os.path.join(rd, f"{prefix}_track.json"), 'w', encoding='utf-8') as fp: json.dump(tj, fp, ensure_ascii=False) return passed if __name__ == "__main__": DEVICE = "SK4T6XZH4PEUOZ99" d = u2.connect(DEVICE) print(f"设备: {DEVICE}") solve_slider(d)