杨豪
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 同步收尾
222 lines
8.4 KiB
Python
222 lines
8.4 KiB
Python
"""一条龙自动过验证码：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))
|