douyin-captcha/try/solo_attempt.py

"""单次干净尝试：刷新→采题→solve（score<=阈值才出手）→force=0.5 拖动→读 verify。

用法: python try/solo_attempt.py [wait_sec] [max_score]
频控敏感：每次尝试间隔 wait_sec（默认 600s）。
"""
import asyncio, json, sys, time, base64, random, 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

CDP = "http://127.0.0.1:9222"
IFRAME_X, IFRAME_Y = 770, 305
SCALE = 340 / 552
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;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);};
  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


async def one_attempt(ws, mid_ref, max_score):
    async def cdp(method, params=None, sid=None):
        mid_ref[0] += 1
        m = {"id": mid_ref[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_ref[0]:
                if "error" in r:
                    raise RuntimeError(f"{method}: {r['error']}")
                return r.get("result", {})

    targets = _json("/json/list") or []
    page = next((t for t in targets if t["type"] == "page" and "douyin" in t.get("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("页面或验证码 iframe 不在", flush=True)
        return None
    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")

    await jse(HOOK)
    for typ in ("mousePressed", "mouseReleased"):
        await cdp("Input.dispatchMouseEvent",
                  {"type": typ, "x": IFRAME_X + 30, "y": IFRAME_Y + 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("no get", flush=True)
        return None
    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"])
    print(f"tip_y={q['tip_y']} solve={ {k: res.get(k) for k in ('x','y','scale','score','confidence')} }", flush=True)
    if not res.get("ok") or res.get("confidence") != "OK" or res.get("score", 99) > max_score:
        print("低质量题，跳过拖动", flush=True)
        return None
    dist_ui = res["x"] * SCALE
    sx, sy = IFRAME_X + 54, IFRAME_Y + 307
    pts = gen_human_path(dist_ui, seed=random.randint(1, 9999))
    await cdp("Input.dispatchMouseEvent",
              {"type": "mousePressed", "x": sx, "y": sy, "button": "left",
               "buttons": 1, "clickCount": 1, "force": 0.5}, psid)
    await asyncio.sleep(0.25)
    cx, cy = sx, sy
    for dx, dy, dt in pts:
        cx += dx
        cy += dy
        await asyncio.sleep(dt / 1000)
        await cdp("Input.dispatchMouseEvent",
                  {"type": "mouseMoved", "x": round(cx, 1), "y": round(cy, 1),
                   "button": "left", "buttons": 1, "force": 0.5}, psid)
    await asyncio.sleep(0.35)
    await cdp("Input.dispatchMouseEvent",
              {"type": "mouseReleased", "x": round(cx, 1), "y": round(cy, 1),
               "button": "left", "buttons": 1, "clickCount": 1, "force": 0.5}, psid)
    await asyncio.sleep(3)
    v = await jse("JSON.stringify(window.__lastVerify)")
    print("verify:", (v or "null")[-90:], flush=True)
    return v


async def main():
    try:
        wait_sec = int(sys.argv[1]) if len(sys.argv) > 1 else 600
        max_score = float(sys.argv[2]) if len(sys.argv) > 2 else 4.2
    except ValueError:
        print("用法: python try/solo_attempt.py [wait_sec] [max_score]", file=sys.stderr)
        return
    ver = _json("/json/version")
    if not ver or not ver.get("webSocketDebuggerUrl"):
        print("无法连接 CDP", file=sys.stderr)
        return
    async with websockets.connect(ver["webSocketDebuggerUrl"], max_size=64 * 1024 * 1024) as ws:
        mid = [0]
        while True:
            try:
                v = await one_attempt(ws, mid, max_score)
                if v and '"code":200' in v:
                    print("*** PASS ***", flush=True)
                    return
            except Exception as e:
                print("attempt error:", e, flush=True)
            print(f"等待 {wait_sec}s（频控冷却）...", flush=True)
            await asyncio.sleep(wait_sec)


if __name__ == "__main__":
    asyncio.run(main())