chuanqi/cq2_sdk/scripts/capture_bypwd.py

"""Capture the byPwd response body to see if we got a real JWT.

Recent debug run showed POST .../byPwd was sent and returned 200, but
we never read the body. If 200 = JWT, save it; if 200 = captcha required,
note the error code.
"""
import asyncio
import base64
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.login.box_996 import Box996Login
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_response"})

    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()

        # Capture byPwd response (Playwright sync API's r.text() is a coroutine
        # and must be awaited; the event handler is sync, so we collect
        # the coroutine and run it after the page closes).
        bypwd_response = {}
        pending_text: list = []

        def on_response(r):
            if "/auth/v3/login/safe/byPwd" in r.url:
                bypwd_response["status"] = r.status
                bypwd_response["url"] = r.url
                pending_text.append(r.text())

        page.on("response", on_response)

        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)
        await page.locator("button.login-form_box__submit_button").first.click(force=True)
        await page.wait_for_timeout(8000)
        await browser.close()

    if pending_text:
        body = pending_text[0]
        if asyncio.iscoroutine(body):
            body = await body
    else:
        body = ""
    print(f"=== byPwd response ===")
    print(f"status: {bypwd_response.get('status')}")
    print(f"body ({len(body)} chars): {body[:1000]}")

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

    try:
        data = json.loads(body)
        print(f"\nparsed JSON keys: {list(data.keys())}")
        token = data.get('data', {}).get('token', '')
        if token:
            print(f"\n*** TOKEN FOUND ({len(token)} chars) ***")
            print(f"  {token[:40]}...")
            # Save to MySQL
            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"response code: {data.get('code')}, msg: {data.get('msg')}")
    except Exception as e:
        print(f"JSON parse error: {e}")


asyncio.run(main())