03e281a810
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
105 lines
4.1 KiB
Python
105 lines
4.1 KiB
Python
"""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()
|