"""test.py 轨迹 + 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, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) 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" IMG_ROOT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "image", "test2_image") OUT = os.path.join(IMG_ROOT, "d") ALIGN_DIR = os.path.join(IMG_ROOT, "align") SUCCESS_DIR = os.path.join(IMG_ROOT, "success") FAILURE_DIR = os.path.join(IMG_ROOT, "failure") 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 = random.uniform(-7,-5) # MICRO_MODE="fixed" 偏移量(正=右,负=左) MICRO_MODE = "fixed" # "fixed"=固定偏移 / "visual"=红绿线对齐 / "none"=不用 def _clamp(v, lo, hi): return max(lo, min(v, hi)) def _pressure_curve(i, n, phase): """压力曲线：开头高→巡航低→末尾高。phase: 'right' 或 'return'。""" t = i / max(1, n - 1) if t < 0.1: return random.randint(60, 90) # 开头按重 elif t > 0.90: return random.randint(55, 85) # 末尾对准按重 else: return random.randint(15, 35) # 巡航轻按 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 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() except Exception: self._fallback = True 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: try: self._dev.touch.move(int(x), int(y)) except Exception: pass return try: with self._lock: self._sock.sendall(self._req(x, y, pressure)) except Exception: self._fallback = True; 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),int(pressure)]}).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 _build_right_track(start_x, start_y, end_x): """test.py 原样 — 钟形鼓包 wobble, EMA 0.55""" dist = abs(end_x - start_x) steps = int(_clamp(dist / random.uniform(6.0, 8.0), 35, 50)) + 20 end_yd = random.randint(-30, 30) wobbles = [(random.uniform(0.15, 0.85), random.uniform(-25, 25)) for _ in range(random.randint(2, 4))] points = [(int(start_x), int(start_y))] smooth_y = float(start_y) for i in range(1, steps + 1): t = i / steps x = start_x + dist * (1.0 - (1.0 - t) ** 2.0) y = start_y + end_yd * (t ** 0.6) if 0.15 < t < 0.9: for wp, wv in wobbles: dw = abs(t - wp) if dw < 0.12: y += wv * (1 - dw / 0.12) smooth_y = smooth_y * 0.55 + y * 0.45 points.append((int(round(x)), int(round(smooth_y)))) return points def _build_human_return_track(start_x, start_y, target_x): """test.py 原样 — wobble, EMA 0.55, blend t>0.5 不打勾""" distance = start_x - target_x if distance <= 0: return [(int(target_x), int(start_y))] # 点数按折回距离自适应，对标真机但控制上限避免等太久 if distance > 600: steps = random.randint(150, 200) # 长折, Y必往下漂 end_dy = random.randint(80, 200); cat = '长折' elif distance > 450: steps = random.randint(120, 170) # 中折 end_dy = random.randint(-20, 40); cat = '中折' else: steps = int(_clamp(distance / random.uniform(3.0, 5.0), 35, 55)) end_dy = random.randint(-20, 40); cat = '短折' print(f' 折回: {cat} distance={distance:.0f}px {steps}点 Y漂={end_dy:+d}') # 两段: 巡航60%点数走88%距离(大步), 减速40%点数走12%距离(密集蠕动) cruise_n = int(steps * 0.60) decel_n = steps - cruise_n target_y = start_y + end_dy r_wobbles = [(random.uniform(0.15, 0.5), random.uniform(-20, 20)) for _ in range(random.randint(2, 4))] track, smooth_y = [], float(start_y) cruise_dist = distance * 0.88 decel_dist = distance * 0.12 # ── 前90%大步巡航 ── for i in range(1, cruise_n + 1): t = i / cruise_n x = start_x - cruise_dist * t y = start_y + (target_y - start_y) * t if 0.1 < t < 0.5: for wp, wv in r_wobbles: dw = abs(t - wp) if dw < 0.12: y += wv * (1 - dw / 0.12) smooth_y = smooth_y * 0.55 + y * 0.45 track.append((int(round(x)), int(round(smooth_y)))) # ── 后10%小步减速对准 ── dsx, dsy = track[-1] for i in range(1, decel_n + 1): tt = i / decel_n ease = 1 - (1 - tt) ** 2 x = dsx - decel_dist * ease y = dsy + (target_y - dsy) * tt track.append((int(round(x)), int(round(y)))) if track: track[-1] = (int(target_x), int(target_y)) return track def save(name, img): cv2.imwrite(os.path.join(OUT, name), img) 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() # 自动判断分辨率: 宽>1000=高分屏(sx=163), 否则=低分屏(sx=87) if sx is None: sx = 163 if W > 1000 else 87 print(f" 屏幕: {W}x{H} → sx={sx} ({'高分' if W>1000 else '低分'})") 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")) if info and len(info) >= 3: sx = info[0] + int(info[2]/2) except: pass 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 + 压感 d.touch.down(sx, sy); time.sleep(0.08) pipe = TouchPipe(d).open() t0 = time.perf_counter(); track_pts = [] right_track = _build_right_track(sx, sy, W-30) for i, (x, y) in enumerate(right_track): p = _pressure_curve(i, len(right_track), 'right') pipe.move(x, y, p) track_pts.append({"x": x, "y": y, "pressure": p, "rel_ms": (time.perf_counter()-t0)*1000}) time.sleep(random.uniform(0.002, 0.004)) # 等设备消化完右滑点再截图 — 每点预留 28ms 处理时间 settle_ms = max(500, len(right_track) * 18) time.sleep(settle_ms / 1000.0) # 展开后的幕布截图（后面叠加轨迹用） unfolded = d.screenshot(format="opencv") # JFBYM crop = d.screenshot(format="opencv") if y_top and slider_bottom: crop = crop[y_top:slider_bottom, :] _, buf = cv2.imencode(".png", crop) gap = None 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) if gap is None: pipe.close(); d.touch.up(W-20, sy); return False # 折回 target_x = gap + OFFSET_COMPENSATE return_track = _build_human_return_track(W-30, right_track[-1][1], target_x) for i, (x, y) in enumerate(return_track): p = _pressure_curve(i, len(return_track), 'return') pipe.move(x, y, p) track_pts.append({"x": x, "y": y, "pressure": p, "rel_ms": (time.perf_counter()-t0)*1000}) # 前60%巡航快发, 后40%减速慢发 frac = i / max(1, len(return_track)-1) if frac > 0.60: time.sleep(random.uniform(0.004, 0.008)) else: time.sleep(random.uniform(0.000, 0.002)) # 等设备消化完折回点 # settle 按距离缩放: 每px约2~4ms, 上下限保护 ret_distance = (W-30) - target_x settle_ms2 = int(ret_distance * random.uniform(2.0, 4.0)) settle_ms2 = max(500, min(3500, settle_ms2)) if ret_distance > 600: scat = '长等' elif ret_distance > 450: scat = '中等' else: scat = '短等' print(f' settle={settle_ms2}ms({settle_ms2/1000:.1f}s)') time.sleep(settle_ms2 / 1000.0) # 微调: 三种模式 cur_x, cur_y = return_track[-1] aligned_img = None 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, 60) 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,60); break prev_pl = pl cur_x = int(_clamp(cur_x-diff,10,W-10)); pipe.move(cur_x,cur_y,50); time.sleep(0.05) # 滑块对准缺口的截图，做轨迹背景 aligned_img = d.screenshot(format="opencv") pipe.close(); d.touch.up(int(cur_x),int(cur_y)); 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 all_track = right_track + return_track 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, "start": {"x": track_pts[0]["x"], "y": track_pts[0]["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 = "NJZX8DZXT47HQGWO" d = u2.connect(DEVICE) print(f"设备: {DEVICE}") solve_slider(d, sx=None) # None=自动判断分辨率