Initial commit.

ocado_grocy/receipt.py

@@ -0,0 +1,161 @@
+"""Parse receipt data only: never treat the live basket as an order."""
+from dataclasses import asdict, dataclass, field
+from datetime import date, datetime
+from zoneinfo import ZoneInfo
+from decimal import Decimal, InvalidOperation
+import json
+
+
+class ImportError(ValueError):
+    """An input or integration error suitable for displaying to the user."""
+
+
+def number(value, label, *, money=False):
+    if isinstance(value, dict):
+        currency = value.get("currency", "GBP")
+        if currency not in ("GBP", "GBX"):
+            raise ImportError(f"Unsupported currency: {currency}")
+        result = number(value.get("amount"), label, money=money)
+        return result / 100 if currency == "GBX" else result
+    text = str(value).strip().replace("£", "").replace(",", "")
+    try:
+        result = Decimal(text)
+    except InvalidOperation as exc:
+        raise ImportError(f"Invalid {label}: {value!r}") from exc
+    if not result.is_finite() or result < 0:
+        raise ImportError(f"Invalid {label}: {value!r}")
+    return result
+
+
+def receipt_date(value):
+    if not value:
+        return None
+    text = str(value).strip()
+    try:
+        return date.fromisoformat(text[:10]).isoformat()
+    except ValueError:
+        pass
+    for fmt in ("%d/%m/%Y", "%d %B %Y", "%d %b %Y", "%A %d %B %Y"):
+        try:
+            return datetime.strptime(text, fmt).date().isoformat()
+        except ValueError:
+            pass
+    raise ImportError(f"Unrecognised date: {value!r}; use YYYY-MM-DD")
+
+
+@dataclass
+class Item:
+    name: str
+    quantity: Decimal
+    total: Decimal
+    product_id: str = ""
+    barcode: str = ""
+    url: str = ""
+    best_before: str | None = None
+    metadata: dict = field(default_factory=dict)
+
+    @property
+    def unit_price(self):
+        return self.total / self.quantity
+
+
+@dataclass
+class Receipt:
+    order_id: str
+    purchased_date: str | None
+    items: list[Item]
+    metadata: dict = field(default_factory=dict)
+
+    def export(self):
+        return asdict(self)
+
+
+def parse_document(data):
+    """Read a normalized cached receipt."""
+    if not isinstance(data, dict) or not isinstance(data.get("items"), list):
+        raise ImportError("Receipt JSON requires an items array")
+    if data.get("currency", "GBP") != "GBP":
+        raise ImportError("Only GBP receipts are supported")
+    order_id = str(data.get("order_id") or "").strip()
+    if not order_id:
+        raise ImportError("Receipt has no order ID")
+    items = []
+    for row in data["items"]:
+        if not isinstance(row, dict):
+            raise ImportError("Receipt items must be objects")
+        status = str(row.get("status", "")).lower()
+        if status in {"unavailable", "cancelled", "rejected", "not delivered"}:
+            continue
+        quantity = number(row.get("delivered_quantity", row.get("quantity")), "quantity")
+        if quantity == 0:
+            continue
+        name = str(row.get("name") or "").strip()
+        if not name:
+            raise ImportError("Receipt item has no name")
+        total = number(row.get("total"), f"line total for {name}", money=True)
+        metadata = dict(row.get("metadata") or {})
+        metadata.update({k: v for k, v in row.items() if k not in {
+            "name", "quantity", "delivered_quantity", "total", "product_id",
+            "barcode", "url", "best_before", "metadata"}})
+        items.append(Item(name, quantity, total, str(row.get("product_id") or ""),
+                          str(row.get("barcode") or ""), str(row.get("url") or ""),
+                          receipt_date(row.get("best_before")), metadata))
+    if not items:
+        raise ImportError("Receipt contains no delivered items")
+    return Receipt(order_id, receipt_date(data.get("purchased_date")), items, data.get("metadata", {}))
+
+
+def product_metadata(product):
+    return {key: product[key] for key in ("brand", "categoryPath", "size", "packSize",
+            "description", "ingredients", "nutritionalInformation", "allergens", "storage",
+            "guaranteedProductLife", "retailerProductId") if key in product}
+
+
+def parse_ocado_order(data, *, expected_order_id=None, product_urls=None):
+    """Parse the order JSON loaded by Playwright, never the page's live basket."""
+    try:
+        order = data["entities"]["order"][data["result"]]
+        order_id = str(order["orderId"])
+        if expected_order_id and order_id != str(expected_order_id):
+            raise ImportError("Ocado API order does not match the requested order")
+        if order["status"] != "DELIVERED":
+            raise ImportError("Ocado order is not delivered")
+        dates = order["dates"]
+        delivered_at = datetime.fromisoformat(dates["deliveryStartDate"].replace("Z", "+00:00"))
+        if delivered_at.tzinfo is not None:
+            delivered_at = delivered_at.astimezone(ZoneInfo(dates.get("timeZoneId", "Europe/London")))
+        purchased_date = delivered_at.date().isoformat()
+        groups = order["groupedProducts"]
+        products = list(groups["products"])
+        for original in groups.get("substitutes", []):
+            for replacement in original.get("substitutes", []):
+                if replacement.get("status") == "ACCEPTED":
+                    products.append({"storageType": original.get("storageType"),
+                                     "substituted_for": original["name"], **replacement})
+        rows = []
+        catalogue = data["entities"].get("product", {})
+        for product in products:
+            info = catalogue.get(product["productId"], {})
+            retailer_id = str(product["retailerProductId"])
+            metadata = {"ocado": product_metadata(info), "storage_type": product.get("storageType"),
+                        "pack_info": product.get("packInfo", {}), "promotions": product.get("promotions", []),
+                        "original_line_price": product["prices"].get("retail"),
+                        "price_per_item": product["prices"].get("pricePerItem"),
+                        "image_url": (product.get("image") or {}).get("src", "")}
+            if product.get("substituted_for"):
+                metadata["substituted_for"] = product["substituted_for"]
+            rows.append({"name": " ".join(product["name"].split()), "quantity": product["quantity"],
+                         "total": product["prices"]["offered"], "product_id": retailer_id,
+                         "best_before": product.get("expirationDate"),
+                         "url": (product_urls or {}).get(retailer_id, ""), "metadata": metadata})
+        # Stable ordering is independent of the browser's equal-expiry sorting.
+        rows.sort(key=lambda row:(row["best_before"] or "9999-12-31", row["product_id"], row["name"]))
+        receipt = parse_document({"order_id":order_id,"purchased_date":purchased_date,"items":rows})
+        if sum(item.quantity for item in receipt.items) != number(order["totalItems"], "receipt item count"):
+            raise ImportError("Delivered quantities do not match the order summary")
+        if sum(item.total for item in receipt.items) != number(order["orderTotals"]["itemPriceAfterPromos"], "receipt product total"):
+            raise ImportError("Delivered line prices do not match the order summary")
+        receipt.metadata = {"source":"ocado-order-api", "order_totals":order["orderTotals"]}
+        return receipt
+    except (KeyError, TypeError, AttributeError) as exc:
+        raise ImportError("Unrecognised Ocado order API structure") from exc