杨豪
cae37465d2
- method_l_shape: OK 候选尺度 ±0.0375 精修（步长 0.0125 + ±6px 窗口）， GT live_165204 Δ 179→176（真人 177），离线 9/10 保持 - 插桩 JSON.stringify 捕获 SDK 日志明文；指针 probe 排除 coalesced/pressure - 拖动不跟手指标作废：post_pos 晚于 SDK 重置滑块，鼠标事件流始终完整 - 剩余 VerifyErr = 候选整体选错（视觉确证），方向：多候选质量评估 - docs: experiments/solution/retrospective 同步收尾
742 lines
33 KiB
Python
742 lines
33 KiB
Python
"""冷启动单次尝试：触发验证码 → 采多题选最优 → force=0.5 单次拖动 → 读 verify。
|
||
|
||
频控对策：verify 只打一次，get/refresh 可多次采题挑最优。
|
||
用法: python try/cold_start_attempt.py [max_questions] [max_score]
|
||
"""
|
||
import asyncio, base64, json, math, os, random, sys, time, urllib.request as ur
|
||
from pathlib import Path
|
||
import websockets
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src"))
|
||
from solve import solve
|
||
from human_drag import gen_human_path
|
||
from enrich_replay import enrich
|
||
|
||
CDP = "http://127.0.0.1:9222"
|
||
SCALE = 340 / 552
|
||
TRACE_JSON = Path(__file__).resolve().parent / "human_trace.json"
|
||
def random_phone():
|
||
"""随机手机号（错误凭证也弹验证码，无需真实存在）。
|
||
用 13x/18x/14x 传统号段：16x/17x 虚拟号段被"验证码登录"拒绝的比例高。"""
|
||
return "1" + random.choice("3458") + "".join(random.choices("0123456789", k=9))
|
||
|
||
|
||
def random_password():
|
||
"""随机密码（满足常见强度要求：大小写+数字+符号）。"""
|
||
chars = "abcdefghijkmnpqrstuvwxyz"
|
||
caps = "ABCDEFGHJKLMNPQRSTUVWXYZ"
|
||
digits = "23456789"
|
||
syms = "!@#$%^&*"
|
||
pool = [random.choice(caps), random.choice(digits), random.choice(syms)]
|
||
pool += [random.choice(chars + caps + digits + syms) for _ in range(9)]
|
||
random.shuffle(pool)
|
||
return "".join(pool)
|
||
CAPT = Path(__file__).resolve().parent / "captures"
|
||
HOOK = """
|
||
(() => {
|
||
window.__lastGet = null; window.__lastVerify = null;
|
||
const oOpen = XMLHttpRequest.prototype.open, oSend = XMLHttpRequest.prototype.send;
|
||
XMLHttpRequest.prototype.open = function(m,u){ this.__u = u; return oOpen.apply(this, arguments); };
|
||
XMLHttpRequest.prototype.send = function(){
|
||
const x = this;
|
||
try {
|
||
if (/captcha\\/verify/.test(x.__u || '')) {
|
||
const body = arguments[0];
|
||
window.__lastVerifyBody = typeof body === 'string' ? body : (body && body.length ? '[form]' + String(body).slice(0, 8000) : 'null');
|
||
}
|
||
} catch (e) {}
|
||
this.addEventListener('load', () => {
|
||
try {
|
||
if (/captcha\\/get/.test(x.__u || '')) window.__lastGet = JSON.parse(x.responseText);
|
||
if (/captcha\\/verify/.test(x.__u || '')) window.__lastVerify = x.responseText.slice(0, 200);
|
||
} catch (e) {}
|
||
});
|
||
return oSend.apply(this, arguments);
|
||
};
|
||
if (!window.__recOn) {
|
||
window.__recOn = 1;
|
||
window.__dragTrace = [];
|
||
// 插桩：捕获 SDK 加密前的明文（captchaBody plain）
|
||
const oStringify = JSON.stringify;
|
||
JSON.stringify = function(...args) {
|
||
try {
|
||
const s = oStringify.apply(this, args);
|
||
if (typeof s === 'string' && s.length > 500 &&
|
||
(s.includes('"mode"') || s.includes('relative_time') || s.includes('drag_width'))) {
|
||
window.__lastPlain = s;
|
||
}
|
||
return s;
|
||
} catch (e) { return oStringify.apply(this, args); }
|
||
};
|
||
const rec = (type) => (e) => {
|
||
if (!e.isTrusted) return;
|
||
window.__dragTrace.push({t: e.timeStamp, type, x: e.clientX, y: e.clientY,
|
||
p: e.pressure, btns: e.buttons});
|
||
};
|
||
document.addEventListener('mousedown', rec('down'), true);
|
||
document.addEventListener('mousemove', rec('move'), true);
|
||
document.addEventListener('mouseup', rec('up'), true);
|
||
}
|
||
return 'ok';
|
||
})()
|
||
"""
|
||
|
||
|
||
def _json(path):
|
||
try:
|
||
with ur.urlopen(f"{CDP}{path}", timeout=5) as r:
|
||
return json.load(r)
|
||
except (OSError, ValueError):
|
||
return None
|
||
|
||
|
||
XDISPLAY = os.environ.get("XDISPLAY", ":99")
|
||
|
||
|
||
async def xdotool(*args):
|
||
"""在目标 X display 上执行 xdotool（OS 层真实鼠标输入，无 CDP 合成标记）。"""
|
||
env = dict(os.environ, DISPLAY=XDISPLAY)
|
||
try:
|
||
proc = await asyncio.create_subprocess_exec(
|
||
"/usr/bin/xdotool", *args,
|
||
stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, env=env)
|
||
_, err = await proc.communicate()
|
||
if proc.returncode != 0:
|
||
print(f"xdotool {args[0]} 失败: {err.decode(errors='replace').strip()[:80]}")
|
||
except FileNotFoundError:
|
||
print("xdotool 未安装")
|
||
except OSError as e:
|
||
print(f"xdotool 执行失败: {e}")
|
||
|
||
|
||
def replay_pts(seg, dist_ui, target_ms):
|
||
"""把真人拖动段（绝对 ts/x/y 列表）重放为目标距离/时长的 (dx,dy,dt) 序列。
|
||
x 线性缩放到 dist_ui（终点精确），时间轴拉伸到 target_ms（保留节奏起伏）。"""
|
||
t0, x0, _y0 = seg[0]
|
||
raw = [(t - t0, x - x0, y - _y0) for (t, x, y) in seg]
|
||
span = max(raw[-1][0], 1.0)
|
||
end_x = raw[-1][1]
|
||
if abs(end_x) < 1.0:
|
||
end_x = max((x for _, x, _ in raw), default=1.0)
|
||
k = dist_ui / end_x
|
||
scale_t = target_ms / span
|
||
out = []
|
||
last_x, last_y = 0.0, 0.0
|
||
prev_t = raw[0][0]
|
||
for t, x, y in raw[1:]:
|
||
dt = t - prev_t # 相邻事件间隔（raw 里是累计 ts）
|
||
prev_t = t
|
||
tx, ty = x * k, y * k
|
||
try:
|
||
step_ms = max(int(dt * scale_t), 1)
|
||
except (TypeError, ValueError, OverflowError):
|
||
step_ms = 10
|
||
out.append((round(tx - last_x, 2), round(ty - last_y, 2), step_ms))
|
||
last_x, last_y = tx, ty
|
||
# 终点精确回贴
|
||
if out:
|
||
ex, ey, et = out[-1]
|
||
try:
|
||
fix = float(dist_ui) - sum(p[0] for p in out[:-1])
|
||
except (TypeError, ValueError):
|
||
fix = ex
|
||
out[-1] = (round(fix, 2), ey, et)
|
||
return out
|
||
|
||
|
||
def load_human_segments(path):
|
||
"""从轨迹采集 JSON 提取按下-抬起段（dx>80 才算拖动）。
|
||
兼容两种采集格式：{ts,type,x,y,buttons,trusted} 与 {t,type,x,y,btns}。"""
|
||
try:
|
||
ev = json.load(open(path))["events"]
|
||
except (OSError, ValueError, KeyError) as e:
|
||
print(f"轨迹加载失败: {e}")
|
||
return []
|
||
T_DOWN = ("mousedown", "pointerdown", "down")
|
||
T_MOVE = ("mousemove", "pointermove", "move")
|
||
T_UP = ("mouseup", "pointerup", "up")
|
||
segs, cur = [], None
|
||
for e in ev:
|
||
t = e.get("t", e.get("ts"))
|
||
if t is None:
|
||
continue
|
||
typ = e.get("type", e.get("t_type", ""))
|
||
x, y = e.get("x"), e.get("y")
|
||
btns = e.get("buttons", e.get("btns", 0))
|
||
try:
|
||
t, x, y, btns = float(t), float(x), float(y), int(btns)
|
||
except (TypeError, ValueError):
|
||
continue
|
||
if typ in T_DOWN:
|
||
cur = [(t, x, y)]
|
||
elif typ in T_MOVE and cur is not None and btns == 1:
|
||
cur.append((t, x, y))
|
||
elif typ in T_UP and cur is not None:
|
||
cur.append((t, x, y))
|
||
dx = cur[-1][1] - cur[0][1]
|
||
if len(cur) > 10 and dx > 80:
|
||
segs.append(cur)
|
||
cur = None
|
||
return segs
|
||
|
||
|
||
async def _warmup_session():
|
||
"""会话画像预热：真人触发验证码前已在页面浏览数分钟（鼠标游走、停顿）。
|
||
我们从页面打开到触发只有几秒 —— 会话级行为画像差异可能是静默降权来源。
|
||
在触发验证码之前模拟 45-75s 慢速游走（xdotool 硬件事件）。"""
|
||
if os.environ.get("XDOTOOL") != "1" or os.environ.get("WARMUP", "1") == "0":
|
||
return
|
||
warm_s = random.uniform(45, 75)
|
||
print(f"会话预热 {warm_s:.0f}s（模拟真人浏览）", flush=True)
|
||
w_t0 = asyncio.get_event_loop().time()
|
||
wx, wy = random.uniform(400, 1400), random.uniform(200, 700)
|
||
while (asyncio.get_event_loop().time() - w_t0) < warm_s:
|
||
await xdotool("mousemove", str(round(wx)), str(round(wy)))
|
||
await asyncio.sleep(random.uniform(1.5, 5.0))
|
||
wx += random.uniform(-120, 120)
|
||
wy += random.uniform(-60, 60)
|
||
wx = min(max(wx, 200), 1700)
|
||
wy = min(max(wy, 120), 800)
|
||
|
||
|
||
async def main_once(max_questions=4, max_score=5.5):
|
||
"""单次完整尝试，返回结果字符串：PASS / 500 / 502 / skip / error..."""
|
||
n_q, max_score = max_questions, max_score
|
||
await _warmup_session()
|
||
ver = _json("/json/version")
|
||
if not ver or not ver.get("webSocketDebuggerUrl"):
|
||
print("无法连接 CDP", file=sys.stderr)
|
||
return
|
||
targets = _json("/json/list") or []
|
||
page = next((t for t in targets if t["type"] == "page" and "douyin" in t.get("url", "")), None)
|
||
if not page:
|
||
print("未找到 douyin 页面", file=sys.stderr)
|
||
return
|
||
async with websockets.connect(ver["webSocketDebuggerUrl"], max_size=64 * 1024 * 1024) as ws:
|
||
mid = [0]
|
||
|
||
async def cdp(method, params=None, sid=None):
|
||
mid[0] += 1
|
||
m = {"id": mid[0], "method": method, "params": params or {}}
|
||
if sid:
|
||
m["sessionId"] = sid
|
||
await ws.send(json.dumps(m))
|
||
while True:
|
||
try:
|
||
r = json.loads(await ws.recv())
|
||
except (json.JSONDecodeError, asyncio.TimeoutError):
|
||
continue
|
||
if r.get("id") == mid[0]:
|
||
if "error" in r:
|
||
raise RuntimeError(f"{method}: {r['error']}")
|
||
return r.get("result", {})
|
||
|
||
psid = (await cdp("Target.attachToTarget", {"targetId": page["id"], "flatten": True}))["sessionId"]
|
||
await cdp("Runtime.enable", sid=psid)
|
||
|
||
async def js(expr):
|
||
r = await cdp("Runtime.evaluate",
|
||
{"expression": expr, "returnByValue": True, "awaitPromise": True}, psid)
|
||
return r.get("result", {}).get("value")
|
||
|
||
async def click(x, y):
|
||
for typ in ("mousePressed", "mouseReleased"):
|
||
await cdp("Input.dispatchMouseEvent",
|
||
{"type": typ, "x": x, "y": y, "button": "left",
|
||
"buttons": 1, "clickCount": 1}, psid)
|
||
|
||
async def type_text(text):
|
||
for ch in text:
|
||
await cdp("Input.dispatchKeyEvent", {"type": "keyDown", "text": ch}, psid)
|
||
await cdp("Input.dispatchKeyEvent", {"type": "keyUp"}, psid)
|
||
await asyncio.sleep(0.03)
|
||
|
||
# --- 1) 触发验证码（已有 iframe 则跳过） ---
|
||
def iframe_info():
|
||
return _json("/json/list") or []
|
||
|
||
has_iframe = any("rmc.bytedance.com" in t.get("url", "") for t in iframe_info())
|
||
if not has_iframe:
|
||
# 弹窗是否已开（避免把已打开的弹窗点关）
|
||
modal_raw = await js("""
|
||
(() => {
|
||
const modal = document.querySelector('[class*=login-panel], [id*=login-panel]');
|
||
const phone = [...document.querySelectorAll('input')].find(i => i.placeholder === '请输入手机号');
|
||
return JSON.stringify({open: !!modal, phoneVisible: !!(phone && phone.offsetWidth > 0)});
|
||
})()
|
||
""")
|
||
try:
|
||
modal_state = json.loads(modal_raw) if modal_raw else {}
|
||
except ValueError:
|
||
modal_state = {}
|
||
if not modal_state.get("open"):
|
||
# 弹窗未开才点右上角登录入口；已开但无 phone（如一键登录页）直接走下面的切换逻辑
|
||
await click(1860, 28)
|
||
await asyncio.sleep(2)
|
||
# 一键登录面板（reload 后默认）：先点"登录其他账号"切到账密表单，再点"密码登录"
|
||
await js("""
|
||
(() => {
|
||
const other = [...document.querySelectorAll('*')].find(
|
||
e => e.children.length === 0 && /登录其他账号/.test(e.textContent || '') && e.offsetWidth > 0);
|
||
if (other) other.click();
|
||
})()
|
||
""")
|
||
await asyncio.sleep(1.5)
|
||
await js("""
|
||
(() => {
|
||
const els = [...document.querySelectorAll('*')].filter(
|
||
e => e.children.length === 0 && /密码登录/.test(e.textContent || '') && e.offsetWidth > 0);
|
||
if (els.length) els[0].click();
|
||
})()
|
||
""")
|
||
await asyncio.sleep(1.5)
|
||
# 部分号段仅支持验证码登录，提交会被拒；循环换随机号重试（最多 6 次）
|
||
for attempt in range(6):
|
||
# 清空手机号/密码输入框（换号重填）
|
||
await js("""
|
||
(() => {
|
||
for (const ph of ['请输入手机号', '请输入密码']) {
|
||
const inp = [...document.querySelectorAll('input')].find(i => i.placeholder === ph);
|
||
if (inp) {
|
||
inp.focus();
|
||
const setter = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
|
||
setter.call(inp, '');
|
||
inp.dispatchEvent(new Event('input', {bubbles: true}));
|
||
}
|
||
}
|
||
})()
|
||
""")
|
||
await asyncio.sleep(0.4)
|
||
geo_raw = await js("""
|
||
(() => {
|
||
const phone = [...document.querySelectorAll('input')].find(i => i.placeholder === '请输入手机号');
|
||
const cb = document.querySelector('span.f64a2kjb');
|
||
if (!phone) return 'null';
|
||
const pr = phone.getBoundingClientRect();
|
||
const cr = cb ? cb.getBoundingClientRect() : null;
|
||
return JSON.stringify({phone: [Math.round(pr.x + pr.width / 2), Math.round(pr.y + pr.height / 2)],
|
||
cb: cr ? [Math.round(cr.x + cr.width / 2), Math.round(cr.y + cr.height / 2)] : null});
|
||
})()
|
||
""")
|
||
try:
|
||
geo = json.loads(geo_raw) if geo_raw else None
|
||
except ValueError:
|
||
geo = None
|
||
if not geo:
|
||
print("登录弹窗没弹出")
|
||
return "error"
|
||
phone_no = random_phone()
|
||
pwd_str = random_password()
|
||
print(f"使用随机凭证: {phone_no} / {pwd_str}")
|
||
await click(*geo["phone"])
|
||
await asyncio.sleep(0.5)
|
||
await type_text(phone_no)
|
||
pwd_raw = await js("""
|
||
(() => {
|
||
const pwd = [...document.querySelectorAll('input')].find(i => i.placeholder === '请输入密码');
|
||
if (!pwd) return 'null';
|
||
const r = pwd.getBoundingClientRect();
|
||
return JSON.stringify([Math.round(r.x + r.width / 2), Math.round(r.y + r.height / 2)]);
|
||
})()
|
||
""")
|
||
try:
|
||
pwd_xy = json.loads(pwd_raw) if pwd_raw else None
|
||
except ValueError:
|
||
pwd_xy = None
|
||
if pwd_xy:
|
||
await click(*pwd_xy)
|
||
await asyncio.sleep(0.5)
|
||
await type_text(pwd_str)
|
||
if geo.get("cb"):
|
||
await click(*geo["cb"])
|
||
await asyncio.sleep(0.5)
|
||
await click(1165, 630)
|
||
captcha_ok = False
|
||
for _ in range(25):
|
||
await asyncio.sleep(1.5)
|
||
if any("rmc.bytedance.com" in t.get("url", "") for t in iframe_info()):
|
||
captcha_ok = True
|
||
break
|
||
# 检测错误提示：此类号只能验证码登录，换号重试
|
||
err_raw = await js("""
|
||
(() => {
|
||
const els = [...document.querySelectorAll('*')].filter(
|
||
e => e.children.length === 0 && e.textContent && e.offsetWidth > 0 &&
|
||
/仅支持验证码|不存在|格式有误|频繁|繁忙|错误/.test(e.textContent));
|
||
return els.length ? els[0].textContent.trim().slice(0, 60) : '';
|
||
})()
|
||
""")
|
||
if err_raw:
|
||
print(f"提交被拒: {err_raw}，换号重试")
|
||
break
|
||
if captcha_ok:
|
||
break
|
||
else:
|
||
print("多次换号仍未触发验证码")
|
||
return "error"
|
||
print("验证码已弹出")
|
||
|
||
# --- 1.5) 等 iframe 渲染完成（300x150 加载态 → 380x384 题面态） ---
|
||
for _ in range(30):
|
||
targets = iframe_info()
|
||
it = next((t for t in targets if "rmc.bytedance.com" in t.get("url", "")), None)
|
||
if not it:
|
||
await asyncio.sleep(1.0)
|
||
continue
|
||
probe = (await cdp("Target.attachToTarget", {"targetId": it["id"], "flatten": True}))["sessionId"]
|
||
await cdp("Runtime.enable", sid=probe)
|
||
r = await cdp("Runtime.evaluate", {"expression": "document.documentElement.offsetWidth",
|
||
"returnByValue": True}, probe)
|
||
w = r.get("result", {}).get("value") or 0
|
||
await cdp("Target.detachFromTarget", {"sessionId": probe})
|
||
if w >= 380:
|
||
break
|
||
await asyncio.sleep(1.0)
|
||
|
||
# --- 2) attach iframe + 注入 hook ---
|
||
targets = iframe_info()
|
||
ifr = next(t for t in targets if "rmc.bytedance.com" in t.get("url", ""))
|
||
isid = (await cdp("Target.attachToTarget", {"targetId": ifr["id"], "flatten": True}))["sessionId"]
|
||
await cdp("Runtime.enable", sid=isid)
|
||
|
||
async def jse(expr, sid=isid):
|
||
r = await cdp("Runtime.evaluate",
|
||
{"expression": expr, "returnByValue": True, "awaitPromise": True}, sid)
|
||
return r.get("result", {}).get("value")
|
||
|
||
# iframe 实时坐标（刷新按钮/滑块用）
|
||
geom_raw = await js("""
|
||
(() => {
|
||
const f = [...document.querySelectorAll('iframe')].find(f => /rmc\\.bytedance\\.com/.test(f.src));
|
||
const r = f.getBoundingClientRect();
|
||
return JSON.stringify({x: Math.round(r.x), y: Math.round(r.y), w: f.offsetWidth, h: f.offsetHeight});
|
||
})()
|
||
""")
|
||
try:
|
||
geom = json.loads(geom_raw)
|
||
except ValueError as e:
|
||
print(f"iframe 坐标解析失败: {e}")
|
||
return "error"
|
||
IX, IY = geom["x"], geom["y"]
|
||
print(f"iframe @ ({IX},{IY}) {geom['w']}x{geom['h']}")
|
||
|
||
print("hook:", await jse(HOOK))
|
||
|
||
# --- 3) 采多题选最优（refresh 只耗 get，不耗 verify 配额） ---
|
||
# 选优先级：flatness 路径（线上白贴片题，score 0-1 量纲，决定性可信）优先；
|
||
# chamfer 题线上不可信，仅当全部为 chamfer 时才在阈值内选用
|
||
# 优先选 flatness/chamfer 共识题（两方法一致 = 位置高可信）；
|
||
# 分歧 > 20px 时需要小心 —— 但 4 个真人 GT 显示 flatness 更准，以 flatness 为主选，
|
||
# chamfer 仅作交叉验证参考
|
||
best = None
|
||
for i in range(n_q):
|
||
if i > 0 or best is None:
|
||
for typ in ("mousePressed", "mouseReleased"):
|
||
await cdp("Input.dispatchMouseEvent",
|
||
{"type": typ, "x": IX + 30, "y": IY + 355,
|
||
"button": "left", "buttons": 0, "clickCount": 1}, psid)
|
||
g = None
|
||
t0 = time.time()
|
||
while time.time() - t0 < 8 and not g:
|
||
await asyncio.sleep(0.8)
|
||
gg = await jse("JSON.stringify(window.__lastGet)")
|
||
if gg and gg != "null":
|
||
try:
|
||
g = json.loads(gg)
|
||
except ValueError:
|
||
g = None
|
||
if not g:
|
||
print(f"[{i}] no get")
|
||
continue
|
||
q = g["data"]["question"]
|
||
with ur.urlopen(q["url1"]) as r:
|
||
big = r.read()
|
||
with ur.urlopen(q["url2"]) as r:
|
||
mark = r.read()
|
||
res = solve(base64.b64encode(big).decode(), base64.b64encode(mark).decode(),
|
||
tip_y=q["tip_y"])
|
||
score = res.get("score", 99)
|
||
ok = res.get("ok") and res.get("confidence") == "OK"
|
||
is_flat = res.get("method") == "flatness"
|
||
# 排序：(共识优先, flatness 优先, score)
|
||
cand = res.get("candidates") or []
|
||
has_consensus = False
|
||
if cand and res.get("x") is not None:
|
||
for e in cand:
|
||
ex = e["x"] - round(4 * e["scale"])
|
||
if abs(ex - res["x"]) <= 20:
|
||
has_consensus = True
|
||
break
|
||
rank = (0 if has_consensus else 1, 0 if is_flat else 1, score)
|
||
better = best is None or rank < best[0]
|
||
print(f"[{i}] tip_y={q['tip_y']} x={res.get('x')} s={res.get('scale')} "
|
||
f"score={score:.3f} {res.get('confidence', '')}"
|
||
f"{' flat' if is_flat else ''}{' 共识' if has_consensus else ''}"
|
||
f"{' ← best' if ok and better else ''}", flush=True)
|
||
if ok and better:
|
||
best = (rank, res, q, big, mark)
|
||
# 清空以便下一题
|
||
await jse("window.__lastGet = null; window.__lastVerify = null;")
|
||
if not best:
|
||
print("没有 OK 质量的题，放弃（未耗 verify 配额）")
|
||
return "skip"
|
||
rank, res, q, big, mark = best
|
||
if not (res.get("method") == "flatness") and rank[1] > max_score:
|
||
print(f"最优分 {rank[1]:.3f} > 阈值 {max_score}，放弃")
|
||
return "skip"
|
||
ts = time.strftime("%H%M%S")
|
||
(CAPT / f"live_{ts}.png").write_bytes(big)
|
||
(CAPT / f"live_{ts}-mark.png").write_bytes(mark)
|
||
print(f"选定题 score={rank[1]:.3f} x={res['x']} method={res.get('method', 'chamfer')} 已存档 live_{ts}")
|
||
if os.environ.get("NO_DRAG") == "1":
|
||
print(f"NO_DRAG: 期望拖动 {res['x']*SCALE:.1f}px（原图 x={res['x']}），留给人工拖动")
|
||
return "nodrag"
|
||
|
||
# --- 4) 单次拖动（force=0.5，全部事件） ---
|
||
# 滑块按钮中心实测
|
||
btn_raw = await jse("""
|
||
(() => {
|
||
const f = parent.document.querySelector('iframe[src*="rmc.bytedance.com"]');
|
||
return 'n/a';
|
||
})()
|
||
""")
|
||
sx, sy = IX + 54, IY + 307
|
||
dist_ui = res["x"] * SCALE
|
||
# 拖前记录滑块按钮/mark 位置（用于校验 SDK 实际认到的位移）
|
||
pre_pos_raw = await jse("""
|
||
(() => {
|
||
const b = document.querySelector('[class*=slider-btn]');
|
||
const m = document.querySelector('[class*=verify-image-slide]');
|
||
const o = {};
|
||
if (b) { const r = b.getBoundingClientRect(); o.btn_x = +r.x.toFixed(1); }
|
||
if (m) { const r = m.getBoundingClientRect(); o.mark_x = +r.x.toFixed(1); }
|
||
return JSON.stringify(o);
|
||
})()
|
||
""")
|
||
pts = gen_human_path(dist_ui, seed=random.randint(1, 9999))
|
||
if os.environ.get("REPLAY") == "1":
|
||
trace_file = Path(os.environ.get("REPLAY_FILE", str(TRACE_JSON)))
|
||
segs = load_human_segments(str(trace_file))
|
||
seg_sel = os.environ.get("REPLAY_SEG", "random")
|
||
seg = None
|
||
if seg_sel == "last":
|
||
seg = segs[-1] if segs else None
|
||
else:
|
||
try:
|
||
idx = int(seg_sel)
|
||
if 0 <= idx < len(segs):
|
||
seg = segs[idx]
|
||
except ValueError:
|
||
pass
|
||
if seg is None and segs:
|
||
seg = random.choice(segs)
|
||
try:
|
||
target_ms = int(os.environ.get("REPLAY_MS", str(random.randint(5000, 10000))))
|
||
except ValueError:
|
||
target_ms = random.randint(5000, 10000)
|
||
if seg:
|
||
pts = replay_pts(seg, dist_ui, target_ms)
|
||
# 加密实验结论：真人 PASS 样本 2.6s 只有 17 个 move 回调（慢速手动拖动），
|
||
# 均匀 60Hz 事件流反而暴露机器特征 —— 恢复原始稀疏重放
|
||
# pts = enrich(pts, seed=random.randint(1, 99999))
|
||
print(f"重放真人轨迹段({trace_file.name}/{seg_sel}): 原始 {len(seg)} 事件 / "
|
||
f"{(seg[-1][0]-seg[0][0])/1000:.2f}s → 加密后 {len(pts)} 步 @ 目标 {target_ms}ms")
|
||
use_xdo = os.environ.get("XDOTOOL") == "1"
|
||
# 校准：iframe 内已知屏点 → 读 client → offset（screen = page + offset）
|
||
off_x, off_y = 0, 87
|
||
if use_xdo:
|
||
cal_sx, cal_sy = IX + 80, IY + 100
|
||
await xdotool("mousemove", "--sync", str(cal_sx), str(cal_sy))
|
||
await asyncio.sleep(0.5)
|
||
cal_raw = await jse("""
|
||
(() => {
|
||
const mv = (window.__dragTrace||[]).filter(e => e.type==='move');
|
||
return mv.length ? JSON.stringify(mv[mv.length-1]) : 'null';
|
||
})()
|
||
""")
|
||
try:
|
||
cal = json.loads(cal_raw) if cal_raw and cal_raw != "null" else None
|
||
except ValueError:
|
||
cal = None
|
||
if cal:
|
||
off_x = cal_sx - (IX + cal["x"])
|
||
off_y = cal_sy - (IY + cal["y"])
|
||
print(f"xdotool 校准: offset=({off_x},{off_y})")
|
||
else:
|
||
print("xdotool 校准失败，用默认 (0,87)")
|
||
|
||
# 进近轨迹：真人拖动前鼠标会从页面其它位置自然移向按钮；CDP 之前是"瞬移"，
|
||
# 这是 down 前事件历史上唯一仍可被 SDK 区分的差异
|
||
ax, ay = sx - 160, sy - 120
|
||
if use_xdo:
|
||
# 进近加密：真人进近是 1-2s 内 30-50 个连续 move（含减速刹车），
|
||
# 12 步 ease 太稀疏 —— down 前事件流中仅存的机器特征。
|
||
# 先快后慢：前 60% 大步快移，后 40% 小步密集（刹车）。
|
||
n_app = random.randint(30, 45)
|
||
for i in range(1, n_app + 1):
|
||
t = i / n_app
|
||
ease = t * t * (3 - 2 * t)
|
||
await asyncio.sleep(random.uniform(0.02, 0.05) if i <= n_app * 0.6
|
||
else random.uniform(0.008, 0.025))
|
||
ix = ax + (sx - ax) * ease + random.uniform(-3, 3) * (1 - t)
|
||
iy = ay + (sy - ay) * ease + random.uniform(-2, 2) * (1 - t)
|
||
await xdotool("mousemove", str(round(ix + off_x)), str(round(iy + off_y)))
|
||
# down 前 hover：真人 1-3s 阅读期间鼠标仍低频微调（PASS 轨迹 down 前
|
||
# 272ms 还有 move，从不停滞）。微动 ±1-2px、间隔 200-600ms。
|
||
hover_ms = random.uniform(1200, 2600)
|
||
h_t0 = asyncio.get_event_loop().time()
|
||
while (asyncio.get_event_loop().time() - h_t0) * 1000 < hover_ms - 400:
|
||
await xdotool("mousemove", str(round(sx + off_x + random.uniform(-2, 2))),
|
||
str(round(sy + off_y + random.uniform(-2, 2))))
|
||
await asyncio.sleep(random.uniform(0.2, 0.6))
|
||
# 最后 <400ms 静止后直接 down（贴合真人 272ms 间隔）
|
||
await asyncio.sleep(random.uniform(0.15, 0.35))
|
||
await xdotool("mousedown", "1")
|
||
await asyncio.sleep(0.15)
|
||
t_drag0 = asyncio.get_event_loop().time()
|
||
cx, cy = sx, sy
|
||
t_plan = 0.0
|
||
for dx, dy, dt in pts:
|
||
t_plan += dt / 1000 # 时间轴连续：跳过位移不跳过时间
|
||
if abs(dx) < 0.3 and abs(dy) < 0.3:
|
||
cx += dx
|
||
cy += dy
|
||
continue
|
||
cx += dx
|
||
cy += dy
|
||
delay = t_plan - (asyncio.get_event_loop().time() - t_drag0)
|
||
if delay > 0:
|
||
await asyncio.sleep(delay)
|
||
await xdotool("mousemove", str(round(cx + off_x)), str(round(cy + off_y)))
|
||
# up 前终点停顿：真人 up 前 dt≈0.7s（终点对齐），不能丢
|
||
up_dt = pts[-1][2] / 1000 if pts else 0.15
|
||
await asyncio.sleep(max(up_dt, 0.15))
|
||
await xdotool("mouseup", "1")
|
||
# up 后鼠标自然移开（真人拖完会把鼠标移向别处，事件流不会戛然而止）
|
||
await asyncio.sleep(random.uniform(0.3, 0.8))
|
||
lx, ly = cx + off_x, cy + off_y
|
||
for i in range(1, 11):
|
||
t = i / 10.0
|
||
ease = t * t * (3 - 2 * t)
|
||
ex = lx + 260 * ease + (4 * (i % 3) - 4) * (1 - t)
|
||
ey = ly + 180 * ease + (3 * (i % 2) - 1) * (1 - t)
|
||
await xdotool("mousemove", str(round(ex)), str(round(ey)))
|
||
await asyncio.sleep(random.uniform(0.04, 0.12))
|
||
print(f"拖动完成(xdotool) {dist_ui:.1f}px（{len(pts)} 步）")
|
||
else:
|
||
for i in range(1, 13):
|
||
t = i / 12.0
|
||
ease = t * t * (3 - 2 * t)
|
||
ix = ax + (sx - ax) * ease + (6 * (i % 3) - 6) * (1 - t)
|
||
iy = ay + (sy - ay) * ease + (4 * (i % 2) - 2) * (1 - t)
|
||
await cdp("Input.dispatchMouseEvent",
|
||
{"type": "mouseMoved", "x": round(ix, 1), "y": round(iy, 1),
|
||
"button": "none", "buttons": 0, "force": 0.5}, psid)
|
||
await asyncio.sleep(0.03 + random.random() * 0.04)
|
||
await asyncio.sleep(0.4)
|
||
await cdp("Input.dispatchMouseEvent",
|
||
{"type": "mousePressed", "x": sx, "y": sy, "button": "left",
|
||
"buttons": 1, "clickCount": 1, "force": 0.5}, psid)
|
||
await asyncio.sleep(0.15)
|
||
t_drag0 = asyncio.get_event_loop().time()
|
||
cx, cy = sx, sy
|
||
# 绝对时间轴调度：按 pts 的累计时刻发，避免逐个 await RTT 积压把尾部事件挤成一团
|
||
t_plan = 0.0
|
||
for dx, dy, dt in pts:
|
||
t_plan += dt / 1000 # 时间轴连续：跳过位移不跳过时间
|
||
if abs(dx) < 0.3 and abs(dy) < 0.3:
|
||
cx += dx
|
||
cy += dy
|
||
continue
|
||
cx += dx
|
||
cy += dy
|
||
delay = t_plan - (asyncio.get_event_loop().time() - t_drag0)
|
||
if delay > 0:
|
||
await asyncio.sleep(delay)
|
||
await cdp("Input.dispatchMouseEvent",
|
||
{"type": "mouseMoved", "x": round(cx, 1), "y": round(cy, 1),
|
||
"button": "left", "buttons": 1, "force": 0.5}, psid)
|
||
# up 前终点停顿：真人 up 前 dt≈0.7s（终点对齐），不能丢
|
||
up_dt = pts[-1][2] / 1000 if pts else 0.15
|
||
await asyncio.sleep(max(up_dt, 0.15))
|
||
await cdp("Input.dispatchMouseEvent",
|
||
{"type": "mouseReleased", "x": round(cx, 1), "y": round(cy, 1),
|
||
"button": "left", "buttons": 1, "clickCount": 1, "force": 0.5}, psid)
|
||
print(f"拖动完成 {dist_ui:.1f}px（{len(pts)} 步）")
|
||
# 释放后立刻量位置（复位前抢采）。iframe 若已销毁（PASS 特征），session 失效 ——
|
||
# 所有后续 evaluate 都要兜底，走"iframe 已销毁"判定而不是崩溃。
|
||
try:
|
||
post_pos_raw = await jse("""
|
||
(() => {
|
||
const b = document.querySelector('[class*=slider-btn]');
|
||
const m = document.querySelector('[class*=verify-image-slide]');
|
||
const o = {};
|
||
if (b) { const r = b.getBoundingClientRect(); o.btn_x = +r.x.toFixed(1); }
|
||
if (m) { const r = m.getBoundingClientRect(); o.mark_x = +r.x.toFixed(1); }
|
||
return JSON.stringify(o);
|
||
})()
|
||
""")
|
||
try:
|
||
pre_pos = json.loads(pre_pos_raw) if pre_pos_raw else {}
|
||
post_pos = json.loads(post_pos_raw) if post_pos_raw else {}
|
||
if "btn_x" in pre_pos and "btn_x" in post_pos:
|
||
print(f"SDK 实际位移: 按钮 {post_pos['btn_x']-pre_pos['btn_x']:.1f}px, "
|
||
f"mark {post_pos.get('mark_x',0)-pre_pos.get('mark_x',0):.1f}px "
|
||
f"（期望 {dist_ui:.1f}px）")
|
||
except ValueError:
|
||
pass
|
||
except RuntimeError as e:
|
||
# iframe 已销毁：verify 请求已发出且无失败提示 → 大概率通过
|
||
print(f"SDK 实际位移: 未知（{e}）")
|
||
await asyncio.sleep(2)
|
||
if not any("rmc.bytedance.com" in t.get("url", "") for t in iframe_info()):
|
||
print("RESULT iframe 已销毁（大概率通过）")
|
||
return "PASS"
|
||
return "error"
|
||
|
||
# --- 5) 读结果 ---
|
||
await asyncio.sleep(3)
|
||
try:
|
||
v = await jse("JSON.stringify(window.__lastVerify)")
|
||
except RuntimeError:
|
||
v = None
|
||
iframe_gone = not any("rmc.bytedance.com" in t.get("url", "") for t in iframe_info())
|
||
if v and v != "null":
|
||
try:
|
||
j = json.loads(v)
|
||
if isinstance(j, str):
|
||
j = json.loads(j)
|
||
code = j.get("code")
|
||
msg = j.get("message", "")
|
||
print(f"RESULT code={code} msg={msg} data={j.get('data')}")
|
||
return "PASS" if code == 200 else str(code)
|
||
except ValueError:
|
||
print("verify 原文:", v[:150])
|
||
return "error"
|
||
if iframe_gone:
|
||
print("RESULT iframe 已销毁（大概率通过）")
|
||
return "PASS"
|
||
print("RESULT 无 verify 响应（拖动未被注册？）")
|
||
return "noresp"
|
||
|
||
|
||
async def main():
|
||
try:
|
||
n_q = int(sys.argv[1]) if len(sys.argv) > 1 else 4
|
||
max_score = float(sys.argv[2]) if len(sys.argv) > 2 else 5.5
|
||
except ValueError:
|
||
print("用法: python try/cold_start_attempt.py [max_questions] [max_score]", file=sys.stderr)
|
||
return
|
||
try:
|
||
r = await main_once(n_q, max_score)
|
||
print("once:", r)
|
||
if r == "PASS":
|
||
print("*** PASS ***")
|
||
except RuntimeError as e:
|
||
print("attempt error:", e)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|