project/tbsg/human_touch.py

# -*- coding: utf-8 -*-
"""
拟人触摸滑动 — 人体关节模型(手腕枢轴)弧线轨迹 + motionevent 连续手势注入
供 main1(列表滑动/说明书上滑) 与 snapshot(快照滚动) 共用。

轨迹模型: 一笔画 = 手腕绕枢轴转动的单向弧(二次贝塞尔),
  起终点横向错开(漂移量与笔画长度相关), 速度 smoothstep 缓起-快中-缓收, 带 ±2px 指尖微抖。
注入方式: 单个 adb shell 会话内连发 input motionevent 触点(手指全程不抬),
  ~45ms/点是未root user机的物理上限; /dev/input(sendevent) 被SELinux封死不可用。
"""
import json
import random
import socket
import subprocess
import sys
import threading
import time

# Windows 默认定时器精度 ~15.6ms, time.sleep(0.004) 实际睡 ~15.6ms,
# 导致滑动触点节奏全部变慢(设计4ms实跑15ms)。进程内调到 1ms 精度,一次性全局生效。
if sys.platform == "win32":
    try:
        import ctypes
        ctypes.windll.winmm.timeBeginPeriod(1)
    except Exception:
        pass

from uiautomator2.core import AdbHTTPConnection


class TouchPipe:
    """uiautomator2 9008端口的keep-alive触点长连接(来自 yzm/tmp_captcha_test6.py, 验证码同款)。
    每个触点 = 一条极小的 injectInputEvent 请求(~1~3ms), 比 input 命令(45ms/点)快20倍。"""

    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

    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:
            self._fallback = True
            self._open_ok = False
        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:
            self._dev.touch.move(int(x), int(y))
            return
        with self._lock:
            self._sock.sendall(self._req(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()
        try:
            if self._sock:
                self._sock.shutdown(socket.SHUT_RDWR)   # 立即打断drainer的recv,不等0.5s超时
        except Exception:
            pass
        try:
            if self._sock:
                self._sock.close()
        except Exception:
            pass
        if self._drainer:
            self._drainer.join(timeout=0.2)


def touchpipe_drag(d, pts: list, step: float = None, hold: float = None,
                   tail: float = None) -> bool:
    """TouchPipe 长连发送一笔弧线(60点/100Hz级, 手指全程不抬, 丝滑+弧线)。
    d = uiautomator2 设备对象。返回 True=成功, False=通道不可用(调用方降级)。
    hold: 按下后停顿(默认0.08~0.14s拟人); tail: 抬起前停顿(默认0.03~0.06s);
    轻扫场景(如图片左滑)传更小的 step/hold/tail。
    注: 按住 ~500ms 不动 App 会判长按(列表页误触"店铺问题"反馈弹窗),
    故 down→首move 间隔超 0.35s 即中止; 异常时 finally 必补 UP, 防手指钉在屏幕上。"""
    if step is None:
        step = random.uniform(0.008, 0.014)
    if hold is None:
        hold = random.uniform(0.08, 0.14)
    if tail is None:
        tail = random.uniform(0.03, 0.06)
    pipe = None
    up_sent = False
    try:
        pipe = TouchPipe(d).open()
        if not pipe._open_ok:
            return False
        t0 = time.time()
        d.touch.down(int(pts[0][0]), int(pts[0][1]))
        time.sleep(hold)                                # 按住停顿(验证码同款手感)
        if time.time() - t0 > 0.35:                     # down延迟过久→一动就成长按
            return False
        for (x, y) in pts[1:-1]:
            time.sleep(step)
            pipe.move(int(x), int(y), 0)
        time.sleep(tail)
        d.touch.up(int(pts[-1][0]), int(pts[-1][1]))
        up_sent = True
        return True
    except Exception:
        return False
    finally:
        if not up_sent:
            try:                                        # 补发UP,防指针残留触发长按
                d.touch.up(int(pts[-1][0]), int(pts[-1][1]))
            except Exception:
                pass
        if pipe is not None:
            pipe.close()


def wrist_arc_pts(w: int, h: int, distance: int, n: int = 10,
                  drift_range=(80, 140)) -> list:
    """生成一笔弧线的触点序列 [(x,y), ...]
    drift_range: 起终点横向漂移范围(短笔画给小值, 如(40,90))"""
    drift = random.choice([-1, 1]) * random.randint(*drift_range)
    x0 = int(w * random.uniform(0.40, 0.60))
    y0 = int(h * random.uniform(0.70, 0.80))
    x2 = x0 + drift
    y2 = max(60, y0 - distance)
    cx = (x0 + x2) / 2 - drift // 2      # 控制点反向偏 → 单向弧
    cy = (y0 + y2) / 2
    pts = []
    prev_y = None
    for i in range(n):
        t = i / (n - 1)
        te = t * t * (3 - 2 * t)         # smoothstep 缓起-快中-缓收
        x = (1 - te) ** 2 * x0 + 2 * (1 - te) * te * cx + te ** 2 * x2
        y = (1 - te) ** 2 * y0 + 2 * (1 - te) * te * cy + te ** 2 * y2
        # 指尖微抖只作用横向; y保证严格上行(密集触点时防1~2px反向)
        x += random.randint(-2, 2)
        y = int(y)
        if prev_y is not None and y >= prev_y:
            y = prev_y - 1
        pts.append((int(x), y))
        prev_y = y
    return pts


# ── 单手拇指侧滑(列表两段式用) ──
# 手性区: 起指在拇指自然落点(屏高74%~86%), 弧线中段扫向对侧, 终点在本侧偏上
RIGHT_SIDE_ZONE = {"x": (0.42, 0.65), "y": (0.74, 0.86), "drift": (40, 140)}
LEFT_SIDE_ZONE = {"x": (0.35, 0.58), "y": (0.74, 0.86), "drift": (-140, -40)}


def wrist_arc_pts_side(w: int, h: int, distance: int, zone: dict,
                       anchor=None, bow_ratio: float = 0.2, n: int = 60) -> list:
    """单手拇指侧滑轨迹(两段式每笔用)。
    anchor: 拇指自然落点(同一滑动的两笔共用,起指位置靠近); None=在zone内随机
    bow_ratio: 弓形强度×距离, 符号=凸向(正=凸右,负=凸左); 右手传负值 → "("形
    起伏基准: 弦偏差≈bow_ratio×distance/2, 生产用0.15~0.25 → 40~60px"""
    drift = random.randint(*zone["drift"])          # 终点横移(方向=手性)
    if anchor is None:
        x0 = int(w * random.uniform(*zone["x"]))
        y0 = int(h * random.uniform(*zone["y"]))
    else:
        x0 = int(anchor[0] + random.randint(-30, 30))
        y0 = int(anchor[1] + random.randint(-30, 30))
    x2 = max(20, min(w - 20, x0 + drift))           # 终点边界保护,防甩出屏幕
    y2 = max(60, y0 - distance)
    bow = bow_ratio * distance                      # 符号=凸向
    cx = (x0 + x2) / 2 + bow
    cy = (y0 + y2) / 2
    pts, prev_y = [], None
    for i in range(n):
        t = i / (n - 1)
        te = t * t * (3 - 2 * t)                    # smoothstep 缓起-快中-缓收
        x = (1 - te) ** 2 * x0 + 2 * (1 - te) * te * cx + te ** 2 * x2
        y = (1 - te) ** 2 * y0 + 2 * (1 - te) * te * cy + te ** 2 * y2
        # 指尖微抖只作用横向; y保证严格上行(密集触点时防1~2px反向)
        x += random.randint(-2, 2)
        y = int(y)
        if prev_y is not None and y >= prev_y:
            y = prev_y - 1
        pts.append((int(x), y))
        prev_y = y
    return pts


def wrist_arc_pts_h(w: int, h: int, distance: int, y_ratio: float = 0.3,
                    n: int = 10, drift_range=(30, 70)) -> list:
    """横向弧线: 从右往左滑(商品图轮播)。纵向漂移 = 手腕横向发力时的上下弓弧"""
    dy = random.choice([-1, 1]) * random.randint(*drift_range)
    x0 = int(w * random.uniform(0.72, 0.85))
    y0 = int(h * y_ratio) + random.randint(-30, 30)
    x2 = max(40, x0 - distance)
    y2 = y0 + dy
    cx = (x0 + x2) / 2
    cy = (y0 + y2) / 2 + random.choice([-1, 1]) * random.randint(20, 50)   # 垂直于运动方向的弓
    pts = []
    prev_x = None
    for i in range(n):
        t = i / (n - 1)
        te = t * t * (3 - 2 * t)
        x = (1 - te) ** 2 * x0 + 2 * (1 - te) * te * cx + te ** 2 * x2
        y = (1 - te) ** 2 * y0 + 2 * (1 - te) * te * cy + te ** 2 * y2
        # 纵向微抖; x保证严格递减(横向轮播滑动不回退)
        y += random.randint(-2, 2)
        x = int(x)
        if prev_x is not None and x >= prev_x:
            x = prev_x - 1
        pts.append((x, int(y)))
        prev_x = x
    return pts


def motionevent_drag(device_id: str, pts: list, step: float = None):
    """把触点序列作为一笔连续手势注入(单shell会话连发)"""
    if step is None:
        step = random.uniform(0.030, 0.045)
    lines = [f"input motionevent DOWN {pts[0][0]} {pts[0][1]}"]
    for (x, y) in pts[1:-1]:
        lines.append(f"sleep {step:.3f}")
        lines.append(f"input motionevent MOVE {x} {y}")
    lines.append(f"input motionevent UP {pts[-1][0]} {pts[-1][1]}")
    subprocess.run(["adb", "-s", device_id, "shell", "sh"],
                   input=("\n".join(lines) + "\n").encode("ascii"),
                   capture_output=True, timeout=60)