jerem
ef6bf9813a
- Endpoints découverts: menu (menus-service) + détails batch (recipes/recipes) - get_menu en 2 temps: menu (ids) -> batch détails (ingrédients/allergènes) - Fix faux positifs: exclusion sur ingrédients/allergènes/nom, plus sur les tags (HelloFresh pose un tag interne 'coconut' sur ~la moitié des recettes) - Token mis en cache (pas de navigateur si frais) - endpoints.json versionné (sans secret), semaine optionnelle (défaut = courante) - Testé: 4 recettes coco/85 détectées, shortlist classée, tous les outils MCP OK - set_selection (écriture) reste à découvrir sur un compte avec box active
117 lines
4.5 KiB
Python
117 lines
4.5 KiB
Python
"""ÉTAPE 0 — Découverte de l'API interne HelloFresh.
|
|
|
|
Lance un navigateur visible, te laisse te connecter et naviguer (menu de la semaine,
|
|
sélection de recettes), puis enregistre TOUTES les requêtes vers le gateway pour en
|
|
déduire les 3 endpoints utiles :
|
|
1. abonnement + semaines éditables
|
|
2. menu d'une semaine (recettes / ingrédients / allergènes)
|
|
3. enregistrement de la sélection de recettes
|
|
|
|
Usage :
|
|
ANTICOCO_HEADLESS=0 python tools/discover_api.py
|
|
|
|
Pendant que la fenêtre est ouverte :
|
|
- connecte-toi,
|
|
- ouvre le menu de la semaine,
|
|
- (optionnel) change une recette pour capturer l'appel d'écriture,
|
|
puis reviens dans le terminal et appuie sur Entrée pour écrire le rapport.
|
|
|
|
Sortie :
|
|
- .session/discovery_log.json : toutes les requêtes gateway observées (debug complet)
|
|
- config/endpoints.json : squelette pré-rempli à compléter/valider à la main
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
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 auth # noqa: E402
|
|
|
|
LOG_PATH = auth.SESSION_DIR / "discovery_log.json"
|
|
ENDPOINTS_PATH = ROOT / "config" / "endpoints.json"
|
|
|
|
|
|
def main() -> None:
|
|
auth.SESSION_DIR.mkdir(parents=True, exist_ok=True)
|
|
requests_seen: list[dict] = []
|
|
|
|
with sync_playwright() as pw:
|
|
ctx = pw.chromium.launch_persistent_context(
|
|
user_data_dir=str(auth.PROFILE_DIR),
|
|
headless=False, # toujours visible : c'est une étape interactive
|
|
locale="fr-FR",
|
|
viewport={"width": 1280, "height": 900},
|
|
)
|
|
page = ctx.pages[0] if ctx.pages else ctx.new_page()
|
|
|
|
def on_request(req):
|
|
if not auth._is_gateway_request(req.url):
|
|
return
|
|
entry = {
|
|
"method": req.method,
|
|
"url": req.url,
|
|
"has_auth": bool(req.headers.get("authorization")),
|
|
}
|
|
if req.method in ("POST", "PUT", "PATCH"):
|
|
try:
|
|
entry["post_data"] = req.post_data
|
|
except Exception:
|
|
entry["post_data"] = None
|
|
requests_seen.append(entry)
|
|
print(f" [{req.method}] {req.url}")
|
|
|
|
page.on("request", on_request)
|
|
|
|
print("Ouvre le menu de la semaine, change une recette si tu veux capturer l'écriture.")
|
|
page.goto(auth.BASE_URL + "/my-account", wait_until="domcontentloaded", timeout=30000)
|
|
|
|
# Si on a un vrai terminal : on attend Entrée. Sinon (lancé en arrière-plan,
|
|
# sans TTY) : on attend une durée fixe pour laisser le temps de se connecter
|
|
# et de naviguer, tout en continuant à logger les requêtes.
|
|
if sys.stdin and sys.stdin.isatty():
|
|
try:
|
|
input("\n>>> Quand tu as fini de naviguer, appuie sur Entrée pour générer le rapport...\n")
|
|
except (EOFError, KeyboardInterrupt):
|
|
pass
|
|
else:
|
|
wait_s = int(os.environ.get("ANTICOCO_DISCOVER_WAIT", "240"))
|
|
print(f"\n>>> Pas de terminal interactif : capture pendant {wait_s}s. "
|
|
"Connecte-toi et navigue dans la fenêtre ouverte...")
|
|
page.wait_for_timeout(wait_s * 1000)
|
|
|
|
ctx.close()
|
|
|
|
LOG_PATH.write_text(json.dumps(requests_seen, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
print(f"\n{len(requests_seen)} requêtes gateway enregistrées dans {LOG_PATH}")
|
|
|
|
# Heuristiques pour pré-remplir le squelette d'endpoints.
|
|
def find(method: str, *needles: str) -> str:
|
|
for r in requests_seen:
|
|
if r["method"] != method:
|
|
continue
|
|
if all(n in r["url"].lower() for n in needles):
|
|
return r["url"]
|
|
return ""
|
|
|
|
skeleton = {
|
|
"_comment": "Endpoints HelloFresh confirmés via discovery. Compléter/valider à la main. Utiliser {week} comme placeholder pour le handle de semaine.",
|
|
"base": "",
|
|
"weeks": find("GET", "subscription") or find("GET", "deliveries") or "",
|
|
"menu": find("GET", "menu") or find("GET", "courses") or "",
|
|
"set_selection": find("PUT", "menu") or find("POST", "menu") or "",
|
|
}
|
|
ENDPOINTS_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
ENDPOINTS_PATH.write_text(json.dumps(skeleton, indent=2, ensure_ascii=False), encoding="utf-8")
|
|
print(f"Squelette d'endpoints écrit dans {ENDPOINTS_PATH} — à vérifier avant usage.")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|