chuanqi/cq2_sdk/scripts/cdp_login_diag.py

"""Diagnose why the login didn't complete after geetest passed.

Enables CDP Network domain, re-does the login flow (fill + click + slider),
and captures:
  - the actual /auth/v3/login/safe/byPwd request + response
  - any console errors
  - page error messages
"""
import asyncio
import json
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
        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)

    def network_events(self, method_filter=None):
        out = []
        for e in self.events:
            if e.get("method", "").startswith("Network."):
                if method_filter is None or method_filter in e["method"]:
                    out.append(e)
        return out


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

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

        # Enable network tracking FIRST
        await page.call("Network.enable")
        print("network tracking enabled")

        # 1. Check current page state — maybe already logged in / error showing?
        state = await page.eval("""
        (function() {
            const err = document.querySelector('.ant-message-error, [class*=error], .ant-form-item-explain-error');
            const msg = document.querySelector('.ant-message-notice, .ant-message');
            const u = document.querySelector('#username');
            const p = document.querySelector('#password');
            return JSON.stringify({
                errText: err ? err.innerText.slice(0,60) : '',
                msgText: msg ? msg.innerText.slice(0,60) : '',
                uLen: u ? (u.value||'').length : -1,
                pLen: p ? (p.value||'').length : -1,
                saveChecked: document.querySelector('#save') ? document.querySelector('#save').checked : null
            });
        })()
        """)
        print(f"page state: {state}")

        # 2. Fill if empty
        st = json.loads(state or "{}")
        if st.get("uLen", 0) < 3:
            await page.eval(f"""
            (function() {{
                const set = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
                const u = document.querySelector('#username');
                const p = document.querySelector('#password');
                set.call(u, '{USERNAME}');
                u.dispatchEvent(new Event('input', {{bubbles:true}}));
                set.call(p, '{PASSWORD}');
                p.dispatchEvent(new Event('input', {{bubbles:true}}));
                return 'filled';
            }})()
            """)
            print("credentials filled")

        # 3. Clear previous network events, click login
        page.events.clear()
        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: ' + btn.innerText.trim();
        })()
        """)
        print(f"login {clicked}")

        # 4. Wait for geetest, solve it
        await asyncio.sleep(2.5)
        for attempt in range(3):
            has_gt = await page.eval("!!document.querySelector('.geetest_box, [class*=geetest_btn]')")
            if not has_gt:
                print(f"attempt {attempt+1}: no geetest visible — checking network...")
                break
            print(f"attempt {attempt+1}: solving slider...")
            info = await page.eval("""
            (function() {
                const sel = ['.geetest_slider_button', '.geetest_btn_drag', '[class*=slider_button]', '.geetest_box .geetest_btn'];
                for (const s of sel) {
                    const el = document.querySelector(s);
                    if (el && el.offsetParent !== null) {
                        const r = el.getBoundingClientRect();
                        return JSON.stringify({x: r.x + r.width/2, y: r.y + r.height/2});
                    }
                }
                return null;
            })()
            """)
            if not info:
                print("  slider not found")
                break
            d = json.loads(info)
            x, y = d["x"], d["y"]
            import math
            await page.mouse("mouseMoved", x, y)
            await asyncio.sleep(0.15)
            await page.mouse("mousePressed", x, y)
            await asyncio.sleep(0.1)
            dist = 280
            for i in range(1, 41):
                progress = i / 40
                eased = 1 - (1 - progress) ** 2.2
                await page.mouse("mouseMoved", x + dist * eased, y + math.sin(i * 0.7) * 2.5)
                await asyncio.sleep(0.028)
            await asyncio.sleep(0.12)
            await page.mouse("mouseReleased", x + dist, y)
            print("  drag done")
            await asyncio.sleep(3)
            # check page state after slide
            after = await page.eval("""
            JSON.stringify({
                url: location.href.slice(0,70),
                err: (document.querySelector('.ant-message-error, [class*=error]')||{}).innerText || ''
            })
            """)
            print(f"  after: {after}")
            if "login" not in json.loads(after or '{"url":"login"}').get("url", ""):
                break

        # 5. Wait and collect network responses for byPwd
        print("\nwaiting 10s for responses...")
        await asyncio.sleep(10)

        # Find requestWillBeSent for byPwd, match to responseReceived
        reqs = {}
        for e in page.network_events():
            m = e["method"]
            p = e.get("params", {})
            if m == "Network.requestWillBeSent":
                url = p.get("request", {}).get("url", "")
                if "byPwd" in url or "login" in url.lower():
                    reqs[p["requestId"]] = {"url": url, "method": p["request"].get("method")}
            elif m == "Network.responseReceived":
                rid = p.get("requestId")
                if rid in reqs:
                    reqs[rid]["status"] = p["response"].get("status")
                    reqs[rid]["respUrl"] = p["response"].get("url", "")[:80]

        print(f"\n=== login-related network requests: {len(reqs)} ===")
        for rid, r in reqs.items():
            print(f"  [{r.get('method','?')}] {r.get('url','')[:75]}")
            print(f"      -> status={r.get('status','?')}")
            # fetch response body via Network.getResponseBody
            try:
                body = await page.call("Network.getResponseBody", {"requestId": rid})
                txt = body.get("body", "")
                if body.get("base64Encoded"):
                    import base64
                    txt = base64.b64decode(txt).decode("utf-8", errors="replace")
                print(f"      body: {txt[:300]}")
            except Exception as ex:
                print(f"      body err: {type(ex).__name__}")

        # 6. Final page state
        final = await page.eval("location.href + ' | ' + document.title")
        print(f"\nfinal page: {final}")

    return 0


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