From 03e281a810adfe5bf0504660709ffd3ccc8d8380 Mon Sep 17 00:00:00 2001 From: jerem Date: Thu, 18 Jun 2026 14:18:40 +0200 Subject: [PATCH] =?UTF-8?q?S=C3=A9lection=20courante=20+=20favoris=20+=20i?= =?UTF-8?q?mages=20servables=20(sortie=20pr=C3=AAte=20Hermes)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deux outils MCP pour qu'Hermes n'ait plus de scripts à écrire : - hf_next_delivery() : prochaine box RÉELLEMENT sélectionnée (≈4 recettes, pas le menu complet) + date/cutoff ; erreur stricte si introuvable (jamais de repli propose). Saute les semaines PAUSED via next_delivery. - hf_favorites() : recettes favorites du compte. Champ is_favorite ajouté partout (hf_get_menu inclus). Endpoints découverts (probe CDP) : - sélection : GET /gw/my-deliveries/menu -> meals[].selection.quantity>0 - favoris : GET /gw/cfs/v2/favorites/recipe -> items[].object_id (GET /gw/v1/carts/{week} renvoie 404 : pas la lecture de sélection.) Images : URLs recettes CloudFront (502) réécrites vers img.hellofresh.com/.../hellofresh_s3/... (hellofresh/images.py), appliqué dans Recipe.summary() -> profite à tous les outils. README : procédure de ré-auth CDP clarifiée (refresh tokens rotatifs, backups inutiles, page /login, profil Chrome dédié). Outils de re-découverte : tools/probe_selection.py, tools/probe_menu_capture.py --- README.md | 39 +++++++-- config/endpoints.json | 7 +- hellofresh/api.py | 63 +++++++++++++++ hellofresh/images.py | 73 +++++++++++++++++ hellofresh/models.py | 6 +- server.py | 52 ++++++++++++ tools/probe_menu_capture.py | 104 ++++++++++++++++++++++++ tools/probe_selection.py | 155 ++++++++++++++++++++++++++++++++++++ 8 files changed, 489 insertions(+), 10 deletions(-) create mode 100644 hellofresh/images.py create mode 100644 tools/probe_menu_capture.py create mode 100644 tools/probe_selection.py diff --git a/README.md b/README.md index aea4e81..b5da2d6 100644 --- a/README.md +++ b/README.md @@ -18,9 +18,15 @@ HTTP 200) — sélection par index de course, ids de compte dérivés dynamiquem ✅ **Auth autonome (pur HTTP)** : le token (30 min) est rafraîchi par un simple `POST /gw/refresh` (le endpoint que la SPA appelle), **sans navigateur**. Le refresh_token roule par fenêtres de 60 j, -remises à zéro à chaque refresh → un homelab allumé reste authentifié **indéfiniment**, sans +remises à zéro à chaque refresh → un homelab **allumé en continu** reste authentifié sans intervention ni re-sync. Le navigateur headless ne sert plus que de filet de secours. +> ⚠️ La chaîne de refresh **peut casser** (homelab éteint trop longtemps, ou refresh_token +> consommé hors `_refresh_session`) → `POST /gw/refresh` renvoie `invalid_grant` et +> `auth_status()` passe à `logged_in:false`. Le seul remède est de **refaire le login CDP** +> (cf. ci-dessous). **Restaurer un ancien `.session/` ne marche pas** : les refresh_token sont +> rotatifs, les anciens sont morts. + > ⚠️ La connexion **directe** automatisée (Playwright/Chromium qui remplit le formulaire) est > bloquée par l'anti-bot HelloFresh. La session se crée donc via **attache CDP à ton vrai Chrome** > (`tools/attach_capture.py`), où le login marche normalement. @@ -44,17 +50,29 @@ playwright install chromium cp .env.example .env ``` -### 2. Créer la session (login via TON Chrome, anti-bot contourné) -Lance ton Chrome avec un port de debug + profil dédié (ta fenêtre Chrome habituelle peut rester -ouverte), connecte-toi à HelloFresh (email + mot de passe), puis attache la capture : +### 2. Créer (ou ré-créer) la session — login via TON Chrome, anti-bot contourné +C'est **la seule méthode de login qui marche** : un navigateur piloté par Playwright (même avec +`channel=chrome`) est bloqué par l'anti-bot HelloFresh. On lance donc le vrai Chrome en mode debug +et on s'y **attache en CDP** (lecture seule). Même procédure pour la 1ʳᵉ session **et** pour une +ré-authentification après `invalid_grant`. + ```bash +# 1) Lance ton vrai Chrome en mode debug avec un PROFIL DÉDIÉ (Chrome ≥149 refuse le debug +# sur le profil par défaut). Ta fenêtre Chrome habituelle peut rester ouverte. "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" \ --remote-debugging-port=9222 --user-data-dir="$HOME/.hf-chrome-debug" \ https://www.hellofresh.fr/my-account/deliveries/menu + +# 2) Connecte-toi dans cette fenêtre (page de login = https://www.hellofresh.fr/login). + +# 3) Pendant que Chrome reste ouvert et connecté, attache la capture : python tools/attach_capture.py # capture trafic + exporte .session/storage_state.json ``` -`storage_state.json` (cookies, ~60 j) est la session réutilisable. `config/endpoints.json` est -déjà rempli ; rejoue `attach_capture` si l'API change (cf. `config/endpoints_discovered.json`). +`attach_capture` doit tourner **pendant** que tu (re)charges la page menu, pour capter le trafic. +Il **écrase** `.session/storage_state.json` avec la session fraîche (cookies ~60 j) — ne pas +restaurer un vieux backup à la place (refresh_token rotatif = anciens morts). Vérifie ensuite : +`python -c "from hellofresh import auth, json; print(auth.auth_status())"` → `logged_in: true`. +`config/endpoints.json` est déjà rempli ; rejoue `attach_capture` si l'API change. ### 3. Tester en local (headless, comme le homelab) ```bash @@ -97,7 +115,9 @@ Enregistrer AntiCoco dans la config MCP de Hermes (côté homelab), URL | `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_next_delivery()` | **prochaine box réellement sélectionnée** (≈4 recettes) + date/cutoff, images servables — prêt Telegram | +| `hf_favorites()` | recettes **favorites** du compte (images servables) | +| `hf_get_menu(week)` | toutes les recettes, avec `contains_excluded` et `is_favorite` | | `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 | @@ -106,4 +126,7 @@ Enregistrer AntiCoco dans la config MCP de Hermes (côté homelab), URL - `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é). +- `config/endpoints.json` — URLs gateway réelles + transfo CDN images (`image_cdn_*` : les URLs + recettes CloudFront `…/0,0/image/X` répondent en 502, réécrites vers `img.hellofresh.com/…/hellofresh_s3/image/X`). + Sélection courante via `my-deliveries/menu`, favoris via `cfs/v2/favorites/recipe`. + Re-découverte : `tools/probe_selection.py` / `tools/probe_menu_capture.py` (attache CDP). diff --git a/config/endpoints.json b/config/endpoints.json index 6df77ee..7e5dad0 100644 --- a/config/endpoints.json +++ b/config/endpoints.json @@ -10,5 +10,10 @@ "subscriptions": "https://www.hellofresh.fr/gw/api/customers/me/subscriptions", "set_selection": "https://www.hellofresh.fr/gw/v1/carts/{week}", "set_selection_method": "PUT", - "selection_preference": "quick" + "selection_preference": "quick", + "current_selection": "https://www.hellofresh.fr/gw/my-deliveries/menu", + "favorites": "https://www.hellofresh.fr/gw/cfs/v2/favorites/recipe", + "image_cdn_host": "img.hellofresh.com", + "image_cdn_transform": "f_auto,fl_lossy,q_auto,w_1200", + "image_cdn_source_hosts": ["d3hvwccx09j84u.cloudfront.net"] } diff --git a/hellofresh/api.py b/hellofresh/api.py index 2b1a6c4..aff72e0 100644 --- a/hellofresh/api.py +++ b/hellofresh/api.py @@ -255,6 +255,69 @@ class HelloFreshClient: recipes.extend(Recipe.from_api(r) for r in raw) return recipes + def get_current_selection(self, week: str) -> list[Recipe]: + """Recettes RÉELLEMENT dans la box de la semaine (sélection courante, ≠ menu complet). + + Source : `GET my-deliveries/menu` (BFF) → `meals[*]` ; un repas est sélectionné quand + `selection.quantity > 0`. On en extrait les ids puis on récupère les détails complets + (ingrédients/allergènes) comme pour le menu. Lève une exception si rien n'est + sélectionné (box vide / non finalisée) — pas de repli vers une proposition. + """ + ep = self._ep.get("current_selection") + if not ep: + raise EndpointsNotConfigured("Endpoint 'current_selection' manquant.") + sub = self._subscription_info() + params = { + "country": self._country, + "locale": self._locale, + "subscription": sub["sub_id"], + "product-sku": sub["sku"], + "week": week, + } + data = self._request("GET", ep, params=params).json() + meals = data.get("meals", []) if isinstance(data, dict) else [] + idx_by_id: dict[str, int | None] = {} + for m in meals: + sel = m.get("selection") + qty = sel.get("quantity") if isinstance(sel, dict) else None + if qty and qty > 0: + rid = (m.get("recipe") or {}).get("id") + if rid: + idx_by_id[str(rid)] = m.get("index") + if not idx_by_id: + raise RuntimeError( + f"Aucune recette sélectionnée pour {week} (box vide ou non finalisée ?)." + ) + recipes = self._fetch_details(list(idx_by_id)) + if not recipes: + raise RuntimeError(f"Détails de la sélection {week} introuvables.") + for r in recipes: + r.course_index = idx_by_id.get(r.id) + return recipes + + def favorite_ids(self) -> set[str]: + """Ids des recettes favorites du compte (best-effort : set() si indispo).""" + ep = self._ep.get("favorites") + if not ep: + return set() + try: + data = self._request("GET", ep, params={ + "country": self._country, "locale": self._locale, "ids": ""}).json() + items = data.get("items", []) if isinstance(data, dict) else (data or []) + return {str(it.get("object_id")) for it in items if it.get("object_id")} + except Exception: + return set() + + def get_favorites(self) -> list[Recipe]: + """Recettes favorites du compte, complètes. Liste vide si aucun favori.""" + ids = self.favorite_ids() + if not ids: + return [] + recipes = self._fetch_details(list(ids)) + for r in recipes: + r.is_favorite = True + return recipes + def selection_indices(self, week: str, recipe_ids: list[str]) -> tuple[list[int], list[str]]: """Mappe des ids de recette vers leurs index de course pour la semaine. diff --git a/hellofresh/images.py b/hellofresh/images.py new file mode 100644 index 0000000..563afc8 --- /dev/null +++ b/hellofresh/images.py @@ -0,0 +1,73 @@ +"""Normalisation des URLs d'images recettes pour un rendu direct (Telegram, etc.). + +L'API recettes renvoie des URLs CloudFront du type + https://d3hvwccx09j84u.cloudfront.net/0,0/image/HF_....jpg +qui répondent en 502 au téléchargement direct (hotlink protégé). Le CDN servable est + https://img.hellofresh.com//hellofresh_s3/image/HF_....jpg +où le segment `0,0` (largeur,hauteur) de CloudFront est remplacé par une transformation +Cloudinary (`f_auto,fl_lossy,q_auto,w_1200`) et le chemin préfixé par `hellofresh_s3`. + +Confirmé par probe (2026-06) : la variante img.hellofresh.com/.../hellofresh_s3/… renvoie 206 + +image/jpeg, alors que l'URL CloudFront brute renvoie 502. Hôte/transfo configurables via +config/endpoints.json (`image_cdn_host`, `image_cdn_transform`, `image_cdn_source_hosts`). +""" + +from __future__ import annotations + +import json +import re +from functools import lru_cache +from urllib.parse import urlsplit + +from . import auth + +_ENDPOINTS_PATH = auth.ROOT / "config" / "endpoints.json" +# Segment de dimensionnement CloudFront en tête de chemin, ex. "/0,0/" ou "/200,200/". +_SIZE_SEG = re.compile(r"^/\d+,\d+(?=/)") + +_DEFAULTS = { + "image_cdn_host": "img.hellofresh.com", + "image_cdn_transform": "f_auto,fl_lossy,q_auto,w_1200", + "image_cdn_source_hosts": ["d3hvwccx09j84u.cloudfront.net"], +} + + +@lru_cache(maxsize=1) +def _cfg() -> tuple[str, str, tuple[str, ...]]: + """(host cible, transfo, hôtes source) lus depuis endpoints.json, avec défauts.""" + data: dict = {} + try: + data = json.loads(_ENDPOINTS_PATH.read_text(encoding="utf-8")) + except Exception: + pass + host = str(data.get("image_cdn_host") or _DEFAULTS["image_cdn_host"]) + transform = str(data.get("image_cdn_transform") or _DEFAULTS["image_cdn_transform"]) + sources = tuple(data.get("image_cdn_source_hosts") or _DEFAULTS["image_cdn_source_hosts"]) + return host, transform, sources + + +def fix_image_url(url: str) -> str: + """Réécrit une URL d'image CloudFront HelloFresh vers le CDN servable. + + - URL vide → "". + - Déjà sur un hôte hellofresh.com → renvoyée telle quelle. + - Hôte CloudFront connu (ou *.cloudfront.net) → host remplacé, segment de taille retiré, + chemin préfixé par `hellofresh_s3` et la transformation insérée. + - Tout autre hôte → renvoyée telle quelle (on ne casse rien). + """ + if not url: + return "" + host, transform, sources = _cfg() + parts = urlsplit(url) + netloc = parts.netloc.lower() + + if "hellofresh.com" in netloc: # déjà servable (img/media.hellofresh.com) + return url + if netloc not in sources and not netloc.endswith(".cloudfront.net"): + return url + + path = _SIZE_SEG.sub("", parts.path) # "/0,0/image/X.jpg" -> "/image/X.jpg" + if not path.startswith("/"): + path = "/" + path + rest = path if path.startswith("/hellofresh_s3/") else "/hellofresh_s3" + path + return f"https://{host}/{transform}{rest}" diff --git a/hellofresh/models.py b/hellofresh/models.py index 5d83a01..fb59d7d 100644 --- a/hellofresh/models.py +++ b/hellofresh/models.py @@ -11,6 +11,8 @@ import re from dataclasses import dataclass, field, asdict from typing import Any +from .images import fix_image_url + _ISO_DURATION = re.compile(r"PT(?:(\d+)H)?(?:(\d+)M)?", re.IGNORECASE) @@ -54,6 +56,7 @@ class Recipe: contains_excluded: bool = False matched_excludes: list[str] = field(default_factory=list) score: float = 0.0 + is_favorite: bool = False # rempli par api (best-effort) depuis le service favoris @classmethod def from_api(cls, raw: dict[str, Any]) -> "Recipe": @@ -117,7 +120,7 @@ class Recipe: "id": self.id, "name": self.name, "headline": self.headline, - "image_url": self.image_url, + "image_url": fix_image_url(self.image_url), "prep_time": self.prep_time, "prep_minutes": _iso_duration_to_minutes(self.prep_time), "allergens": self.allergens, @@ -125,6 +128,7 @@ class Recipe: "matched_excludes": self.matched_excludes, "score": self.score, "tags": self.tags, + "is_favorite": self.is_favorite, } diff --git a/server.py b/server.py index 86cc401..681827a 100644 --- a/server.py +++ b/server.py @@ -73,6 +73,55 @@ async def hf_account_info() -> dict: return await anyio.to_thread.run_sync(_impl) +@mcp.tool() +async def hf_next_delivery() -> dict: + """Prochaine livraison + recettes RÉELLEMENT sélectionnées (prêt à mettre en forme). + + Renvoie la semaine, la date de livraison, le cutoff, et les ~4 recettes de la box avec + image servable (URL corrigée), temps de prépa, allergènes, tags, `is_favorite`. Erreur + stricte si la sélection est introuvable — ne propose JAMAIS de recettes non sélectionnées. + """ + def _impl() -> dict: + with api.HelloFreshClient() as client: + acct = client.account_info() + nd = acct.get("next_delivery") or {} + week = nd.get("week") + if not week: + raise RuntimeError("Aucune prochaine livraison (abonnement en pause ?).") + recipes = client.get_current_selection(week) # lève si vide / KO + favs = client.favorite_ids() + hf_filter.annotate(recipes) + for r in recipes: + r.is_favorite = r.id in favs + return { + "week": week, + "delivery_date": nd.get("date"), + "cutoff": nd.get("cutoff"), + "count": len(recipes), + "recipes": [r.summary() for r in recipes], + } + + return await anyio.to_thread.run_sync(_impl) + + +@mcp.tool() +async def hf_favorites() -> dict: + """Recettes favorites du compte, complètes (image corrigée, allergènes, `is_favorite`). + + Lecture seule. Liste vide si aucun favori. Indépendant de la semaine (favoris globaux). + """ + def _impl() -> dict: + with api.HelloFreshClient() as client: + recipes = client.get_favorites() + hf_filter.annotate(recipes) + return { + "count": len(recipes), + "recipes": [r.summary() for r in recipes], + } + + return await anyio.to_thread.run_sync(_impl) + + @mcp.tool() async def hf_get_menu(week: str = "") -> dict: """Toutes les recettes proposées pour une semaine, chacune annotée. @@ -84,7 +133,10 @@ async def hf_get_menu(week: str = "") -> dict: w = week or api.current_week() with api.HelloFreshClient() as client: recipes = client.get_menu(w) + favs = client.favorite_ids() # best-effort (set() si indispo) hf_filter.annotate(recipes) + for r in recipes: + r.is_favorite = r.id in favs return { "week": w, "count": len(recipes), diff --git a/tools/probe_menu_capture.py b/tools/probe_menu_capture.py new file mode 100644 index 0000000..848231a --- /dev/null +++ b/tools/probe_menu_capture.py @@ -0,0 +1,104 @@ +"""Capture ciblée de l'appel « sélection courante » de la page menu (via CDP). + +Contrairement à attach_capture (passif), ce script PILOTE la page : il navigue lui-même +vers la page menu de la semaine cible, ce qui force le rechargement et fait partir l'appel +qui porte la sélection (probablement GET /gw/cart/{uuid}). Il isole les appels pertinents, +puis, s'il voit un GET /gw/cart/{uuid}, le rejoue côté client authentifié et dump le corps. + +Prérequis : Chrome lancé en debug + connecté (cf. README §2) : + "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" \\ + --remote-debugging-port=9222 --user-data-dir="$HOME/.hf-chrome-debug" \\ + https://www.hellofresh.fr/my-account/deliveries/menu + +Usage : + python tools/probe_menu_capture.py [WEEK] # WEEK ex. 2026-W27 (défaut: prochaine livraison) +""" + +from __future__ import annotations + +import json +import re +import sys +import time +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 api, auth # noqa: E402 + +CDP_URL = "http://localhost:9222" +OUT = auth.SESSION_DIR / "_probe" +RELEVANT = ("/cart", "/v1/carts", "menus-service", "/meal", "selection", "/box") +UUID_RE = re.compile(r"/gw/cart/([0-9a-f]{8}-[0-9a-f-]{27,})") + + +def main() -> None: + OUT.mkdir(parents=True, exist_ok=True) + week = sys.argv[1] if len(sys.argv) > 1 else None + if not week: + with api.HelloFreshClient() as c: + week = (c.account_info().get("next_delivery") or {}).get("week") or api.current_week() + menu_url = f"{auth.BASE_URL}/my-account/deliveries/menu?weekId={week}" + print(f"[probe] semaine cible = {week}") + + hits: list[dict] = [] + + def on_request(req): + u = req.url + if not auth._is_gateway_request(u): + return + if any(k in u for k in RELEVANT): + hits.append({"method": req.method, "url": u, + "auth": bool(req.headers.get("authorization"))}) + print(f" [{req.method}] {u[:150]}") + + with sync_playwright() as pw: + print(f"[probe] connexion CDP {CDP_URL} …") + browser = pw.chromium.connect_over_cdp(CDP_URL) + ctx = browser.contexts[0] + for p in ctx.pages: + p.on("request", on_request) + ctx.on("page", lambda p: p.on("request", on_request)) + + page = ctx.pages[0] if ctx.pages else ctx.new_page() + print(f"[probe] navigation -> {menu_url}") + try: + page.goto(menu_url, wait_until="domcontentloaded", timeout=45000) + except Exception as e: # noqa: BLE001 + print(" (goto a levé, on continue d'écouter)", e) + for _ in range(20): + page.wait_for_timeout(1000) + + print(f"\n[probe] {len(hits)} appels pertinents capturés.") + cart_uuids = [] + for h in hits: + m = UUID_RE.search(h["url"]) + if m and h["method"] == "GET": + cart_uuids.append(m.group(1)) + cart_uuids = list(dict.fromkeys(cart_uuids)) + (OUT / "menu_capture_hits.json").write_text( + json.dumps(hits, ensure_ascii=False, indent=2), encoding="utf-8") + + if cart_uuids: + print(f"[probe] cart UUID(s) détecté(s): {cart_uuids}") + with api.HelloFreshClient() as c: + for uid in cart_uuids: + for suffix in ("", "/items"): + url = f"{auth.BASE_URL}/gw/cart/{uid}{suffix}" + try: + r = c._request("GET", url, params={"country": "FR", "locale": "fr-FR"}) + name = f"cart_{uid[:8]}{suffix.replace('/', '_')}" + (OUT / f"{name}.json").write_text( + json.dumps(r.json(), ensure_ascii=False, indent=2), encoding="utf-8") + print(f" {r.status_code} {url} -> {name}.json") + except Exception as e: # noqa: BLE001 + print(f" ERR {url}: {e}") + else: + print("[probe] aucun GET /gw/cart/{uuid} vu. Voir menu_capture_hits.json pour les autres pistes.") + + +if __name__ == "__main__": + main() diff --git a/tools/probe_selection.py b/tools/probe_selection.py new file mode 100644 index 0000000..e412191 --- /dev/null +++ b/tools/probe_selection.py @@ -0,0 +1,155 @@ +"""ÉTAPE 1 — Probe de découverte (jetable). + +Sonde, sur le compte abonné réel, trois inconnues impossibles à lever en lecture seule : + 1. SÉLECTION : quel GET renvoie les recettes RÉELLEMENT dans la box de la prochaine + livraison (≠ menu complet), et sous quelle forme (ids vs index de course). + 2. IMAGE : quelle transformation rend une URL d'image téléchargeable (les URLs brutes + cloudfront renvoient des 502). + 3. FAVORIS : quel endpoint gateway sert la page /recipes/favorites, et sa forme. + +Réutilise HelloFreshClient (token + re-auth 401 déjà gérés). Ce sont des GET → lecture seule. + +Usage : + python tools/probe_selection.py +""" + +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 hellofresh import api # noqa: E402 + + +def _short(obj, n: int = 1400) -> str: + """JSON tronqué pour garder la sortie lisible.""" + s = json.dumps(obj, ensure_ascii=False, indent=2) + return s if len(s) <= n else s[:n] + f"\n… (+{len(s) - n} chars)" + + +def _keys(obj) -> str: + if isinstance(obj, dict): + return "dict keys=" + ", ".join(list(obj.keys())[:25]) + if isinstance(obj, list): + head = obj[0] if obj else None + return f"list len={len(obj)} head={_keys(head) if isinstance(head, (dict, list)) else type(head).__name__}" + return type(obj).__name__ + + +def _get(client: api.HelloFreshClient, url: str, params: dict) -> None: + print(f"\n>>> GET {url}\n params={params}") + try: + resp = client._client.request("GET", url, params=params) + if resp.status_code == 401: + from hellofresh import auth + client._token = auth.get_token(force=True) + client._client.headers["Authorization"] = f"Bearer {client._token}" + resp = client._client.request("GET", url, params=params) + print(f" -> {resp.status_code} {resp.headers.get('content-type', '')}") + if resp.status_code >= 400: + print(f" body[:300]={resp.text[:300]!r}") + return + try: + data = resp.json() + except Exception: + print(f" (non-JSON) body[:300]={resp.text[:300]!r}") + return + print(f" shape: {_keys(data)}") + print(" " + _short(data).replace("\n", "\n ")) + except Exception as e: # noqa: BLE001 + print(f" !! exception: {type(e).__name__}: {e}") + + +def main() -> None: + base = "https://www.hellofresh.fr/gw" + with api.HelloFreshClient() as client: + country, locale = client._country, client._locale + sub = client._subscription_info() + acct = client.account_info() + nd = acct.get("next_delivery") or {} + week = nd.get("week") or api.current_week() + deliv = next((d for d in client._deliveries() if d.week == week), None) + cutoff = deliv.cutoff_date if deliv else (nd.get("cutoff") or "") + + print("=" * 70) + print(f"sub_id={sub['sub_id']} customer_id={sub['customer_id']} sku={sub['sku']}") + print(f"next_delivery week={week} date={nd.get('date')} cutoff={cutoff}") + print(f"meals attendus = {acct.get('subscription', {}).get('meals')}") + print("=" * 70) + + # --- 1. SÉLECTION --------------------------------------------------- + full_params = { + "customer": sub["customer_id"], "subscription": sub["sub_id"], + "product-sku": sub["sku"], "week": week, "cutoff_time": cutoff, + "country": country, "locale": locale, + } + print("\n########## 1. SÉLECTION ##########") + # Candidat A — symétrique du PUT /gw/v1/carts/{week} + _get(client, f"{base}/v1/carts/{week}", full_params) + _get(client, f"{base}/v1/carts/{week}", {"country": country, "locale": locale}) + # Candidat C — /deliveries porte peut-être la sélection : dump brut du Delivery + print("\n>>> RAW /deliveries (cherche meals/courses/selectedRecipes/cart par semaine)") + draw = client._request("GET", client._ep["weeks"], params={ + "country": country, "locale": locale, + "rangeStart": api.current_week(0), "rangeEnd": api.current_week(10), + }).json() + items = draw.get("items", []) if isinstance(draw, dict) else (draw or []) + match = next((w for w in items if str(w.get("id") or w.get("week")) == week), None) + print(" " + _short(match or {"_no_match_for_week": week}).replace("\n", "\n ")) + # Candidat B — cartId réel de l'abonné, cherché dans /subscriptions brut + print("\n>>> RAW /subscriptions (cherche un id de cart hebdo)") + sraw = client._request("GET", client._ep["subscriptions"], + params={"country": country}).json() + sitems = sraw.get("items", []) if isinstance(sraw, dict) else (sraw or []) + print(" " + _short(sitems[0] if sitems else sraw).replace("\n", "\n ")) + + # --- 2. IMAGE ------------------------------------------------------- + print("\n########## 2. IMAGE ##########") + recipes = client.get_menu(week) + raw_url = next((r.image_url for r in recipes if r.image_url), "") + print(f"image brute = {raw_url}") + if raw_url: + from urllib.parse import urlsplit, urlunsplit + parts = urlsplit(raw_url) + for host in ("media.hellofresh.com", "img.hellofresh.com"): + cand = urlunsplit((parts.scheme, host, parts.path, parts.query, parts.fragment)) + try: + r = client._client.request("GET", cand, headers={"Range": "bytes=0-0"}) + print(f" {host:24} -> {r.status_code} {r.headers.get('content-type', '')}") + except Exception as e: # noqa: BLE001 + print(f" {host:24} -> !! {type(e).__name__}: {e}") + try: + r = client._client.request("GET", raw_url, headers={"Range": "bytes=0-0"}) + print(f" {'(brute)':24} -> {r.status_code} {r.headers.get('content-type', '')}") + except Exception as e: # noqa: BLE001 + print(f" {'(brute)':24} -> !! {type(e).__name__}: {e}") + + # --- 3. FAVORIS (endpoint réel capturé via CDP) -------------------- + print("\n########## 3. FAVORIS ##########") + _get(client, f"{base}/cfs/v2/favorites/recipe", + {"country": country, "locale": locale, "ids": ""}) + + # --- 4. PLAN (candidat sélection) ---------------------------------- + print("\n########## 4. PLAN (candidat sélection) ##########") + # planId vu dans /gw/api/plans/{id} ; on le cherche dans subscriptions/me brut + plan_id = None + for cand in (sitems[0] if sitems else {}, sraw if isinstance(sraw, dict) else {}): + if isinstance(cand, dict): + for k in ("planId", "plan_id"): + if cand.get(k): + plan_id = cand[k] + meraw = client._request("GET", f"{base}/api/customers/me", + params={"country": country, "locale": locale}).json() + print(" me keys:", _keys(meraw)) + if plan_id: + _get(client, f"{base}/api/plans/{plan_id}", {}) + else: + print(" (planId introuvable dans subscriptions/me — voir dump deliveries ci-dessus)") + + +if __name__ == "__main__": + main()