meal-order-manager/scripts/fill_redefine_cart.py

"""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:
            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 = []
        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)

        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())