`scripts/cdp_login_and_extract.py`

168 lines
#!/usr/bin/env python3
"""
Drive the debug Chrome on :9222 via raw CDP (websocket-client, no Playwright):
  1. Find the Spoonflower tab (or open one if missing)
  2. If not signed in, fill #email/#password and click #submit
  3. Read all Spoonflower cookies via Network.getAllCookies
  4. Save to ~/.spoonflower-cookies.json (Playwright cookie shape)

This sidesteps Playwright's Browser.setDownloadBehavior issue with
externally-launched Chrome.
"""
import json, os, sys, time, urllib.request, pathlib
import websocket  # type: ignore

DEBUG_PORT = 9222
COOKIES_PATH = pathlib.Path.home() / ".spoonflower-cookies.json"


def list_targets():
    with urllib.request.urlopen(f"http://127.0.0.1:{DEBUG_PORT}/json", timeout=4) as r:
        return json.loads(r.read())


def open_new_tab(url):
    with urllib.request.urlopen(
        f"http://127.0.0.1:{DEBUG_PORT}/json/new?{url}", timeout=4
    ) as r:
        return json.loads(r.read())


class CDP:
    def __init__(self, ws_url):
        self.ws = websocket.create_connection(ws_url, timeout=30)
        self._id = 0

    def call(self, method, params=None, timeout=20):
        self._id += 1
        mid = self._id
        self.ws.send(json.dumps({"id": mid, "method": method, "params": params or {}}))
        deadline = time.time() + timeout
        while time.time() < deadline:
            self.ws.settimeout(max(0.1, deadline - time.time()))
            try:
                msg = json.loads(self.ws.recv())
            except websocket.WebSocketTimeoutException:
                break
            if msg.get("id") == mid:
                if "error" in msg:
                    raise RuntimeError(f"CDP error {method}: {msg['error']}")
                return msg.get("result", {})
        raise TimeoutError(f"CDP {method} timed out")

    def close(self):
        try:
            self.ws.close()
        except Exception:
            pass


def get_creds():
    sys.path.insert(0, str(pathlib.Path(__file__).parent))
    from spoonflower_lib import creds
    return creds()


def find_spoonflower_target():
    targets = list_targets()
    for t in targets:
        if t.get("type") == "page" and "spoonflower" in t.get("url", ""):
            return t
    # No spoonflower tab — open one
    print("[cdp] no spoonflower tab — opening one", flush=True)
    open_new_tab("https://www.spoonflower.com/login")
    time.sleep(4)
    for t in list_targets():
        if t.get("type") == "page" and "spoonflower" in t.get("url", ""):
            return t
    return None


def main():
    email, password = get_creds()

    target = find_spoonflower_target()
    if not target:
        print(json.dumps({"ok": False, "error": "no spoonflower tab and could not open one"}))
        return 1

    cdp = CDP(target["webSocketDebuggerUrl"])
    try:
        cdp.call("Page.enable")
        cdp.call("Network.enable")

        url = target.get("url", "")
        print(f"[cdp] attached to {url}", flush=True)

        # If we're on /login, drive it
        if "/login" in url:
            print("[cdp] driving login form", flush=True)
            cdp.call("Runtime.evaluate", {
                "expression": (
                    f"document.querySelector('#email').value = {json.dumps(email)};"
                    f"document.querySelector('#password').value = {json.dumps(password)};"
                    "var f = document.querySelector('#email').form;"
                    "if (f) f.submit(); else document.querySelector('#submit').click();"
                ),
                "awaitPromise": False,
            })
            # Wait for navigation
            time.sleep(8)

        # Check signed-in via get-user-stats inline
        stats_js = """
        fetch('https://cart.spoonflower.com/api/spoonflower/get-user-stats', {credentials:'include'})
          .then(r => r.text()).then(t => t).catch(e => 'ERR:'+e)
        """
        r = cdp.call("Runtime.evaluate", {
            "expression": stats_js, "awaitPromise": True, "returnByValue": True,
        })
        body = r.get("result", {}).get("value", "")
        print(f"[cdp] get-user-stats body: {body[:300]}", flush=True)
        try:
            stats = json.loads(body).get("data_layer", {})
        except Exception:
            stats = {}
        user_id = stats.get("userID")
        print(f"[cdp] userID={user_id}  designForSale={stats.get('designForSale')}",
              flush=True)

        # Pull cookies regardless — even partial cookies help debug
        cks = cdp.call("Network.getAllCookies").get("cookies", [])
        # Normalize to Playwright shape
        pw_cookies = []
        for c in cks:
            if "spoonflower" not in c.get("domain", ""):
                continue
            pw_cookies.append({
                "name": c["name"],
                "value": c["value"],
                "domain": c["domain"],
                "path": c.get("path", "/"),
                "secure": bool(c.get("secure", True)),
                "httpOnly": bool(c.get("httpOnly", False)),
                "sameSite": c.get("sameSite", "Lax"),
                "expires": int(c.get("expires") or -1),
            })

        COOKIES_PATH.write_text(json.dumps(pw_cookies, indent=2))
        print(f"[cdp] wrote {len(pw_cookies)} spoonflower cookies to {COOKIES_PATH}",
              flush=True)

        # Print summary as last JSON line for the caller
        print(json.dumps({
            "ok": bool(user_id and user_id != "null"),
            "userID": user_id,
            "designForSale": stats.get("designForSale"),
            "numOrders": stats.get("numOrders"),
            "cookies_saved": len(pw_cookies),
            "cookies_path": str(COOKIES_PATH),
        }))
        return 0 if user_id else 2
    finally:
        cdp.close()


if __name__ == "__main__":
    sys.exit(main())