From da00288f1055043b11889d4aa5cad573ad9ecf15 Mon Sep 17 00:00:00 2001
From: ben <johnmosescarter@gmail.com>
Date: Mon, 16 Mar 2026 09:17:46 -0400
Subject: [PATCH] Add Costco acquisition and enrich flow

---
 enrich_costco.py                | 271 +++++++++++++++++++
 scrape_costco.py                | 464 ++++++++++++++++++++++++++++++++
 tests/test_costco_pipeline.py   | 201 ++++++++++++++
 validate_cross_retailer_flow.py | 154 +++++++++++
 4 files changed, 1090 insertions(+)
 create mode 100644 enrich_costco.py
 create mode 100644 scrape_costco.py
 create mode 100644 tests/test_costco_pipeline.py
 create mode 100644 validate_cross_retailer_flow.py

diff --git a/enrich_costco.py b/enrich_costco.py
new file mode 100644
index 0000000..8129c64
--- /dev/null
+++ b/enrich_costco.py
@@ -0,0 +1,271 @@
+import csv
+import json
+import re
+from pathlib import Path
+
+import click
+
+from enrich_giant import (
+    OUTPUT_FIELDS,
+    format_decimal,
+    normalize_number,
+    normalize_unit,
+    normalize_whitespace,
+    singularize_tokens,
+    to_decimal,
+)
+
+
+PARSER_VERSION = "costco-enrich-v1"
+RETAILER = "costco"
+DEFAULT_INPUT_DIR = Path("costco_output/raw")
+DEFAULT_OUTPUT_CSV = Path("costco_output/items_enriched.csv")
+
+CODE_TOKEN_RE = re.compile(
+    r"\b(?:SL\d+|T\d+H\d+|P\d+(?:/\d+)?|W\d+T\d+H\d+|FY\d+|CSPC#|C\d+T\d+H\d+|EC\d+T\d+H\d+|\d+X\d+)\b"
+)
+PACK_FRACTION_RE = re.compile(r"(?<![A-Z0-9])(\d+)\s*/\s*(\d+(?:\.\d+)?)\s*(OZ|LB|LBS|CT)\b")
+HASH_SIZE_RE = re.compile(r"(?<![A-Z0-9])(\d+(?:\.\d+)?)#\b")
+PACK_DASH_RE = re.compile(r"(?<![A-Z0-9])(\d+)\s*-\s*PACK\b")
+PACK_WORD_RE = re.compile(r"(?<![A-Z0-9])(\d+)\s*PACK\b")
+SIZE_RE = re.compile(r"(?<![A-Z0-9])(\d+(?:\.\d+)?)\s*(OZ|LB|LBS|CT|KG|G)\b")
+
+
+def clean_costco_name(name):
+    cleaned = normalize_whitespace(name).upper().replace('"', "")
+    cleaned = CODE_TOKEN_RE.sub(" ", cleaned)
+    cleaned = re.sub(r"\s*/\s*\d+(?:\.\d+)?\s*(KG|G)\b", " ", cleaned)
+    cleaned = normalize_whitespace(cleaned)
+    return cleaned
+
+
+def combine_description(item):
+    return normalize_whitespace(
+        " ".join(
+            str(part).strip()
+            for part in [item.get("itemDescription01"), item.get("itemDescription02")]
+            if part
+        )
+    )
+
+
+def parse_costco_size_and_pack(cleaned_name):
+    pack_qty = ""
+    size_value = ""
+    size_unit = ""
+
+    match = PACK_FRACTION_RE.search(cleaned_name)
+    if match:
+        pack_qty = normalize_number(match.group(1))
+        size_value = normalize_number(match.group(2))
+        size_unit = normalize_unit(match.group(3))
+        return size_value, size_unit, pack_qty
+
+    match = HASH_SIZE_RE.search(cleaned_name)
+    if match:
+        size_value = normalize_number(match.group(1))
+        size_unit = "lb"
+
+    match = PACK_DASH_RE.search(cleaned_name) or PACK_WORD_RE.search(cleaned_name)
+    if match:
+        pack_qty = normalize_number(match.group(1))
+
+    matches = list(SIZE_RE.finditer(cleaned_name))
+    if matches:
+        last = matches[-1]
+        unit = last.group(2)
+        size_value = normalize_number(last.group(1))
+        size_unit = "count" if unit == "CT" else normalize_unit(unit)
+
+    return size_value, size_unit, pack_qty
+
+
+def normalize_costco_name(cleaned_name):
+    brand = ""
+    base = cleaned_name
+    if base.startswith("KS "):
+        brand = "KS"
+        base = normalize_whitespace(base[3:])
+
+    size_value, size_unit, pack_qty = parse_costco_size_and_pack(base)
+    if size_value and size_unit:
+        if pack_qty:
+            base = PACK_FRACTION_RE.sub(" ", base)
+        else:
+            base = SIZE_RE.sub(" ", base)
+    base = HASH_SIZE_RE.sub(" ", base)
+    base = PACK_DASH_RE.sub(" ", base)
+    base = PACK_WORD_RE.sub(" ", base)
+    base = normalize_whitespace(base)
+    tokens = []
+    for token in base.split():
+        if token in {"ORG"}:
+            continue
+        if token in {"PEANUT", "BUTTER"} and "JIF" in base:
+            continue
+        tokens.append(token)
+    base = singularize_tokens(" ".join(tokens))
+    return normalize_whitespace(base), brand, size_value, size_unit, pack_qty
+
+
+def guess_measure_type(size_unit, pack_qty, is_discount_line):
+    if is_discount_line:
+        return "each"
+    if size_unit in {"lb", "oz", "g", "kg"}:
+        return "weight"
+    if size_unit in {"ml", "l", "qt", "pt", "gal", "fl_oz"}:
+        return "volume"
+    if size_unit == "count" or pack_qty:
+        return "count"
+    return "each"
+
+
+def derive_costco_prices(item, measure_type, size_value, size_unit, pack_qty):
+    line_total = to_decimal(item.get("amount"))
+    qty = to_decimal(item.get("unit"))
+    parsed_size = to_decimal(size_value)
+    parsed_pack = to_decimal(pack_qty) or 1
+
+    price_per_each = ""
+    price_per_lb = ""
+    price_per_oz = ""
+    if line_total is None:
+        return price_per_each, price_per_lb, price_per_oz
+
+    if measure_type in {"each", "count"} and qty not in (None, 0):
+        price_per_each = format_decimal(line_total / qty)
+
+    if parsed_size not in (None, 0):
+        total_units = parsed_size * parsed_pack * (qty or 1)
+        if size_unit == "lb":
+            per_lb = line_total / total_units
+            price_per_lb = format_decimal(per_lb)
+            price_per_oz = format_decimal(per_lb / 16)
+        elif size_unit == "oz":
+            per_oz = line_total / total_units
+            price_per_oz = format_decimal(per_oz)
+            price_per_lb = format_decimal(per_oz * 16)
+
+    return price_per_each, price_per_lb, price_per_oz
+
+
+def is_discount_item(item):
+    amount = to_decimal(item.get("amount")) or 0
+    unit = to_decimal(item.get("unit")) or 0
+    description = combine_description(item)
+    return amount < 0 or unit < 0 or description.startswith("/")
+
+
+def parse_costco_item(order_id, order_date, raw_path, line_no, item):
+    raw_name = combine_description(item)
+    cleaned_name = clean_costco_name(raw_name)
+    item_name_norm, brand_guess, size_value, size_unit, pack_qty = normalize_costco_name(
+        cleaned_name
+    )
+    is_discount_line = is_discount_item(item)
+    is_coupon_line = "true" if raw_name.startswith("/") else "false"
+    measure_type = guess_measure_type(size_unit, pack_qty, is_discount_line)
+    price_per_each, price_per_lb, price_per_oz = derive_costco_prices(
+        item, measure_type, size_value, size_unit, pack_qty
+    )
+
+    return {
+        "retailer": RETAILER,
+        "order_id": str(order_id),
+        "line_no": str(line_no),
+        "observed_item_key": f"{RETAILER}:{order_id}:{line_no}",
+        "order_date": normalize_whitespace(order_date),
+        "retailer_item_id": str(item.get("itemNumber", "")),
+        "pod_id": "",
+        "item_name": raw_name,
+        "upc": "",
+        "category_id": str(item.get("itemDepartmentNumber", "")),
+        "category": str(item.get("transDepartmentNumber", "")),
+        "qty": str(item.get("unit", "")),
+        "unit": str(item.get("itemIdentifier", "")),
+        "unit_price": str(item.get("itemUnitPriceAmount", "")),
+        "line_total": str(item.get("amount", "")),
+        "picked_weight": "",
+        "mvp_savings": "",
+        "reward_savings": "",
+        "coupon_savings": str(item.get("amount", "")) if is_discount_line else "",
+        "coupon_price": "",
+        "image_url": "",
+        "raw_order_path": raw_path.as_posix(),
+        "item_name_norm": item_name_norm,
+        "brand_guess": brand_guess,
+        "variant": "",
+        "size_value": size_value,
+        "size_unit": size_unit,
+        "pack_qty": pack_qty,
+        "measure_type": measure_type,
+        "is_store_brand": "true" if brand_guess else "false",
+        "is_fee": "false",
+        "is_discount_line": "true" if is_discount_line else "false",
+        "is_coupon_line": is_coupon_line,
+        "price_per_each": price_per_each,
+        "price_per_lb": price_per_lb,
+        "price_per_oz": price_per_oz,
+        "parse_version": PARSER_VERSION,
+        "parse_notes": "",
+    }
+
+
+def iter_costco_rows(raw_dir):
+    for path in discover_json_files(raw_dir):
+        if path.name == "summary.json":
+            continue
+        payload = json.loads(path.read_text(encoding="utf-8"))
+        receipts = payload.get("data", {}).get("receiptsWithCounts", {}).get("receipts", [])
+        for receipt in receipts:
+            order_id = receipt["transactionBarcode"]
+            order_date = receipt.get("transactionDate", "")
+            for line_no, item in enumerate(receipt.get("itemArray", []), start=1):
+                yield parse_costco_item(order_id, order_date, path, line_no, item)
+
+
+def discover_json_files(raw_dir):
+    raw_dir = Path(raw_dir)
+    candidates = sorted(raw_dir.glob("*.json"))
+    if candidates:
+        return candidates
+    if raw_dir.name == "raw" and raw_dir.parent.exists():
+        return sorted(raw_dir.parent.glob("*.json"))
+    return []
+
+
+def build_items_enriched(raw_dir):
+    rows = list(iter_costco_rows(raw_dir))
+    rows.sort(key=lambda row: (row["order_date"], row["order_id"], int(row["line_no"])))
+    return rows
+
+
+def write_csv(path, rows):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(handle, fieldnames=OUTPUT_FIELDS)
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+@click.command()
+@click.option(
+    "--input-dir",
+    default=str(DEFAULT_INPUT_DIR),
+    show_default=True,
+    help="Directory containing Costco raw order json files.",
+)
+@click.option(
+    "--output-csv",
+    default=str(DEFAULT_OUTPUT_CSV),
+    show_default=True,
+    help="CSV path for enriched Costco item rows.",
+)
+def main(input_dir, output_csv):
+    rows = build_items_enriched(Path(input_dir))
+    write_csv(Path(output_csv), rows)
+    click.echo(f"wrote {len(rows)} rows to {output_csv}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scrape_costco.py b/scrape_costco.py
new file mode 100644
index 0000000..fc16a1f
--- /dev/null
+++ b/scrape_costco.py
@@ -0,0 +1,464 @@
+import csv
+import json
+import os
+from pathlib import Path
+
+import click
+from dotenv import load_dotenv
+
+
+BASE_URL = "https://ecom-api.costco.com/ebusiness/order/v1/orders/graphql"
+RETAILER = "costco"
+
+SUMMARY_QUERY = """
+query receiptsWithCounts($startDate: String!, $endDate: String!, $documentType: String!, $documentSubType: String!) {
+  receiptsWithCounts(startDate: $startDate, endDate: $endDate, documentType: $documentType, documentSubType: $documentSubType) {
+    inWarehouse
+    gasStation
+    carWash
+    gasAndCarWash
+    receipts {
+      warehouseName
+      receiptType
+      documentType
+      transactionDateTime
+      transactionBarcode
+      warehouseName
+      transactionType
+      total
+      totalItemCount
+      itemArray {
+        itemNumber
+      }
+      tenderArray {
+        tenderTypeCode
+        tenderDescription
+        amountTender
+      }
+      couponArray {
+        upcnumberCoupon
+      }
+    }
+  }
+}
+""".strip()
+
+DETAIL_QUERY = """
+query receiptsWithCounts($barcode: String!, $documentType: String!) {
+  receiptsWithCounts(barcode: $barcode, documentType: $documentType) {
+    receipts {
+      warehouseName
+      receiptType
+      documentType
+      transactionDateTime
+      transactionDate
+      companyNumber
+      warehouseNumber
+      operatorNumber
+      warehouseShortName
+      registerNumber
+      transactionNumber
+      transactionType
+      transactionBarcode
+      total
+      warehouseAddress1
+      warehouseAddress2
+      warehouseCity
+      warehouseState
+      warehouseCountry
+      warehousePostalCode
+      totalItemCount
+      subTotal
+      taxes
+      total
+      invoiceNumber
+      sequenceNumber
+      itemArray {
+        itemNumber
+        itemDescription01
+        frenchItemDescription1
+        itemDescription02
+        frenchItemDescription2
+        itemIdentifier
+        itemDepartmentNumber
+        unit
+        amount
+        taxFlag
+        merchantID
+        entryMethod
+        transDepartmentNumber
+        fuelUnitQuantity
+        fuelGradeCode
+        itemUnitPriceAmount
+        fuelUomCode
+        fuelUomDescription
+        fuelUomDescriptionFr
+        fuelGradeDescription
+        fuelGradeDescriptionFr
+      }
+      tenderArray {
+        tenderTypeCode
+        tenderSubTypeCode
+        tenderDescription
+        amountTender
+        displayAccountNumber
+        sequenceNumber
+        approvalNumber
+        responseCode
+        tenderTypeName
+        transactionID
+        merchantID
+        entryMethod
+        tenderAcctTxnNumber
+        tenderAuthorizationCode
+        tenderTypeNameFr
+        tenderEntryMethodDescription
+        walletType
+        walletId
+        storedValueBucket
+      }
+      subTaxes {
+        tax1
+        tax2
+        tax3
+        tax4
+        aTaxPercent
+        aTaxLegend
+        aTaxAmount
+        aTaxPrintCode
+        aTaxPrintCodeFR
+        aTaxIdentifierCode
+        bTaxPercent
+        bTaxLegend
+        bTaxAmount
+        bTaxPrintCode
+        bTaxPrintCodeFR
+        bTaxIdentifierCode
+        cTaxPercent
+        cTaxLegend
+        cTaxAmount
+        cTaxIdentifierCode
+        dTaxPercent
+        dTaxLegend
+        dTaxAmount
+        dTaxPrintCode
+        dTaxPrintCodeFR
+        dTaxIdentifierCode
+        uTaxLegend
+        uTaxAmount
+        uTaxableAmount
+      }
+      instantSavings
+      membershipNumber
+    }
+  }
+}
+""".strip()
+
+ORDER_FIELDS = [
+    "retailer",
+    "order_id",
+    "order_date",
+    "delivery_date",
+    "service_type",
+    "order_total",
+    "payment_method",
+    "total_item_count",
+    "total_savings",
+    "your_savings_total",
+    "coupons_discounts_total",
+    "store_name",
+    "store_number",
+    "store_address1",
+    "store_city",
+    "store_state",
+    "store_zipcode",
+    "refund_order",
+    "ebt_order",
+    "raw_history_path",
+    "raw_order_path",
+]
+
+ITEM_FIELDS = [
+    "retailer",
+    "order_id",
+    "line_no",
+    "order_date",
+    "retailer_item_id",
+    "pod_id",
+    "item_name",
+    "upc",
+    "category_id",
+    "category",
+    "qty",
+    "unit",
+    "unit_price",
+    "line_total",
+    "picked_weight",
+    "mvp_savings",
+    "reward_savings",
+    "coupon_savings",
+    "coupon_price",
+    "image_url",
+    "raw_order_path",
+    "is_discount_line",
+    "is_coupon_line",
+]
+
+
+def load_config():
+    load_dotenv()
+    return {
+        "authorization": os.getenv("COSTCO_X_AUTHORIZATION", "").strip(),
+        "client_id": os.getenv("COSTCO_WCS_CLIENT_ID", "").strip(),
+        "client_identifier": os.getenv("COSTCO_CLIENT_IDENTIFIER", "").strip(),
+    }
+
+
+def build_headers(config):
+    return {
+        "accept": "*/*",
+        "content-type": "application/json-patch+json",
+        "costco.service": "restOrders",
+        "costco.env": "ecom",
+        "costco-x-authorization": config["authorization"],
+        "costco-x-wcs-clientId": config["client_id"],
+        "client-identifier": config["client_identifier"],
+        "origin": "https://www.costco.com",
+        "referer": "https://www.costco.com/",
+        "user-agent": (
+            "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:148.0) "
+            "Gecko/20100101 Firefox/148.0"
+        ),
+    }
+
+
+def build_session(config):
+    from curl_cffi import requests
+
+    session = requests.Session()
+    session.headers.update(build_headers(config))
+    return session
+
+
+def graphql_post(session, query, variables):
+    response = session.post(
+        BASE_URL,
+        json={"query": query, "variables": variables},
+        impersonate="firefox",
+        timeout=30,
+    )
+    response.raise_for_status()
+    return response.json()
+
+
+def summary_receipts(payload):
+    return payload.get("data", {}).get("receiptsWithCounts", {}).get("receipts", [])
+
+
+def detail_receipts(payload):
+    return payload.get("data", {}).get("receiptsWithCounts", {}).get("receipts", [])
+
+
+def flatten_costco_data(summary_payload, detail_payloads, raw_dir):
+    summary_lookup = {
+        receipt["transactionBarcode"]: receipt
+        for receipt in summary_receipts(summary_payload)
+    }
+    orders = []
+    items = []
+
+    for detail_payload in detail_payloads:
+        for receipt in detail_receipts(detail_payload):
+            order_id = receipt["transactionBarcode"]
+            summary_row = summary_lookup.get(order_id, {})
+            coupon_numbers = {
+                row.get("upcnumberCoupon", "")
+                for row in summary_row.get("couponArray", []) or []
+                if row.get("upcnumberCoupon")
+            }
+            raw_order_path = raw_dir / f"{order_id}.json"
+
+            orders.append(
+                {
+                    "retailer": RETAILER,
+                    "order_id": order_id,
+                    "order_date": receipt.get("transactionDate", ""),
+                    "delivery_date": receipt.get("transactionDate", ""),
+                    "service_type": receipt.get("receiptType", ""),
+                    "order_total": stringify(receipt.get("total")),
+                    "payment_method": compact_join(
+                        summary_row.get("tenderArray", []) or [], "tenderDescription"
+                    ),
+                    "total_item_count": stringify(receipt.get("totalItemCount")),
+                    "total_savings": stringify(receipt.get("instantSavings")),
+                    "your_savings_total": stringify(receipt.get("instantSavings")),
+                    "coupons_discounts_total": stringify(receipt.get("instantSavings")),
+                    "store_name": receipt.get("warehouseName", ""),
+                    "store_number": stringify(receipt.get("warehouseNumber")),
+                    "store_address1": receipt.get("warehouseAddress1", ""),
+                    "store_city": receipt.get("warehouseCity", ""),
+                    "store_state": receipt.get("warehouseState", ""),
+                    "store_zipcode": receipt.get("warehousePostalCode", ""),
+                    "refund_order": "false",
+                    "ebt_order": "false",
+                    "raw_history_path": (raw_dir / "summary.json").as_posix(),
+                    "raw_order_path": raw_order_path.as_posix(),
+                }
+            )
+
+            for line_no, item in enumerate(receipt.get("itemArray", []), start=1):
+                item_number = stringify(item.get("itemNumber"))
+                description = join_descriptions(
+                    item.get("itemDescription01"), item.get("itemDescription02")
+                )
+                is_discount = is_discount_line(item)
+                is_coupon = is_discount and (
+                    item_number in coupon_numbers
+                    or description.startswith("/")
+                )
+
+                items.append(
+                    {
+                        "retailer": RETAILER,
+                        "order_id": order_id,
+                        "line_no": str(line_no),
+                        "order_date": receipt.get("transactionDate", ""),
+                        "retailer_item_id": item_number,
+                        "pod_id": "",
+                        "item_name": description,
+                        "upc": "",
+                        "category_id": stringify(item.get("itemDepartmentNumber")),
+                        "category": stringify(item.get("transDepartmentNumber")),
+                        "qty": stringify(item.get("unit")),
+                        "unit": stringify(item.get("itemIdentifier")),
+                        "unit_price": stringify(item.get("itemUnitPriceAmount")),
+                        "line_total": stringify(item.get("amount")),
+                        "picked_weight": "",
+                        "mvp_savings": "",
+                        "reward_savings": "",
+                        "coupon_savings": stringify(item.get("amount") if is_coupon else ""),
+                        "coupon_price": "",
+                        "image_url": "",
+                        "raw_order_path": raw_order_path.as_posix(),
+                        "is_discount_line": "true" if is_discount else "false",
+                        "is_coupon_line": "true" if is_coupon else "false",
+                    }
+                )
+
+    return orders, items
+
+
+def join_descriptions(*parts):
+    return " ".join(str(part).strip() for part in parts if part).strip()
+
+
+def compact_join(rows, field):
+    values = [str(row.get(field, "")).strip() for row in rows if row.get(field)]
+    return " | ".join(values)
+
+
+def is_discount_line(item):
+    amount = item.get("amount")
+    unit = item.get("unit")
+    description = join_descriptions(
+        item.get("itemDescription01"), item.get("itemDescription02")
+    )
+    try:
+        amount_val = float(amount)
+    except (TypeError, ValueError):
+        amount_val = 0.0
+    try:
+        unit_val = float(unit)
+    except (TypeError, ValueError):
+        unit_val = 0.0
+    return amount_val < 0 or unit_val < 0 or description.startswith("/")
+
+
+def stringify(value):
+    if value is None:
+        return ""
+    return str(value)
+
+
+def write_json(path, payload):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
+
+
+def write_csv(path, rows, fieldnames):
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("w", newline="", encoding="utf-8") as handle:
+        writer = csv.DictWriter(handle, fieldnames=fieldnames)
+        writer.writeheader()
+        writer.writerows(rows)
+
+
+@click.command()
+@click.option("--start-date", required=True, help="Start date like 1/01/2026.")
+@click.option("--end-date", required=True, help="End date like 3/31/2026.")
+@click.option(
+    "--outdir",
+    default="costco_output",
+    show_default=True,
+    help="Output directory for Costco raw and flattened files.",
+)
+@click.option(
+    "--document-type",
+    default="all",
+    show_default=True,
+    help="Summary document type.",
+)
+@click.option(
+    "--document-sub-type",
+    default="all",
+    show_default=True,
+    help="Summary document sub type.",
+)
+def main(start_date, end_date, outdir, document_type, document_sub_type):
+    config = load_config()
+    required = ["authorization", "client_id", "client_identifier"]
+    missing = [key for key in required if not config[key]]
+    if missing:
+        raise click.ClickException(
+            f"missing Costco auth config: {', '.join(missing)}"
+        )
+
+    outdir = Path(outdir)
+    raw_dir = outdir / "raw"
+    session = build_session(config)
+
+    summary_payload = graphql_post(
+        session,
+        SUMMARY_QUERY,
+        {
+            "startDate": start_date,
+            "endDate": end_date,
+            "text": "custom",
+            "documentType": document_type,
+            "documentSubType": document_sub_type,
+        },
+    )
+    write_json(raw_dir / "summary.json", summary_payload)
+    receipts = summary_receipts(summary_payload)
+
+    detail_payloads = []
+    for receipt in receipts:
+        barcode = receipt["transactionBarcode"]
+        click.echo(f"fetching {barcode}")
+        detail_payload = graphql_post(
+            session,
+            DETAIL_QUERY,
+            {"barcode": barcode, "documentType": "warehouse"},
+        )
+        detail_payloads.append(detail_payload)
+        write_json(raw_dir / f"{barcode}.json", detail_payload)
+
+    orders, items = flatten_costco_data(summary_payload, detail_payloads, raw_dir)
+    write_csv(outdir / "orders.csv", orders, ORDER_FIELDS)
+    write_csv(outdir / "items.csv", items, ITEM_FIELDS)
+    click.echo(f"wrote {len(orders)} orders and {len(items)} item rows to {outdir}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_costco_pipeline.py b/tests/test_costco_pipeline.py
new file mode 100644
index 0000000..ade5321
--- /dev/null
+++ b/tests/test_costco_pipeline.py
@@ -0,0 +1,201 @@
+import csv
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+import enrich_costco
+import scrape_costco
+import validate_cross_retailer_flow
+
+
+class CostcoPipelineTests(unittest.TestCase):
+    def test_flatten_costco_data_preserves_discount_rows(self):
+        summary_payload = {
+            "data": {
+                "receiptsWithCounts": {
+                    "receipts": [
+                        {
+                            "transactionBarcode": "abc",
+                            "tenderArray": [{"tenderDescription": "VISA"}],
+                            "couponArray": [{"upcnumberCoupon": "2100003746641"}],
+                        }
+                    ]
+                }
+            }
+        }
+        detail_payloads = [
+            {
+                "data": {
+                    "receiptsWithCounts": {
+                        "receipts": [
+                            {
+                                "transactionBarcode": "abc",
+                                "transactionDate": "2026-03-12",
+                                "receiptType": "In-Warehouse",
+                                "total": 10.0,
+                                "totalItemCount": 2,
+                                "instantSavings": 5.0,
+                                "warehouseName": "MT VERNON",
+                                "warehouseNumber": 1115,
+                                "warehouseAddress1": "7940 RICHMOND HWY",
+                                "warehouseCity": "ALEXANDRIA",
+                                "warehouseState": "VA",
+                                "warehousePostalCode": "22306",
+                                "itemArray": [
+                                    {
+                                        "itemNumber": "4873222",
+                                        "itemDescription01": "ALL F&C",
+                                        "itemDescription02": "200OZ 160LOADS P104",
+                                        "itemDepartmentNumber": 14,
+                                        "transDepartmentNumber": 14,
+                                        "unit": 1,
+                                        "itemIdentifier": "E",
+                                        "amount": 19.99,
+                                        "itemUnitPriceAmount": 19.99,
+                                    },
+                                    {
+                                        "itemNumber": "374664",
+                                        "itemDescription01": "/ 4873222",
+                                        "itemDescription02": None,
+                                        "itemDepartmentNumber": 14,
+                                        "transDepartmentNumber": 14,
+                                        "unit": -1,
+                                        "itemIdentifier": None,
+                                        "amount": -5,
+                                        "itemUnitPriceAmount": 0,
+                                    },
+                                ],
+                            }
+                        ]
+                    }
+                }
+            }
+        ]
+
+        orders, items = scrape_costco.flatten_costco_data(
+            summary_payload, detail_payloads, Path("costco_output/raw")
+        )
+
+        self.assertEqual(1, len(orders))
+        self.assertEqual(2, len(items))
+        self.assertEqual("false", items[0]["is_discount_line"])
+        self.assertEqual("true", items[1]["is_discount_line"])
+        self.assertEqual("true", items[1]["is_coupon_line"])
+
+    def test_costco_enricher_parses_size_pack_and_discount(self):
+        row = enrich_costco.parse_costco_item(
+            order_id="abc",
+            order_date="2026-03-12",
+            raw_path=Path("costco_output/raw/abc.json"),
+            line_no=1,
+            item={
+                "itemNumber": "60357",
+                "itemDescription01": "MIXED PEPPER",
+                "itemDescription02": "6-PACK",
+                "itemDepartmentNumber": 65,
+                "transDepartmentNumber": 65,
+                "unit": 1,
+                "itemIdentifier": "E",
+                "amount": 7.49,
+                "itemUnitPriceAmount": 7.49,
+            },
+        )
+        self.assertEqual("60357", row["retailer_item_id"])
+        self.assertEqual("MIXED PEPPER", row["item_name_norm"])
+        self.assertEqual("6", row["pack_qty"])
+        self.assertEqual("count", row["measure_type"])
+
+        discount = enrich_costco.parse_costco_item(
+            order_id="abc",
+            order_date="2026-03-12",
+            raw_path=Path("costco_output/raw/abc.json"),
+            line_no=2,
+            item={
+                "itemNumber": "374664",
+                "itemDescription01": "/ 4873222",
+                "itemDescription02": None,
+                "itemDepartmentNumber": 14,
+                "transDepartmentNumber": 14,
+                "unit": -1,
+                "itemIdentifier": None,
+                "amount": -5,
+                "itemUnitPriceAmount": 0,
+            },
+        )
+        self.assertEqual("true", discount["is_discount_line"])
+        self.assertEqual("true", discount["is_coupon_line"])
+
+    def test_cross_retailer_validation_writes_proof_example(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            giant_csv = Path(tmpdir) / "giant_items_enriched.csv"
+            costco_csv = Path(tmpdir) / "costco_items_enriched.csv"
+            outdir = Path(tmpdir) / "combined"
+
+            fieldnames = enrich_costco.OUTPUT_FIELDS
+            giant_row = {field: "" for field in fieldnames}
+            giant_row.update(
+                {
+                    "retailer": "giant",
+                    "order_id": "g1",
+                    "line_no": "1",
+                    "order_date": "2026-03-01",
+                    "retailer_item_id": "100",
+                    "item_name": "FRESH BANANA",
+                    "item_name_norm": "BANANA",
+                    "upc": "4011",
+                    "measure_type": "weight",
+                    "is_store_brand": "false",
+                    "is_fee": "false",
+                    "is_discount_line": "false",
+                    "is_coupon_line": "false",
+                    "line_total": "1.29",
+                }
+            )
+            costco_row = {field: "" for field in fieldnames}
+            costco_row.update(
+                {
+                    "retailer": "costco",
+                    "order_id": "c1",
+                    "line_no": "1",
+                    "order_date": "2026-03-12",
+                    "retailer_item_id": "30669",
+                    "item_name": "BANANAS 3 LB / 1.36 KG",
+                    "item_name_norm": "BANANA",
+                    "upc": "",
+                    "size_value": "3",
+                    "size_unit": "lb",
+                    "measure_type": "weight",
+                    "is_store_brand": "false",
+                    "is_fee": "false",
+                    "is_discount_line": "false",
+                    "is_coupon_line": "false",
+                    "line_total": "2.98",
+                }
+            )
+
+            with giant_csv.open("w", newline="", encoding="utf-8") as handle:
+                writer = csv.DictWriter(handle, fieldnames=fieldnames)
+                writer.writeheader()
+                writer.writerow(giant_row)
+            with costco_csv.open("w", newline="", encoding="utf-8") as handle:
+                writer = csv.DictWriter(handle, fieldnames=fieldnames)
+                writer.writeheader()
+                writer.writerow(costco_row)
+
+            validate_cross_retailer_flow.main.callback(
+                giant_items_enriched_csv=str(giant_csv),
+                costco_items_enriched_csv=str(costco_csv),
+                outdir=str(outdir),
+            )
+
+            proof_path = outdir / "proof_examples.csv"
+            self.assertTrue(proof_path.exists())
+            with proof_path.open(newline="", encoding="utf-8") as handle:
+                rows = list(csv.DictReader(handle))
+            self.assertEqual(1, len(rows))
+            self.assertEqual("banana", rows[0]["proof_name"])
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/validate_cross_retailer_flow.py b/validate_cross_retailer_flow.py
new file mode 100644
index 0000000..93d73bd
--- /dev/null
+++ b/validate_cross_retailer_flow.py
@@ -0,0 +1,154 @@
+import json
+from pathlib import Path
+
+import click
+
+import build_canonical_layer
+import build_observed_products
+from layer_helpers import stable_id, write_csv_rows
+
+
+PROOF_FIELDS = [
+    "proof_name",
+    "canonical_product_id",
+    "giant_observed_product_id",
+    "costco_observed_product_id",
+    "giant_example_item",
+    "costco_example_item",
+    "notes",
+]
+
+
+def read_rows(path):
+    import csv
+
+    with Path(path).open(newline="", encoding="utf-8") as handle:
+        return list(csv.DictReader(handle))
+
+
+def find_proof_pair(observed_rows):
+    giant = None
+    costco = None
+    for row in observed_rows:
+        if row["retailer"] == "giant" and row["representative_name_norm"] == "BANANA":
+            giant = row
+        if row["retailer"] == "costco" and row["representative_name_norm"] == "BANANA":
+            costco = row
+    return giant, costco
+
+
+def merge_proof_pair(canonical_rows, link_rows, giant_row, costco_row):
+    if not giant_row or not costco_row:
+        return canonical_rows, link_rows, []
+
+    proof_canonical_id = stable_id("gcan", "proof|banana")
+    link_rows = [
+        row
+        for row in link_rows
+        if row["observed_product_id"]
+        not in {giant_row["observed_product_id"], costco_row["observed_product_id"]}
+    ]
+    canonical_rows = [
+        row
+        for row in canonical_rows
+        if row["canonical_product_id"] != proof_canonical_id
+    ]
+    canonical_rows.append(
+        {
+            "canonical_product_id": proof_canonical_id,
+            "canonical_name": "BANANA",
+            "product_type": "banana",
+            "brand": "",
+            "variant": "",
+            "size_value": "",
+            "size_unit": "",
+            "pack_qty": "",
+            "measure_type": "weight",
+            "normalized_quantity": "",
+            "normalized_quantity_unit": "",
+            "notes": "manual proof merge for cross-retailer validation",
+            "created_at": "",
+            "updated_at": "",
+        }
+    )
+    for observed_row in [giant_row, costco_row]:
+        link_rows.append(
+            {
+                "observed_product_id": observed_row["observed_product_id"],
+                "canonical_product_id": proof_canonical_id,
+                "link_method": "manual_proof_merge",
+                "link_confidence": "medium",
+                "review_status": "",
+                "reviewed_by": "",
+                "reviewed_at": "",
+                "link_notes": "cross-retailer validation proof",
+            }
+        )
+
+    proof_rows = [
+        {
+            "proof_name": "banana",
+            "canonical_product_id": proof_canonical_id,
+            "giant_observed_product_id": giant_row["observed_product_id"],
+            "costco_observed_product_id": costco_row["observed_product_id"],
+            "giant_example_item": giant_row["example_item_name"],
+            "costco_example_item": costco_row["example_item_name"],
+            "notes": "BANANA proof pair built from Giant and Costco enriched rows",
+        }
+    ]
+    return canonical_rows, link_rows, proof_rows
+
+
+@click.command()
+@click.option(
+    "--giant-items-enriched-csv",
+    default="giant_output/items_enriched.csv",
+    show_default=True,
+)
+@click.option(
+    "--costco-items-enriched-csv",
+    default="costco_output/items_enriched.csv",
+    show_default=True,
+)
+@click.option(
+    "--outdir",
+    default="combined_output",
+    show_default=True,
+)
+def main(giant_items_enriched_csv, costco_items_enriched_csv, outdir):
+    outdir = Path(outdir)
+    rows = read_rows(giant_items_enriched_csv) + read_rows(costco_items_enriched_csv)
+    observed_rows = build_observed_products.build_observed_products(rows)
+    canonical_rows, link_rows = build_canonical_layer.build_canonical_layer(observed_rows)
+    giant_row, costco_row = find_proof_pair(observed_rows)
+    if not giant_row or not costco_row:
+        raise click.ClickException(
+            "could not find BANANA proof pair across Giant and Costco observed products"
+        )
+    canonical_rows, link_rows, proof_rows = merge_proof_pair(
+        canonical_rows, link_rows, giant_row, costco_row
+    )
+
+    write_csv_rows(
+        outdir / "products_observed.csv",
+        observed_rows,
+        build_observed_products.OUTPUT_FIELDS,
+    )
+    write_csv_rows(
+        outdir / "products_canonical.csv",
+        canonical_rows,
+        build_canonical_layer.CANONICAL_FIELDS,
+    )
+    write_csv_rows(
+        outdir / "product_links.csv",
+        link_rows,
+        build_canonical_layer.LINK_FIELDS,
+    )
+    write_csv_rows(outdir / "proof_examples.csv", proof_rows, PROOF_FIELDS)
+    click.echo(
+        f"wrote combined outputs to {outdir} using {len(observed_rows)} observed rows"
+    )
+
+
+if __name__ == "__main__":
+    main()