chuanqi/cq2_sdk/scripts/cdp_puzzle_solve.py

"""Solve the geetest v4 PUZZLE slider properly:
  1. Screenshot the captcha widget region
  2. Pixel-analyze to find the puzzle GAP (the notch where the piece fits)
  3. Drag the slider to EXACTLY that x-offset (not a fixed distance)
  4. Human-like trajectory

Then click login and monitor for success + game launch.
"""
import asyncio
import base64
import io
import json
import math
import subprocess
import sys

import httpx
from PIL import Image
import websockets

USERNAME = "longwu988"
PASSWORD = "Longwu1988"
CDP = "http://127.0.0.1:9222"


async def get_targets():
    async with httpx.AsyncClient(timeout=5) as c:
        r = await c.get(f"{CDP}/json/list")
        return r.json()


class Page:
    def __init__(self, ws):
        self.ws = ws
        self.mid = 0
        self.events = []

    async def call(self, method, params=None):
        self.mid += 1
        await self.ws.send(json.dumps({"id": self.mid, "method": method, "params": params or {}}))
        while True:
            msg = json.loads(await self.ws.recv())
            if msg.get("id") == self.mid:
                return msg.get("result", {})
            elif "method" in msg:
                self.events.append(msg)

    async def eval(self, expr):
        r = await self.call("Runtime.evaluate", {"expression": expr, "returnByValue": True})
        return r.get("result", {}).get("value")

    async def mouse(self, typ, x, y):
        params = {"type": typ, "x": x, "y": y, "button": "left"}
        if typ == "mousePressed":
            params["clickCount"] = 1
            params["buttons"] = 1
        if typ == "mouseReleased":
            params["clickCount"] = 1
            params["buttons"] = 0
        await self.call("Input.dispatchMouseEvent", params)

    async def screenshot(self) -> Image.Image | None:
        r = await self.call("Page.captureScreenshot", {"format": "png"})
        if r.get("data"):
            return Image.open(io.BytesIO(base64.b64decode(r["data"])))
        return None


def find_gap_x(img: Image.Image, widget: dict) -> int | None:
    """Find the puzzle gap x-position in the widget screenshot.

    The gap is a rectangular notch with distinctly different brightness
    from the surrounding image. Scan the image area (top 60% of widget)
    column-by-column for a vertical brightness discontinuity.
    """
    # widget coords are in page space; screenshot is full page
    x0, y0 = int(widget["x"]), int(widget["y"])
    w, h = int(widget["w"]), int(widget["h"])
    # the puzzle image is typically the upper portion
    img_y0 = y0 + 40  # skip header
    img_y1 = y0 + int(h * 0.55)
    img_x0 = x0 + 5
    img_x1 = x0 + w - 5

    if img_x1 <= img_x0 or img_y1 <= img_y0:
        return None

    crop = img.crop((img_x0, img_y0, img_x1, img_y1)).convert("RGB")
    cw, ch = crop.size
    px = crop.load()

    # Column brightness signature
    col_avg = []
    for x in range(cw):
        s = 0
        for y in range(0, ch, 2):
            r, g, b = px[x, y]
            s += (r + g + b) / 3
        col_avg.append(s / max(1, ch // 2))

    # The gap: several consecutive columns much darker (or brighter) than neighbors.
    # Find the strongest left-edge discontinuity in the middle 80% of the image
    best_x = None
    best_score = 0
    for x in range(int(cw * 0.15), int(cw * 0.9)):
        left_avg = sum(col_avg[max(0, x - 6) : x]) / 6
        right_avg = sum(col_avg[x : x + 6]) / 6
        diff = abs(left_avg - right_avg)
        if diff > best_score:
            best_score = diff
            best_x = x

    if best_x is None or best_score < 8:  # too subtle
        return None

    # gap position in PAGE coordinates
    return img_x0 + best_x


def cq2_pid():
    try:
        out = subprocess.run(
            ["powershell", "-NoProfile", "-Command",
             "Get-Process cq2 -ErrorAction SilentlyContinue | Select-Object -First 1 -ExpandProperty Id"],
            capture_output=True, text=True, timeout=5)
        return out.stdout.strip() or None
    except Exception:
        return None


def gateway_conns():
    try:
        out = subprocess.run(
            ["powershell", "-NoProfile", "-Command",
             "(netstat -ano | Select-String '47.118.|:12000' | Measure-Object).Count"],
            capture_output=True, text=True, timeout=5)
        return int(out.stdout.strip() or 0)
    except Exception:
        return 0


async def main():
    targets = await get_targets()
    login = next((t for t in targets if "login" in t.get("url", "") and t["type"] == "page"), None)
    if not login:
        print("no login page")
        for t in targets:
            print(f"  [{t['type']}] {t['url'][:70]}")
        return 1

    async with websockets.connect(login["webSocketDebuggerUrl"], max_size=10**7) as ws:
        page = Page(ws)
        await page.call("Network.enable")

        # 1. ensure credentials
        await page.eval(f"""
        (function() {{
            const set = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
            const u = document.querySelector('#username');
            const p = document.querySelector('#password');
            if (u && !(u.value||'').length) {{ set.call(u, '{USERNAME}'); u.dispatchEvent(new Event('input', {{bubbles:true}})); }}
            if (p && !(p.value||'').length) {{ set.call(p, '{PASSWORD}'); p.dispatchEvent(new Event('input', {{bubbles:true}})); }}
            return 'ok';
        }})()
        """)

        # 2. click login
        print("clicking login...")
        await page.eval("""
        (function() {
            const btn = document.querySelector('button.ant-btn.login-btn, button.login-btn, button.ant-btn');
            if (btn) btn.click();
        })()
        """)
        await asyncio.sleep(2.5)

        # 3. captcha loop (up to 5 attempts)
        for attempt in range(5):
            # is captcha visible?
            gt = await page.eval("""
            (function() {
                const g = document.querySelector('.geetest_box, [class*=geetest_panel]');
                if (!g || g.offsetParent === null) return null;
                const r = g.getBoundingClientRect();
                const slider = g.querySelector('[class*=slider_button], .geetest_btn_drag, [class*=btn_drag], .geetest_btn');
                const sr = slider ? slider.getBoundingClientRect() : null;
                const state = (g.innerText||'').slice(0,100);
                return JSON.stringify({x:r.x, y:r.y, w:r.width, h:r.height,
                    sx: sr?sr.x+sr.width/2:null, sy: sr?sr.y+sr.height/2:null,
                    state});
            })()
            """)
            if not gt:
                print(f"attempt {attempt+1}: captcha not visible")
                # maybe logged in already?
                url = await page.eval("location.href")
                if "login" not in (url or ""):
                    print(f"*** NAVIGATED AWAY: {url} — LOGIN SUCCESS ***")
                    break
                break

            g = json.loads(gt)
            print(f"attempt {attempt+1}: captcha visible {g['w']:.0f}x{g['h']:.0f}")
            print(f"  state: {g.get('state','')[:60]}")

            if "验证失败" in g.get("state", "") or "请重新" in g.get("state", ""):
                print("  previous attempt failed; refreshing captcha...")
                await page.eval("""
                (function() {
                    const r = document.querySelector('[class*=refresh], .geetest_refresh, [class*=reset]');
                    if (r) r.click();
                })()
                """)
                await asyncio.sleep(1.5)

            # 4. screenshot & find the gap
            shot = await page.screenshot()
            if shot is None:
                print("  screenshot failed")
                continue
            shot.save(f"captcha_attempt{attempt+1}.png")

            gap_x = find_gap_x(shot, g)
            if gap_x is None:
                print("  gap not found — trying center heuristic")
                gap_x = int(g["x"] + g["w"] * 0.6)
            print(f"  gap at page-x={gap_x}")

            # 5. slider position
            if g.get("sx") is None:
                print("  slider handle not found")
                continue
            sx, sy = g["sx"], g["sy"]
            dist = gap_x - sx + 12  # small overshoot correction
            print(f"  slider at ({sx:.0f},{sy:.0f}), drag dist={dist:.0f}px")

            # 6. human-like drag
            await page.mouse("mouseMoved", sx, sy)
            await asyncio.sleep(0.12)
            await page.mouse("mousePressed", sx, sy)
            await asyncio.sleep(0.08)
            steps = 45
            for i in range(1, steps + 1):
                progress = i / steps
                eased = 1 - (1 - progress) ** 2.4
                cx = sx + dist * eased
                cy = sy + math.sin(i * 0.6) * 2.0
                await page.mouse("mouseMoved", cx, cy)
                await asyncio.sleep(0.024)
            await asyncio.sleep(0.10)
            # tiny settle-back (real hands overshoot then correct)
            back = 3
            await page.mouse("mouseMoved", sx + dist - back, sy + 1)
            await asyncio.sleep(0.08)
            await page.mouse("mouseReleased", sx + dist - back, sy + 1)
            print("  drag complete")

            await asyncio.sleep(3.5)

            # 7. check outcome
            url = await page.eval("location.href")
            if "login" not in (url or ""):
                print(f"\n*** LOGIN SUCCESS! url={url[:60]} ***")
                break
            state2 = await page.eval("""
            (document.querySelector('.geetest_box')||{innerText:''}).innerText.slice(0,80)
            """)
            print(f"  post-drag captcha state: {state2}")
            if "验证成功" in (state2 or "") or "成功" in (state2 or ""):
                print("  captcha SOLVED, waiting for redirect...")
                await asyncio.sleep(4)
                url2 = await page.eval("location.href")
                if "login" not in (url2 or ""):
                    print(f"*** LOGIN SUCCESS! url={url2[:60]} ***")
                    break

        # 8. if logged in: monitor for game launch
        url = await page.eval("location.href")
        if "login" not in (url or ""):
            print("\n=== monitoring 90s for game launch + traffic ===")
            for i in range(30):
                await asyncio.sleep(3)
                pid = cq2_pid()
                conns = gateway_conns()
                t = (i + 1) * 3
                flags = []
                if pid:
                    flags.append(f"CQ2={pid}")
                if conns:
                    flags.append(f"conns={conns}")
                print(f"  t+{t:>3}s {' '.join(flags)}")
                if pid and conns:
                    print("\n*** REAL TRAFFIC — capture is recording ***")
                    await asyncio.sleep(25)
                    return 0
        else:
            print("\nlogin did not complete")

    return 0


sys.exit(asyncio.run(main()))