"""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 Recipe, Week, _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}" 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 get_editable_weeks(self) -> list[Week]: """Semaines de livraison de l'abonnement (vide si pas d'abonnement actif).""" 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 [Week.from_api(w) for w in items] def get_menu(self, week: str | None = None) -> list[Recipe]: """Recettes proposées pour une semaine (défaut : semaine courante), complètes. Deux appels : le menu (ids) puis le batch de détails (ingrédients/allergènes). """ week = week or current_week() menu_params = { "country": self._country, "locale": self._locale, "product": self._product, "weeks": week, } menu = self._request("GET", self._ep["menu"], params=menu_params).json() items = menu.get("items", []) if isinstance(menu, dict) else [] if not items: return [] courses = items[0].get("courses", []) ids: list[str] = [] for c in courses: rid = (c.get("recipe") or {}).get("id") if rid and rid not in ids: ids.append(rid) return self._fetch_details(ids) 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 set_selection(self, week: str, recipe_ids: list[str]) -> dict[str, Any]: """ÉCRIT la sélection de recettes pour une semaine (après confirmation). Endpoint à découvrir sur un compte avec abonnement actif (cf. set_selection vide). """ url = self._ep.get("set_selection") or "" if not url: raise EndpointsNotConfigured( "Endpoint 'set_selection' inconnu : à capturer via discover_api.py sur un " "compte avec box active (modifier une recette pour observer l'appel PUT/POST)." ) url = url.replace("{week}", str(week)) method = self._ep.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}