AntiCoco/hellofresh/auth.py

"""Authentification HelloFresh via Playwright + capture du bearer token.

Stratégie (cf. plan) :
- **Login local d'abord** : sur le Mac, fenêtre visible (`headless=False`) → l'utilisateur
  se connecte (captcha/2FA gérés à la main). La session est persistée dans `.session/profile`.
- **Homelab** : `headless=True`, réutilise la session synchronisée. `HF_EMAIL`/`HF_PASSWORD`
  servent uniquement de fallback de re-login auto si la session a expiré.
- **Token** : on intercepte l'en-tête `Authorization: Bearer …` envoyé aux hôtes gateway et on
  le met en cache dans `.session/token.json`. `api.py` le réutilise pour les appels httpx.

Pattern persistant repris d'`Automood/scraper.py` (`launch_persistent_context`).
"""

from __future__ import annotations

import base64
import json
import os
import time
import urllib.parse
from pathlib import Path

from playwright.sync_api import sync_playwright

ROOT = Path(__file__).resolve().parent.parent
SESSION_DIR = ROOT / ".session"
PROFILE_DIR = SESSION_DIR / "profile"
TOKEN_CACHE = SESSION_DIR / "token.json"
STATE_PATH = SESSION_DIR / "storage_state.json"

BASE_URL = "https://www.hellofresh.fr"
# Page qui déclenche des appels gateway authentifiés (menu de la semaine).
MENU_PAGE = f"{BASE_URL}/my-account/deliveries/menu"
ACCOUNT_PAGE = f"{BASE_URL}/my-account"

ATTENTE_LOGIN_S = 180  # temps laissé pour un login manuel (captcha / 2FA)
TOKEN_TTL_S = 30 * 60  # on rafraîchit le token au-delà de 30 min par prudence


def _headless() -> bool:
    return os.environ.get("ANTICOCO_HEADLESS", "1") not in ("0", "false", "False", "")


def _channel() -> str | None:
    """Canal navigateur : 'chrome' en local (évite les blocages du Chromium bundlé),
    vide/None sur le homelab headless (image Docker = chromium Playwright).

    Défini par ANTICOCO_BROWSER_CHANNEL ('chrome', 'msedge', ou '' pour chromium).
    """
    ch = os.environ.get("ANTICOCO_BROWSER_CHANNEL", "")
    return ch or None


def _is_gateway_request(url: str) -> bool:
    return ("/gw/" in url or url.startswith("https://gw.")) and "hellofresh" in url


def _is_logged_in(page) -> bool:
    """Pas connecté = un champ mot de passe est visible (page de login).

    Détection volontairement indépendante de la locale (sélecteur CSS, pas de texte).
    """
    try:
        return page.locator('input[type="password"]').count() == 0
    except Exception:
        return False


def _auto_login(page) -> bool:
    """Tente un login automatique avec HF_EMAIL/HF_PASSWORD. Best-effort.

    Les sélecteurs exacts du formulaire HelloFresh sont à confirmer ; on cible les
    champs standards. En cas d'échec (captcha, sélecteurs changés), renvoie False et
    on retombe sur le login manuel.
    """
    email = os.environ.get("HF_EMAIL")
    password = os.environ.get("HF_PASSWORD")
    if not email or not password:
        return False
    try:
        page.fill('input[type="email"], input[name="email"], input#email', email, timeout=8000)
        page.fill('input[type="password"], input[name="password"]', password, timeout=8000)
        page.click('button[type="submit"], button[data-test-id="login-submit"]', timeout=8000)
        page.wait_for_timeout(4000)
        return _is_logged_in(page)
    except Exception:
        return False


def _open_context(pw):
    kwargs = dict(
        user_data_dir=str(PROFILE_DIR),
        headless=_headless(),
        locale="fr-FR",
        viewport={"width": 1280, "height": 900},
    )
    channel = _channel()
    if channel:
        kwargs["channel"] = channel
    return pw.chromium.launch_persistent_context(**kwargs)


def ensure_logged_in() -> bool:
    """Garantit une session connectée dans le profil persistant.

    - Si déjà connecté : retourne True immédiatement.
    - Sinon, tente l'auto-login (env) ; à défaut attend un login manuel (fenêtre visible).
    Retourne True si la session est établie.
    """
    SESSION_DIR.mkdir(parents=True, exist_ok=True)
    with sync_playwright() as pw:
        ctx = _open_context(pw)
        try:
            page = ctx.pages[0] if ctx.pages else ctx.new_page()
            page.goto(ACCOUNT_PAGE, wait_until="domcontentloaded", timeout=30000)
            page.wait_for_timeout(2000)
            if _is_logged_in(page):
                return True

            # Fallback 1 : auto-login si identifiants fournis.
            if _auto_login(page):
                page.wait_for_timeout(2000)
                return True

            # Fallback 2 : login manuel (uniquement utile en fenêtre visible).
            if _headless():
                raise RuntimeError(
                    "Session HelloFresh expirée et auto-login impossible en headless. "
                    "Refaire le login en local (ANTICOCO_HEADLESS=0) puis re-sync .session/."
                )
            debut = time.time()
            while time.time() - debut < ATTENTE_LOGIN_S:
                if _is_logged_in(page):
                    page.wait_for_timeout(2000)
                    return True
                page.wait_for_timeout(2000)
            return False
        finally:
            ctx.close()


def _capture_from_context(ctx, roll_state: bool) -> dict:
    """Charge le menu dans `ctx` et capture le bearer token (refraîchi par la SPA).

    `roll_state=True` : ré-exporte storage_state.json après coup, pour faire « rouler »
    la session (le refresh_token tourne ~60 j) sans intervention manuelle.
    """
    observed = {"token": None, "gateways": set()}
    page = ctx.pages[0] if ctx.pages else ctx.new_page()

    def on_request(req):
        try:
            if not _is_gateway_request(req.url):
                return
            base = req.url.split("/gw/")[0] + "/gw" if "/gw/" in req.url else req.url
            observed["gateways"].add(base)
            a = req.headers.get("authorization") or req.headers.get("Authorization")
            if a and a.lower().startswith("bearer "):
                observed["token"] = a.split(" ", 1)[1].strip()
        except Exception:
            pass

    page.on("request", on_request)
    page.goto(MENU_PAGE, wait_until="domcontentloaded", timeout=45000)
    # SPA lourde : on attend que les appels gateway (et un éventuel refresh) partent,
    # plutôt que networkidle qui ne se déclenche jamais.
    for _ in range(15):
        page.wait_for_timeout(1000)
        if observed["token"]:
            page.wait_for_timeout(1000)
            break

    if not observed["token"]:
        raise RuntimeError(
            "Aucun bearer token capturé (session expirée ? refaire le login + re-sync). "
        )

    if roll_state:
        try:
            ctx.storage_state(path=str(STATE_PATH))
        except Exception:
            pass

    result = {
        "token": observed["token"],
        "gateways": sorted(observed["gateways"]),
        "captured_at": time.time(),
    }
    TOKEN_CACHE.write_text(json.dumps(result, indent=2), encoding="utf-8")
    return result


def capture_token(force: bool = False) -> dict:
    """Renvoie un bearer token frais (et les hôtes gateway), via navigateur.

    Préfère `storage_state.json` (cookies 60 j) → fonctionne en **headless** sur le homelab :
    la SPA rafraîchit elle-même le token (contourne la protection anti-bot des endpoints
    OAuth bruts). À défaut, retombe sur le profil persistant (login interactif).
    """
    cached = _read_token_cache()
    if cached and not force and (time.time() - cached.get("captured_at", 0)) < TOKEN_TTL_S:
        return cached

    SESSION_DIR.mkdir(parents=True, exist_ok=True)
    with sync_playwright() as pw:
        if STATE_PATH.exists():
            launch = {"headless": _headless()}
            if _channel():
                launch["channel"] = _channel()
            browser = pw.chromium.launch(**launch)
            ctx = browser.new_context(storage_state=str(STATE_PATH), locale="fr-FR")
            try:
                return _capture_from_context(ctx, roll_state=True)
            finally:
                ctx.close()
                browser.close()
        ctx = _open_context(pw)
        try:
            return _capture_from_context(ctx, roll_state=False)
        finally:
            ctx.close()


def _read_token_cache() -> dict | None:
    if TOKEN_CACHE.exists():
        try:
            return json.loads(TOKEN_CACHE.read_text(encoding="utf-8"))
        except Exception:
            return None
    return None


def _jwt_exp(token: str) -> float:
    """Renvoie le timestamp d'expiration d'un JWT (0 si indéterminable)."""
    try:
        payload = token.split(".")[1]
        payload += "=" * (-len(payload) % 4)  # padding base64url
        data = json.loads(base64.urlsafe_b64decode(payload))
        return float(data.get("exp", 0))
    except Exception:
        return 0.0


def token_from_storage_state() -> str | None:
    """Extrait l'access_token du cookie `apiV2Auth` de .session/storage_state.json.

    Permet de s'authentifier SANS navigateur (utile en headless sur le homelab),
    tant que la session synchronisée n'a pas expiré. Renvoie None si absent/expiré.
    """
    if not STATE_PATH.exists():
        return None
    try:
        state = json.loads(STATE_PATH.read_text(encoding="utf-8"))
        for c in state.get("cookies", []):
            if c.get("name") == "apiV2Auth":
                data = json.loads(urllib.parse.unquote(c["value"]))
                tok = data.get("access_token")
                if tok and _jwt_exp(tok) > time.time() + 60:
                    return tok
    except Exception:
        return None
    return None


def get_token(force: bool = False) -> str:
    """Renvoie un bearer token valide, par ordre de préférence :

    1. token en cache encore frais ;
    2. cookie `apiV2Auth` de la session synchronisée (sans navigateur) ;
    3. capture via navigateur (login si besoin) — impossible en headless si session morte.
    """
    if not force:
        cached = _read_token_cache()
        if cached and (time.time() - cached.get("captured_at", 0)) < TOKEN_TTL_S:
            return cached["token"]
        tok = token_from_storage_state()
        if tok:
            return tok
    # Refresh nécessaire : si on a une session synchronisée, le navigateur (même headless)
    # la rafraîchit ; sinon login interactif via le profil persistant.
    if STATE_PATH.exists():
        return capture_token(force=force)["token"]
    ensure_logged_in()
    return capture_token(force=force)["token"]


def auth_status() -> dict:
    """État de connexion sans ouvrir de fenêtre si un token en cache est encore frais."""
    cached = _read_token_cache()
    if cached and (time.time() - cached.get("captured_at", 0)) < TOKEN_TTL_S:
        age = int(time.time() - cached["captured_at"])
        return {"logged_in": True, "source": "cache", "token_age_s": age, "gateways": cached.get("gateways", [])}
    if token_from_storage_state():
        return {"logged_in": True, "source": "storage_state"}
    try:
        ok = ensure_logged_in()
        return {"logged_in": bool(ok), "source": "browser"}
    except Exception as e:
        return {"logged_in": False, "error": str(e)}