chuanqi/cq2_sdk/scripts/cdp_search_launch.py

"""Search for 玖扬复古176 via CDP, open its page, click 启动, monitor traffic.

Flow:
  1. On home page (pc.hzwlan.com/home/homePage), find the search input
  2. Type 玖扬复古176 (native setter + keyboard events to trigger suggestions)
  3. Click the matching game suggestion/card
  4. On the game detail page, click the 启动 button
  5. Monitor: cq.exe/cq2.exe spawn? gateway :12000 connections? capture lines?
"""
import asyncio
import base64
import json
import subprocess
import sys
import time

import httpx
import websockets

CDP = "http://127.0.0.1:9222"
GAME_NAME = "玖扬复古176"


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 click(self, x, y):
        await self.call("Input.dispatchMouseEvent", {"type": "mousePressed", "x": x, "y": y, "button": "left", "clickCount": 1, "buttons": 1})
        await asyncio.sleep(0.06)
        await self.call("Input.dispatchMouseEvent", {"type": "mouseReleased", "x": x, "y": y, "button": "left", "clickCount": 1, "buttons": 0})

    async def key(self, char, raw_key=None, code=None, key_code=None):
        params = {"type": "keyDown", "text": char, "key": raw_key or char}
        if code:
            params["code"] = code
        if key_code:
            params["windowsVirtualKeyCode"] = key_code
        await self.call("Input.dispatchKeyEvent", params)
        await asyncio.sleep(0.04)
        params["type"] = "keyUp"
        await self.call("Input.dispatchKeyEvent", 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 cq_procs():
    try:
        out = subprocess.run(
            ["powershell", "-NoProfile", "-Command",
             "Get-Process | Where-Object { $_.Name -match '^cq' } | ForEach-Object { \"$($_.Name):$($_.Id)\" }"],
            capture_output=True, text=True, timeout=5)
        return [l.strip() for l in out.stdout.splitlines() if l.strip()] or []
    except Exception:
        return []


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


def capture_lines():
    try:
        out = subprocess.run(
            ["powershell", "-NoProfile", "-Command",
             "(Get-Content 'D:\\传奇\\cq2_sdk\\real_login.jsonl' -ErrorAction SilentlyContinue | 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()
    home = next((t for t in targets if "homePage" in t.get("url", "") and t["type"] == "page"), None)
    if not home:
        # fallback: any hzwlan page that's not login
        home = next((t for t in targets if "hzwlan" in t.get("url", "") and "login" not in t["url"] and t["type"] == "page"), None)
    if not home:
        print("no home page target:")
        for t in targets:
            print(f"  [{t['type']}] {t['url'][:70]}")
        return 1

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

        # 1. locate search input
        sr = await page.eval("""
        (function() {
            const inputs = document.querySelectorAll('input');
            for (const inp of inputs) {
                const r = inp.getBoundingClientRect();
                if (r.width > 100 && r.y < 150 && inp.offsetParent) {
                    return JSON.stringify({x: r.x + r.width/2, y: r.y + r.height/2, w: r.width, ph: inp.placeholder || ''});
                }
            }
            return null;
        })()
        """)
        if not sr:
            print("search input not found")
            await page.screenshot("no_search.png")
            return 1
        s = json.loads(sr)
        print(f"search input @ ({s['x']:.0f},{s['y']:.0f}) w={s['w']:.0f} ph='{s['ph'][:20]}'")

        # 2. click it and type the game name
        await page.click(s["x"], s["y"])
        await asyncio.sleep(0.4)
        # clear existing
        await page.key("a", raw_key="a", code="KeyA", key_code=65)
        # type via input events (Chinese chars don't map to keys — use direct value set + suggestions trigger)
        typed = await page.eval(f"""
        (function() {{
            const inp = document.activeElement;
            if (!inp || inp.tagName !== 'INPUT') {{
                // find again
                const inputs = document.querySelectorAll('input');
                for (const i of inputs) {{
                    const r = i.getBoundingClientRect();
                    if (r.width > 100 && r.y < 150 && i.offsetParent) {{ inp = i; break; }}
                }}
            }}
            if (!inp) return 'no input';
            const set = Object.getOwnPropertyDescriptor(window.HTMLInputElement.prototype, 'value').set;
            set.call(inp, '{GAME_NAME}');
            inp.dispatchEvent(new Event('input', {{bubbles: true}}));
            inp.dispatchEvent(new Event('change', {{bubbles: true}}));
            inp.dispatchEvent(new KeyboardEvent('keyup', {{key: 'a', bubbles: true}}));
            return 'typed: ' + inp.value;
        }})()
        """)
        print(f"typed: {typed}")
        await asyncio.sleep(2.5)

        # 3. look for suggestions / search results with 玖扬
        sugg = await page.eval(f"""
        (function() {{
            const results = [];
            const els = document.querySelectorAll('*');
            for (const el of els) {{
                if (el.children.length <= 3) {{
                    const t = (el.innerText || '').trim();
                    if (t.includes('玖扬') && t.length < 80 && !results.some(r => r.text === t)) {{
                        const r = el.getBoundingClientRect();
                        if (r.width > 30 && r.height > 10 && r.y > 0) {{
                            results.push({{text: t.slice(0,50), x: r.x + r.width/2, y: r.y + r.height/2}});
                        }}
                    }}
                }}
                if (results.length >= 6) break;
            }}
            return JSON.stringify(results);
        }})()
        """)
        print(f"\n=== suggestions mentioning 玖扬 ===")
        try:
            sugs = json.loads(sugg or "[]")
            for sg in sugs:
                print(f"  '{sg['text'][:40]}' @ ({sg['x']:.0f},{sg['y']:.0f})")
        except Exception:
            print(sugg[:300])
            sugs = []

        await page.screenshot("search_results.png")

        if not sugs:
            # press Enter to submit search
            await page.key("\r", raw_key="Enter", code="Enter", key_code=13)
            print("no suggestions — pressed Enter to search")
            await asyncio.sleep(3.5)
            sugg2 = await page.eval(f"""
            (function() {{
                const results = [];
                const els = document.querySelectorAll('*');
                for (const el of els) {{
                    if (el.children.length <= 3) {{
                        const t = (el.innerText || '').trim();
                        if (t.includes('玖扬') && t.length < 80 && !results.some(r => r.text === t)) {{
                            const r = el.getBoundingClientRect();
                            if (r.width > 30 && r.height > 10 && r.y > 0) {{
                                results.push({{text: t.slice(0,50), x: r.x + r.width/2, y: r.y + r.height/2}});
                            }}
                        }}
                    }}
                    if (results.length >= 6) break;
                }}
                return JSON.stringify(results);
            }})()
            """)
            try:
                sugs = json.loads(sugg2 or "[]")
                print(f"after Enter: {len(sugs)} matches")
                for sg in sugs:
                    print(f"  '{sg['text'][:40]}' @ ({sg['x']:.0f},{sg['y']:.0f})")
            except Exception:
                sugs = []
            await page.screenshot("search_results2.png")

        if not sugs:
            print("game not found via search — aborting")
            return 1

        # 4. click the first 玖扬 match
        g0 = sugs[0]
        print(f"\nclicking '{g0['text'][:30]}'...")
        await page.click(g0["x"], g0["y"])
        await asyncio.sleep(4)

        # 5. on the game page, find 启动
        for round in range(4):
            launch = await page.eval("""
            (function() {
                const btns = document.querySelectorAll('button, [class*=btn], [class*=start], [class*=launch], a');
                const cands = [];
                for (const b of btns) {
                    const t = (b.innerText || '').trim();
                    if (/启动|开始游戏|进入游戏|继续游戏|开始/.test(t) && t.length < 12 && b.offsetParent) {
                        const r = b.getBoundingClientRect();
                        if (r.width > 40) {
                            cands.push({text: t, x: r.x + r.width/2, y: r.y + r.height/2, w: r.width});
                        }
                    }
                }
                return JSON.stringify(cands.slice(0, 5));
            })()
            """)
            try:
                lc = json.loads(launch or "[]")
            except Exception:
                lc = []
            url_now = await page.eval("location.href.slice(0,80)")
            print(f"round {round+1}: url={url_now}  launch_btns={len(lc)}")
            if lc:
                b0 = lc[0]
                print(f"  LAUNCH: '{b0['text']}' @ ({b0['x']:.0f},{b0['y']:.0f})")
                await page.screenshot("before_launch.png")
                await page.click(b0["x"], b0["y"])
                print("*** 启动 CLICKED ***")
                break
            await asyncio.sleep(3)
        else:
            print("no launch button found")
            await page.screenshot("no_launch_btn.png")
            return 1

    # 6. monitor
    print("\n=== monitoring 240s: procs / gw conns / capture ===")
    t0 = time.time()
    saw_traffic = False
    while time.time() - t0 < 240:
        await asyncio.sleep(5)
        el = int(time.time() - t0)
        procs = cq_procs()
        conns = gateway_conns()
        lines = capture_lines()
        flags = []
        if procs:
            flags.append("procs=" + ",".join(p.split(":")[0] for p in procs))
        if conns:
            flags.append(f"gw={conns}")
        if lines:
            flags.append(f"cap={lines}")
            saw_traffic = True
        print(f"  t+{el:>3}s {' '.join(flags) or '...'}")
        if lines >= 2:
            print("\n*** REAL TCP TRAFFIC CAPTURED ***")
            await asyncio.sleep(15)  # let more login frames accumulate
            return 0
        if procs and conns and el > 60:
            print("\n*** game connected to gateway ***")
            await asyncio.sleep(30)
            return 0

    print(f"\ndone. captured={saw_traffic}")
    return 0 if saw_traffic else 1


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