Sélection courante + favoris + images servables (sortie prête Hermes)

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
This commit is contained in:
jerem
2026-06-18 14:18:40 +02:00
parent 4b1eb9f52c
commit 03e281a810
8 changed files with 489 additions and 10 deletions

View File

@@ -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` **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, (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. 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 > ⚠️ 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** > 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. > (`tools/attach_capture.py`), où le login marche normalement.
@@ -44,17 +50,29 @@ playwright install chromium
cp .env.example .env cp .env.example .env
``` ```
### 2. Créer la session (login via TON Chrome, anti-bot contourné) ### 2. Créer (ou ré-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 C'est **la seule méthode de login qui marche** : un navigateur piloté par Playwright (même avec
ouverte), connecte-toi à HelloFresh (email + mot de passe), puis attache la capture : `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 ```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" \ "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" \
--remote-debugging-port=9222 --user-data-dir="$HOME/.hf-chrome-debug" \ --remote-debugging-port=9222 --user-data-dir="$HOME/.hf-chrome-debug" \
https://www.hellofresh.fr/my-account/deliveries/menu 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 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 `attach_capture` doit tourner **pendant** que tu (re)charges la page menu, pour capter le trafic.
déjà rempli ; rejoue `attach_capture` si l'API change (cf. `config/endpoints_discovered.json`). 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) ### 3. Tester en local (headless, comme le homelab)
```bash ```bash
@@ -97,7 +115,9 @@ Enregistrer AntiCoco dans la config MCP de Hermes (côté homelab), URL
| `hf_auth_status()` | état de connexion | | `hf_auth_status()` | état de connexion |
| `hf_login()` | (re)connexion + capture token | | `hf_login()` | (re)connexion + capture token |
| `hf_list_weeks()` | semaines modifiables | | `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_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_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 | | `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/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/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).

View File

@@ -10,5 +10,10 @@
"subscriptions": "https://www.hellofresh.fr/gw/api/customers/me/subscriptions", "subscriptions": "https://www.hellofresh.fr/gw/api/customers/me/subscriptions",
"set_selection": "https://www.hellofresh.fr/gw/v1/carts/{week}", "set_selection": "https://www.hellofresh.fr/gw/v1/carts/{week}",
"set_selection_method": "PUT", "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"]
} }

View File

@@ -255,6 +255,69 @@ class HelloFreshClient:
recipes.extend(Recipe.from_api(r) for r in raw) recipes.extend(Recipe.from_api(r) for r in raw)
return recipes 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]]: 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. """Mappe des ids de recette vers leurs index de course pour la semaine.

73
hellofresh/images.py Normal file
View File

@@ -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/<transfo>/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}"

View File

@@ -11,6 +11,8 @@ import re
from dataclasses import dataclass, field, asdict from dataclasses import dataclass, field, asdict
from typing import Any from typing import Any
from .images import fix_image_url
_ISO_DURATION = re.compile(r"PT(?:(\d+)H)?(?:(\d+)M)?", re.IGNORECASE) _ISO_DURATION = re.compile(r"PT(?:(\d+)H)?(?:(\d+)M)?", re.IGNORECASE)
@@ -54,6 +56,7 @@ class Recipe:
contains_excluded: bool = False contains_excluded: bool = False
matched_excludes: list[str] = field(default_factory=list) matched_excludes: list[str] = field(default_factory=list)
score: float = 0.0 score: float = 0.0
is_favorite: bool = False # rempli par api (best-effort) depuis le service favoris
@classmethod @classmethod
def from_api(cls, raw: dict[str, Any]) -> "Recipe": def from_api(cls, raw: dict[str, Any]) -> "Recipe":
@@ -117,7 +120,7 @@ class Recipe:
"id": self.id, "id": self.id,
"name": self.name, "name": self.name,
"headline": self.headline, "headline": self.headline,
"image_url": self.image_url, "image_url": fix_image_url(self.image_url),
"prep_time": self.prep_time, "prep_time": self.prep_time,
"prep_minutes": _iso_duration_to_minutes(self.prep_time), "prep_minutes": _iso_duration_to_minutes(self.prep_time),
"allergens": self.allergens, "allergens": self.allergens,
@@ -125,6 +128,7 @@ class Recipe:
"matched_excludes": self.matched_excludes, "matched_excludes": self.matched_excludes,
"score": self.score, "score": self.score,
"tags": self.tags, "tags": self.tags,
"is_favorite": self.is_favorite,
} }

View File

@@ -73,6 +73,55 @@ async def hf_account_info() -> dict:
return await anyio.to_thread.run_sync(_impl) 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() @mcp.tool()
async def hf_get_menu(week: str = "") -> dict: async def hf_get_menu(week: str = "") -> dict:
"""Toutes les recettes proposées pour une semaine, chacune annotée. """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() w = week or api.current_week()
with api.HelloFreshClient() as client: with api.HelloFreshClient() as client:
recipes = client.get_menu(w) recipes = client.get_menu(w)
favs = client.favorite_ids() # best-effort (set() si indispo)
hf_filter.annotate(recipes) hf_filter.annotate(recipes)
for r in recipes:
r.is_favorite = r.id in favs
return { return {
"week": w, "week": w,
"count": len(recipes), "count": len(recipes),

104
tools/probe_menu_capture.py Normal file
View File

@@ -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()

155
tools/probe_selection.py Normal file
View File

@@ -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()