"""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 UnicodeDecodeError: print("CSV must be UTF-8.", file=sys.stderr) return 1 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 return _fill_browser(browser, plan, menu_url) def _fill_browser(browser, plan: FillPlan, menu_url: str) -> int: try: try: page = browser.new_page() except PlaywrightError as exc: print(f"Failed: {exc}") print(f"Added 0. Failed 1. Skipped {len(plan.skipped)}.") return 1 return _fill_page(page, plan, menu_url) finally: try: browser.close() except PlaywrightError: pass def _fill_page(page, plan: FillPlan, menu_url: str) -> int: failed: list[str] = [] added = 0 ready = False 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 posted = [] page_closed = False for index, item in enumerate(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 except PlaywrightError as exc: failed.append(f"{item.name}: {exc}") print(f"Failed: {item.name}: {exc}") for rest in plan.adds[index + 1 :]: message = f"{rest.name}: not attempted" failed.append(message) print(f"Failed: {message}") page_closed = True break posted.append(item) if not page_closed: 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) ready = True except (RuntimeError, PlaywrightError) as exc: failed.append(str(exc)) print(f"Failed: {exc}") print(f"Added {added}. Failed {len(failed)}. Skipped {len(plan.skipped)}.") if ready: _wait_for_review(page) return 1 if failed or plan.skipped else 0 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())