Écriture de sélection câblée (PUT cart) + auth par cookie storage_state

Découvert via attache CDP au vrai Chrome (contourne le blocage automation) :
- set_selection = PUT /gw/v1/carts/{week}, body {meals:[{index,quantity}], extras:[]}
  sélection par index de course, params (customer/subscription/sku/cutoff) dérivés
  dynamiquement de /subscriptions + /deliveries (aucun id en dur)
- Recipe.course_index conservé depuis le menu pour le mapping id->index
- get_editable_weeks via /deliveries (modèle Delivery: cutoff, status, editable)
- Token lu depuis le cookie apiV2Auth (storage_state) -> auth sans navigateur, headless OK
- hf_confirm_selection: garde-fou coco + dry_run; tool attach_capture.py ajouté
- Dry-run validé: requête identique à l'appel réel capturé
This commit is contained in:
2026-06-15 22:57:36 +02:00
parent 30b950ec41
commit 051ecb50d8
9 changed files with 388 additions and 69 deletions

View File

@@ -43,12 +43,17 @@ def main() -> None:
requests_seen: list[dict] = []
with sync_playwright() as pw:
ctx = pw.chromium.launch_persistent_context(
kwargs = dict(
user_data_dir=str(auth.PROFILE_DIR),
headless=False, # toujours visible : c'est une étape interactive
locale="fr-FR",
viewport={"width": 1280, "height": 900},
)
channel = auth._channel()
if channel:
kwargs["channel"] = channel
print(f"Navigateur : {channel}")
ctx = pw.chromium.launch_persistent_context(**kwargs)
page = ctx.pages[0] if ctx.pages else ctx.new_page()
def on_request(req):
@@ -67,49 +72,71 @@ def main() -> None:
requests_seen.append(entry)
print(f" [{req.method}] {req.url}")
closed = {"v": False}
ctx.on("close", lambda *_: closed.update(v=True))
page.on("close", lambda *_: closed.update(v=True))
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)
try:
page.goto(auth.BASE_URL + "/my-account", wait_until="domcontentloaded", timeout=30000)
except Exception as e:
print("goto initial échoué (on continue quand même):", e)
# 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():
# On capture jusqu'à : Entrée (si TTY), fermeture de la fenêtre, ou fin du délai.
# Le rapport est TOUJOURS écrit (finally) même si la fenêtre est fermée en cours.
try:
if sys.stdin and sys.stdin.isatty():
try:
input("\n>>> Quand tu as fini, appuie sur Entrée (ou ferme la fenêtre)...\n")
except (EOFError, KeyboardInterrupt):
pass
else:
wait_s = int(os.environ.get("ANTICOCO_DISCOVER_WAIT", "240"))
print(f"\n>>> Capture pendant {wait_s}s max. Connecte-toi (email + mot de passe, "
"PAS Google), ouvre le menu, change une recette. Ferme la fenêtre quand fini.")
waited = 0
while waited < wait_s and not closed["v"]:
try:
page.wait_for_timeout(2000)
except Exception:
break # page/contexte fermé
waited += 2
finally:
_save_report(requests_seen)
try:
input("\n>>> Quand tu as fini de naviguer, appuie sur Entrée pour générer le rapport...\n")
except (EOFError, KeyboardInterrupt):
if not closed["v"]:
ctx.close()
except Exception:
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()
def _save_report(requests_seen: list[dict]) -> None:
"""Écrit le log complet + un squelette d'endpoints découverts (sans écraser le vrai)."""
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):
if r["method"] == method and all(n in r["url"].lower() for n in needles):
return r["url"]
return ""
noise = ("otlp/traces", "/login", "auth/email", "translations")
writes = [r for r in requests_seen
if r["method"] in ("POST", "PUT", "PATCH")
and not any(n in r["url"] for n in noise)]
discovered = ENDPOINTS_PATH.parent / "endpoints_discovered.json"
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 "",
"_comment": "Brouillon issu de la dernière discovery (ne remplace pas config/endpoints.json).",
"menu": find("GET", "menus-service") or find("GET", "menu"),
"weeks": find("GET", "deliveries") or find("GET", "subscriptions"),
"set_selection_candidates": [f"[{r['method']}] {r['url']}" for r in writes],
}
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.")
discovered.parent.mkdir(parents=True, exist_ok=True)
discovered.write_text(json.dumps(skeleton, indent=2, ensure_ascii=False), encoding="utf-8")
print(f"Candidats d'écriture ({len(writes)}) -> {discovered}")
if __name__ == "__main__":