"""一条龙自动过验证码：CDP websocket 直连，全程 <45s。 流程：轮询最新 net_*_get.json（由 captcha_net_capture.py 落盘） → 等大图/mark 落盘 → live_solve(tip_y 约束) → 拟人路径 → CDP 鼠标事件拖动 前置： 1. Chrome 带 --remote-debugging-port=9222 2. 验证码已弹出（登录页触发法） 3. try/captcha_net_capture.py 在后台跑（抓 get 响应与图片字节） 用法: .venv/bin/python try/auto_solve_drag.py [--max-runs 3] 每轮：点刷新（模拟坐标）→ 等 get → solve → 拖 → 读 verify 结果（从落盘 JSON）。 """ import asyncio import glob import json import os import sys import time import urllib.request from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "src")) sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "try")) import cv2 # noqa: E402 import websockets # noqa: E402 import importlib.util # noqa: E402 from human_drag import gen_human_path # noqa: E402 def _load_local(name: str): """从本目录加载模块（try/ 名与 Python 关键字冲突，无法静态导入）。""" spec = importlib.util.spec_from_file_location( name, Path(__file__).resolve().parent / f"{name}.py") if spec is None or spec.loader is None: raise ImportError(f"无法加载模块 {name}") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod live_solve = _load_local("live_solve").live_solve CDP_HTTP = "http://127.0.0.1:9222" CAPTURES = Path(__file__).resolve().parent / "captures" # 验证码 UI 常量（380x384 iframe） UI_IFRAME = (770, 305) # iframe 左上角视口坐标（如变了用 --iframe-x/y 覆盖） UI_SLIDER_BTN = (22 + 32, 287 + 20) # 滑块按钮中心（iframe 内坐标） UI_REFRESH = (30, 355) # 刷新按钮（iframe 内坐标） SCALE = 340 / 552 # UI 显示 / 原图 next_msg_id = [1000] async def cdp(ws, method, params=None, session_id=None): next_msg_id[0] += 1 mid = next_msg_id[0] msg = {"id": mid, "method": method} if params: msg["params"] = params if session_id: msg["sessionId"] = session_id await ws.send(json.dumps(msg)) # 等到对应 id 的响应（跳过事件） while True: try: raw = await ws.recv() m = json.loads(raw) except (json.JSONDecodeError, websockets.ConnectionClosed) as e: raise RuntimeError(f"CDP 消息处理失败: {e}") from e if m.get("id") == mid: if "error" in m: raise RuntimeError(f"{method}: {m['error']}") return m.get("result", {}) def latest_pair(before_ts): """找比 before_ts 新的 get.json 与对应图片对，返回 (ts, big, mark, tip_y) 或 None。""" for gj in sorted(glob.glob(str(CAPTURES / "net_*_get.json")), reverse=True): ts = os.path.basename(gj).split("_")[1] if ts <= before_ts: continue big = mark = None for f in glob.glob(f"{CAPTURES}/net_{ts}_*"): if f.endswith("get.json"): continue im = cv2.imread(f, cv2.IMREAD_UNCHANGED) if im is None: continue if im.shape[:2] == (344, 552): big = f elif im.shape[:2] == (110, 110): mark = f if big and mark: try: with open(gj) as fh: tip = json.load(fh)["data"]["question"]["tip_y"] except (OSError, KeyError, ValueError): continue return ts, big, mark, tip return None async def drag(ws, session_id, dist_ui, seed): try: start_x = float(UI_IFRAME[0] + UI_SLIDER_BTN[0]) start_y = float(UI_IFRAME[1] + UI_SLIDER_BTN[1]) except (TypeError, ValueError): start_x, start_y = 824.0, 612.0 pts = gen_human_path(dist_ui, duration=2600, seed=seed) mid_idx = len(pts) * 7 // 10 cx, cy = start_x, start_y # force=0.5：CDP 鼠标事件缺省 pressure=0，真鼠标恒 0.5（W3C UI Events）， # 不补齐会被 SDK 行为风控判 5014（见 docs/browser-automation/cdp-drag-force-pressure.md） await cdp(ws, "Input.dispatchMouseEvent", {"type": "mousePressed", "x": start_x, "y": start_y, "button": "left", "buttons": 1, "clickCount": 1, "force": 0.5}, session_id) await asyncio.sleep(0.3) for i, (dx, dy, dt) in enumerate(pts): cx += dx cy += dy await asyncio.sleep(dt / 1000.0) await cdp(ws, "Input.dispatchMouseEvent", {"type": "mouseMoved", "x": round(cx, 1), "y": round(cy, 1), "button": "left", "buttons": 1, "force": 0.5}, session_id) if i == mid_idx: await asyncio.sleep(0.6) await asyncio.sleep(0.3) await cdp(ws, "Input.dispatchMouseEvent", {"type": "mouseReleased", "x": round(cx, 1), "y": round(cy, 1), "button": "left", "buttons": 1, "clickCount": 1, "force": 0.5}, session_id) def verify_result(after_time): """读 verify 落盘 JSON（由 captcha_verify_watch.py / captcha_net_capture.py 写）。""" cands = sorted(glob.glob(str(CAPTURES / "net_*_verify.json")), reverse=True) if cands: try: mtime = os.path.getmtime(cands[0]) except OSError: return None if mtime > after_time: try: with open(cands[0]) as fh: return json.load(fh) except (OSError, ValueError): return None return None async def main(max_runs: int): try: with urllib.request.urlopen(f"{CDP_HTTP}/json/version", timeout=5) as r: ver = json.load(r) with urllib.request.urlopen(f"{CDP_HTTP}/json/list", timeout=5) as r: targets = json.load(r) except (OSError, ValueError) as e: print(f"无法连接 CDP {CDP_HTTP}: {e}", file=sys.stderr) return async with websockets.connect(ver["webSocketDebuggerUrl"], max_size=64 * 1024 * 1024) as ws: # attach 主页面 try: page = next(t for t in targets if t["type"] == "page" and "douyin.com" in t.get("url", "")) except StopIteration: print("未找到 douyin.com 页面", file=sys.stderr) return sid = (await cdp(ws, "Target.attachToTarget", {"targetId": page["id"], "flatten": True}))["sessionId"] last_ts = "" ok = fail = 0 for run in range(max_runs): print(f"===== run {run + 1}/{max_runs} =====", flush=True) # 点刷新（用 CDP 鼠标事件，坐标 = iframe + 刷新按钮） rx = UI_IFRAME[0] + UI_REFRESH[0] ry = UI_IFRAME[1] + UI_REFRESH[1] for typ in ("mousePressed", "mouseReleased"): await cdp(ws, "Input.dispatchMouseEvent", {"type": typ, "x": rx, "y": ry, "button": "left", "buttons": 1, "clickCount": 1}, sid) t0 = time.time() # 轮询新样本（最多 12s） got = None while time.time() - t0 < 12 and not got: await asyncio.sleep(0.5) got = latest_pair(last_ts) if not got: print("超时未拿到新题", flush=True) continue ts, big_f, mark_f, tip_y = got last_ts = ts result = live_solve(big_f, mark_f, tip_y) if not result: print("solve 失败", flush=True) continue print(f"[{ts}] tip_y={tip_y} x={result['x']} score={result['score']} " f"(+{time.time() - t0:.1f}s)", flush=True) await drag(ws, sid, result["x"] * SCALE, seed=result["x"]) print(f"拖动完成 (+{time.time() - t0:.1f}s)", flush=True) await asyncio.sleep(4) vr = verify_result(time.time() - 12) print("verify:", vr, flush=True) msg = (vr or {}).get("message", "") if "通过" in msg: ok += 1 break # 通过即止 fail += 1 print(f"完成：通过 {ok} / 失败 {fail}", flush=True) if __name__ == "__main__": import argparse ap = argparse.ArgumentParser() ap.add_argument("--max-runs", type=int, default=3) args = ap.parse_args() asyncio.run(main(args.max_runs))