butubb
cf075ba4e1
用户反馈"滑动太像机器人""批量执行时每台设备都一样"。 core/humanize.py（新）分两个层次： - **每次不同**：起止点/幅度/时长抖动、弧线方向随机——最容易被识别的不是"慢"， 是"每次都一模一样"； - **每台设备不同**：由 crc32(serial) 派生稳定的"性格"（手速 0.82~1.32、 幅度 0.86~1.16、弧度、常用横坐标 ±9% 屏宽、停顿 0.80~1.35）。同设备多次运行 风格一致，不同设备明显不同——13 台批量跑看着像 13 个人各刷各的。 用独立 Random 实例播种，不碰全局 random（多线程 worker 会打乱取值顺序）。 轨迹用二次贝塞尔走 d.swipe_points（曲线），异常时自动退回直线 d.swipe。 **点数固定 4 个**：swipe_points 在慢设备上每多一个点约多 1 秒（实测 2 点 1.4s / 6 点 6.0s / 10 点 10.6s），设备自己会插值几十步，4 点已足够弯。 顺带修一个真机上的老毛病：滑动原本每次都要读 d.info，而它在部分设备上要 **14 秒**。改用 humanize.screen_size()（走 window_size，同设备 0.6s，缓存 120s）——所以哪怕多了弧线，真机单次滑动反而从 ~15s 降到 ~5s。 - tasks/generic/task.py：swipe / swipe_until 走拟人（新增 distance_ratio、 jitter、humanize 参数，默认开）；swipe_until 每轮停顿也抖动；wait 步骤新增 可选 vary_pace（默认关，按设备节奏缩放 0.8~1.35 倍）； - static/admin/editor.js：滑动类步骤参数面板加"幅度/抖动/拟人轨迹"+ 说明； - 文档：TASK_DEV.md §8.4（含"点数别调大""别用 d.info"两个坑）、步骤表、 README 能力表与结构。 自测：假设备单测（曲线/抖动/设备间差异/退化路径/不越界/尺寸缓存）+ 真机联调 （4 次滑动全部 ok，标注"弧线"、坐标每次不同、对照的 humanize=false 仍是直线）。
187 lines
8.2 KiB
Python
187 lines
8.2 KiB
Python
"""拟人化操作：滑动不机械，且**每台设备有自己的手感**。
|
||
|
||
分两个层次，别混在一起：
|
||
|
||
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"]
|