chuanqi/cq2_sdk/scripts/cdp_auto_login.py

"""Full auto-login via CDP including Geetest v4 slider solve.

Flow:
  1. (capture must be running in background)
  2. Fill #username / #password
  3. Click login button -> Geetest slider appears
  4. Solve the slider: find the drag handle, dispatch realistic mouse drag
     (Input.dispatchMouseEvent with human-like trajectory over ~1.5s)
  5. Monitor: login success (page navigates) -> find game -> click 启动
  6. cq2.exe spawns -> TCP login traffic hits our capture
"""
import asyncio
import base64
import json
import subprocess
import sys

import httpx
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

    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", {})

    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, button="left", buttons=1):
        params = {"type": typ, "x": x, "y": y, "button": button}
        if typ in ("mousePressed",):
            params["clickCount"] = 1
            params["buttons"] = buttons
        if typ in ("mouseReleased",):
            params["clickCount"] = 1
            params["buttons"] = 0
        await self.call("Input.dispatchMouseEvent", params)

    async def screenshot(self, path):
        r = await self.call("Page.captureScreenshot", {"format": "png"})
        if r.get("data"):
            with open(path, "wb") as f:
                f.write(base64.b64decode(r["data"]))


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 solve_geetest(page: Page) -> bool:
    """Drag the geetest slider with a human-like trajectory."""
    # Find the slider button position (geetest v4 renders .geetest_slider_button or similar)
    info = await page.eval("""
    (function() {
        // geetest v4 widget
        const candidates = [
            '.geetest_slider_button',
            '.geetest_btn_drag',
            '[class*=slider_button]',
            '[class*=btn_drag]',
            '.geetest_box .geetest_btn',
        ];
        for (const sel of candidates) {
            const el = document.querySelector(sel);
            if (el && el.offsetParent !== null) {
                const r = el.getBoundingClientRect();
                return JSON.stringify({sel, x: r.x + r.width/2, y: r.y + r.height/2, w: r.width});
            }
        }
        // fallback: any visible geetest element
        const g = document.querySelector('[class*=geetest]');
        if (g) {
            const r = g.getBoundingClientRect();
            return JSON.stringify({sel: 'generic-geetest', x: r.x + 50, y: r.y + r.height/2, w: 40});
        }
        return null;
    })()
    """)
    if not info:
        print("  no geetest slider found in DOM")
        return False

    d = json.loads(info)
    x, y = d["x"], d["y"]
    print(f"  slider @{d['sel']} ({x:.0f},{y:.0f})")

    # Track distance: typical geetest v4 slide is ~260-300px
    dist = 280

    # Human-like drag: accelerate then decelerate, slight y wobble
    await page.mouse("mouseMoved", x, y)
    await asyncio.sleep(0.15)
    await page.mouse("mousePressed", x, y)
    await asyncio.sleep(0.1)

    import math
    steps = 40
    for i in range(1, steps + 1):
        # ease-out curve (fast start, slow end) like a real finger
        progress = i / steps
        eased = 1 - (1 - progress) ** 2.2
        cx = x + dist * eased
        cy = y + math.sin(i * 0.7) * 2.5  # wobble
        await page.mouse("mouseMoved", cx, cy)
        await asyncio.sleep(0.028)  # ~1.1s total drag

    await asyncio.sleep(0.12)
    await page.mouse("mouseReleased", x + dist, y)
    print("  drag complete")
    await asyncio.sleep(2.5)

    # Check result: did the captcha pass?
    passed = await page.eval("""
    (function() {
        // success indicator: geetest result icon or the widget disappearing
        const ok = document.querySelector('.geetest_success, [class*=success], .geetest_result_success');
        const widget = document.querySelector('[class*=geetest_panel], .geetest_box');
        const retry = document.querySelector('[class*=retry], .geetest_reset');
        return JSON.stringify({
            successIcon: !!ok,
            widgetGone: !widget || widget.offsetParent === null,
            retryVisible: !!retry
        });
    })()
    """)
    print(f"  result: {passed}")
    try:
        r = json.loads(passed or "{}")
        return r.get("successIcon") or r.get("widgetGone")
    except Exception:
        return False


async def main() -> int:
    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 — maybe already logged in? targets:")
        for t in targets:
            print(f"  [{t['type']}] {t['url'][:70]}")
        return 1

    print(f"login page: {login['url'][:60]}")
    async with websockets.connect(login["webSocketDebuggerUrl"], max_size=10**7) as ws:
        page = Page(ws)

        # fill credentials
        print("filling credentials...")
        await page.eval(f"""
        (function() {{
            const u = document.querySelector('#username');
            const p = document.querySelector('#password');
            if (!u || !p) return 'missing fields';
            const set = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
            set.call(u, '{USERNAME}');
            u.dispatchEvent(new Event('input', {{bubbles:true}}));
            set.call(p, '{PASSWORD}');
            p.dispatchEvent(new Event('input', {{bubbles:true}}));
            return 'filled';
        }})()
        """)
        await asyncio.sleep(0.5)

        # click login
        clicked = await page.eval("""
        (function() {
            const btn = document.querySelector('button.ant-btn.login-btn, button.login-btn, button.ant-btn');
            if (!btn) return 'no button';
            btn.click();
            return 'clicked';
        })()
        """)
        print(f"login click: {clicked}")
        await asyncio.sleep(2.5)

        # geetest appears? try solving up to 3 times
        for attempt in range(3):
            has_gt = await page.eval("""
            !!document.querySelector('[class*=geetest], iframe[src*=geetest]')
            """)
            if not has_gt:
                print(f"attempt {attempt+1}: no geetest visible")
                break
            print(f"attempt {attempt+1}: solving geetest...")
            ok = await solve_geetest(page)
            if ok:
                print("geetest PASSED")
                break
            print("geetest retry...")
            await asyncio.sleep(1.5)

        # monitor for login success + game launch
        print()
        print("=== monitoring 120s: login -> game -> traffic ===")
        for i in range(40):
            await asyncio.sleep(3)
            state = await page.eval("""
            JSON.stringify({url: location.href.slice(0,70)})
            """)
            try:
                st = json.loads(state or "{}")
            except Exception:
                st = {}
            pid = cq2_pid()
            conns = gateway_conns()
            t = (i + 1) * 3
            flags = []
            if "login" not in st.get("url", "login"):
                flags.append("NAVIGATED!")
            if pid:
                flags.append(f"CQ2={pid}")
            if conns:
                flags.append(f"conns={conns}")
            print(f"  t+{t:>3}s {st.get('url','?')[:60]:<60} {' '.join(flags)}")
            if pid and conns >= 1:
                print()
                print("*** REAL GAME TRAFFIC LIVE — letting capture run 30 more seconds ***")
                await asyncio.sleep(30)
                return 0

    return 0


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