commit b881111504a167d5386e039d2da3f6266c447d24 Author: jerem Date: Mon Jun 15 22:09:11 2026 +0200 AntiCoco: serveur MCP HelloFresh sans noix de coco - Auth Playwright (login local, session persistee, capture du bearer token) - Client httpx vers l'API interne (endpoints via discover_api.py) - Filtre d'exclusion insensible aux accents (coco & co) - Serveur FastMCP (streamable-http) + outils hf_* - Docker + compose pour deploiement homelab diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..c210a57 --- /dev/null +++ b/.env.example @@ -0,0 +1,11 @@ +# Compte HelloFresh (région France : hellofresh.fr) +# Optionnels : utilisés seulement comme fallback de re-login auto si la session +# Playwright (.session/) a expiré. Le login principal se fait en local, fenêtre visible. +HF_EMAIL= +HF_PASSWORD= + +# Port d'écoute du serveur MCP (streamable-http). Exposé sur 127.0.0.1 côté homelab. +ANTICOCO_PORT=9200 + +# headless=1 sur le homelab (pas d'écran), headless=0 en local pour le 1er login. +ANTICOCO_HEADLESS=1 diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..7132f64 --- /dev/null +++ b/.gitignore @@ -0,0 +1,15 @@ +# Secrets & session — JAMAIS versionnés (syncés à la main vers le homelab) +.env +.session/ + +# Données découvertes localement (regénérables via tools/discover_api.py) +config/endpoints.json + +# Python +__pycache__/ +*.pyc +.venv/ +venv/ + +# OS +.DS_Store diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..2c54a0d --- /dev/null +++ b/Dockerfile @@ -0,0 +1,16 @@ +# Image Playwright officielle : Chromium + polices + deps système déjà présents. +FROM mcr.microsoft.com/playwright/python:v1.49.1-noble + +WORKDIR /app + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY . . + +ENV PYTHONUNBUFFERED=1 +# Sur le homelab : headless obligatoire (pas d'écran). +ENV ANTICOCO_HEADLESS=1 +EXPOSE 9200 + +CMD ["python", "server.py"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..05e4c39 --- /dev/null +++ b/README.md @@ -0,0 +1,90 @@ +# AntiCoco 🥥🚫 + +Serveur **MCP** qui donne accès à ton compte **HelloFresh** (France, `hellofresh.fr`) pour : +lire le menu de la semaine, **exclure des ingrédients** (la **noix de coco** en priorité), +**proposer une shortlist** de recettes, puis **enregistrer ta sélection** après confirmation. + +Client MCP visé : **Hermes** (Nous Research), qui tourne sur le même homelab. + +> ⚠️ HelloFresh n'a pas d'API publique. AntiCoco s'appuie sur l'API interne `gw/` du site +> (non documentée, susceptible de changer) — **usage strictement personnel**. + +## Architecture + +``` +Hermes ──HTTP──▶ server.py (FastMCP, :9200/mcp) + ├─ hellofresh/auth.py login Playwright + capture du bearer token + ├─ hellofresh/api.py appels httpx vers le gateway HelloFresh + ├─ hellofresh/filter.py exclusion (coco !) + scoring préférences + └─ config/ excludes.json · prefs.json · endpoints.json +``` + +## Mise en route + +### 1. Installer +```bash +pip install -r requirements.txt +playwright install chromium +cp .env.example .env # remplir HF_EMAIL / HF_PASSWORD (optionnels, fallback re-login) +``` + +### 2. Découvrir les endpoints (étape 0, en local, fenêtre visible) +```bash +ANTICOCO_HEADLESS=0 python tools/discover_api.py +``` +Connecte-toi, ouvre le menu de la semaine, change une recette pour capturer l'écriture, puis +Entrée. Vérifie/complète ensuite `config/endpoints.json` (généré depuis le trafic capturé, +voir aussi `.session/discovery_log.json`). + +### 3. Tester en local +```bash +ANTICOCO_HEADLESS=0 python server.py # 1er run : login dans la fenêtre si besoin +``` +Le login crée `.session/profile` (profil Playwright persistant) — réutilisé ensuite headless. + +## Déploiement homelab (Docker) + +`.session/` et `.env` ne sont **jamais** versionnés. Workflow : + +```bash +# 1. Sur le Mac : générer une session connectée (fenêtre visible) +ANTICOCO_HEADLESS=0 python server.py # se connecter, puis Ctrl-C + +# 2. Pousser le code +git add -A && git commit -m "..." && git push + +# 3. Synchroniser la session vers le homelab (NON versionnée) +scp -r .session config/endpoints.json jerem@192.168.0.43:/AntiCoco/ + +# 4. Sur le homelab : déployer +ssh homelab +cd /AntiCoco && git pull && docker compose up -d --build +``` + +Vérifier : `curl -s http://127.0.0.1:9200/mcp` (le serveur répond au handshake MCP). + +## Intégration Hermes + +Enregistrer AntiCoco dans la config MCP de Hermes (côté homelab), URL +`http://127.0.0.1:9200/mcp` (transport streamable-http). + +> Si Hermes n'accepte que le **stdio**, changer la dernière ligne de `server.py` +> (`mcp.run(transport="stdio")`) et lancer le serveur en sous-processus — le reste est identique. + +## Outils MCP exposés + +| Outil | Rôle | +|-------|------| +| `hf_auth_status()` | état de connexion | +| `hf_login()` | (re)connexion + capture token | +| `hf_list_weeks()` | semaines modifiables | +| `hf_get_menu(week)` | toutes les recettes, avec flag `contains_excluded` | +| `hf_propose(week, count=0)` | shortlist **sans coco**, classée par préférences | +| `hf_confirm_selection(week, recipe_ids)` | **écrit** la sélection (refuse la coco) | +| `hf_get_excludes()` / `hf_add_exclude(term)` / `hf_remove_exclude(term)` | gérer la liste d'exclusion | + +## Configuration + +- `config/excludes.json` — ingrédients bannis (matching insensible casse/accents). Coco déjà listée. +- `config/prefs.json` — mots-clés `liked`/`disliked` pour classer les propositions. +- `config/endpoints.json` — URLs gateway réelles (généré par `discover_api.py`, non versionné). diff --git a/config/excludes.json b/config/excludes.json new file mode 100644 index 0000000..62a999a --- /dev/null +++ b/config/excludes.json @@ -0,0 +1,16 @@ +{ + "_comment": "Ingrédients à exclure. Matching insensible à la casse et aux accents, sur le nom des ingrédients ET les allergènes. La noix de coco est l'ennemi numéro 1.", + "exclude": [ + "noix de coco", + "coco", + "coconut", + "lait de coco", + "creme de coco", + "huile de coco", + "sucre de coco", + "farine de coco", + "eau de coco", + "rape de coco", + "noix de coco rapee" + ] +} diff --git a/config/prefs.json b/config/prefs.json new file mode 100644 index 0000000..040fbc4 --- /dev/null +++ b/config/prefs.json @@ -0,0 +1,14 @@ +{ + "_comment": "Préférences optionnelles pour classer les recettes proposées. 'liked'/'disliked' sont des mots-clés cherchés dans le nom, le titre et les ingrédients de la recette. Un match 'liked' augmente le score, un 'disliked' le baisse (sans exclure — pour exclure, utiliser excludes.json).", + "liked": [ + "boeuf", + "poulet", + "pates", + "fromage", + "champignon" + ], + "disliked": [ + "epinard", + "betterave" + ] +} diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..d621713 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,18 @@ +services: + anticoco: + build: . + container_name: anticoco + restart: unless-stopped + # Exposé uniquement en local sur le homelab — Hermes (même machine) s'y connecte. + ports: + - "127.0.0.1:9200:9200" + env_file: + - .env + environment: + ANTICOCO_PORT: "9200" + ANTICOCO_HEADLESS: "1" + volumes: + # Session Playwright (cookies/login) syncée à la main depuis le Mac : NON versionnée. + - ./.session:/app/.session + # Liste d'exclusion + préférences + endpoints découverts, modifiables sans rebuild. + - ./config:/app/config diff --git a/hellofresh/__init__.py b/hellofresh/__init__.py new file mode 100644 index 0000000..6298ed4 --- /dev/null +++ b/hellofresh/__init__.py @@ -0,0 +1,6 @@ +"""Accès personnel au compte HelloFresh (région France). + +L'API interne `gw/` de hellofresh.fr n'est pas une API publique documentée et peut +changer ; usage strictement personnel. Les endpoints réels sont découverts via +`tools/discover_api.py` (capture du trafic d'une session connectée). +""" diff --git a/hellofresh/api.py b/hellofresh/api.py new file mode 100644 index 0000000..bc7c786 --- /dev/null +++ b/hellofresh/api.py @@ -0,0 +1,150 @@ +"""Client httpx vers l'API interne HelloFresh. + +Les URLs réelles ne sont pas codées en dur : elles viennent de `config/endpoints.json`, +généré/validé via `tools/discover_api.py`. Tant que ce fichier n'est pas rempli, les +appels lèvent une erreur explicite. + +Le token bearer est fourni par `auth.get_token()`. Les réponses (forme variable) sont +mappées vers `models.Recipe` / `models.Week` de façon tolérante. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +import httpx + +from . import auth +from .models import Recipe, Week, _first + +ENDPOINTS_PATH = auth.ROOT / "config" / "endpoints.json" + +DEFAULT_HEADERS = { + "Accept": "application/json", + "Content-Type": "application/json", + "Origin": auth.BASE_URL, + "Referer": auth.BASE_URL + "/", + "User-Agent": ( + "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 " + "(KHTML, like Gecko) Chrome/124.0 Safari/537.36" + ), +} + + +class EndpointsNotConfigured(RuntimeError): + pass + + +def _load_endpoints() -> dict: + if not ENDPOINTS_PATH.exists(): + raise EndpointsNotConfigured( + f"{ENDPOINTS_PATH} absent. Lancer `python tools/discover_api.py` pour le générer." + ) + data = json.loads(ENDPOINTS_PATH.read_text(encoding="utf-8")) + missing = [k for k in ("weeks", "menu", "set_selection") if not data.get(k)] + if missing: + raise EndpointsNotConfigured( + f"Endpoints manquants dans {ENDPOINTS_PATH}: {missing}. " + "Compléter via tools/discover_api.py." + ) + return data + + +class HelloFreshClient: + def __init__(self, token: str | None = None): + self._endpoints = _load_endpoints() + self._token = token or auth.get_token() + self._client = httpx.Client( + headers={**DEFAULT_HEADERS, "Authorization": f"Bearer {self._token}"}, + timeout=30.0, + follow_redirects=True, + ) + + def close(self) -> None: + self._client.close() + + def __enter__(self) -> "HelloFreshClient": + return self + + def __exit__(self, *exc) -> None: + self.close() + + # --- requêtes bas niveau avec re-auth transparent sur 401 --------------- + def _request(self, method: str, url: str, **kwargs) -> httpx.Response: + resp = self._client.request(method, url, **kwargs) + if resp.status_code == 401: + # Token expiré → on en capture un neuf et on rejoue une fois. + self._token = auth.get_token(force=True) + self._client.headers["Authorization"] = f"Bearer {self._token}" + resp = self._client.request(method, url, **kwargs) + resp.raise_for_status() + return resp + + # --- API métier --------------------------------------------------------- + def get_editable_weeks(self) -> list[Week]: + """Liste les semaines de l'abonnement encore modifiables.""" + resp = self._request("GET", self._endpoints["weeks"]) + data = resp.json() + raw_weeks = _extract_list(data, "weeks", "deliveries", "items", "data") + weeks = [Week.from_api(w) for w in raw_weeks] + editable = [w for w in weeks if w.editable] or weeks + return editable + + def get_menu(self, week: str) -> list[Recipe]: + """Recettes proposées pour une semaine donnée.""" + url = self._endpoints["menu"].replace("{week}", str(week)) + resp = self._request("GET", url, params=None if "{week}" in self._endpoints["menu"] else {"week": week}) + data = resp.json() + raw_recipes = _extract_recipes(data) + return [Recipe.from_api(r) for r in raw_recipes] + + def set_selection(self, week: str, recipe_ids: list[str]) -> dict[str, Any]: + """Enregistre la sélection de recettes pour une semaine (écriture). + + N'est appelé qu'après confirmation côté serveur MCP. La forme du payload + dépend de l'endpoint découvert ; on envoie une structure courante, à ajuster + selon discover_api.py si nécessaire. + """ + url = self._endpoints["set_selection"].replace("{week}", str(week)) + method = self._endpoints.get("set_selection_method", "PUT").upper() + payload = { + "week": week, + "recipes": [{"id": rid, "quantity": 1} for rid in recipe_ids], + } + resp = self._request(method, url, json=payload) + try: + return resp.json() + except Exception: + return {"status": resp.status_code, "ok": True} + + +def _extract_list(data: Any, *keys: str) -> list[dict]: + if isinstance(data, list): + return data + if isinstance(data, dict): + found = _first(data, *keys, default=None) + if isinstance(found, list): + return found + # Parfois imbriqué sous "data"/"items" + for v in data.values(): + if isinstance(v, list): + return v + return [] + + +def _extract_recipes(data: Any) -> list[dict]: + """Extrait la liste de recettes, qui peut être imbriquée dans des 'courses'.""" + if isinstance(data, dict): + courses = _first(data, "courses", "modules", "items", default=None) + if isinstance(courses, list): + recipes = [] + for c in courses: + if isinstance(c, dict) and "recipe" in c and isinstance(c["recipe"], dict): + recipes.append(c["recipe"]) + elif isinstance(c, dict): + recipes.append(c) + if recipes: + return recipes + return _extract_list(data, "recipes", "items", "data") diff --git a/hellofresh/auth.py b/hellofresh/auth.py new file mode 100644 index 0000000..866825e --- /dev/null +++ b/hellofresh/auth.py @@ -0,0 +1,204 @@ +"""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 json +import os +import time +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" + +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-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 _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): + return pw.chromium.launch_persistent_context( + user_data_dir=str(PROFILE_DIR), + headless=_headless(), + locale="fr-FR", + viewport={"width": 1280, "height": 900}, + ) + + +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_token(force: bool = False) -> dict: + """Capture (ou relit depuis le cache) le bearer token et les hôtes gateway observés. + + Retourne {"token": str, "gateways": [str], "captured_at": float}. + """ + 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) + observed = {"token": None, "gateways": set()} + + with sync_playwright() as pw: + ctx = _open_context(pw) + try: + 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) + auth = req.headers.get("authorization") or req.headers.get("Authorization") + if auth and auth.lower().startswith("bearer "): + observed["token"] = auth.split(" ", 1)[1].strip() + except Exception: + pass + + page.on("request", on_request) + page.goto(MENU_PAGE, wait_until="networkidle", timeout=45000) + page.wait_for_timeout(3000) + + if not _is_logged_in(page): + raise RuntimeError( + "Non connecté lors de la capture du token. Lancer ensure_logged_in() d'abord." + ) + if not observed["token"]: + raise RuntimeError( + "Aucun bearer token capturé. Vérifier MENU_PAGE / le pattern gateway, " + "ou rejouer tools/discover_api.py." + ) + + 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 + 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 get_token(force: bool = False) -> str: + """Renvoie un bearer token valide, en s'assurant d'être connecté au préalable.""" + 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", [])} + try: + ok = ensure_logged_in() + return {"logged_in": bool(ok), "source": "browser"} + except Exception as e: + return {"logged_in": False, "error": str(e)} diff --git a/hellofresh/filter.py b/hellofresh/filter.py new file mode 100644 index 0000000..9dcbcd8 --- /dev/null +++ b/hellofresh/filter.py @@ -0,0 +1,111 @@ +"""Filtrage des recettes : exclusion d'ingrédients (coco !) + scoring par préférences. + +Matching insensible à la casse ET aux accents : « Noix de Coco », « noix de coco rapée » +et « creme de coco » matchent tous l'entrée « coco ». On compare des mots normalisés sur +le nom des ingrédients, les allergènes, le nom et le titre de la recette. +""" + +from __future__ import annotations + +import json +import unicodedata +from pathlib import Path + +from .auth import ROOT +from .models import Recipe + +EXCLUDES_PATH = ROOT / "config" / "excludes.json" +PREFS_PATH = ROOT / "config" / "prefs.json" + + +def normalize(s: str) -> str: + """Minuscule + suppression des accents (NFD → drop des diacritiques).""" + s = unicodedata.normalize("NFD", s or "") + s = "".join(c for c in s if unicodedata.category(c) != "Mn") + return s.lower().strip() + + +# --- gestion de la liste d'exclusion --------------------------------------- +def load_excludes() -> list[str]: + if not EXCLUDES_PATH.exists(): + return [] + data = json.loads(EXCLUDES_PATH.read_text(encoding="utf-8")) + return list(data.get("exclude", [])) + + +def save_excludes(terms: list[str]) -> None: + existing = {} + if EXCLUDES_PATH.exists(): + existing = json.loads(EXCLUDES_PATH.read_text(encoding="utf-8")) + existing["exclude"] = terms + EXCLUDES_PATH.write_text(json.dumps(existing, indent=2, ensure_ascii=False), encoding="utf-8") + + +def add_exclude(term: str) -> list[str]: + terms = load_excludes() + if normalize(term) not in {normalize(t) for t in terms}: + terms.append(term) + save_excludes(terms) + return terms + + +def remove_exclude(term: str) -> list[str]: + nt = normalize(term) + terms = [t for t in load_excludes() if normalize(t) != nt] + save_excludes(terms) + return terms + + +def load_prefs() -> dict: + if not PREFS_PATH.exists(): + return {"liked": [], "disliked": []} + data = json.loads(PREFS_PATH.read_text(encoding="utf-8")) + return {"liked": data.get("liked", []), "disliked": data.get("disliked", [])} + + +# --- application aux recettes ---------------------------------------------- +def _recipe_haystack(recipe: Recipe) -> str: + parts = [recipe.name, recipe.headline, *recipe.ingredients, *recipe.allergens, *recipe.tags] + return normalize(" | ".join(p for p in parts if p)) + + +def mark_excluded(recipe: Recipe, excludes: list[str] | None = None) -> Recipe: + """Remplit `contains_excluded` et `matched_excludes` sur la recette.""" + excludes = excludes if excludes is not None else load_excludes() + hay = _recipe_haystack(recipe) + matched = [term for term in excludes if normalize(term) and normalize(term) in hay] + recipe.matched_excludes = matched + recipe.contains_excluded = bool(matched) + return recipe + + +def score(recipe: Recipe, prefs: dict | None = None) -> float: + prefs = prefs or load_prefs() + hay = _recipe_haystack(recipe) + s = 0.0 + for kw in prefs.get("liked", []): + if normalize(kw) and normalize(kw) in hay: + s += 1.0 + for kw in prefs.get("disliked", []): + if normalize(kw) and normalize(kw) in hay: + s -= 1.0 + recipe.score = s + return s + + +def annotate(recipes: list[Recipe]) -> list[Recipe]: + """Marque exclusions + score sur une liste de recettes (in place).""" + excludes = load_excludes() + prefs = load_prefs() + for r in recipes: + mark_excluded(r, excludes) + score(r, prefs) + return recipes + + +def propose(recipes: list[Recipe], count: int | None = None) -> list[Recipe]: + """Retire les recettes exclues (coco…) et classe le reste par score décroissant.""" + annotate(recipes) + safe = [r for r in recipes if not r.contains_excluded] + safe.sort(key=lambda r: r.score, reverse=True) + return safe[:count] if count else safe diff --git a/hellofresh/models.py b/hellofresh/models.py new file mode 100644 index 0000000..87f2e32 --- /dev/null +++ b/hellofresh/models.py @@ -0,0 +1,124 @@ +"""Modèles de données HelloFresh, indépendants de la forme exacte de l'API interne. + +Les dataclasses sont volontairement tolérantes : `Recipe.from_api` mappe au mieux les +champs des réponses gateway (qui varient selon les endpoints découverts) vers une forme +stable utilisée par le filtre et le serveur MCP. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field, asdict +from typing import Any + + +def _first(d: dict, *keys, default=None): + """Renvoie la première clé présente et non vide parmi `keys`.""" + for k in keys: + v = d.get(k) + if v not in (None, "", [], {}): + return v + return default + + +@dataclass +class Recipe: + id: str + name: str + headline: str = "" + ingredients: list[str] = field(default_factory=list) + allergens: list[str] = field(default_factory=list) + tags: list[str] = field(default_factory=list) + image_url: str = "" + prep_time: str = "" + # Champs calculés par le filtre (remplis plus tard) + contains_excluded: bool = False + matched_excludes: list[str] = field(default_factory=list) + score: float = 0.0 + + @classmethod + def from_api(cls, raw: dict[str, Any]) -> "Recipe": + """Construit une Recipe depuis un objet recette brut de l'API gateway. + + Les noms de champs HelloFresh varient ; on tente plusieurs alias. À ajuster + une fois la forme réelle confirmée via discover_api.py. + """ + ingredients = [] + for ing in _first(raw, "ingredients", default=[]) or []: + if isinstance(ing, dict): + name = _first(ing, "name", "label", "title") + if name: + ingredients.append(str(name)) + elif isinstance(ing, str): + ingredients.append(ing) + + allergens = [] + for al in _first(raw, "allergens", default=[]) or []: + if isinstance(al, dict): + name = _first(al, "name", "label", "title") + if name: + allergens.append(str(name)) + elif isinstance(al, str): + allergens.append(al) + + tags = [] + for tg in _first(raw, "tags", "labels", default=[]) or []: + if isinstance(tg, dict): + name = _first(tg, "name", "label", "text") + if name: + tags.append(str(name)) + elif isinstance(tg, str): + tags.append(tg) + + image = _first(raw, "imageLink", "image", "imageUrl", "cardLink", default="") + if isinstance(image, dict): + image = _first(image, "url", "link", default="") + + return cls( + id=str(_first(raw, "id", "uuid", "recipeId", default="")), + name=str(_first(raw, "name", "title", default="(sans nom)")), + headline=str(_first(raw, "headline", "subtitle", "description", default="")), + ingredients=ingredients, + allergens=allergens, + tags=tags, + image_url=str(image or ""), + prep_time=str(_first(raw, "prepTime", "totalTime", "time", default="")), + ) + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + def summary(self) -> dict[str, Any]: + """Version compacte pour les réponses MCP (moins de tokens).""" + return { + "id": self.id, + "name": self.name, + "headline": self.headline, + "contains_excluded": self.contains_excluded, + "matched_excludes": self.matched_excludes, + "score": self.score, + "tags": self.tags, + } + + +@dataclass +class Week: + id: str # handle de semaine, ex. "2026-W25" + delivery_date: str = "" + editable: bool = False + max_selectable: int = 0 + recipes: list[Recipe] = field(default_factory=list) + + @classmethod + def from_api(cls, raw: dict[str, Any], recipes: list[Recipe] | None = None) -> "Week": + return cls( + id=str(_first(raw, "id", "week", "handle", "yearWeek", default="")), + delivery_date=str(_first(raw, "deliveryDate", "date", default="")), + editable=bool(_first(raw, "editable", "isEditable", "menuEditable", default=False)), + max_selectable=int(_first(raw, "maxSelectable", "numberOfSelections", default=0) or 0), + recipes=recipes or [], + ) + + def to_dict(self) -> dict[str, Any]: + d = asdict(self) + d["recipes"] = [r.summary() for r in self.recipes] + return d diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..b3e06cd --- /dev/null +++ b/requirements.txt @@ -0,0 +1,4 @@ +mcp>=1.2 +playwright>=1.45 +httpx>=0.28 +python-dotenv>=1.0 diff --git a/server.py b/server.py new file mode 100644 index 0000000..ace6d67 --- /dev/null +++ b/server.py @@ -0,0 +1,134 @@ +"""Serveur MCP AntiCoco — accès personnel à HelloFresh, sans noix de coco. + +Client visé : Hermes (Nous Research), sur le même homelab. Transport streamable-HTTP +sur 127.0.0.1:$ANTICOCO_PORT/mcp. Si Hermes n'accepte que le stdio, changer le +`transport=` de `mcp.run()` ci-dessous (le reste du code est identique). + +Les outils sont synchrones : FastMCP les exécute dans un thread worker, ce qui permet +d'utiliser l'API *synchrone* de Playwright (auth.py) sans conflit de boucle asyncio. +""" + +from __future__ import annotations + +import os + +from dotenv import load_dotenv +from mcp.server.fastmcp import FastMCP + +load_dotenv() + +from hellofresh import api, auth, filter as hf_filter # noqa: E402 + +PORT = int(os.environ.get("ANTICOCO_PORT", "9200")) +mcp = FastMCP("AntiCoco", host="0.0.0.0", port=PORT) + + +@mcp.tool() +def hf_auth_status() -> dict: + """État de la connexion HelloFresh (utilise le token en cache s'il est frais).""" + return auth.auth_status() + + +@mcp.tool() +def hf_login() -> dict: + """S'assure d'être connecté et capture un bearer token frais. + + En headless (homelab), échoue si la session a expiré → refaire le login local. + """ + ok = auth.ensure_logged_in() + if not ok: + return {"logged_in": False, "error": "login non établi (timeout ou session expirée)"} + info = auth.capture_token(force=True) + return {"logged_in": True, "gateways": info.get("gateways", [])} + + +@mcp.tool() +def hf_list_weeks() -> list[dict]: + """Liste les semaines de l'abonnement encore modifiables (handle + date livraison).""" + with api.HelloFreshClient() as client: + weeks = client.get_editable_weeks() + return [{"week": w.id, "delivery_date": w.delivery_date, "editable": w.editable, + "max_selectable": w.max_selectable} for w in weeks] + + +@mcp.tool() +def hf_get_menu(week: str) -> dict: + """Toutes les recettes proposées pour une semaine, chacune annotée. + + Chaque recette porte `contains_excluded` (true si elle contient un ingrédient banni, + coco en tête) et `matched_excludes` (quels termes ont matché). + """ + with api.HelloFreshClient() as client: + recipes = client.get_menu(week) + hf_filter.annotate(recipes) + return { + "week": week, + "count": len(recipes), + "recipes": [r.summary() for r in recipes], + } + + +@mcp.tool() +def hf_propose(week: str, count: int = 0) -> dict: + """Shortlist de recettes SANS ingrédient exclu, classée par préférences. + + `count=0` renvoie toutes les recettes sûres. Étape « je propose » : rien n'est écrit + sur le compte ici — utiliser hf_confirm_selection() ensuite. + """ + with api.HelloFreshClient() as client: + recipes = client.get_menu(week) + safe = hf_filter.propose(recipes, count=count or None) + excluded = [r.summary() for r in recipes if r.contains_excluded] + return { + "week": week, + "proposed": [r.summary() for r in safe], + "excluded_for_coco_etc": excluded, + "note": "Aucune écriture effectuée. Confirme avec hf_confirm_selection(week, recipe_ids).", + } + + +@mcp.tool() +def hf_confirm_selection(week: str, recipe_ids: list[str]) -> dict: + """ÉCRIT la sélection de recettes dans la box de la semaine (après confirmation). + + Garde-fou : refuse une recette contenant un ingrédient exclu. + """ + with api.HelloFreshClient() as client: + menu = client.get_menu(week) + hf_filter.annotate(menu) + by_id = {r.id: r for r in menu} + bad = [rid for rid in recipe_ids if rid in by_id and by_id[rid].contains_excluded] + if bad: + return { + "ok": False, + "error": "Sélection refusée : recette(s) avec ingrédient exclu (coco ?).", + "offending_ids": bad, + } + unknown = [rid for rid in recipe_ids if rid not in by_id] + if unknown: + return {"ok": False, "error": "Recette(s) inconnue(s) pour cette semaine.", + "unknown_ids": unknown} + result = client.set_selection(week, recipe_ids) + return {"ok": True, "week": week, "selected": recipe_ids, "api_response": result} + + +@mcp.tool() +def hf_get_excludes() -> list[str]: + """Liste actuelle des ingrédients exclus.""" + return hf_filter.load_excludes() + + +@mcp.tool() +def hf_add_exclude(term: str) -> list[str]: + """Ajoute un ingrédient à exclure. Renvoie la nouvelle liste.""" + return hf_filter.add_exclude(term) + + +@mcp.tool() +def hf_remove_exclude(term: str) -> list[str]: + """Retire un ingrédient de la liste d'exclusion. Renvoie la nouvelle liste.""" + return hf_filter.remove_exclude(term) + + +if __name__ == "__main__": + mcp.run(transport="streamable-http") diff --git a/tools/discover_api.py b/tools/discover_api.py new file mode 100644 index 0000000..1870a43 --- /dev/null +++ b/tools/discover_api.py @@ -0,0 +1,106 @@ +"""ÉTAPE 0 — Découverte de l'API interne HelloFresh. + +Lance un navigateur visible, te laisse te connecter et naviguer (menu de la semaine, +sélection de recettes), puis enregistre TOUTES les requêtes vers le gateway pour en +déduire les 3 endpoints utiles : + 1. abonnement + semaines éditables + 2. menu d'une semaine (recettes / ingrédients / allergènes) + 3. enregistrement de la sélection de recettes + +Usage : + ANTICOCO_HEADLESS=0 python tools/discover_api.py + +Pendant que la fenêtre est ouverte : + - connecte-toi, + - ouvre le menu de la semaine, + - (optionnel) change une recette pour capturer l'appel d'écriture, +puis reviens dans le terminal et appuie sur Entrée pour écrire le rapport. + +Sortie : + - .session/discovery_log.json : toutes les requêtes gateway observées (debug complet) + - config/endpoints.json : squelette pré-rempli à compléter/valider à la main +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from playwright.sync_api import sync_playwright # noqa: E402 +from hellofresh import auth # noqa: E402 + +LOG_PATH = auth.SESSION_DIR / "discovery_log.json" +ENDPOINTS_PATH = ROOT / "config" / "endpoints.json" + + +def main() -> None: + auth.SESSION_DIR.mkdir(parents=True, exist_ok=True) + requests_seen: list[dict] = [] + + with sync_playwright() as pw: + ctx = pw.chromium.launch_persistent_context( + user_data_dir=str(auth.PROFILE_DIR), + headless=False, # toujours visible : c'est une étape interactive + locale="fr-FR", + viewport={"width": 1280, "height": 900}, + ) + page = ctx.pages[0] if ctx.pages else ctx.new_page() + + def on_request(req): + if not auth._is_gateway_request(req.url): + return + entry = { + "method": req.method, + "url": req.url, + "has_auth": bool(req.headers.get("authorization")), + } + if req.method in ("POST", "PUT", "PATCH"): + try: + entry["post_data"] = req.post_data + except Exception: + entry["post_data"] = None + requests_seen.append(entry) + print(f" [{req.method}] {req.url}") + + page.on("request", on_request) + + print("Ouvre le menu de la semaine, change une recette si tu veux capturer l'écriture.") + page.goto(auth.BASE_URL + "/my-account", wait_until="domcontentloaded", timeout=30000) + + try: + input("\n>>> Quand tu as fini de naviguer, appuie sur Entrée pour générer le rapport...\n") + except (EOFError, KeyboardInterrupt): + pass + + ctx.close() + + LOG_PATH.write_text(json.dumps(requests_seen, indent=2, ensure_ascii=False), encoding="utf-8") + print(f"\n{len(requests_seen)} requêtes gateway enregistrées dans {LOG_PATH}") + + # Heuristiques pour pré-remplir le squelette d'endpoints. + def find(method: str, *needles: str) -> str: + for r in requests_seen: + if r["method"] != method: + continue + if all(n in r["url"].lower() for n in needles): + return r["url"] + return "" + + skeleton = { + "_comment": "Endpoints HelloFresh confirmés via discovery. Compléter/valider à la main. Utiliser {week} comme placeholder pour le handle de semaine.", + "base": "", + "weeks": find("GET", "subscription") or find("GET", "deliveries") or "", + "menu": find("GET", "menu") or find("GET", "courses") or "", + "set_selection": find("PUT", "menu") or find("POST", "menu") or "", + } + ENDPOINTS_PATH.parent.mkdir(parents=True, exist_ok=True) + ENDPOINTS_PATH.write_text(json.dumps(skeleton, indent=2, ensure_ascii=False), encoding="utf-8") + print(f"Squelette d'endpoints écrit dans {ENDPOINTS_PATH} — à vérifier avant usage.") + + +if __name__ == "__main__": + main()