"""拟人化操作：滑动不机械，且**每台设备有自己的手感**。 分两个层次，别混在一起： 1. **每次都不一样**（抖动）：起点/终点位置、滑动幅度、时长、弧线方向都随机—— 最容易被识别的特征不是"慢"，而是"同一帧重复播放"。 2. **每台设备稳定不同**（性格）：由 `serial` 派生一份固定的偏好（手速、幅度、 弧度、停顿、常用触点的横向位置）。同一台设备多次运行风格一致，不同设备之间 明显不同——批量跑 13 台时看起来像 13 个人各刷各的，而不是 13 台同步机器人。 **性格稳定、抖动随机**，两者都由 serial 播种，不引入全局状态、不做任何 IO。 性格怎么来的：`crc32(serial)` 作种子 → 几个围绕 1.0 的比例因子（见 `profile()`）。 换设备换风格，改名字**不改**风格（按 serial 而不是设备名），换地址会换风格—— 这符合直觉：地址代表"这一台"。 想手工看一眼某台设备的性格：`python -c "from core import humanize; print(humanize.profile('192.168.20.55:5555'))"` """ import random import time import zlib from core.logger import get_logger _log = get_logger("core.humanize") _PROFILE_CACHE = {} # serial -> (w, h, ts)：屏幕尺寸缓存（见 screen_size） _SIZE_CACHE = {} _SIZE_TTL = 120 def profile(serial): """设备性格（按 serial 缓存，恒定）。字段都是围绕 1.0 的比例因子。""" key = str(serial or "") p = _PROFILE_CACHE.get(key) if p is not None: return p # 用独立的 Random 实例而不是全局 random：worker 是多线程的，全局 random 的 # 取值顺序会被其它线程打乱 —— 那样"每台设备的性格"就串味了 r = random.Random(zlib.crc32(("human:" + key).encode("utf-8"))) p = { "speed": r.uniform(0.82, 1.32), # <1 划得快（时长短），>1 划得慢 "distance": r.uniform(0.86, 1.16), # 有人习惯一划到底，有人小步快跑 "curve": r.uniform(0.4, 1.6), # 弧线弯度系数 "x_bias": r.uniform(-0.09, 0.09), # 常用触点的横向偏移（屏幕宽比例） "pause": r.uniform(0.80, 1.35), # 停顿倍率（wait 步骤可选启用） } _PROFILE_CACHE[key] = p return p def screen_size(d, serial=""): """屏幕尺寸，带缓存。 为什么不用 `d.info`：它在部分设备上**极慢**（实测 14s，见 doc/backlog），而 尺寸在一次运行里根本不会变。`d.window_size()` 快得多（同设备 0.6s）。 缓存 120s 过期：够躲开慢调用，又能在屏幕旋转/换分辨率后自愈。 """ key = str(serial or "") now = time.time() hit = _SIZE_CACHE.get(key) if hit and now - hit[2] < _SIZE_TTL: return hit[0], hit[1] try: w, h = d.window_size() except Exception: # 老 agent 没有 window_size 就退回 info info = d.info w, h = info["displayWidth"], info["displayHeight"] w, h = int(w), int(h) if w > 0 and h > 0: _SIZE_CACHE[key] = (w, h, now) return w, h def _bezier(p0, p1, p2, n): """二次贝塞尔采样——手指划出来是弧线，不是尺子画的直线。""" pts = [] for i in range(n + 1): t = i / n x = (1 - t) ** 2 * p0[0] + 2 * (1 - t) * t * p1[0] + t * t * p2[0] y = (1 - t) ** 2 * p0[1] + 2 * (1 - t) * t * p1[1] + t * t * p2[1] pts.append((int(round(x)), int(round(y)))) # 同样的点会连续出现（曲线很平时）——去掉，免得下发冗余坐标 out = [pts[0]] for p in pts[1:]: if p != out[-1]: out.append(p) return out def swipe(d, direction, serial="", duration_min=0.25, duration_max=0.50, distance_ratio=0.6, jitter=0.15, humanize=True): """按人类习惯滑一下（上/下/左/右）。 参数： direction 上滑 up / 下滑 down / 左滑 left / 右滑 right serial 用来取本设备的性格（不传就是"普通人"） duration_min/max 手指接触屏幕的时长区间（秒），再乘本设备手速系数 distance_ratio 滑动幅度占屏幕高（竖滑）或宽（横滑）的比例，默认 0.6 jitter 抖动幅度 0~0.4：位置/幅度/时长各抖这么多；0 = 每次都一样 humanize False = 退回老的直线滑动（完全规则，用于对照/排障） 返回实际参数 dict（时长/起止点），方便日志与排障。异常向上抛，由调用方兜。 """ w, h = screen_size(d, serial) prof = profile(serial) vertical = direction in ("up", "down") j = max(0.0, min(float(jitter or 0), 0.4)) # ---- 时长：用户给的范围 × 本设备手速 × 本次抖动 ---- dur = random.uniform(float(duration_min), float(duration_max)) dur *= prof["speed"] if j: dur *= 1 + random.uniform(-j, j) * 0.6 dur = max(0.08, round(dur, 3)) # ---- 幅度与中心点 ---- span = (h if vertical else w) * float(distance_ratio) * prof["distance"] if j: span *= 1 + random.uniform(-j, j) span = max(0.08, min(span, (h if vertical else w) * 0.92)) # 起点不在正中：人不会每次都在屏幕同一条线上划 if vertical: cx = w * (0.5 + prof["x_bias"] + (random.uniform(-j, j) * 0.6 if j else 0)) cx = int(max(w * 0.10, min(cx, w * 0.90))) cy = h * 0.5 if j: cy += h * random.uniform(-j, j) * 0.5 sign = -1 if direction == "up" else 1 p0 = (cx, int(max(h * 0.03, min(cy - sign * span / 2, h * 0.97)))) p2 = (cx, int(max(h * 0.03, min(cy + sign * span / 2, h * 0.97)))) else: cy = int(h * (0.5 + prof["x_bias"] + (random.uniform(-j, j) * 0.6 if j else 0))) cy = int(max(h * 0.10, min(cy, h * 0.90))) cx = w * 0.5 if j: cx += w * random.uniform(-j, j) * 0.5 sign = -1 if direction == "left" else 1 p0 = (int(max(w * 0.03, min(cx - sign * span / 2, w * 0.97))), cy) p2 = (int(max(w * 0.03, min(cx + sign * span / 2, w * 0.97))), cy) if not humanize: # 老行为：正中直线、无抖动（点与点之间由设备等分插值） if vertical: s, e = (0.8, 0.2) if direction == "up" else (0.2, 0.8) base = (int(w * 0.5), int(h * s), int(w * 0.5), int(h * e)) else: s, e = (0.8, 0.2) if direction == "left" else (0.2, 0.8) base = (int(w * s), int(h * 0.5), int(w * e), int(h * 0.5)) d.swipe(*base, random.uniform(float(duration_min), float(duration_max))) return {"duration": dur, "from": base[:2], "to": base[2:], "human": False} # ---- 弧线：控制点在中点、垂直于滑动方向偏移 ---- amp = span * 0.05 * prof["curve"] if j: amp *= random.uniform(0.3, 1.0) amp *= random.choice((-1, 1)) # 这次往左弯还是往右弯 else: # jitter=0 时不引入任何随机：弯向由设备性格决定，便于对照排障 amp *= -1 if prof["curve"] < 1.0 else 1 mx, my = (p0[0] + p2[0]) / 2, (p0[1] + p2[1]) / 2 if vertical: ctrl = (mx + amp, my) else: ctrl = (mx, my + amp) # 点数刻意压到 4：`swipe_points` 在慢设备上**每多一个点就多约 1s** # （实测 2 点 1.4s / 6 点 6.0s / 10 点 10.6s），而设备自己会在点之间 # 插值 70 步，4 个点的贝塞尔已经足够弯 pts = _bezier(p0, ctrl, p2, 4) if len(pts) < 2: # 幅度太小时退化成直线 pts = [p0, p2] try: d.swipe_points([[x, y] for x, y in pts], dur) except Exception as e: # 老设备/agent 不支持曲线就退回直线 _log.warning("swipe_points 不可用，退回直线滑动: %s", e) d.swipe(p0[0], p0[1], p2[0], p2[1], dur) return {"duration": dur, "from": p0, "to": p2, "curve": amp, "human": True} def pace(serial, seconds, enabled=True): """按本设备节奏微调一个停顿（wait 步骤可选启用）。enabled=False 原样返回。""" if not enabled: return float(seconds) return float(seconds) * profile(serial)["pause"]