"""冷启动单次尝试：触发验证码 → 采多题选最优 → 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())