61ee7f02a4
- Refus des recettes payantes (chargeSetting) à la sélection, override allow_premium - Recipe.surcharge_cents/is_premium exposés dans summary(); propose() les exclut - hellofresh/webui.py : page d'admin + API JSON montées sur FastMCP (/, /api/*) édition à chaud des excludes et préférences (liked/disliked)
385 lines
16 KiB
Python
385 lines
16 KiB
Python
"""Client httpx vers l'API interne HelloFresh (région FR).
|
|
|
|
Flux confirmé par discovery + tests (cf. config/endpoints.json) :
|
|
1. MENU GET menus-service/menus?country=FR&locale=fr-FR&weeks={week}&product=classic-box
|
|
-> items[0].courses[*].recipe (ids + nom, MAIS sans ingrédients)
|
|
2. DÉTAILS GET recipes/recipes?ids=a,b,c&country=FR&locale=fr-FR&take=N
|
|
-> items[*] (recettes complètes : ingrédients + allergènes) — 1 seul appel
|
|
3. SEMAINES GET api/customers/me/deliveries (vide si pas d'abonnement actif)
|
|
|
|
Le token bearer vient d'`auth.get_token()`. `set_selection` (écriture) reste à découvrir
|
|
sur un compte avec abonnement actif → l'appel lève une erreur explicite tant qu'il est vide.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import datetime
|
|
import json
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from . import auth
|
|
from .models import Delivery, Recipe, _first
|
|
|
|
ENDPOINTS_PATH = auth.ROOT / "config" / "endpoints.json"
|
|
BATCH_SIZE = 80 # nb max d'ids par appel détails (le menu en compte ~85)
|
|
|
|
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"))
|
|
for key in ("menu", "recipe_details"):
|
|
if not data.get(key):
|
|
raise EndpointsNotConfigured(
|
|
f"Endpoint '{key}' manquant dans {ENDPOINTS_PATH}. Rejouer discover_api.py."
|
|
)
|
|
return data
|
|
|
|
|
|
def current_week(offset: int = 0) -> str:
|
|
"""Handle de semaine ISO 'YYYY-Www' (offset en nb de semaines)."""
|
|
d = datetime.date.today() + datetime.timedelta(weeks=offset)
|
|
y, w, _ = d.isocalendar()
|
|
return f"{y}-W{w:02d}"
|
|
|
|
|
|
def _dedupe_by_name(recipes: list[Recipe]) -> list[Recipe]:
|
|
"""Fusionne les variantes du même plat (même nom, ids différents : 2/4 pers., etc.).
|
|
|
|
On garde la première occurrence ; les ids des variantes restent valides pour la
|
|
sélection puisqu'ils pointent vers la même recette.
|
|
"""
|
|
seen: set[str] = set()
|
|
out: list[Recipe] = []
|
|
for r in recipes:
|
|
key = " ".join((r.name or "").lower().split())
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(r)
|
|
return out
|
|
|
|
|
|
class HelloFreshClient:
|
|
def __init__(self, token: str | None = None):
|
|
self._ep = _load_endpoints()
|
|
self._country = self._ep.get("country", "FR")
|
|
self._locale = self._ep.get("locale", "fr-FR")
|
|
self._product = self._ep.get("product", "classic-box")
|
|
self._token = token or auth.get_token()
|
|
self._client = httpx.Client(
|
|
headers={**DEFAULT_HEADERS, "Authorization": f"Bearer {self._token}"},
|
|
timeout=60.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ête 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:
|
|
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 _deliveries(self) -> list[Delivery]:
|
|
params = {
|
|
"country": self._country,
|
|
"locale": self._locale,
|
|
"rangeStart": current_week(0),
|
|
"rangeEnd": current_week(10),
|
|
}
|
|
data = self._request("GET", self._ep["weeks"], params=params).json()
|
|
items = data.get("items", []) if isinstance(data, dict) else (data or [])
|
|
return [Delivery.from_api(w) for w in items]
|
|
|
|
def get_editable_weeks(self) -> list[Delivery]:
|
|
"""Semaines de livraison encore modifiables (vide si pas d'abonnement actif)."""
|
|
return [d for d in self._deliveries() if d.editable]
|
|
|
|
def _subscription_info(self) -> dict[str, Any]:
|
|
"""Récupère {sub_id, customer_id, sku} depuis l'abonnement (pas de valeurs en dur)."""
|
|
ep = self._ep.get("subscriptions")
|
|
if not ep:
|
|
raise EndpointsNotConfigured("Endpoint 'subscriptions' manquant.")
|
|
data = self._request("GET", ep, params={"country": self._country}).json()
|
|
if isinstance(data, dict) and "id" in data:
|
|
sub = data
|
|
else:
|
|
items = data.get("items", []) if isinstance(data, dict) else (data or [])
|
|
if not items:
|
|
raise RuntimeError("Aucun abonnement actif sur ce compte.")
|
|
sub = items[0]
|
|
cust = sub.get("customer") or {}
|
|
prod = sub.get("product") or {}
|
|
return {"sub_id": str(sub.get("id", "")),
|
|
"customer_id": str(cust.get("id", "")),
|
|
"sku": str(prod.get("sku", ""))}
|
|
|
|
def account_info(self) -> dict[str, Any]:
|
|
"""Résumé lisible du compte : client, abonnement, adresse, prochaine livraison."""
|
|
ep = self._ep.get("subscriptions")
|
|
if not ep:
|
|
raise EndpointsNotConfigured("Endpoint 'subscriptions' manquant.")
|
|
data = self._request("GET", ep, params={"country": self._country}).json()
|
|
items = data.get("items", []) if isinstance(data, dict) else (data or [])
|
|
if not items:
|
|
raise RuntimeError("Aucun abonnement actif sur ce compte.")
|
|
sub = items[0]
|
|
cust = sub.get("customer") or {}
|
|
ptype = sub.get("productType") or {}
|
|
prod = sub.get("product") or {}
|
|
ship = sub.get("shippingAddress") or {}
|
|
|
|
def _addr(a: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"name": " ".join(p for p in (a.get("firstName"), a.get("lastName")) if p),
|
|
"address": a.get("address1"),
|
|
"postcode": a.get("postcode"),
|
|
"city": a.get("city"),
|
|
"country": (a.get("country") or {}).get("iso2Code"),
|
|
"phone": a.get("phone"),
|
|
}
|
|
|
|
unit_price = prod.get("unitPrice")
|
|
return {
|
|
"customer": {
|
|
"id": cust.get("id"),
|
|
"email": cust.get("email"),
|
|
"first_name": cust.get("firstName"),
|
|
"last_name": cust.get("lastName"),
|
|
"locale": cust.get("locale"),
|
|
"loyalty_points": (cust.get("loyalty") or {}).get("value"),
|
|
},
|
|
"subscription": {
|
|
"id": sub.get("id"),
|
|
"active": sub.get("isActive"),
|
|
"paused_at": sub.get("pausedAt"),
|
|
"canceled_at": sub.get("canceledAt"),
|
|
"blocked": sub.get("isBlocked"),
|
|
"sku": prod.get("sku") or ptype.get("handle"),
|
|
"product_name": ptype.get("productName"),
|
|
"meals": (ptype.get("specs") or {}).get("meals"),
|
|
"people": (ptype.get("specs") or {}).get("size"),
|
|
"box_price_eur": unit_price / 100 if isinstance(unit_price, int) else None,
|
|
"preset": sub.get("preset"),
|
|
"delivery_weekday": sub.get("deliveryWeekday"),
|
|
"delivery_interval": sub.get("deliveryInterval"),
|
|
"payment_method": sub.get("paymentMethod"),
|
|
},
|
|
"shipping_address": _addr(ship),
|
|
"next_delivery": {
|
|
"week": sub.get("nextDeliveryWeek"),
|
|
"date": sub.get("nextDelivery"),
|
|
"cutoff": sub.get("nextCutoffDate"),
|
|
},
|
|
}
|
|
|
|
def _menu_courses(self, week: str) -> list[dict]:
|
|
"""Courses bruts du menu (chacun : index + recipe). Base de l'écriture."""
|
|
params = {
|
|
"country": self._country,
|
|
"locale": self._locale,
|
|
"product": self._product,
|
|
"weeks": week,
|
|
}
|
|
menu = self._request("GET", self._ep["menu"], params=params).json()
|
|
items = menu.get("items", []) if isinstance(menu, dict) else []
|
|
return items[0].get("courses", []) if items else []
|
|
|
|
def get_menu(self, week: str | None = None) -> list[Recipe]:
|
|
"""Recettes proposées pour une semaine (défaut : courante), complètes.
|
|
|
|
Deux appels : menu (ids + index de course) puis batch de détails
|
|
(ingrédients/allergènes). L'`index` de course est conservé pour l'écriture.
|
|
"""
|
|
week = week or current_week()
|
|
courses = self._menu_courses(week)
|
|
if not courses:
|
|
return []
|
|
id_index: dict[str, int] = {}
|
|
charge_by_id: dict[str, dict] = {}
|
|
ids: list[str] = []
|
|
for c in courses:
|
|
rid = (c.get("recipe") or {}).get("id")
|
|
if rid and rid not in id_index:
|
|
id_index[rid] = c.get("index")
|
|
cs = c.get("chargeSetting")
|
|
if isinstance(cs, dict):
|
|
charge_by_id[rid] = cs
|
|
ids.append(rid)
|
|
recipes = self._fetch_details(ids)
|
|
for r in recipes:
|
|
r.course_index = id_index.get(r.id)
|
|
cs = charge_by_id.get(r.id)
|
|
if cs:
|
|
r.surcharge_cents = int(cs.get("amount") or 0)
|
|
r.surcharge_reason = str(cs.get("reason") or "")
|
|
return _dedupe_by_name(recipes)
|
|
|
|
def _fetch_details(self, ids: list[str]) -> list[Recipe]:
|
|
"""Récupère les recettes complètes par batch d'ids."""
|
|
recipes: list[Recipe] = []
|
|
for i in range(0, len(ids), BATCH_SIZE):
|
|
chunk = ids[i:i + BATCH_SIZE]
|
|
params = {
|
|
"ids": ",".join(chunk),
|
|
"country": self._country,
|
|
"locale": self._locale,
|
|
"take": str(len(chunk)),
|
|
}
|
|
data = self._request("GET", self._ep["recipe_details"], params=params).json()
|
|
raw = data.get("items", []) if isinstance(data, dict) else (data or [])
|
|
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.
|
|
|
|
Renvoie (indices_trouvés, ids_inconnus).
|
|
"""
|
|
id_index = {(c.get("recipe") or {}).get("id"): c.get("index")
|
|
for c in self._menu_courses(week)}
|
|
indices, unknown = [], []
|
|
for rid in recipe_ids:
|
|
idx = id_index.get(rid)
|
|
if idx is None:
|
|
unknown.append(rid)
|
|
else:
|
|
indices.append(idx)
|
|
return indices, unknown
|
|
|
|
def set_selection(self, week: str, recipe_ids: list[str], dry_run: bool = False) -> dict[str, Any]:
|
|
"""ÉCRIT la sélection de recettes pour une semaine (PUT du cart hebdo).
|
|
|
|
La sélection se fait par `index` de course (mappé depuis les ids de recette).
|
|
`dry_run=True` construit la requête sans l'envoyer (pour vérification).
|
|
"""
|
|
url = (self._ep.get("set_selection") or "").replace("{week}", str(week))
|
|
if not url:
|
|
raise EndpointsNotConfigured("Endpoint 'set_selection' non configuré.")
|
|
|
|
indices, unknown = self.selection_indices(week, recipe_ids)
|
|
if unknown:
|
|
raise ValueError(f"Recette(s) absente(s) du menu {week}: {unknown}")
|
|
|
|
sub = self._subscription_info()
|
|
deliv = next((d for d in self._deliveries() if d.week == week), None)
|
|
if deliv and not deliv.editable:
|
|
raise ValueError(f"Semaine {week} non modifiable (statut {deliv.status}).")
|
|
|
|
params = {
|
|
"customer": sub["customer_id"],
|
|
"subscription": sub["sub_id"],
|
|
"product-sku": sub["sku"],
|
|
"week": week,
|
|
"cutoff_time": deliv.cutoff_date if deliv else "",
|
|
"update_quantity": "true",
|
|
"ignore_addons": "false",
|
|
"preference": self._ep.get("selection_preference", "quick"),
|
|
}
|
|
body = {"meals": [{"index": i, "quantity": 1} for i in indices], "extras": []}
|
|
|
|
if dry_run:
|
|
return {"dry_run": True, "url": url, "params": params, "body": body}
|
|
|
|
method = self._ep.get("set_selection_method", "PUT").upper()
|
|
resp = self._request(method, url, params=params, json=body)
|
|
try:
|
|
return {"ok": True, "status": resp.status_code, "response": resp.json()}
|
|
except Exception:
|
|
return {"ok": True, "status": resp.status_code}
|