chuanqi/cq2_sdk/scripts/cdp_a1_login.py

"""A1 password login automation: real Edge browser (Playwright) + jfbym cloud slider
recognition + Playwright realistic mouse drag (Bezier curve) + HTTP 996box byPwd.

Why this exists:
  Pure HTTP w-parameter generation gets rejected by 996box geetest server
  (result=fail: synthetic trajectory detected). Solution: a real browser drags the
  slider with human-like motion; jfbym decides *where* to drag, browser does
  *how* to drag. The captchaOutput from the page is then plugged into the
  996box byPwd HTTP call we already wrote.
"""
from __future__ import annotations

import argparse
import asyncio
import base64
import json
import os
import random
import sys
import time
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.jfbym import JfbymSliderSolver
from cq2_sdk.login.box_996 import Box996Error, CaptchaRequiredError, LOGIN_BASE
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"
LOGIN_URL = "https://pc.hzwlan.com/login/account"


def human_drag(start_x: int, start_y: int, end_x: int, end_y: int,
                steps: int = 25, jitter_px: float = 1.5) -> list[tuple[int, int, int]]:
    """Build a realistic mouse-drag trajectory: bezier-curve + variable speed + micro-jitter.

    Returns: list of (x, y, dt_ms) tuples for Input.dispatchMouseEvent sequence.
    """
    pts = []
    # bezier control points to make a slight arc (humans don't drag in straight line)
    cx1 = start_x + (end_x - start_x) * 0.3 + random.uniform(-15, 15)
    cy1 = start_y - random.uniform(8, 18)
    cx2 = start_x + (end_x - start_x) * 0.7 + random.uniform(-10, 10)
    cy2 = start_y + random.uniform(-5, 5)
    total_ms = random.randint(380, 650)  # human drag duration: 0.4-0.65s
    dt_per = total_ms // steps
    # variable speed: slower at start (overcome inertia), fast in middle, slow at end
    for i in range(steps + 1):
        t = i / steps
        # ease-in-out cubic
        s = 3*t*t - 2*t*t*t if t < 0.5 else 1 - (-2*t + 2)**3 / 2
        # de Casteljau bezier
        def bez(t, p0, p1, p2, p3):
            return (1-t)**3*p0 + 3*(1-t)**2*t*p1 + 3*(1-t)*t**2*p2 + t**3*p3
        bx = bez(s, start_x, cx1, cx2, end_x)
        by = bez(s, start_y, cy1, cy2, end_y)
        # micro jitter
        bx += random.uniform(-jitter_px, jitter_px)
        by += random.uniform(-jitter_px/2, jitter_px/2)
        # timing: extra dwell at start + end (humans pause to confirm)
        if i in (0, 1, steps - 1, steps):
            dt_per += random.randint(15, 45)
        pts.append((int(bx), int(by), int(dt_per)))
    return pts


async def solve_captcha_and_login(
    username: str, password: str, jfbym_token: str, log: DebugLogger
) -> str:
    from playwright.async_api import async_playwright

    solver = JfbymSliderSolver(token=jfbym_token)

    async with async_playwright() as p:
        # Use system Edge (real browser, full CDP stack)
        browser = await p.chromium.launch(
            executable_path=EDGE,
            headless=True,
            args=["--disable-blink-features=AutomationControlled"],
        )
        ctx = await browser.new_context(
            viewport={"width": 1366, "height": 768},
            user_agent="Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 "
                       "(KHTML, like Gecko) Chrome/108.0.0.0 Safari/537.36 "
                       "Edg/108.0.1462.54",
            locale="zh-CN",
        )
        # stealth: hide webdriver flag
        await ctx.add_init_script("""
            Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
        """)
        page = await ctx.new_page()
        log.info("opening_login_page", url=LOGIN_URL)
        await page.goto(LOGIN_URL, wait_until="domcontentloaded", timeout=30000)

        # page.evaluate takes ONE arg; bundle the two strings into a list and
        # destructure inside the JS function. Then set the React-friendly
        # native value via the prototype setter, dispatch input + change so
        # Ant Design's controlled input + Form re-validates and clears the
        # disabled class on the submit button.
        await page.evaluate(
            "([u, p]) => {"
            "  const setVal = (sel, v) => {"
            "    const el = document.querySelector(sel);"
            "    const nativeSetter = Object.getOwnPropertyDescriptor("
            "        window.HTMLInputElement.prototype, 'value').set;"
            "    nativeSetter.call(el, v);"
            "    el.dispatchEvent(new Event('input', {bubbles: true}));"
            "    el.dispatchEvent(new Event('change', {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'); }"
            "}",
            [username, password],
        )
        await page.wait_for_timeout(300)

        # Click via Playwright (DOM click still dispatches).
        login_btn = page.locator("button.login-form_box__submit_button").first
        await login_btn.click(force=True)
        log.info("login_button_clicked")

        # Wait for the geetest widget (iframe/canvas) to appear
        # geetest v4 typical: canvas with class .geetest_canvas_slice / .geetest_canvas_bg
        try:
            await page.wait_for_selector(
                ".geetest_canvas_bg, .geetest_slice, canvas",
                timeout=15000,
            )
        except Exception as e:
            log.warn("geetest_widget_not_found", error=str(e)[:80])
        log.info("geetest_widget_visible")

        # Let the puzzle fully render
        await asyncio.sleep(0.8)

        # 1) Screenshot bg image (the background with the gap)
        bg_path = _HERE / "jfbym_diag" / "a1_cdp_bg.png"
        bg_path.parent.mkdir(exist_ok=True)
        bg_locator = page.locator(".geetest_canvas_bg").first
        await bg_locator.screenshot(path=str(bg_path))
        log.info("bg_screenshot_saved", path=str(bg_path))

        # 2) Get slider track and pointer for the drag start
        slider_loc = page.locator(".geetest_canvas_slice, .geetest_slider_button").first
        box = await slider_loc.bounding_box()
        if not box:
            raise RuntimeError("slider track not found")
        start_x = int(box["x"] + box["width"] / 2)
        start_y = int(box["y"] + box["height"] / 2)
        log.info("slider_start", x=start_x, y=start_y)

        # 3) Upload bg to jfbym for distance
        with open(bg_path, "rb") as f:
            bg_bytes = f.read()
        gap_x = await asyncio.to_thread(
            solver.solve_distance_sync,
            base64.b64encode(bg_bytes).decode(),
            "true",  # extra=true mode: "background only" -> return gap x
        )
        # The slider starts at left edge (typically a few px in). gap_x is
        # the absolute x of the gap. Drag distance = gap_x - slider_start_x.
        drag_distance = int(gap_x) - start_x + int(box["x"])  # gap_x is page-relative
        if drag_distance <= 0:
            drag_distance = 50  # sane fallback
        log.info("jfbym_gap_calculated", gap_x=int(gap_x), drag=drag_distance)

        # 4) Realistic mouse drag using bezier + variable speed
        trajectory = human_drag(start_x, start_y, start_x + drag_distance, start_y)
        log.info("mouse_drag_start", points=len(trajectory), total_dx=drag_distance)
        await page.mouse.move(start_x, start_y)
        await page.mouse.down()
        for x, y, dt in trajectory:
            await page.mouse.move(x, y)
            await asyncio.sleep(dt / 1000.0)
        await page.mouse.up()
        log.info("mouse_drag_done")

        # 5) Wait for geetest verification (network round trip to gcaptcha4)
        # The seccode gets stored in window.captchaObj or similar
        await asyncio.sleep(3.5)  # generous — geetest verify can be 1-3s

        # 6) Extract seccode from the page's JS context
        seccode_json = await page.evaluate("""() => {
            // geetest v4 stores result in window.captchaObj.captchaResult
            // or accessible via the iframe's contentWindow
            if (window.captchaObj && window.captchaObj.captchaResult) {
                return JSON.stringify(window.captchaObj.captchaResult);
            }
            const ifr = document.querySelector('iframe[src*="geetest"]')
                     || document.querySelector('#geetest-iframe')
                     || document.querySelector('iframe');
            if (ifr && ifr.contentWindow) {
                try {
                    if (ifr.contentWindow.captchaObj) {
                        return JSON.stringify(ifr.contentWindow.captchaObj);
                    }
                } catch (e) { return JSON.stringify({error: e.message}); }
            }
            return JSON.stringify({error: 'captchaObj not found'});
        }""")
        log.info("seccode_extracted", preview=seccode_json[:200])
        seccode = json.loads(seccode_json)
        if "captcha_output" not in seccode:
            raise RuntimeError(f"failed to extract seccode: {seccode}")

        # 7) Close browser (don't need it anymore)
        await browser.close()

        # 8) POST to 996box byPwd with extracted captchaOutput
        captcha_payload = {
            "captchaId": seccode["captcha_id"],
            "captchaOutput": seccode["captcha_output"],
            "genTime": seccode["gen_time"],
            "lotNumber": seccode["lot_number"],
            "passToken": seccode["pass_token"],
        }
        captcha_b64 = base64.b64encode(
            json.dumps(captcha_payload, separators=(",", ":")).encode()
        ).decode()

        log.info("calling_996box_bypwd", captcha_id=seccode["captcha_id"])
        from cq2_sdk.login.box_996 import Box996Login
        box = Box996Login(
            geetest=None, device_id="00:15:5d:0a:04:00",
        )
        session = await box.login_with_password(
            username, password,
            captcha={
                "loginInfo": "",
                "channel_id": "",
                "game_id": "",
                "inviter_uid": "",
                "needConfirmLogin": True,
                "phone": "",
                "code": "",
                "_captcha_": captcha_b64,
            },
        )
        log.info("login_success", uid=session.uid)
        return session.token


async def main_async() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--user", default=os.environ.get("CQ2_ACCOUNT"))
    ap.add_argument("--password", default=os.environ.get("CQ2_PASSWORD"))
    ap.add_argument("--jfbym-token", default=os.environ.get("CQ2_JFBYM_TOKEN"))
    ap.add_argument("--check", action="store_true",
                    help="just verify environment (no browser launch)")
    args = ap.parse_args()

    log = DebugLogger(extra={"task": "cdp_a1_login", "user": args.user})

    if args.check:
        ok = True
        print("[check] playwright:", end=" ")
        try:
            from playwright.async_api import async_playwright
            print("OK")
        except ImportError:
            print("MISSING")
            ok = False
        print("[check] Edge binary:", end=" ")
        from pathlib import Path
        print("OK" if Path(EDGE).exists() else "MISSING")
        print("[check] jfbym token:", end=" ")
        print("OK" if args.jfbym_token else "MISSING (set CQ2_JFBYM_TOKEN)")
        return 0 if ok else 1

    if not all([args.user, args.password, args.jfbym_token]):
        print("ERROR: missing --user / --password / --jfbym-token")
        return 1

    try:
        token = await solve_captcha_and_login(
            args.user, args.password, args.jfbym_token, log,
        )
    except Exception as e:
        log.error("a1_failed", error=str(e)[:200])
        print(f"[FAIL] {type(e).__name__}: {e}")
        return 1

    # Persist
    with connection() as conn:
        u = UserStore(conn).get_by_username(args.user)
        if u is None:
            print(f"[FAIL] user {args.user!r} not in MySQL (run scaffold first)")
            return 1
        UserStore(conn).update_token(user_id=u.id, token=token)
    print(f"\n[OK] A1 password login via real browser + jfbym succeeded")
    print(f"  user    = {args.user}")
    print(f"  token   = {token[:30]}...")
    print(f"  saved to MySQL users.id={u.id}")
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main_async()))