161 lines
7.3 KiB
Python
161 lines
7.3 KiB
Python
"""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
|