杨豪
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 同步收尾
100 lines
4.1 KiB
Python
100 lines
4.1 KiB
Python
# 监听 captcha/verify 请求+响应全文（含 postData），落盘 /tmp/verify_captures/
|
||
# 用途：对比真人 vs 自动化拖动的 verify 参数差异
|
||
import asyncio, json, time, urllib.request, os, sys
|
||
import websockets
|
||
|
||
CDP_HTTP = "http://127.0.0.1:9222"
|
||
OUT = "/tmp/verify_captures"
|
||
|
||
|
||
def _cdp_json(path: str):
|
||
"""拉取 CDP HTTP 接口的 JSON；失败返回 None。"""
|
||
try:
|
||
with urllib.request.urlopen(f"{CDP_HTTP}{path}", timeout=5) as r:
|
||
return json.load(r)
|
||
except (OSError, ValueError):
|
||
return None
|
||
|
||
|
||
async def main(minutes: float = 10.0) -> None:
|
||
ver = _cdp_json("/json/version")
|
||
if not ver or "webSocketDebuggerUrl" not in ver:
|
||
print("无法连接 CDP 9222（Chrome 需带 --remote-debugging-port=9222）", file=sys.stderr)
|
||
return
|
||
try:
|
||
os.makedirs(OUT, exist_ok=True)
|
||
except OSError as e:
|
||
print(f"无法创建输出目录 {OUT}: {e}", file=sys.stderr)
|
||
return
|
||
|
||
async with websockets.connect(ver["webSocketDebuggerUrl"], max_size=64 * 1024 * 1024) as ws:
|
||
msg_id = 10
|
||
|
||
async def send(method, params=None, sid=None):
|
||
nonlocal msg_id
|
||
msg_id += 1
|
||
m = {"id": msg_id, "method": method, "params": params or {}}
|
||
if sid:
|
||
m["sessionId"] = sid
|
||
await ws.send(json.dumps(m))
|
||
return msg_id
|
||
|
||
await send("Target.setAutoAttach", {"autoAttach": True, "waitForDebuggerOnStart": False, "flatten": True})
|
||
for t in (_cdp_json("/json/list") or []):
|
||
if t.get("type") in ("page", "iframe"):
|
||
await send("Target.attachToTarget", {"targetId": t["id"], "flatten": True})
|
||
|
||
pend = {}
|
||
deadline = time.time() + minutes * 60
|
||
print("watching verify (full)...", flush=True)
|
||
while time.time() < deadline:
|
||
try:
|
||
raw = await asyncio.wait_for(ws.recv(), timeout=1)
|
||
m = json.loads(raw)
|
||
except asyncio.TimeoutError:
|
||
continue
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if m.get("method") == "Target.attachedToTarget":
|
||
await send("Network.enable", sid=m["params"]["sessionId"])
|
||
meth = m.get("method", "")
|
||
if meth == "Network.requestWillBeSent" and "captcha/verify" in m["params"]["request"]["url"]:
|
||
ts = time.strftime("%H%M%S")
|
||
rec = {"ts": ts, "url": m["params"]["request"]["url"][:300],
|
||
"method": m["params"]["request"]["method"],
|
||
"post": m["params"]["request"].get("postData", "")}
|
||
rid = m["params"]["requestId"]
|
||
body_id = await send("Network.getResponseBody", {"requestId": rid}, sid=m.get("sessionId"))
|
||
rec["req_id"] = rid
|
||
rec["url_full"] = m["params"]["request"]["url"]
|
||
pend[body_id] = rec
|
||
# 同时存下请求体全文（含 captchaBody 轨迹密文）
|
||
path_req = f"{OUT}/verify_{ts}_req.json"
|
||
try:
|
||
with open(path_req, "w") as f:
|
||
json.dump(rec, f, ensure_ascii=False, indent=1)
|
||
except OSError as e:
|
||
print(f"落盘失败 {path_req}: {e}", file=sys.stderr)
|
||
print(f"[{ts}] VERIFY-REQ post_len={len(rec['post'])}", flush=True)
|
||
elif "id" in m and m["id"] in pend and "result" in m:
|
||
rec = pend.pop(m["id"])
|
||
rec["resp"] = m["result"].get("body", "")
|
||
path = f"{OUT}/verify_{rec['ts']}.json"
|
||
try:
|
||
with open(path, "w") as f:
|
||
json.dump(rec, f, ensure_ascii=False, indent=1)
|
||
except OSError as e:
|
||
print(f"落盘失败 {path}: {e}", file=sys.stderr)
|
||
print(f"[{rec['ts']}] VERIFY-RESP: {rec['resp'][:150]}", flush=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
minutes = float(sys.argv[1]) if len(sys.argv) > 1 else 10.0
|
||
except ValueError:
|
||
minutes = 10.0
|
||
try:
|
||
asyncio.run(main(minutes))
|
||
except KeyboardInterrupt:
|
||
pass
|