fetch fresh cielo home tokens (needs playwright)

get_cielo_tokens.py

#!/usr/bin/env python3
"""Fetch fresh Cielo Home tokens for the cielo_home custom integration.

Cielo's web login (home.cielowigle.com) is protected by a reCAPTCHA, so the
tokens the integration needs cannot be obtained with a plain HTTP request.
This script opens a real browser, lets you log in (solving the captcha if it
appears), then captures and prints the five values you paste into the
integration's config flow:

    access_token, refresh_token, session_id, user_id, x_api_key

The x_api_key in particular is a per-session value returned in the login
response, which is why it has to be captured from a real login rather than
hard-coded.

Setup (one time):

    pip install playwright
    playwright install chromium

Usage:

    python get_cielo_tokens.py
    python get_cielo_tokens.py --email you@example.com   # pre-fill email
    python get_cielo_tokens.py --no-verify               # skip the test API call

A browser window opens. Your credentials are pre-filled if provided; otherwise
type them in. Tick the captcha box if it shows up, then click Login. The script
prints the values and exits.
"""

from __future__ import annotations

import argparse
import getpass
import json
import sys
import time

LOGIN_PAGE = "https://home.cielowigle.com/"
API_HOST = "https://api.smartcielo.com"
LOGIN_TIMEOUT_S = 300  # how long to wait for a successful login

# Best-effort selectors for the login form. If they miss, just type manually.
USER_SELECTOR = (
    "input[type='email'], input[name*='user' i], "
    "input[id*='user' i], input[type='text']"
)
PASS_SELECTOR = "input[type='password']"


def extract_tokens(login_body: dict) -> dict | None:
    """Pull the five config values out of a /auth/login response body."""
    data = login_body.get("data") or {}
    user = data.get("user") or {}
    if not user.get("accessToken"):
        return None
    return {
        "access_token": user.get("accessToken", ""),
        "refresh_token": user.get("refreshToken", ""),
        "session_id": user.get("sessionId", ""),
        "user_id": user.get("userId", ""),
        # per-session key returned alongside the user object
        "x_api_key": data.get("x-api-key", ""),
    }


def verify(request_ctx, tokens: dict) -> str:
    """Hit /web/devices with the captured tokens to confirm they work."""
    resp = request_ctx.get(
        f"{API_HOST}/web/devices?limit=420",
        headers={
            "authorization": tokens["access_token"],
            "x-api-key": tokens["x_api_key"],
            "content-type": "application/json; charset=UTF-8",
            "origin": LOGIN_PAGE.rstrip("/"),
            "referer": LOGIN_PAGE,
        },
    )
    if resp.status != 200:
        return f"verify FAILED: HTTP {resp.status}"
    body = resp.json()
    if body.get("status") != 200:
        return f"verify FAILED: {body.get('message')}"
    devices = (body.get("data") or {}).get("listDevices") or []
    names = ", ".join(
        f"{d.get('deviceName')} ({d.get('deviceType')}, applianceId {d.get('applianceId')})"
        for d in devices
    )
    return f"verify OK: {len(devices)} device(s): {names or '(none)'}"


def main() -> int:
    parser = argparse.ArgumentParser(description="Grab fresh Cielo Home tokens.")
    parser.add_argument("--email", default=None, help="email to pre-fill")
    parser.add_argument("--password", default=None, help="password to pre-fill (insecure on shared machines)")
    parser.add_argument("--no-verify", action="store_true", help="skip the test API call")
    parser.add_argument("--headless", action="store_true", help="run without a visible window (captcha will likely fail)")
    args = parser.parse_args()

    try:
        from playwright.sync_api import sync_playwright
    except ImportError:
        print("Playwright is not installed. Run:\n  pip install playwright\n  playwright install chromium", file=sys.stderr)
        return 1

    email = args.email or input("Cielo email: ").strip()
    password = args.password or getpass.getpass("Cielo password (hidden): ")

    captured: dict = {}

    def on_response(resp):
        if "/auth/login" not in resp.url or resp.status != 200:
            return
        try:
            body = resp.json()
        except Exception:
            return
        tokens = extract_tokens(body)
        if tokens:
            captured["tokens"] = tokens

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=args.headless)
        context = browser.new_context()
        page = context.new_page()
        page.on("response", on_response)

        print(f"Opening {LOGIN_PAGE} ...")
        page.goto(LOGIN_PAGE, wait_until="domcontentloaded")
        page.wait_for_timeout(2500)

        # Best-effort pre-fill; ignore failures and let the user type.
        try:
            user_field = page.locator(USER_SELECTOR).first
            if user_field.count():
                user_field.fill(email)
            pass_field = page.locator(PASS_SELECTOR).first
            if pass_field.count():
                pass_field.fill(password)
        except Exception:
            pass

        print("\n>>> In the browser: tick the captcha box if it appears, then click Login.")
        print(f">>> Waiting up to {LOGIN_TIMEOUT_S}s for a successful login...\n")

        deadline = time.time() + LOGIN_TIMEOUT_S
        while time.time() < deadline and "tokens" not in captured:
            page.wait_for_timeout(1000)

        if "tokens" not in captured:
            print("Did not capture a successful login. Check your credentials/captcha and try again.", file=sys.stderr)
            browser.close()
            return 1

        tokens = captured["tokens"]

        verify_msg = "verify skipped"
        if not args.no_verify:
            try:
                verify_msg = verify(context.request, tokens)
            except Exception as e:  # noqa: BLE001
                verify_msg = f"verify errored: {e}"

        browser.close()

    print("\n" + "=" * 70)
    print("Paste these into Home Assistant:")
    print("  Settings -> Devices & Services -> Cielo Home -> Configure")
    print("=" * 70)
    for field in ("access_token", "refresh_token", "session_id", "user_id", "x_api_key"):
        print(f"{field}: {tokens[field]}")
    print("=" * 70)
    print(verify_msg)
    print("\nAlso copy/paste-friendly JSON:")
    print(json.dumps(tokens))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())