project/tbsg/yzm/tmp_captcha_test2.py

"""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=自动判断分辨率