chuanqi/cq2_sdk/scripts/cdp_recon.py

"""Drive 996box via CDP (Chrome DevTools Protocol):
  1. Connect to the login page (pc.hzwlan.com/login/account)
  2. Inspect the login form (field names/selectors)
  3. Screenshot for visual confirmation

Does NOT submit yet — recon only (we need the exact selectors first).
"""
import asyncio
import base64
import json
import sys

import httpx
import websockets


async def get_targets():
    async with httpx.AsyncClient(timeout=5) as c:
        r = await c.get("http://127.0.0.1:9222/json/list")
        return r.json()


async def cdp_eval(ws_url: str, expr: str) -> dict:
    async with websockets.connect(ws_url, max_size=10 * 1024 * 1024) as ws:
        await ws.send(json.dumps({
            "id": 1,
            "method": "Runtime.evaluate",
            "params": {"expression": expr, "returnByValue": True},
        }))
        while True:
            msg = json.loads(await ws.recv())
            if msg.get("id") == 1:
                return msg.get("result", {})


async def cdp_screenshot(ws_url: str, path: str) -> None:
    async with websockets.connect(ws_url, max_size=10 * 1024 * 1024) as ws:
        await ws.send(json.dumps({"id": 2, "method": "Page.captureScreenshot", "params": {"format": "png"}}))
        while True:
            msg = json.loads(await ws.recv())
            if msg.get("id") == 2:
                data = msg["result"]["data"]
                with open(path, "wb") as f:
                    f.write(base64.b64decode(data))
                return


async def main():
    targets = await get_targets()
    login_page = next((t for t in targets if "login" in t.get("url", "")), None)
    if not login_page:
        print("no login page target found")
        for t in targets:
            print(f"  [{t['type']}] {t['url'][:80]}")
        return 1
    ws_url = login_page["webSocketDebuggerUrl"]
    print(f"login page: {login_page['url']}")
    print(f"ws: {ws_url[:60]}...")
    print()

    # 1. Recon: form fields
    print("=== form recon ===")
    expr = """
    (function() {
        const inputs = Array.from(document.querySelectorAll('input')).map(i => ({
            tag: 'input', type: i.type, name: i.name, id: i.id,
            placeholder: i.placeholder, cls: i.className.slice(0,40),
            visible: i.offsetParent !== null
        }));
        const buttons = Array.from(document.querySelectorAll('button, [role=button], .btn, a[class*=btn]')).slice(0,12).map(b => ({
            tag: b.tagName, text: (b.innerText||'').trim().slice(0,30),
            cls: (b.className||'').toString().slice(0,40), id: b.id,
            visible: b.offsetParent !== null
        }));
        return JSON.stringify({inputs, buttons}, null, 1);
    })()
    """
    res = await cdp_eval(ws_url, expr)
    val = res.get("result", {}).get("value", "")
    if val:
        data = json.loads(val)
        print("--- inputs ---")
        for i in data["inputs"]:
            vis = "OK " if i["visible"] else "hid"
            print(f"  [{vis}] type={i['type']:<10} name={i['name']:<16} id={i['id']:<20} ph={i['placeholder'][:20]}")
        print("--- buttons ---")
        for b in data["buttons"]:
            vis = "OK " if b["visible"] else "hid"
            print(f"  [{vis}] {b['tag']:<8} '{b['text']}'  cls={b['cls'][:30]}")
    else:
        print("eval failed:", json.dumps(res)[:300])

    # 2. Screenshot
    await cdp_screenshot(ws_url, "cdp_login_page.png")
    print("\nscreenshot: cdp_login_page.png")
    return 0


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