# /// script
# requires-python = ">=3.10"
# dependencies = ["playwright==1.55.0"]
# ///
"""
Veoluna: link your Garmin Connect account (feature P-18).

Opens Garmin's own sign-in page in a browser window. You sign in there (with the two-step code if
Garmin asks for one); your password never goes through this script or Veoluna. The script then
saves oauth1_token.json next to itself: choose that file in Veoluna (Account > Garmin Connect),
then delete it.

Usage:
    uv run garmin-login.py            # Garmin accounts everywhere except mainland China
    uv run garmin-login.py --china    # garmin.cn accounts

The first run downloads Chromium for Playwright (about 150 MB).

Adapted from the browser sign-in script used by zwift-offline
(https://gist.github.com/coleman8er/5c8e192d2aa3c8a3a6220c5702e8a5e6): same flow, standard-library
HTTP and OAuth 1 signing, only the OAuth 1 token is written, and nothing is printed that could be
copied by mistake.
"""

from __future__ import annotations

import argparse
import base64
import hashlib
import hmac
import json
import os
import re
import secrets
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

# The OAuth consumer of the Garmin Connect mobile app, as published by the garth project. The Veoluna
# server must be configured with the same one (GARMIN_CONSUMER_KEY / GARMIN_CONSUMER_SECRET).
CONSUMER_URL = "https://thegarth.s3.amazonaws.com/oauth_consumer.json"
USER_AGENT = "com.garmin.android.apps.connectmobile"
LOGIN_TIMEOUT_S = 300
TICKET = re.compile(r"ticket=(ST-[A-Za-z0-9\-]+)")


def pct(s: str) -> str:
    """RFC 3986 percent-encoding, as OAuth 1 requires."""
    return urllib.parse.quote(s, safe="-._~")


def oauth1_header(method: str, url: str, consumer: dict, token: dict | None = None, body: dict | None = None,
                  nonce: str | None = None, timestamp: int | None = None) -> str:
    """Authorization header for one request, HMAC-SHA1 (RFC 5849); query and form body are signed."""
    parts = urllib.parse.urlsplit(url)
    oauth = {
        "oauth_consumer_key": consumer["consumer_key"],
        "oauth_nonce": nonce or secrets.token_hex(16),
        "oauth_signature_method": "HMAC-SHA1",
        "oauth_timestamp": str(timestamp or int(time.time())),
        "oauth_version": "1.0",
    }
    if token:
        oauth["oauth_token"] = token["oauth_token"]
    params = list(oauth.items()) + urllib.parse.parse_qsl(parts.query, keep_blank_values=True) + list((body or {}).items())
    encoded = sorted((pct(k), pct(v)) for k, v in params)
    base_url = f"{parts.scheme}://{parts.netloc}{parts.path}"
    base = "&".join([method.upper(), pct(base_url), pct("&".join(f"{k}={v}" for k, v in encoded))])
    key = f"{pct(consumer['consumer_secret'])}&{pct(token['oauth_token_secret'] if token else '')}"
    signature = base64.b64encode(hmac.new(key.encode(), base.encode(), hashlib.sha1).digest()).decode()
    oauth["oauth_signature"] = signature
    return "OAuth " + ", ".join(f'{pct(k)}="{pct(v)}"' for k, v in oauth.items())


def http(method: str, url: str, headers: dict | None = None, data: bytes | None = None) -> str:
    req = urllib.request.Request(url, data=data, method=method, headers={"User-Agent": USER_AGENT, **(headers or {})})
    try:
        with urllib.request.urlopen(req, timeout=30) as res:
            return res.read().decode()
    except urllib.error.HTTPError as err:
        hint = " Garmin is limiting sign-ins: wait an hour and try again." if err.code == 429 else ""
        raise SystemExit(f"Garmin answered HTTP {err.code} for {urllib.parse.urlsplit(url).path}.{hint}") from None
    except urllib.error.URLError as err:
        raise SystemExit(f"Could not reach {urllib.parse.urlsplit(url).netloc}: {err.reason}") from None


def browser_login(domain: str) -> str:
    """Opens Garmin's sign-in page and waits for the service ticket that ends the sign-in."""
    from playwright.sync_api import Error as PlaywrightError, sync_playwright

    sso = f"https://sso.{domain}/sso"
    url = f"{sso}/embed?" + urllib.parse.urlencode({
        "id": "gauth-widget",
        "embedWidget": "true",
        "gauthHost": sso,
        "clientId": "GarminConnect",
        "locale": "en_US",
        "redirectAfterAccountLoginUrl": f"{sso}/embed",
        "service": f"{sso}/embed",
    })
    with sync_playwright() as p:
        try:
            browser = p.chromium.launch(headless=False)
        except PlaywrightError:
            print("Downloading the browser (first run only)…")
            subprocess.run([sys.executable, "-m", "playwright", "install", "chromium"], check=True)
            browser = p.chromium.launch(headless=False)
        page = browser.new_context().new_page()
        page.goto(url)
        print("\nA browser window is open on Garmin's sign-in page.")
        print("Sign in there (with the code if Garmin asks for one). It closes by itself when done.\n")
        deadline = time.time() + LOGIN_TIMEOUT_S
        ticket = None
        while time.time() < deadline and ticket is None:
            try:
                m = TICKET.search(page.url) or TICKET.search(page.content())
                if m:
                    ticket = m.group(1)
            except PlaywrightError:
                pass  # page between two loads
            if page.is_closed():
                break
            page.wait_for_timeout(500)
        browser.close()
    if ticket is None:
        raise SystemExit("No sign-in within 5 minutes (or the window was closed). Run the script again.")
    return ticket


def main() -> None:
    parser = argparse.ArgumentParser(description="Link a Garmin Connect account to Veoluna.")
    parser.add_argument("--china", action="store_true", help="Garmin account in mainland China (garmin.cn)")
    parser.add_argument("--out", type=Path, default=Path(__file__).resolve().parent / "oauth1_token.json",
                        help="where to save the token (default: next to this script)")
    args = parser.parse_args()
    domain = "garmin.cn" if args.china else "garmin.com"
    api = f"https://connectapi.{domain}"

    consumer = json.loads(http("GET", CONSUMER_URL))
    ticket = browser_login(domain)

    print("Signed in. Getting the token…")
    url = f"{api}/oauth-service/oauth/preauthorized?" + urllib.parse.urlencode({
        "ticket": ticket,
        "login-url": f"https://sso.{domain}/sso/embed",
        "accepts-mfa-tokens": "true",
    })
    oauth1 = dict(urllib.parse.parse_qsl(http("GET", url, {"Authorization": oauth1_header("GET", url, consumer)})))
    if not oauth1.get("oauth_token") or not oauth1.get("oauth_token_secret"):
        raise SystemExit("Garmin did not return a token. Run the script again.")
    oauth1["domain"] = domain

    # Same check as the server: the token must give an access token.
    url = f"{api}/oauth-service/oauth/exchange/user/2.0"
    body = {"mfa_token": oauth1["mfa_token"]} if oauth1.get("mfa_token") else {}
    oauth2 = json.loads(http("POST", url, {
        "Authorization": oauth1_header("POST", url, consumer, oauth1, body),
        "Content-Type": "application/x-www-form-urlencoded",
    }, urllib.parse.urlencode(body).encode()))
    profile = json.loads(http("GET", f"{api}/userprofile-service/socialProfile", {"Authorization": f"Bearer {oauth2['access_token']}"}) or "{}")

    out: Path = args.out
    out.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(out, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        json.dump(oauth1, f, indent=2)
    print(f"\nGarmin account: {profile.get('fullName') or profile.get('displayName') or 'signed in'}")
    print(f"Token saved to: {out}")
    print("\nNext: in Veoluna, open Account > Garmin Connect > Link with token file and choose this file.")
    print("Then delete it: anyone holding it can upload to (and read) your Garmin account.")


if __name__ == "__main__":
    main()