"""真人轨迹采集 + CDP 回放（hook 路线的落地实验） 原理： 1. rec 阶段：在验证码 iframe document 上挂 capture 阶段监听，记录真人拖动产生的 pointerdown/move/up + mousedown/move/up 完整事件流（含原生时序）。 2. replay 阶段：CDP Input.dispatchMouseEvent 按录制的坐标与相对时序重放。 CDP 事件走浏览器输入管道，isTrusted=true，事件密度/节奏与真人一致。 用法： .venv/bin/python try/replay_human.py record # 进入录制模式，等用户手动拖 .venv/bin/python try/replay_human.py replay # 刷新→solve→回放录制的真人轨迹 """ import asyncio import base64 import glob import json import os import sys import time import urllib.request 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")) CDP_HTTP = "http://127.0.0.1:9222" CAPT_DIR = Path(__file__).resolve().parent / "captures" REPLAY_FILE = Path(__file__).resolve().parent / "human_trace.json" IFRAME_X, IFRAME_Y = 770, 305 # 验证码 iframe 左上角视口坐标 SCALE = 340 / 552 # UI 显示 / 原图 def _json(path): try: with urllib.request.urlopen(f"{CDP_HTTP}{path}", timeout=5) as r: return json.load(r) except (OSError, ValueError): return None HOOK_JS = """ (() => { if (window.__traceHook) return 'already'; window.__traceHook = 1; window.__trace = []; const types = ['pointerdown','pointermove','pointerup', 'mousedown','mousemove','mouseup']; for (const t of types) { document.addEventListener(t, e => { window.__trace.push({ t, x: e.clientX, y: e.clientY, ts: Math.round(e.timeStamp * 100) / 100, buttons: e.buttons, trusted: e.isTrusted, pressure: e.pressure || 0, movementX: e.movementX || 0, movementY: e.movementY || 0, }); }, {capture: true, passive: true}); } return 'trace hook on'; })() """ async def main(mode: str): ver = _json("/json/version") targets = _json("/json/list") or [] if not ver: print("CDP 9222 不可用") return page = next((t for t in targets if t["type"] == "page" and "douyin" in t["url"]), None) ifr = next((t for t in targets if "rmc.bytedance.com" in t.get("url", "")), None) if not page or not ifr: print("找不到 douyin 页面或验证码 iframe") 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: 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"] 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") # 强制重挂（页面可能刷新过，旧 hook 已失效；先拆旧 listener） await jse(""" (() => { window.__trace = []; if (window.__traceFn) { for (const [t, f] of window.__traceFn) document.removeEventListener(t, f, {capture: true}); } window.__traceFn = []; const types = ['pointerdown','pointermove','pointerup','mousedown','mousemove','mouseup']; for (const t of types) { const f = e => { window.__trace.push({t, x: e.clientX, y: e.clientY, ts: Math.round(e.timeStamp*100)/100, buttons: e.buttons, trusted: e.isTrusted}); }; document.addEventListener(t, f, {capture: true, passive: true}); window.__traceFn.push([t, f]); } return 'rehooked'; })() """) if mode == "record": print("== 录制模式：请在 20s 内用鼠标手动完成一次滑块拖动 ==") await asyncio.sleep(20) raw = await jse("JSON.stringify(window.__trace || [])") try: trace = json.loads(raw or "[]") except (TypeError, ValueError) as e: print(f"trace 解析失败: {e}") return mv = [e for e in trace if e["t"] == "mousemove"] dn = [e for e in trace if e["t"] == "mousedown"] print(f"共 {len(trace)} 事件: mousedown={len(dn)} mousemove={len(mv)}") if len(dn) >= 1 and len(mv) > 10: # 以第一次 mousedown 为拖动起点，截取到 mouseup start_ts = dn[0]["ts"] ups = [e["ts"] for e in trace if e["t"] == "mouseup" and e["ts"] > start_ts] end_ts = ups[0] if ups else trace[-1]["ts"] drag = [e for e in trace if start_ts - 50 <= e["ts"] <= end_ts] try: with open(REPLAY_FILE, "w") as f: json.dump({"iframe_origin": [IFRAME_X, IFRAME_Y], "events": drag}, f) except OSError as e: print(f"保存失败: {e}") return print(f"已保存 {len(drag)} 事件（{end_ts - start_ts:.0f}ms）→ {REPLAY_FILE}") else: print("没采到拖动，重试") return # ---- replay 模式 ---- if not REPLAY_FILE.exists(): print("没有录制文件，先跑 record") return try: with open(REPLAY_FILE) as f: rec = json.load(f) except (OSError, ValueError) as e: print(f"读取录制文件失败: {e}") return events = rec["events"] ox, oy = rec["iframe_origin"] # 找 mousedown 起点 dn = next(e for e in events if e["t"] == "mousedown") start_x, start_y = ox + dn["x"], oy + dn["y"] t0 = dn["ts"] ups = [e["ts"] for e in events if e["t"] == "mouseup"] total_ms = (ups[0] - t0) if ups else 1000 print(f"回放 {len(events)} 事件，拖动 {total_ms:.0f}ms") # 刷新拿新题 rx, ry = IFRAME_X + 30, IFRAME_Y + 355 for typ in ("mousePressed", "mouseReleased"): await cdp("Input.dispatchMouseEvent", {"type": typ, "x": rx, "y": ry, "button": "left", "buttons": 1, "clickCount": 1}, psid) got = None t0w = time.time() while time.time() - t0w < 14 and not got: await asyncio.sleep(0.5) fs = sorted(glob.glob(str(CAPT_DIR / "net_*_get.json")), reverse=True) if fs and time.time() - os.path.getmtime(fs[0]) < 10: ts = os.path.basename(fs[0]).split("_")[1] await asyncio.sleep(2.5) jpgs = [f for f in glob.glob(str(CAPT_DIR / f"net_{ts}_*")) if f.endswith((".jpg", ".jpeg"))] pngs = [f for f in glob.glob(str(CAPT_DIR / f"net_{ts}_*")) if f.endswith(".png")] if jpgs and pngs: got = (ts, jpgs[0], pngs[0]) if not got: print("没等到题") return ts, big_f, mark_f = got try: with open(f"{CAPT_DIR}/net_{ts}_get.json") as f: tip_y = json.load(f)["data"]["question"]["tip_y"] except (OSError, KeyError, ValueError) as e: print(f"get.json 读取失败: {e}") return from solve import solve try: with open(big_f, "rb") as f: big_b64 = base64.b64encode(f.read()).decode() with open(mark_f, "rb") as f: mark_b64 = base64.b64encode(f.read()).decode() except OSError as e: print(f"图片读取失败: {e}") return result = solve(big_b64, mark_b64) print(f"[{ts}] tip_y={tip_y} solve:", {k: result[k] for k in ("ok", "x", "score") if k in result}) if not result.get("ok"): return dist_ui = result["x"] * SCALE print(f"需拖 {dist_ui:.1f}px(UI)，录制拖动 {total_ms:.0f}ms") # 回放：mousedown → 按 ts 差调度 mousemove → mouseup await cdp("Input.dispatchMouseEvent", {"type": "mousePressed", "x": start_x, "y": start_y, "button": "left", "buttons": 1, "clickCount": 1}, psid) t_start = time.time() for e in events: if e["t"] != "mousemove" or e["ts"] <= t0: continue delay = (e["ts"] - t0) / 1000.0 - (time.time() - t_start) if delay > 0: await asyncio.sleep(delay) await cdp("Input.dispatchMouseEvent", {"type": "mouseMoved", "x": ox + e["x"], "y": oy + e["y"], "button": "left", "buttons": 1}, psid) await asyncio.sleep(0.25) await cdp("Input.dispatchMouseEvent", {"type": "mouseReleased", "x": ox + events[-1]["x"], "y": oy + events[-1]["y"], "button": "left", "buttons": 1, "clickCount": 1}, psid) print("回放完成") await asyncio.sleep(4) if __name__ == "__main__": mode = sys.argv[1] if len(sys.argv) > 1 else "record" asyncio.run(main(mode))