# -*- coding: utf-8 -*- """test7 — test6 的A/B试验版: 落点补偿 -3px(实测落点系统性偏右2~6px, 甜蜜点0~+4)。 改动: ①OFFSET_COMPENSATE=-2(实测容忍窗不对称: 左仅~5px/右~8px, 0±2最稳) ②归档独立 ③离群标记 ④右滑提速。其余与test6一致。 2026-09-22 改(guiji_new 19条真人轨迹实测校准): ⑤右段换真人速度模板(爆发起步→长弧刹车, 原为匀速) ⑥Y漂移校准: 右段改向下(-2%~+10%), 折回改单一向下0.20~0.38(原±28px比真人小6倍) ⑦确认阶段到位后静默不重发 ⑧释放前静默停顿40~120ms ⑨折回改为真人事件率时间步进采样(dt 4.0~7.5ms重尾, 巡航段偶发30~80ms微犹豫, 贴入缺口段连续不停顿)——实测尾段是一像素一像素磨进缺口的, 旧的100点均匀采样会在贴入段变成4~5px一跳 ⑩初始按压收窄到60~120ms(实测中位75ms) ⑪发送层: Windows定时器提升到1ms+忙等, 落后时整体后移时间表而非丢点——丢点会让轨迹出现"卡顿后飞过去"的瞬移点。⑫折回速度剖面重写(逐条实测速度曲线): 巡航到距缺口50~90px → 缓降到~25%峰值 → 最后12~25px急剧降到爬行速度, 再用~10px极慢磨进缺口——真人样本"100px:1.05 → 20px:0.23 → 10px:0.03 → 5px:0.03"; 前两版(长弧衰减 / 巡航到10px才刹车)都与实测不符。⑬释放前慢速微调: 用幕布左边缘量出API gap与真实缺口的剩余距离(≤30px), 1px/30~90ms慢慢挪过去再释放。""" 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" if sys.platform == "win32": # Windows 默认定时器精度约15.6ms: sleep(4~7ms)会随机超睡, 导致发送 # 卡顿和跳点(轨迹飞点)。提升到1ms后真人事件间隔才能稳定发出。 try: import ctypes ctypes.windll.winmm.timeBeginPeriod(1) except Exception: pass sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # 项目根(tbsg): detect_slider_button 等 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", "test7_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.86, 0.93) # 折回主段使用固定的快速节奏，不再直接照搬某一条可能很慢的真人模板。 # 采样间隔保持在触摸事件可稳定消费的范围，距离越长只增加点数，不拉长点间隔。 RETURN_SAMPLE_INTERVAL_RANGE = (0.0042, 0.0052) RETURN_BRAKE_TIME_RANGE = (0.75, 0.88) # (保留兼容, 剖面模式下未用) # ── 真人折回速度剖面(2026-09-22 13条轨迹实测): 按路程位置(10段)的速度区间 px/ms ── # 0%=折返点旁 100%=缺口旁; 形状 = 起步→峰值→长弧渐进减速→爬行入缺口 RETURN_SPEED_PROFILE = ( (0.24, 1.68), (0.08, 3.56), (0.13, 3.98), (0.15, 3.69), (0.15, 3.07), (0.08, 2.17), (0.07, 1.62), (0.09, 1.51), (0.04, 1.34), (0.02, 1.17), ) RETURN_CURVE_LIMIT_1220 = 42 # 2026-09-22 19条实测: 折回Y漂移全部向下(比值0.19~0.39, 绝对值90~239px), # 旧±28px上限比真人小6倍且无方向偏好; 上限按实测最大值239px放宽。 RETURN_Y_CORRIDOR_1220 = 280 RETURN_END_DRIFT_LIMIT_1220 = 240 # 方案一参数：按真人样本的量级修正点密度与阶段时长。 RIGHT_SAMPLE_INTERVAL_RANGE = (0.0038, 0.0052) RIGHT_POINT_LIMITS = (60, 110) # 右滑移动本身按 test5 的真人节奏控制；到最右端后的幕布展开等待 # 由 RIGHT_SETTLE_RANGE 单独负责，不计入右滑阶段。 RIGHT_SPEED_RANGE = (0.45, 1.55) # 右滑速度 px/ms(13条真人轨迹 0.40~1.66, 覆盖全部样本) TRACK_SKIP_LATE_S = 0.010 # 真人样本的右端折返点集中在约 1023~1109 px；不要每次都固定在 1190。 # 下限略高于滑块可确认的右端，避免随机到太靠左导致幕布未完全展开。 RIGHT_END_RANGE_1220 = (1060, 1105) # 真人右滑Y漂移: 2026-09-22 19条实测均值+3.6%向下(范围-2.3%~+11.7%), # 旧数据"轻微向上收尾"的结论作废, 方向改为向下为主。 RIGHT_Y_DRIFT_RATIO_RANGE = (-0.02, 0.10) # 真人右段时长实测444~991ms(19条, 800~930px跨度), 模板时长按距离缩放后钳位在此区间 RIGHT_TEMPLATE_DURATION_RANGE = (0.30, 1.20) # 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_new") _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 confirmed = False # 首次确认到位后静默 last_push = 0.0 def _needs_push(center): if not confirmed: return True if center is None: return False if expected_center is not None: return abs(center - expected_center) > SLIDER_POSITION_TOLERANCE * 2 if min_center is not None: return center < min_center - SLIDER_POSITION_TOLERANCE return False while time.perf_counter() - started < timeout_s: send_ms = 0.0 if _needs_push(last_center) and time.perf_counter() - last_push >= 0.08: last_push = time.perf_counter() 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 if ok: confirmed = 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 _subdivide_track(points, schedule, max_step_px=16.0, min_dt_s=0.007): """大跨步拆成多步: 真人事件间隔~4.4ms, 高速段单步只有15~25px; 主机采样间隔10~14ms会让峰值段一步40px+, 视觉上"加速过猛"。 只在步长>max_step_px且间隔>=min_dt_s时拆分, 时间表同步对半插值。""" if schedule is None or len(points) < 2 or len(schedule) != len(points): return points, schedule new_pts = [points[0]] new_sched = [schedule[0]] for i in range(1, len(points)): x0, y0 = points[i - 1] x1, y1 = points[i] t0, t1 = schedule[i - 1], schedule[i] step = math.hypot(x1 - x0, y1 - y0) if step > max_step_px and (t1 - t0) >= min_dt_s: n = int(math.ceil(step / max_step_px)) # 子步间隔不低于3.5ms, 超出主机稳定发送能力 n = min(n, max(1, int((t1 - t0) / 0.0035))) for k in range(1, n): f = k / n new_pts.append((int(round(x0 + (x1 - x0) * f)), int(round(y0 + (y1 - y0) * f)))) new_sched.append(t0 + (t1 - t0) * f) new_pts.append(points[i]) new_sched.append(schedule[i]) return new_pts, new_sched 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): """折回总时长(人速校准): 5条真人轨迹折回速度 0.07~0.89 px/ms (70~890 px/s)。 主段巡航 + 0.15~0.35s 收尾, 上限2.6s(含末端爬行微调)。""" speed = random.uniform(400.0, 900.0) * max(scale, 0.8) # px/s(实测: 慢于400px/s失败率61%, 快于800仅26%) return _clamp(distance / speed + random.uniform(0.15, 0.30), 0.35, 2.6) 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, "resyncs": 0, "shifted_ms": 0.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 resyncs = 0 shifted_ms = 0.0 for i, (x, y) in enumerate(track): deadline = started + schedule[i] wait_s = deadline - time.perf_counter() if wait_s > 0.002: time.sleep(wait_s - 0.001) # Windows定时器已提升到1ms, 剩余用忙等 while time.perf_counter() < deadline: pass lag_s = max(0.0, time.perf_counter() - deadline) # 落后超过阈值: 不丢中间点, 把剩余时间表整体后移"全部"落后量。 # 只移超出阈值的部分会残留~10ms落后, 之后每个点都会再次触发 # 重同步(级联放大, 一次卡顿变几十次)。最后一点不做处理。 if i < len(track) - 1 and lag_s > TRACK_SKIP_LATE_S: shift = lag_s for j in range(i + 1, len(schedule)): schedule[j] += shift resyncs += 1 shifted_ms += shift * 1000 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, "resyncs": resyncs, "shifted_ms": shifted_ms, "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): # select 节流: 手机回包到达时才读, 避免热循环 recv 每秒唤醒上百次 # 与发送循环抢 GIL(会放大发送循环的调度卡顿)。 import select while not self._stop.is_set(): try: ready, _, _ = select.select([self._sock], [], [], 0.05) if ready and 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 # 门槛单位=屏幕像素(与运行时选桶一致)。老录制器写原始触摸单位(1px≈16单位), # 阈值1000实际只有62px; 新录制器x直接是像素, 真人右滑实际跨度~850~930px if span < 300: 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 # 像素门槛: 真人折回跨度实测146~796px, 100px以下才算退化数据 if span < 100: 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_velocity_template(points, phase): """Extract horizontal progress as a function of elapsed phase time. phase='right' uses touch-down -> turn point; 'return' uses turn point -> release. Stationary holds (initial press hold, turn-point hold, endpoint plateau) are trimmed so the template only contains real motion; micro-hesitations in between stay in the raw arrays. """ if not points or len(points) < 20: return None turn_idx = max(range(len(points)), key=lambda i: float(points[i].get("x", 0))) if phase == "right": seg = points[:turn_idx + 1] min_span = 300.0 sign = 1.0 else: seg = points[turn_idx:] min_span = 100.0 sign = -1.0 x0 = float(seg[0].get("x", 0)) x1 = float(seg[-1].get("x", 0)) full_span = (x1 - x0) * sign if full_span < min_span: return None # Skip the stationary start: the initial press hold on the right phase # and the hold at the turn point on the return phase are pauses, not # motion, and are emitted separately by the runtime. threshold = max(10.0, full_span * 0.01) start_idx = None for i in range(1, len(seg)): if (float(seg[i].get("x", 0)) - x0) * sign >= threshold: start_idx = i break if start_idx is None or len(seg) - start_idx < 12: return None motion = seg[start_idx:] mx0 = float(motion[0].get("x", 0)) mx1 = float(motion[-1].get("x", 0)) span = (mx1 - mx0) * sign t0 = float(motion[0].get("rel_ms", 0.0)) t1 = float(motion[-1].get("rel_ms", 0.0)) duration_ms = t1 - t0 if span < 100 or duration_ms < 100: return None raw_t = np.asarray([ (float(p.get("rel_ms", 0.0)) - t0) / duration_ms for p in motion ], dtype=np.float64) raw_u = np.asarray([ (float(p.get("x", 0)) - mx0) * sign / span for p in motion ], 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 finger # has already stopped. That endpoint plateau is a hold phase, not real # 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 { "phase": phase, "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), "span": span, "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": []}, "velocity_right": [], } 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_velocity_template(points, "return") right_velocity = _normalize_velocity_template(points, "right") if right: templates["right"].append(right) if ret: span = max(float(points[i]["x"]) for i in range(len(points))) - float(points[-1]["x"]) # 分桶单位=像素, 与运行时 _build_human_return_track 的450/600px分界一致 if span < 450: key = "short" elif span < 600: key = "medium" else: key = "long" templates[key].append(ret) if velocity: templates["velocity"][key].append(velocity) if right_velocity: templates["velocity_right"].append(right_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={}/{}/{} velocity_right={}".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"]), len(templates["velocity_right"]))) 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, velocity_template=None, duration_s=None): """右滑轨迹：连续低频起伏，避免逐点随机造成锯齿。 velocity_template 给定时, 几何进度和发送时刻直接取自真人右段 速度模板的原始事件(event_progress/event_t): 匀速采样会变成 "爆发起步→长弧刹车"的真人节奏并保留原始微犹豫。返回 (points, schedule); 模板不可用时返回 (points, None)。 """ dist = end_x - start_x if dist <= 0: return [(int(start_x), int(start_y))], None # 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. 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)) # 2026-09-22 实测: 右滑段Y漂移轻微向下(-2%~+12%), 不再生成向上的漂移。 drift = random.uniform(*RIGHT_Y_DRIFT_RATIO_RANGE) * dist schedule = None if velocity_template is not None: points = [] schedule = [] last_keep_t = -1.0 # 保险下限6ms: 右段模板已在7~9ms网格上重采样, 这里只挡异常密集 # 的原始事件(老模板路径), 不再起滤稀作用 min_gap = 0.006 / max(float(duration_s or 1.0), 1e-6) for t, u in zip(velocity_template["event_t"], velocity_template["event_progress"]): if t - last_keep_t < min_gap: continue 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 point = (int(round(x)), int(round(y))) if points and point == points[-1]: continue points.append(point) schedule.append(t * float(duration_s or 1.0)) last_keep_t = t if len(points) < 12: velocity_template = None # 退化, 走匀速分支 points = [] schedule = None if velocity_template is None: steps = max(1, int(point_count) - 1) 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]) points = _dedupe_track(points) schedule = None else: points[-1] = (int(end_x), points[-1][1]) if len(schedule) > 1: schedule[0] = 0.0 return points, schedule # 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" elif distance > 450 * scale: cat = "medium" else: cat = "short" # 2026-09-22 19条实测: 折回Y漂移全部向下(比值0.19~0.39, 90~240px); # 旧三档±小漂移与真人方向相反, 改为单一向下分布+硬上限。 end_ratio = random.triangular(0.20, 0.38, 0.30) 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)) # ── 折回速度剖面(2026-09-23 真人19条按距离分桶实测) ── # 距离越长巡航越快(短0.75/中1.07/长1.95 px/ms), 减速越早越狠, # 总时长稳定在2.3~2.8s。锚点链(距缺口px, 点速度/巡航)对数线性 # 插值; 点速度由实测的"0~N px区间均值"反推(区间均值≠边界点速度)。 # 长折按真人逐点曲线校准(150px:0.81 → 100px:0.46 → 70px:0.25 → # 50px:0.18 → 30px:0.09 → 20px:0.03): 减速段占总时长47~69%; # 旧链(100px就掉到0.08)减速段膨胀到81%, 太慢。 # 减速起点与锚点距离按屏幕宽度缩放: 真人样本来自1220宽屏, 小屏上 # 同样的绝对px会让减速区占比过大(720屏上256px折回65%在减速)。 if cat == "long": v_peak = random.uniform(1.6, 2.4) * max(scale, 0.8) chain = ((100, 0.40), (70, 0.25), (50, 0.16), (30, 0.08), (20, 0.04), (10, 0.02), (5, 0.01)) elif cat == "medium": v_peak = random.uniform(0.7, 1.3) * max(scale, 0.8) chain = ((100, 0.26), (50, 0.15), (20, 0.10), (10, 0.07), (5, 0.01)) else: v_peak = random.uniform(0.45, 0.9) * max(scale, 0.8) # 短折按200~300px样本(100px:0.48, 50px:0.31, 20px:0.09, 10px:0.04)校准 chain = ((100, 0.45), (50, 0.28), (20, 0.12), (10, 0.04), (5, 0.02)) chain = tuple((max(2.0, round(d * scale)), r) for d, r in chain) if distance >= 250: dec_start = 150.0 * scale else: dec_start = max(80.0 * scale, distance * 0.55) anchors = [(dec_start, 1.0)] + [(d, r) for d, r in chain if d < dec_start] ramp_end = distance * random.uniform(0.20, 0.30) # 起步加速完成点(已走px, 采样一次) def _smoothstep(f): f = min(1.0, max(0.0, f)) return f * f * (3 - 2 * f) # C1连续, 无折角 def _v_of_u(u): d = (1.0 - u) * distance # 距缺口 px if d >= dec_start: # 巡航段: 前25~38%路程内加速到峰值(真人加速快), 之后巡航保持 progressed = distance - d f = progressed / max(ramp_end, 1e-6) return v_peak * (0.25 + 0.75 * _smoothstep(f)) if distance < 150: # 超短折回: 全程巡航, 最后8px收一下 return v_peak * 0.06 if d <= 8 else v_peak for (d1, r1), (d2, r2) in zip(anchors, anchors[1:]): if d <= d1 and d > d2: f = min(1.0, (d1 - d) / max(d1 - d2, 1e-6)) return v_peak * math.exp(math.log(r1) + math.log(r2 / r1) * f) return v_peak * anchors[-1][1] # d<=5px: 爬行 prof_u, prof_t = [], [] t_cur = 0.0 u_cur = 0.0 # 按真人事件率时间步进采样: 实测折回段dt中位4.4ms(重尾3.8~7.5ms), # 全程连续出事件, 尾段一像素一像素磨进缺口。均匀路程采样(旧100点) # 会把尾段变成4~5px一跳/90ms一顿, 与真人相反。 while u_cur < 1.0 - 1e-9: v = _v_of_u(u_cur) * random.uniform(0.97, 1.03) # ±3%微扰 # 巡航段偶发30~80ms微犹豫(约1.5%事件, 实测偶见); 最后15%路程 # 贴入缺口时保持连续, 不插任何停顿。 if random.random() < 0.015 and u_cur < 0.85: dt_ms = random.uniform(30.0, 80.0) else: # 真人事件间隔中位4.4ms, 但主机注入层稳定维持的下限实测约10ms # (Windows偶发10~360ms抢占; test6用15~30ms间隔从不卡顿)。 # 降到10~14ms: 保留真人速度剖面与尾段1px连续贴入, 给调度留余量。 dt_ms = random.uniform(10.0, 14.0) # 刹车锚点链首段只有50px宽: 接近时限制单步距离, 避免一步跳过 # (长折巡航步长可达25~30px)。dt收缩但不低于3.5ms。 d_cur = (1.0 - u_cur) * distance if d_cur < dec_start + 20.0: step_px = v * dt_ms if step_px > 3.0: dt_ms = max(dt_ms * 3.0 / step_px, 3.5) u_cur = min(1.0, u_cur + v * dt_ms / distance) t_cur += dt_ms prof_u.append(u_cur) prof_t.append(t_cur) event_u = np.asarray(prof_u, dtype=np.float64) event_t = np.asarray(prof_t, dtype=np.float64) / max(1.0, prof_t[-1]) # 归一化0~1 duration_s = prof_t[-1] / 1000.0 # 剖面自然时长(s) 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") 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 _timing_selfcheck(): """开跑前自检主机计时稳定性: 100次12ms sleep, 统计超睡>8ms的比例。 机器负载窗口(杀毒/磁盘/USB抢占)时 sleep 超睡可达30~70ms, 该窗口 内发送必然卡顿, 提前告警比跑完看轨迹划算。""" overs = 0 for _ in range(100): s = time.perf_counter() time.sleep(0.012) if (time.perf_counter() - s) - 0.012 > 0.008: overs += 1 if overs <= 2: print(f" 计时自检: OK ({overs}/100 次超睡)") return True print(f" ⚠ 计时自检: 不稳 ({overs}/100 次超睡>8ms), 主机处于负载窗口, " f"轨迹会卡顿, 建议稍后再跑") return False 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}") _timing_selfcheck() 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 Exception as exc: print(f" detect_slider_button 失败: {exc}") 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 or "请按住滑块" 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: texts = [it[1] for it in (ocr_r[0] if ocr_r and ocr_r[0] else [])] print(f" 未检测到滑块提示文字, 本次不执行") print(f" 当前屏幕OCR文本: {texts[:15]}") cv2.imwrite(os.path.join(OUT, "_no_captcha.png"), screen) print(f" 已保存截图: {os.path.join(OUT, '_no_captcha.png')} (确认验证码是否已弹出)") 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.06, 0.12) # 实测初始按压停顿中位75ms(62~122ms) time.sleep(down_hold_s) pipe = TouchPipe(d).open() t0 = time.perf_counter(); track_pts = [] # 优先使用真人右段速度模板(爆发起步→长弧刹车, 含原始微犹豫); # 模板不可用时退回匀速+抖动方案。 rvel_list = _load_guiji_templates().get("velocity_right") or [] right_velocity = random.choice(rvel_list) if rvel_list else None if right_velocity is not None: right_target_s = _clamp( right_velocity["duration_ms"] / 1000.0 * (right_end_x - sx) / max(right_velocity["span"], 1.0), *RIGHT_TEMPLATE_DURATION_RANGE) # 模板曲线按7~9ms均匀时间网格重采样: 保留"爆发→刹车"的速度 # 剖面, 密度与主机注入能力匹配(直接过滤原始事件会因模板而异, # 曾出现整段只剩32点的情况; 4~5ms间隔在负载窗口下不稳) right_point_count = max(12, int(right_target_s / random.uniform(0.007, 0.009)) + 1) t_grid = [i / (right_point_count - 1) for i in range(right_point_count)] u_grid = [float(np.interp(ti, right_velocity["t"], right_velocity["progress"])) for ti in t_grid] right_velocity = {"event_t": t_grid, "event_progress": u_grid, "duration_ms": right_velocity["duration_ms"], "span": right_velocity["span"]} right_sample_interval_s = 0.0 else: right_target_s = _clamp((right_end_x - sx) / random.uniform(*RIGHT_SPEED_RANGE) / 1000.0, 0.18, 1.3) 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, right_schedule = _build_right_track( sx, sy, right_end_x, right_point_count, velocity_template=right_velocity, duration_s=right_target_s, ) # 峰值段单步40px+太猛, 拆成<=16px的小步(时间表同步拆分) right_track, right_schedule = _subdivide_track(right_track, right_schedule) right_elapsed_s, right_emit = _emit_track( pipe, right_track, 'right', track_pts, t0, right_target_s, schedule_override=right_schedule ) right_emit["requested_points"] = right_point_count right_emit["sample_interval_ms"] = right_sample_interval_s * 1000 right_emit["schedule_mode"] = "real_template" if right_velocity is not None else "uniform" right_finished = time.perf_counter() print(f" 右滑[{right_emit['schedule_mode']}]: 计划{len(right_track)}点/实发{right_emit['sent_points']}点 " f"重同步{right_emit['resyncs']}次(+{right_emit['shifted_ms']:.0f}ms) " f"{right_elapsed_s*1000:.0f}ms 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}") # 折返点停顿: 真人中位0.2~0.6s(37~1014ms), 旧值2.5~3.2s比真人长 # 5~10倍(旧注释"该档失败率仅12%", 但现在轨迹已全面真人化, 重新A/B)。 # 打码API必须在停顿期间完成(实测0.6~2.0s), 因此改为API返回后再补 # 0.15~0.35s余量, 总停顿≈API耗时+小余量(约1.0~2.4s)。 time.sleep(random.uniform(0.15, 0.35)) # 折回 — 按距离确定点数和阶段时长 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 ) # 巡航段单步可达20px+, 拆成<=16px的小步让加速段更平滑 return_track, return_schedule = _subdivide_track(return_track, return_schedule) 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['resyncs']}次(+{return_emit['shifted_ms']:.0f}ms) " f"{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 '未检测到'}") if return_center is not None: _err = return_center - (target_x + handle_center_offset) if abs(_err) > 20: print(f" ⚠ 落点离群 {_err:+.0f}px (目标{target_x + handle_center_offset:.0f}, 实际{return_center:.0f}) — 打码gap可能错误") 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] # ── 释放前慢速微调: API gap 与真实缺口(幕布左边缘)常有几px差距 ── # 用幕布左边缘量出剩余距离, 像真人一样1px/30~90ms慢慢挪过去。 micro_px = 0 micro_ms = 0.0 micro_edge = None try: check = d.screenshot(format="opencv") micro_edge, _ = detect_captcha_left_edge(check) micro_px = int(round(micro_edge - cur_x)) except Exception as exc: print(f" 幕布边缘检测失败({exc}), 跳过微调") if 2 < abs(micro_px) <= 12: direction = 1 if micro_px > 0 else -1 start_x = cur_x micro_started = time.perf_counter() for _ in range(abs(micro_px)): cur_x += direction send_ms = 0.0 try: send_ms = pipe.move(int(cur_x), int(cur_y), 0) except Exception: pass track_pts.append({"x": int(cur_x), "y": int(cur_y), "pressure": 0, "phase": "micro", "confirm": False, "scheduled_ms": None, "lag_ms": None, "send_ms": send_ms, "rel_ms": (time.perf_counter()-t0)*1000}) time.sleep(random.uniform(0.03, 0.09)) micro_ms = (time.perf_counter() - micro_started) * 1000 print(f" 微调: 幕布边缘{micro_edge:.0f} vs 当前位置{start_x} " f"差{micro_px:+d}px, {abs(micro_px)}步慢挪 {micro_ms:.0f}ms") elif abs(micro_px) > 12: print(f" 幕布边缘偏差{micro_px:+d}px过大(可能检测错误), 不做微调") phase_timing["micro_adjust_px"] = micro_px phase_timing["micro_adjust_ms"] = micro_ms 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 # 真人释放前静默停顿40~120ms(实测中位43ms), 期间不发送任何触摸事件 release_pause_s = random.uniform(0.04, 0.12) time.sleep(release_pause_s) phase_timing["release_pause_ms"] = release_pause_s * 1000 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 = "8XHEJBHMZHTKYTHM" d = u2.connect(DEVICE) print(f"设备: {DEVICE}") ok = solve_slider(d) print(f"退出: {'通过' if ok else '未通过/未执行'}") sys.exit(0 if ok else 1)