chuanqi/cq2_sdk/scripts/capture_bypwd_v2.py

"""Capture byPwd response via page.expect_response (works while page is still open).

Returns the actual response body so we can see if we got a real JWT.
"""
import asyncio
import json
import sys
from pathlib import Path

_HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(_HERE.parent / "src"))

from cq2_sdk.env_loader import load_env
load_env(_HERE.parent / ".env")

from cq2_sdk.store import UserStore, connection
from cq2_sdk.debug_log import DebugLogger

EDGE = r"C:\Program Files (x86)\Microsoft\Edge\Application\msedge.exe"
URL = "https://pc.hzwlan.com/login/account"


async def main():
    log = DebugLogger(extra={"task": "capture_bypwd_v2"})

    from playwright.async_api import async_playwright

    async with async_playwright() as p:
        browser = await p.chromium.launch(executable_path=EDGE, headless=True)
        ctx = await browser.new_context(viewport={"width": 1366, "height": 768})
        page = await ctx.new_page()

        await page.goto(URL, wait_until="networkidle", timeout=30000)
        await page.evaluate(
            "([u, p]) => {"
            "  const setVal = (sel, v) => {"
            "    const el = document.querySelector(sel);"
            "    Object.getOwnPropertyDescriptor("
            "        window.HTMLInputElement.prototype, 'value').set.call(el, v);"
            "    el.dispatchEvent(new Event('input', {bubbles: true}));"
            "  };"
            "  setVal('input#username', u);"
            "  setVal('input#password', p);"
            "  const btn = document.querySelector("
            "      'button.login-form_box__submit_button');"
            "  if (btn) { btn.disabled = false; btn.classList.remove('disabled'); }"
            "}",
            ["longwu988", "Longwu1988"],
        )
        await page.wait_for_timeout(300)

        # expect_response captures while page is alive; the response handle
        # is awaitable after the context exits.
        async with page.expect_response("**/auth/v3/login/safe/byPwd") as resp_info:
            await page.locator(
                "button.login-form_box__submit_button"
            ).first.click(force=True)
            # Don't wait_timeout here - we want to catch the response when it arrives
        response = await resp_info.value
        status = response.status
        body = await response.text()
        await browser.close()

    print(f"=== byPwd response ===")
    print(f"status: {status}")
    print(f"body ({len(body)} chars): {body[:1500]}")

    if not body:
        print("no body")
        return

    try:
        data = json.loads(body)
        print(f"\nparsed keys: {list(data.keys())}")
        token = data.get("data", {}).get("token", "")
        if token:
            print(f"\n*** TOKEN FOUND ({len(token)} chars) ***")
            print(f"  preview: {token[:50]}...")
            with connection() as conn:
                u = UserStore(conn).get_by_username("longwu988")
                if u:
                    UserStore(conn).update_token(user_id=u.id, token=token)
                    print(f"  saved to MySQL users.id={u.id}")
                else:
                    print("  user longwu988 not in MySQL; run scaffold first")
        else:
            print(f"\nno token in response: code={data.get('code')}, msg={data.get('msg')}")
    except Exception as e:
        print(f"\nJSON parse error: {e}")


asyncio.run(main())