"""Fill a Redefine Meals guest cart from an admin order-list CSV. Opens a visible browser, adds each matched meal, then stops on the cart page. Log in and check out in that window. Press Enter when finished, or close the window. The script does not store a password and does not open checkout itself. python3 scripts/fill_redefine_cart.py order-list-2026-W39.csv """ from __future__ import annotations import argparse import sys import threading from pathlib import Path from urllib.parse import urlsplit ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) from playwright.sync_api import Error as PlaywrightError # noqa: E402 from playwright.sync_api import sync_playwright # noqa: E402 from scraper.fill_cart import ( # noqa: E402 FillCartError, FillPlan, build_plan, cart_lines, error_text, quantity_for, ) from scraper.parse_menu import ( # noqa: E402 MenuParseError, extract_catalog_products, fetch_menu_html, ) DEFAULT_MENU_URL = "https://www.redefinemeals.com/menu" _CART_JS = """ async ({ method, path, body }) => { const headers = { Accept: "application/json", "X-Requested-With": "XMLHttpRequest", }; const match = document.cookie.match(/(?:^|; )XSRF-TOKEN=([^;]*)/); if (match) { headers["X-XSRF-TOKEN"] = decodeURIComponent(match[1]); } const init = { method, credentials: "same-origin", headers }; if (body !== null && body !== undefined) { headers["Content-Type"] = "application/json"; init.body = JSON.stringify(body); } const response = await fetch(path, init); const text = await response.text(); let data = null; if (text) { try { data = JSON.parse(text); } catch (error) { data = { message: text.slice(0, 200) }; } } return { ok: response.ok, status: response.status, data }; } """ def main(argv: list[str] | None = None) -> int: sys.stdout.reconfigure(line_buffering=True) sys.stderr.reconfigure(line_buffering=True) parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("csv_path", type=Path, help="Admin order-list CSV") parser.add_argument("--menu-url", default=DEFAULT_MENU_URL) args = parser.parse_args(argv) try: csv_text = args.csv_path.read_text(encoding="utf-8-sig") menu_html = fetch_menu_html(args.menu_url) products = extract_catalog_products(menu_html) plan = build_plan(csv_text, products) except (OSError, FillCartError, MenuParseError) as exc: print(exc, file=sys.stderr) return 1 _print_plan(plan) if not plan.adds: print("Nothing matched. Not opening a browser.") return 1 if plan.skipped else 0 return _fill(plan, args.menu_url) def _fill(plan: FillPlan, menu_url: str) -> int: with sync_playwright() as playwright: try: browser = playwright.chromium.launch(headless=False) except PlaywrightError as exc: print(f"Browser failed: {exc}", file=sys.stderr) print( "Install the browser with: playwright install chromium", file=sys.stderr, ) return 1 page = browser.new_page() try: page.goto(menu_url, wait_until="domcontentloaded", timeout=60000) cart = _cart_request(page, "GET", "/api/cart") existing = cart_lines(cart) if existing: print("The cart already has items:") for line in existing: print(f" {line.name} x{line.quantity}") if not _confirm_clear(): print("Exiting without adding.") return 1 cleared = _cart_request(page, "POST", "/api/cart/clear", {}) if cart_lines(cleared): print( "The cart still has items after clear. Exiting without adding." ) return 1 failed: list[str] = [] posted = [] for item in plan.adds: try: _cart_request( page, "POST", "/api/cart/add", { "uuid": item.uuid, "quantity": item.quantity, "properties": None, }, ) except RuntimeError as exc: failed.append(f"{item.name}: {exc}") print(f"Failed: {item.name}: {exc}") continue posted.append(item) added = 0 try: current = cart_lines(_cart_request(page, "GET", "/api/cart")) except RuntimeError as exc: failed.append(f"Could not read the cart: {exc}") print(f"Failed: Could not read the cart: {exc}") else: for item in posted: found = quantity_for(current, item.uuid) if found != item.quantity: message = ( f"{item.name}: requested {item.quantity}, cart has {found}" ) failed.append(message) print(f"Failed: {message}") else: added += 1 origin = _origin(page.url or menu_url) page.goto(f"{origin}/cart", wait_until="domcontentloaded", timeout=60000) print(f"Added {added}. Failed {len(failed)}. Skipped {len(plan.skipped)}.") _wait_for_review(page) return 1 if failed or plan.skipped else 0 finally: try: browser.close() except PlaywrightError: pass def _cart_request(page, method: str, path: str, body: object | None = None) -> object: result = page.evaluate(_CART_JS, {"method": method, "path": path, "body": body}) if not isinstance(result, dict) or not result.get("ok"): status = result.get("status") if isinstance(result, dict) else 0 data = result.get("data") if isinstance(result, dict) else None raise RuntimeError(error_text(int(status or 0), data)) return result.get("data") def _confirm_clear() -> bool: if not sys.stdin.isatty(): print("Refusing to clear a cart without a typed yes.") return False answer = input("Type yes to clear the cart and add this order: ") return answer.strip().casefold() == "yes" def _wait_for_review(page) -> None: print( "Cart is ready. Log in and check out in this browser window.\n" "Press Enter here after you are done. Closing the window also stops the script." ) finished = threading.Event() def _wait_for_enter() -> None: try: input() except EOFError: pass finished.set() if sys.stdin.isatty(): threading.Thread(target=_wait_for_enter, daemon=True).start() while not finished.is_set(): try: closed = page.is_closed() except PlaywrightError: return if closed: return try: page.wait_for_timeout(250) except PlaywrightError: return def _print_plan(plan: FillPlan) -> None: if plan.ignored_zero: print(f"Ignored {plan.ignored_zero} items with quantity 0.") if plan.adds: print("Adding:") for item in plan.adds: print(f" {item.name} x{item.quantity}") if plan.skipped: print("Skipped:") for line in plan.skipped: print(f" {line.name}: {line.reason}") def _origin(url: str) -> str: parts = urlsplit(url) return f"{parts.scheme}://{parts.netloc}" if __name__ == "__main__": raise SystemExit(main())