added 14.2 and 14.3 for refactor prep

README.md

@@ -12,6 +12,7 @@ Run each script step-by-step from the terminal.
 4. `enrich_costco.py`: normalize Costco line items
 5. `build_purchases.py`: combine retailer outputs into one purchase table
 6. `review_products.py`: review unresolved product matches in the terminal
+7. `report_pipeline_status.py`: show how many rows survive each stage
 
 ## Requirements
 
@@ -31,6 +32,7 @@ pip install -r requirements.txt
 Current version works best with `.env` in the project root.  The scraper will prompt for these values if they are not found in the current browser session.  
 - `scrape_giant` prompts if `GIANT_USER_ID` or `GIANT_LOYALTY_NUMBER` is missing.
 - `scrape_costco` tries `.env` first, then Firefox local storage for session-backed values; `COSTCO_CLIENT_IDENTIFIER` should still be set explicitly.
+- Costco discount matching happens later in `enrich_costco.py`; you do not need to pre-clean discount lines by hand.
 
 ```env
 GIANT_USER_ID=...
@@ -53,6 +55,8 @@ python enrich_costco.py
 python build_purchases.py
 python review_products.py
 python build_purchases.py
+python review_products.py --refresh-only
+python report_pipeline_status.py
 ```
 
 Why run `build_purchases.py` twice:
@@ -66,6 +70,12 @@ If you only want to refresh the queue without reviewing interactively:
 python review_products.py --refresh-only
 ```
 
+If you want a quick stage-by-stage accountability check:
+
+```bash
+python report_pipeline_status.py
+```
+
 ## Key Outputs
 
 Giant:
@@ -77,6 +87,7 @@ Costco:
 - `costco_output/orders.csv`
 - `costco_output/items.csv`
 - `costco_output/items_enriched.csv`
+- `costco_output/items_enriched.csv` now preserves raw totals and matched net discount fields
 
 Combined:
 - `combined_output/purchases.csv`
@@ -85,6 +96,8 @@ Combined:
 - `combined_output/canonical_catalog.csv`
 - `combined_output/product_links.csv`
 - `combined_output/comparison_examples.csv`
+- `combined_output/pipeline_status.csv`
+- `combined_output/pipeline_status.json`
 
 ## Review Workflow
 
@@ -95,9 +108,14 @@ Run `review_products.py` to cleanup unresolved or weakly unified items:
 - skip it for later
 Decisions are saved and reused on later runs.
 
+The review step is intentionally conservative:
+- weak exact-name matches stay in the queue instead of auto-creating canonical products
+- canonical names should describe stable product identity, not retailer packaging text
+
 ## Notes
 - This project is designed around fragile retailer scraping flows, so the code favors explicit retailer-specific steps over heavy abstraction.
 - `scrape_giant.py` and `scrape_costco.py` are meant to work as standalone acquisition scripts.
+- Costco discount rows are preserved for auditability and also matched back to purchased items during enrichment.
 - `validate_cross_retailer_flow.py` is a proof/check script, not a required production step.
 
 ## Test

build_canonical_layer.py

@@ -1,4 +1,5 @@
 import click
+import re
 
 from layer_helpers import read_csv_rows, representative_value, stable_id, write_csv_rows
 
@@ -20,6 +21,8 @@ CANONICAL_FIELDS = [
     "updated_at",
 ]
 
+CANONICAL_DROP_TOKENS = {"CT", "COUNT", "COUNTS", "DOZ", "DOZEN", "DOZ.", "PACK"}
+
 LINK_FIELDS = [
     "observed_product_id",
     "canonical_product_id",
@@ -91,26 +94,24 @@ def auto_link_rule(observed_row):
             "high",
         )
 
-    if (
-        observed_row.get("representative_name_norm")
-        and not observed_row.get("representative_size_value")
-        and not observed_row.get("representative_size_unit")
-        and not observed_row.get("representative_pack_qty")
-    ):
-        return (
-            "exact_name",
-            "|".join(
-                [
-                    f"name={observed_row['representative_name_norm']}",
-                    f"measure={observed_row['representative_measure_type']}",
-                ]
-            ),
-            "medium",
-        )
-
     return "", "", ""
 
 
+def clean_canonical_name(name):
+    tokens = []
+    for token in re.sub(r"[^A-Z0-9\s]", " ", (name or "").upper()).split():
+        if token.isdigit():
+            continue
+        if token in CANONICAL_DROP_TOKENS:
+            continue
+        if re.fullmatch(r"\d+(?:PK|PACK)", token):
+            continue
+        if re.fullmatch(r"\d+DZ", token):
+            continue
+        tokens.append(token)
+    return " ".join(tokens).strip()
+
+
 def canonical_row_for_group(canonical_product_id, group_rows, link_method):
     quantity_value, quantity_unit = normalized_quantity(
         {
@@ -130,7 +131,10 @@ def canonical_row_for_group(canonical_product_id, group_rows, link_method):
     )
     return {
         "canonical_product_id": canonical_product_id,
-        "canonical_name": representative_value(group_rows, "representative_name_norm"),
+        "canonical_name": clean_canonical_name(
+            representative_value(group_rows, "representative_name_norm")
+        )
+        or representative_value(group_rows, "representative_name_norm"),
         "product_type": "",
         "brand": representative_value(group_rows, "representative_brand"),
         "variant": representative_value(group_rows, "representative_variant"),

build_purchases.py

@@ -33,6 +33,8 @@ PURCHASE_FIELDS = [
     "measure_type",
     "line_total",
     "unit_price",
+    "matched_discount_amount",
+    "net_line_total",
     "store_name",
     "store_number",
     "store_city",
@@ -94,7 +96,7 @@ def decimal_or_zero(value):
 
 
 def derive_metrics(row):
-    line_total = to_decimal(row.get("line_total"))
+    line_total = to_decimal(row.get("net_line_total") or row.get("line_total"))
     qty = to_decimal(row.get("qty"))
     pack_qty = to_decimal(row.get("pack_qty"))
     size_value = to_decimal(row.get("size_value"))
@@ -292,6 +294,8 @@ def build_purchase_rows(
                 "measure_type": row["measure_type"],
                 "line_total": row["line_total"],
                 "unit_price": row["unit_price"],
+                "matched_discount_amount": row.get("matched_discount_amount", ""),
+                "net_line_total": row.get("net_line_total", ""),
                 "store_name": order_row.get("store_name", ""),
                 "store_number": order_row.get("store_number", ""),
                 "store_city": order_row.get("store_city", ""),

collect_costco_web.py

@@ -0,0 +1,65 @@
+import click
+
+import scrape_costco
+
+
+@click.command()
+@click.option(
+    "--outdir",
+    default="data/costco-web",
+    show_default=True,
+    help="Directory for Costco raw and collected outputs.",
+)
+@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.",
+)
+@click.option(
+    "--window-days",
+    default=92,
+    show_default=True,
+    type=int,
+    help="Maximum number of days to request per summary window.",
+)
+@click.option(
+    "--months-back",
+    default=36,
+    show_default=True,
+    type=int,
+    help="How many months of receipts to enumerate back from today.",
+)
+@click.option(
+    "--firefox-profile-dir",
+    default=None,
+    help="Firefox profile directory to use for cookies and session storage.",
+)
+def main(
+    outdir,
+    document_type,
+    document_sub_type,
+    window_days,
+    months_back,
+    firefox_profile_dir,
+):
+    scrape_costco.run_collection(
+        outdir=outdir,
+        document_type=document_type,
+        document_sub_type=document_sub_type,
+        window_days=window_days,
+        months_back=months_back,
+        firefox_profile_dir=firefox_profile_dir,
+        orders_filename="collected_orders.csv",
+        items_filename="collected_items.csv",
+    )
+
+
+if __name__ == "__main__":
+    main()

collect_giant_web.py

@@ -0,0 +1,34 @@
+import click
+
+import scrape_giant
+
+
+@click.command()
+@click.option("--user-id", default=None, help="Giant user id.")
+@click.option("--loyalty", default=None, help="Giant loyalty number.")
+@click.option(
+    "--outdir",
+    default="data/giant-web",
+    show_default=True,
+    help="Directory for raw json and collected csv outputs.",
+)
+@click.option(
+    "--sleep-seconds",
+    default=1.5,
+    show_default=True,
+    type=float,
+    help="Delay between order detail requests.",
+)
+def main(user_id, loyalty, outdir, sleep_seconds):
+    scrape_giant.run_collection(
+        user_id,
+        loyalty,
+        outdir,
+        sleep_seconds,
+        orders_filename="collected_orders.csv",
+        items_filename="collected_items.csv",
+    )
+
+
+if __name__ == "__main__":
+    main()

enrich_costco.py

@@ -1,13 +1,17 @@
 import csv
 import json
 import re
+from collections import defaultdict
 from pathlib import Path
 
 import click
 
 from enrich_giant import (
     OUTPUT_FIELDS,
+    derive_normalized_quantity,
+    derive_price_fields,
     format_decimal,
+    normalization_identity,
     normalize_number,
     normalize_unit,
     normalize_whitespace,
@@ -29,6 +33,7 @@ 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")
+DISCOUNT_TARGET_RE = re.compile(r"^/\s*(\d+)\b")
 
 
 def clean_costco_name(name):
@@ -156,6 +161,13 @@ def is_discount_item(item):
     return amount < 0 or unit < 0 or description.startswith("/")
 
 
+def discount_target_id(raw_name):
+    match = DISCOUNT_TARGET_RE.match(normalize_whitespace(raw_name))
+    if not match:
+        return ""
+    return match.group(1)
+
+
 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)
@@ -168,12 +180,42 @@ def parse_costco_item(order_id, order_date, raw_path, line_no, item):
     price_per_each, price_per_lb, price_per_oz = derive_costco_prices(
         item, measure_type, size_value, size_unit, pack_qty
     )
+    normalized_row_id = f"{RETAILER}:{order_id}:{line_no}"
+    normalized_quantity, normalized_quantity_unit = derive_normalized_quantity(
+        size_value,
+        size_unit,
+        pack_qty,
+        measure_type,
+    )
+    identity_key, normalization_basis = normalization_identity(
+        {
+            "retailer": RETAILER,
+            "normalized_row_id": normalized_row_id,
+            "upc": "",
+            "retailer_item_id": str(item.get("itemNumber", "")),
+            "item_name_norm": item_name_norm,
+            "size_value": size_value,
+            "size_unit": size_unit,
+            "pack_qty": pack_qty,
+        }
+    )
+    price_fields = derive_price_fields(
+        price_per_each,
+        price_per_lb,
+        price_per_oz,
+        str(item.get("amount", "")),
+        str(item.get("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}",
+        "normalized_row_id": normalized_row_id,
+        "normalized_item_id": f"cnorm:{identity_key}",
+        "normalization_basis": normalization_basis,
+        "observed_item_key": normalized_row_id,
         "order_date": normalize_whitespace(order_date),
         "retailer_item_id": str(item.get("itemNumber", "")),
         "pod_id": "",
@@ -190,6 +232,8 @@ def parse_costco_item(order_id, order_date, raw_path, line_no, item):
         "reward_savings": "",
         "coupon_savings": str(item.get("amount", "")) if is_discount_line else "",
         "coupon_price": "",
+        "matched_discount_amount": "",
+        "net_line_total": str(item.get("amount", "")) if not is_discount_line else "",
         "image_url": "",
         "raw_order_path": raw_path.as_posix(),
         "item_name_norm": item_name_norm,
@@ -199,18 +243,64 @@ def parse_costco_item(order_id, order_date, raw_path, line_no, item):
         "size_unit": size_unit,
         "pack_qty": pack_qty,
         "measure_type": measure_type,
+        "normalized_quantity": normalized_quantity,
+        "normalized_quantity_unit": normalized_quantity_unit,
         "is_store_brand": "true" if brand_guess else "false",
+        "is_item": "false" if is_discount_line else "true",
         "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_fields,
-        "price_per_lb": price_per_lb,
-        "price_per_oz": price_per_oz,
         "parse_version": PARSER_VERSION,
         "parse_notes": "",
     }
 
 
+def match_costco_discounts(rows):
+    rows_by_order = defaultdict(list)
+    for row in rows:
+        rows_by_order[row["order_id"]].append(row)
+
+    for order_rows in rows_by_order.values():
+        purchase_rows_by_item_id = defaultdict(list)
+        for row in order_rows:
+            if row.get("is_discount_line") == "true":
+                continue
+            retailer_item_id = row.get("retailer_item_id", "")
+            if retailer_item_id:
+                purchase_rows_by_item_id[retailer_item_id].append(row)
+
+        for row in order_rows:
+            if row.get("is_discount_line") != "true":
+                continue
+            target_id = discount_target_id(row.get("item_name", ""))
+            if not target_id:
+                continue
+            matches = purchase_rows_by_item_id.get(target_id, [])
+            if len(matches) != 1:
+                row["parse_notes"] = normalize_whitespace(
+                    f"{row.get('parse_notes', '')};discount_target_unmatched={target_id}"
+                ).strip(";")
+                continue
+
+            purchase_row = matches[0]
+            matched_discount = to_decimal(row.get("line_total"))
+            gross_total = to_decimal(purchase_row.get("line_total"))
+            existing_discount = to_decimal(purchase_row.get("matched_discount_amount")) or 0
+            if matched_discount is None or gross_total is None:
+                continue
+
+            total_discount = existing_discount + matched_discount
+            purchase_row["matched_discount_amount"] = format_decimal(total_discount)
+            purchase_row["net_line_total"] = format_decimal(gross_total + total_discount)
+            purchase_row["parse_notes"] = normalize_whitespace(
+                f"{purchase_row.get('parse_notes', '')};matched_discount={target_id}"
+            ).strip(";")
+            row["parse_notes"] = normalize_whitespace(
+                f"{row.get('parse_notes', '')};matched_to_item={target_id}"
+            ).strip(";")
+
+
 def iter_costco_rows(raw_dir):
     for path in discover_json_files(raw_dir):
         if path.name in {"summary.json", "summary_requests.json"}:
@@ -238,6 +328,7 @@ def discover_json_files(raw_dir):
 
 def build_items_enriched(raw_dir):
     rows = list(iter_costco_rows(raw_dir))
+    match_costco_discounts(rows)
     rows.sort(key=lambda row: (row["order_date"], row["order_id"], int(row["line_no"])))
     return rows
 
@@ -264,6 +355,7 @@ def write_csv(path, rows):
     help="CSV path for enriched Costco item rows.",
 )
 def main(input_dir, output_csv):
+    click.echo("legacy entrypoint: prefer normalize_costco_web.py for data-model outputs")
     rows = build_items_enriched(Path(input_dir))
     write_csv(Path(output_csv), rows)
     click.echo(f"wrote {len(rows)} rows to {output_csv}")

enrich_giant.py

@@ -16,6 +16,9 @@ OUTPUT_FIELDS = [
     "retailer",
     "order_id",
     "line_no",
+    "normalized_row_id",
+    "normalized_item_id",
+    "normalization_basis",
     "observed_item_key",
     "order_date",
     "retailer_item_id",
@@ -33,6 +36,8 @@ OUTPUT_FIELDS = [
     "reward_savings",
     "coupon_savings",
     "coupon_price",
+    "matched_discount_amount",
+    "net_line_total",
     "image_url",
     "raw_order_path",
     "item_name_norm",
@@ -42,13 +47,21 @@ OUTPUT_FIELDS = [
     "size_unit",
     "pack_qty",
     "measure_type",
+    "normalized_quantity",
+    "normalized_quantity_unit",
     "is_store_brand",
+    "is_item",
     "is_fee",
     "is_discount_line",
     "is_coupon_line",
     "price_per_each",
+    "price_per_each_basis",
+    "price_per_count",
+    "price_per_count_basis",
     "price_per_lb",
+    "price_per_lb_basis",
     "price_per_oz",
+    "price_per_oz_basis",
     "parse_version",
     "parse_notes",
 ]
@@ -327,6 +340,65 @@ def derive_prices(item, measure_type, size_value="", size_unit="", pack_qty=""):
     return price_per_each, price_per_lb, price_per_oz
 
 
+def derive_normalized_quantity(size_value, size_unit, pack_qty, measure_type):
+    parsed_size = to_decimal(size_value)
+    parsed_pack = to_decimal(pack_qty) or Decimal("1")
+
+    if parsed_size not in (None, Decimal("0")) and size_unit:
+        return format_decimal(parsed_size * parsed_pack), size_unit
+    if parsed_pack not in (None, Decimal("0")) and measure_type == "count":
+        return format_decimal(parsed_pack), "count"
+    if measure_type == "each":
+        return "1", "each"
+    return "", ""
+
+
+def derive_price_fields(price_per_each, price_per_lb, price_per_oz, line_total, qty, pack_qty):
+    line_total_decimal = to_decimal(line_total)
+    qty_decimal = to_decimal(qty)
+    pack_decimal = to_decimal(pack_qty)
+    price_per_count = ""
+    price_per_count_basis = ""
+    if line_total_decimal is not None and qty_decimal not in (None, Decimal("0")) and pack_decimal not in (
+        None,
+        Decimal("0"),
+    ):
+        price_per_count = format_decimal(line_total_decimal / (qty_decimal * pack_decimal))
+        price_per_count_basis = "line_total_over_pack_qty"
+
+    return {
+        "price_per_each": price_per_each,
+        "price_per_each_basis": "line_total_over_qty" if price_per_each else "",
+        "price_per_count": price_per_count,
+        "price_per_count_basis": price_per_count_basis,
+        "price_per_lb": price_per_lb,
+        "price_per_lb_basis": "parsed_or_picked_weight" if price_per_lb else "",
+        "price_per_oz": price_per_oz,
+        "price_per_oz_basis": "parsed_or_picked_weight" if price_per_oz else "",
+    }
+
+
+def normalization_identity(row):
+    if row.get("upc"):
+        return f"{row['retailer']}|upc={row['upc']}", "exact_upc"
+    if row.get("retailer_item_id"):
+        return f"{row['retailer']}|retailer_item_id={row['retailer_item_id']}", "exact_retailer_item_id"
+    if row.get("item_name_norm"):
+        return (
+            "|".join(
+                [
+                    row["retailer"],
+                    f"name={row['item_name_norm']}",
+                    f"size={row.get('size_value', '')}",
+                    f"unit={row.get('size_unit', '')}",
+                    f"pack={row.get('pack_qty', '')}",
+                ]
+            ),
+            "exact_name_size_pack",
+        )
+    return row["normalized_row_id"], "row_identity"
+
+
 def parse_item(order_id, order_date, raw_path, line_no, item):
     cleaned_name = clean_item_name(item.get("itemName", ""))
     size_value, size_unit, pack_qty = parse_size_and_pack(cleaned_name)
@@ -350,11 +422,42 @@ def parse_item(order_id, order_date, raw_path, line_no, item):
     if size_value and not size_unit:
         parse_notes.append("size_without_unit")
 
+    normalized_row_id = f"{RETAILER}:{order_id}:{line_no}"
+    normalized_quantity, normalized_quantity_unit = derive_normalized_quantity(
+        size_value,
+        size_unit,
+        pack_qty,
+        measure_type,
+    )
+    identity_key, normalization_basis = normalization_identity(
+        {
+            "retailer": RETAILER,
+            "normalized_row_id": normalized_row_id,
+            "upc": stringify(item.get("primUpcCd")),
+            "retailer_item_id": stringify(item.get("podId")),
+            "item_name_norm": normalized_name,
+            "size_value": size_value,
+            "size_unit": size_unit,
+            "pack_qty": pack_qty,
+        }
+    )
+    price_fields = derive_price_fields(
+        price_per_each,
+        price_per_lb,
+        price_per_oz,
+        stringify(item.get("groceryAmount")),
+        stringify(item.get("shipQy")),
+        pack_qty,
+    )
+
     return {
         "retailer": RETAILER,
         "order_id": str(order_id),
         "line_no": str(line_no),
-        "observed_item_key": f"{RETAILER}:{order_id}:{line_no}",
+        "normalized_row_id": normalized_row_id,
+        "normalized_item_id": f"gnorm:{identity_key}",
+        "normalization_basis": normalization_basis,
+        "observed_item_key": normalized_row_id,
         "order_date": normalize_whitespace(order_date),
         "retailer_item_id": stringify(item.get("podId")),
         "pod_id": stringify(item.get("podId")),
@@ -371,6 +474,8 @@ def parse_item(order_id, order_date, raw_path, line_no, item):
         "reward_savings": stringify(item.get("rewardSavings")),
         "coupon_savings": stringify(item.get("couponSavings")),
         "coupon_price": stringify(item.get("couponPrice")),
+        "matched_discount_amount": "",
+        "net_line_total": stringify(item.get("totalPrice")),
         "image_url": extract_image_url(item),
         "raw_order_path": raw_path.as_posix(),
         "item_name_norm": normalized_name,
@@ -380,13 +485,14 @@ def parse_item(order_id, order_date, raw_path, line_no, item):
         "size_unit": size_unit,
         "pack_qty": pack_qty,
         "measure_type": measure_type,
+        "normalized_quantity": normalized_quantity,
+        "normalized_quantity_unit": normalized_quantity_unit,
         "is_store_brand": "true" if bool(prefix) else "false",
+        "is_item": "false" if is_fee else "true",
         "is_fee": "true" if is_fee else "false",
         "is_discount_line": "false",
         "is_coupon_line": "false",
-        "price_per_each": price_per_each,
+        **price_fields,
-        "price_per_lb": price_per_lb,
-        "price_per_oz": price_per_oz,
         "parse_version": PARSER_VERSION,
         "parse_notes": ";".join(parse_notes),
     }
@@ -439,6 +545,7 @@ def write_csv(path, rows):
     help="CSV path for enriched Giant item rows.",
 )
 def main(input_dir, output_csv):
+    click.echo("legacy entrypoint: prefer normalize_giant_web.py for data-model outputs")
     raw_dir = Path(input_dir)
     output_path = Path(output_csv)
 

normalize_costco_web.py

@@ -0,0 +1,28 @@
+from pathlib import Path
+
+import click
+
+import enrich_costco
+
+
+@click.command()
+@click.option(
+    "--input-dir",
+    default="data/costco-web/raw",
+    show_default=True,
+    help="Directory containing Costco raw order json files.",
+)
+@click.option(
+    "--output-csv",
+    default="data/costco-web/normalized_items.csv",
+    show_default=True,
+    help="CSV path for normalized Costco item rows.",
+)
+def main(input_dir, output_csv):
+    rows = enrich_costco.build_items_enriched(Path(input_dir))
+    enrich_costco.write_csv(Path(output_csv), rows)
+    click.echo(f"wrote {len(rows)} rows to {output_csv}")
+
+
+if __name__ == "__main__":
+    main()

normalize_giant_web.py

@@ -0,0 +1,28 @@
+from pathlib import Path
+
+import click
+
+import enrich_giant
+
+
+@click.command()
+@click.option(
+    "--input-dir",
+    default="data/giant-web/raw",
+    show_default=True,
+    help="Directory containing Giant raw order json files.",
+)
+@click.option(
+    "--output-csv",
+    default="data/giant-web/normalized_items.csv",
+    show_default=True,
+    help="CSV path for normalized Giant item rows.",
+)
+def main(input_dir, output_csv):
+    rows = enrich_giant.build_items_enriched(Path(input_dir))
+    enrich_giant.write_csv(Path(output_csv), rows)
+    click.echo(f"wrote {len(rows)} rows to {output_csv}")
+
+
+if __name__ == "__main__":
+    main()

pm/data-model.org

@@ -1,133 +1,131 @@
-* grocery data model and file layout
+* Grocery data model and file layout
 
 This document defines the shared file layout and stable CSV schemas for the
-grocery pipeline. The goal is to keep retailer-specific ingest separate from
+grocery pipeline.
-cross-retailer product modeling so Giant-specific quirks do not become the
+Goals:
-system of record.
+- Ensure data gathering is separate from analysis
-
+- Enable multiple data gathering methods
-** design rules
+- One layer for review and analysis  
 
+** Design Rules
 - Raw retailer exports remain the source of truth.
 - Retailer parsing is isolated to retailer-specific files and ids.
-- Cross-retailer product layers begin only after retailer-specific enrichment.
+- Cross-retailer product layers begin only after retailer-specific normalization.
 - CSV schemas are stable and additive: new columns may be appended, but
    existing columns should not be repurposed.
 - Unknown values should be left blank rather than guessed.
 
-** directory layout
+*** Retailer-specific data:
-
-Use one top-level data root:
-
-#+begin_example
-data/
-  giant/
-    raw/
-      history.json
-      orders/
-        <order_id>.json
-    orders.csv
-    items_raw.csv
-    items_enriched.csv
-    products_observed.csv
-  costco/
-    raw/
-      ...
-    orders.csv
-    items_raw.csv
-    items_enriched.csv
-    products_observed.csv
-  shared/
-    products_canonical.csv
-    product_links.csv
-    review_queue.csv
-#+end_example
-
-** layer responsibilities
-
-- `data/<retailer>/raw/`
-  Stores unmodified retailer payloads exactly as fetched.
-- `data/<retailer>/orders.csv`
-  One row per retailer order or visit, flattened from raw order data.
-- `data/<retailer>/items_raw.csv`
-  One row per retailer line item, preserving retailer-native values needed for
-  reruns and debugging.
-- `data/<retailer>/items_enriched.csv`
-  Parsed retailer line items with normalized fields and derived guesses, still
-  retailer-specific.
-- `data/<retailer>/products_observed.csv`
-  Distinct retailer-facing observed products aggregated from enriched items.
-- `data/shared/products_canonical.csv`
-  Cross-retailer canonical product entities used for comparison.
-- `data/shared/product_links.csv`
-  Links from retailer observed products to canonical products.
-- `data/shared/review_queue.csv`
-  Human review queue for unresolved or low-confidence matching/parsing cases.
-
-** retailer-specific versus shared
-
-Retailer-specific:
-
 - raw json payloads
 - retailer order ids
 - retailer line numbers
 - retailer category ids and names
 - retailer item names
 - retailer image urls
-- parsed guesses derived from one retailer feed
-- observed products scoped to one retailer
-
-Shared:
-
-- canonical products
-- observed-to-canonical links
-- human review state for unresolved cases
 - comparison-ready normalized quantity basis fields
   
-Observed products are the boundary between retailer-specific parsing and
+*** Review/Combined data:
-cross-retailer canonicalization. Nothing upstream of `products_observed.csv`
+- catalog of reviewed products
-should require knowledge of another retailer.
+- links from normalized retailer items to catalog
+- human review state for unresolved cases
 
-** schema: `data/<retailer>/orders.csv`
 
-One row per order or visit.
+* Pipeline
+Each step can be run alone if its dependents exist.
+Each retail provider script must produce deterministic line-item outputs, and
+normalization may assign within-retailer product identity only when the
+retailer itself provides strong evidence.
 
-| column | meaning |
+Key: 
-|-
+- (1) input
-| `retailer` | retailer slug such as `giant` |
+- [1] output
-| `order_id` | retailer order or visit id |
-| `order_date` | order date in `YYYY-MM-DD` when available |
-| `delivery_date` | fulfillment date in `YYYY-MM-DD` when available |
-| `service_type` | retailer service type such as `INSTORE` |
-| `order_total` | order total as provided by retailer |
-| `payment_method` | retailer payment label |
-| `total_item_count` | total line count or item count from retailer |
-| `total_savings` | total savings as provided by retailer |
-| `your_savings_total` | savings field from retailer when present |
-| `coupons_discounts_total` | coupon/discount total from retailer |
-| `store_name` | retailer store name |
-| `store_number` | retailer store number |
-| `store_address1` | street address |
-| `store_city` | city |
-| `store_state` | state or province |
-| `store_zipcode` | postal code |
-| `refund_order` | retailer refund flag |
-| `ebt_order` | retailer EBT flag |
-| `raw_history_path` | relative path to source history payload |
-| `raw_order_path` | relative path to source order payload |
  
-Primary key:
+** 1. Collect
+Get raw receipt/visit and item data from a retailer.
+Scraping is unique to a Retailer and method (e.g., Giant-Web and Giant-Scan).
+Preserve complete raw data and preserve fidelity.
+Avoid interpretation beyond basic data flattening.
+ - (1) Source access (Varies, eg header data, auth for API access)
+ - [1] collected visits from each retailer
+ - [2] collected items from each retailer
+ - [3] any other raw data that supports [1] and [2]; explicit source (eventual receipt scan?)
    
-- (`retailer`, `order_id`)
+** 2. Normalize
+Parse and extract structured facts from retailer-specific raw data
+  to create a standardized item format for that retailer.
+Strictly dependent on Collect method and output.
+ - Extract quantity, size, pack, pricing, variant
+ - Add discount line items to product line items using upc/retail_item_id and concurrence
+ - Cleanup naming to facilitate later matching
+ - Assign retailer-level `normalized_item_id` only when evidence is deterministic
+ - Never use fuzzy or semantic matching here
+ - (1) collected items from each retailer
+ - (2) collected visits from each retailer
+ - [1] normalized items from each retailer
 
-** schema: `data/<retailer>/items_raw.csv`
+** 3. Review/Combine (Canonicalization)
+Decide whether two normalized retailer items are "the same product";
+ match items across retailers using algo/logic and human review.
+Create catalog linked to normalized retailer items.
+ - Review operates on distinct `normalized_item_id` values, not individual purchase rows
+ - Cross-retailer identity decisions happen only here
+ - Asking human to create a canonical/catalog item with:
+   - friendly/catalog_name: "bell pepper"; "milk"
+   - category: "produce"; "dairy"
+   - product_type: "pepper"; "milk"
+   - ? variant? "whole, "skim", "2pct"
+ - Then link the group of items to that catalog item.
+ - (1) normalized items from each retailer
+ - [1] review queue of items to be reviewed
+ - [2] catalog (lookup table) of confirmed normalized retailer items and catalog_id
+ - [3] purchase list of normalized items , pivot-ready
    
+** Unresolved Issues
+1. need central script to orchestrate; metadata belongs there and nowhere else
+2. `LIME` and `LIME . / .` appearing in the catalog: names must come from review-approved names, not raw strings
+
+
+* Directory Layout
+Use one top-level data root:
+#+begin_example
+main.py
+collect_<retailer>_<method>.py
+normalize_<retailer>_<method>.py
+review.py
+data/
+  <retailer-method>/
+    raw/  # unmodified retailer payloads exactly as fetched
+      <order_id.json> 
+    collected_items.csv # one row per retailer line item w/ retailer-native values
+    collected_orders.csv # one row per receipt/visit, flattened from raw order data
+    normalized_items.csv # parsed retailer-specific line items with normalized fields
+  costco-web/ # sample
+    raw/
+      orders/
+        history.json
+        <order_id>.json
+    collected_items.csv
+    collected_orders.csv
+    normalized_items.csv
+  review/
+    review_queue.csv # Human review queue for unresolved matching/parsing cases.
+    product_links.csv # Links from normalized retailer items to catalog items.
+  catalog.csv  # Cross-retailer product catalog entities used for comparison.
+  purchases.csv
+#+end_example
+
+Notes:
+- The current repo still uses transitional root-level scripts and output folders.
+- This layout is the target structure for the refactor, not a claim that migration is already complete.
+
+* Schemas
+** `data/<retailer-method>/collected_items.csv`
 One row per retailer line item.
-
+| key                | definition                                 |
-| column           | meaning                                 |
+|--------------------+--------------------------------------------|
-|------------------+-----------------------------------------|
+| `retailer` PK      | retailer slug                              |
-| `retailer`       | retailer slug                           |
+| `order_id` PK      | retailer order id                          |
-| `order_id`       | retailer order id                       |
+| `line_no`  PK      | stable line number within order export     |
-| `line_no`        | stable line number within order export  |
 | `order_date`       | copied from order when available           |
 | `retailer_item_id` | retailer-native item id when available     |
 | `pod_id`           | retailer pod/item id                       |
@@ -149,135 +147,110 @@ One row per retailer line item.
 | `is_discount_line` | retailer adjustment or discount-line flag  |
 | `is_coupon_line`   | coupon-like line flag when distinguishable |
 
-Primary key:
+** `data/<retailer-method>/collected_orders.csv`
+One row per order/visit/receipt.
+| key                       | definition                                      |
+|---------------------------+-------------------------------------------------|
+| `retailer` PK             | retailer slug such as `giant`                   |
+| `order_id` PK             | retailer order or visit id                      |
+| `order_date`              | order date in `YYYY-MM-DD` when available       |
+| `delivery_date`           | fulfillment date in `YYYY-MM-DD` when available |
+| `service_type`            | retailer service type such as `INSTORE`         |
+| `order_total`             | order total as provided by retailer             |
+| `payment_method`          | retailer payment label                          |
+| `total_item_count`        | total line count or item count from retailer    |
+| `total_savings`           | total savings as provided by retailer           |
+| `your_savings_total`      | savings field from retailer when present        |
+| `coupons_discounts_total` | coupon/discount total from retailer             |
+| `store_name`              | retailer store name                             |
+| `store_number`            | retailer store number                           |
+| `store_address1`          | street address                                  |
+| `store_city`              | city                                            |
+| `store_state`             | state or province                               |
+| `store_zipcode`           | postal code                                     |
+| `refund_order`            | retailer refund flag                            |
+| `ebt_order`               | retailer EBT flag                               |
+| `raw_history_path`        | relative path to source history payload         |
+| `raw_order_path`          | relative path to source order payload           |
 
-- (`retailer`, `order_id`, `line_no`)
+** `data/<retailer-method>/normalized_items.csv`
+One row per retailer line item after deterministic parsing. Preserve raw
+fields from `collected_items.csv` and add parsed fields that make later review
+and grouping easier. Normalization may assign retailer-level identity when the
+evidence is deterministic and retailer-scoped.
 
-** schema: `data/<retailer>/items_enriched.csv`
+| key                        | definition                                                       |
-
+|----------------------------+------------------------------------------------------------------|
-One row per retailer line item after deterministic parsing. Preserve the raw
+| `retailer` PK              | retailer slug                                                    |
-fields from `items_raw.csv` and add parsed fields.
+| `order_id` PK              | retailer order id                                                |
-
+| `line_no` PK               | line number within order                                         |
-| column              | meaning                                                     |
+| `normalized_row_id`        | stable row key, typically `<retailer>:<order_id>:<line_no>`      |
-|---------------------+-------------------------------------------------------------|
+| `normalized_item_id`       | stable retailer-level item identity when deterministic grouping is supported |
-| `retailer`          | retailer slug                                               |
+| `normalization_basis`      | basis used to assign `normalized_item_id`                        |
-| `order_id`          | retailer order id                                           |
-| `line_no`           | line number within order                                    |
-| `observed_item_key` | stable row key, typically `<retailer>:<order_id>:<line_no>` |
 | `retailer_item_id`         | retailer-native item id                                          |
 | `item_name`                | raw retailer item name                                           |
-| `item_name_norm`    | normalized item name                                        |
+| `item_name_norm`           | normalized retailer item name                                    |
 | `brand_guess`              | parsed brand guess                                               |
 | `variant`                  | parsed variant text                                              |
 | `size_value`               | parsed numeric size value                                        |
 | `size_unit`                | parsed size unit such as `oz`, `lb`, `fl_oz`                     |
 | `pack_qty`                 | parsed pack or count guess                                       |
 | `measure_type`             | `each`, `weight`, `volume`, `count`, or blank                    |
+| `normalized_quantity`      | numeric comparison basis derived during normalization            |
+| `normalized_quantity_unit` | basis unit such as `oz`, `lb`, `count`, or blank                 |
+| `is_item`                  | item flag                                                        |
 | `is_store_brand`           | store-brand guess                                                |
 | `is_fee`                   | fee or non-product flag                                          |
 | `is_discount_line`         | discount or adjustment-line flag                                 |
 | `is_coupon_line`           | coupon-like line flag                                            |
+| `matched_discount_amount`  | matched discount value carried onto purchased row when supported |
+| `net_line_total`           | line total after matched discount when supported                 |
 | `price_per_each`           | derived per-each price when supported                            |
+| `price_per_each_basis`     | source basis for `price_per_each`                                |
+| `price_per_count`          | derived per-count price when supported                           |
+| `price_per_count_basis`    | source basis for `price_per_count`                               |
 | `price_per_lb`             | derived per-pound price when supported                           |
+| `price_per_lb_basis`       | source basis for `price_per_lb`                                  |
 | `price_per_oz`             | derived per-ounce price when supported                           |
+| `price_per_oz_basis`       | source basis for `price_per_oz`                                  |
 | `image_url`                | best available retailer image url                                |
+| `raw_order_path`           | relative path to source order payload                            |
 | `parse_version`            | parser version string for reruns                                 |
 | `parse_notes`              | optional non-fatal parser notes                                  |
 
-Primary key:
+Notes:
+- `normalized_row_id` identifies the purchase row; `normalized_item_id` identifies a repeated retailer item when strong retailer evidence supports grouping.
+- Valid `normalization_basis` values should be explicit, e.g. `exact_upc`, `exact_retailer_item_id`, `exact_name_size_pack`, or `approved_retailer_alias`.
+- Do not use fuzzy or semantic matching to assign `normalized_item_id`.
+- Discount/coupon rows may remain as standalone normalized rows for auditability even when their amounts are attached to a purchased row via `matched_discount_amount`.
+- Cross-retailer identity is handled later in review/combine via `catalog.csv` and `product_links.csv`.
 
-- (`retailer`, `order_id`, `line_no`)
+** `data/review/product_links.csv`
+One row per review-approved link from a normalized retailer item to a catalog item.
+Many normalized retailer items may link to the same catalog item.
 
-** schema: `data/<retailer>/products_observed.csv`
+| key                     | definition                                  |
-
+|-------------------------+---------------------------------------------|
-One row per distinct retailer-facing observed product.
+| `normalized_item_id` PK | normalized retailer item id                 |
-
+| `catalog_id` PK         | linked catalog product id                   |
-| column                        | meaning                                                        |
+| `link_method`           | `manual`, `exact_upc`, `exact_name_size`, etc. |
-|-------------------------------+----------------------------------------------------------------|
-| `observed_product_id`         | stable observed product id                                     |
-| `retailer`                    | retailer slug                                                  |
-| `observed_key`                | deterministic grouping key used to create the observed product |
-| `representative_retailer_item_id` | best representative retailer-native item id               |
-| `representative_upc`          | best representative UPC/PLU                                    |
-| `representative_item_name`    | representative raw retailer name                               |
-| `representative_name_norm`    | representative normalized name                                 |
-| `representative_brand`        | representative brand guess                                     |
-| `representative_variant`      | representative variant                                         |
-| `representative_size_value`   | representative size value                                      |
-| `representative_size_unit`    | representative size unit                                       |
-| `representative_pack_qty`     | representative pack/count                                      |
-| `representative_measure_type` | representative measure type                                    |
-| `representative_image_url`    | representative image url                                       |
-| `is_store_brand`              | representative store-brand flag                                |
-| `is_fee`                      | representative fee flag                                        |
-| `is_discount_line`            | representative discount-line flag                              |
-| `is_coupon_line`              | representative coupon-line flag                                |
-| `first_seen_date`             | first order date seen                                          |
-| `last_seen_date`              | last order date seen                                           |
-| `times_seen`                  | number of enriched item rows grouped here                      |
-| `example_order_id`            | one example retailer order id                                  |
-| `example_item_name`           | one example raw item name                                      |
-| `distinct_retailer_item_ids_count` | count of distinct retailer-native item ids               |
-
-Primary key:
-
-- (`observed_product_id`)
-
-** schema: `data/shared/products_canonical.csv`
-
-One row per cross-retailer canonical product.
-
-| column                     | meaning                                          |
-|----------------------------+--------------------------------------------------|
-| `canonical_product_id`     | stable canonical product id                      |
-| `canonical_name`           | canonical human-readable name                    |
-| `product_type`             | broad class such as `apple`, `milk`, `trash_bag` |
-| `brand`                    | canonical brand when applicable                  |
-| `variant`                  | canonical variant                                |
-| `size_value`               | normalized size value                            |
-| `size_unit`                | normalized size unit                             |
-| `pack_qty`                 | normalized pack/count                            |
-| `measure_type`             | normalized measure type                          |
-| `normalized_quantity`      | numeric comparison basis value                   |
-| `normalized_quantity_unit` | basis unit such as `oz`, `lb`, `count`           |
-| `notes`                    | optional human notes                             |
-| `created_at`               | creation timestamp or date                       |
-| `updated_at`               | last update timestamp or date                    |
-
-Primary key:
-
-- (`canonical_product_id`)
-
-** schema: `data/shared/product_links.csv`
-
-One row per observed-to-canonical relationship.
-
-| column | meaning |
-|-
-| `observed_product_id` | retailer observed product id |
-| `canonical_product_id` | linked canonical product id |
-| `link_method` | `manual`, `exact_upc`, `exact_name`, etc. |
 | `link_confidence`       | optional confidence label                   |
 | `review_status`         | `pending`, `approved`, `rejected`, or blank |
 | `reviewed_by`           | reviewer id or initials                     |
 | `reviewed_at`           | review timestamp or date                    |
 | `link_notes`            | optional notes                              |
 
-Primary key:
+** `data/review/review_queue.csv`
-
-- (`observed_product_id`, `canonical_product_id`)
-
-** schema: `data/shared/review_queue.csv`
-
 One row per issue needing human review.
 
-| column | meaning |
+| key                  | definition                                          |
-|-
+|----------------------+-----------------------------------------------------|
-| `review_id` | stable review row id |
+| `review_id` PK       | stable review row id                                |
-| `queue_type` | `observed_product`, `link_candidate`, `parse_issue` |
+| `queue_type`         | `link_candidate`, `parse_issue`, `catalog_cleanup`  |
 | `retailer`           | retailer slug when applicable                       |
-| `observed_product_id` | observed product id when applicable |
+| `normalized_item_id` | normalized retailer item id when review is item-level |
-| `canonical_product_id` | candidate canonical id when applicable |
+| `normalized_row_id`  | normalized row id when review is row-specific       |
+| `catalog_id`         | candidate canonical id                              |
 | `reason_code`        | machine-readable review reason                      |
 | `priority`           | optional priority label                             |
 | `raw_item_names`     | compact list of example raw names                   |
@@ -290,20 +263,84 @@ One row per issue needing human review.
 | `resolution_notes`   | reviewer notes                                      |
 | `created_at`         | creation timestamp or date                          |
 | `updated_at`         | last update timestamp or date                       |
+** `data/catalog.csv`
+One row per cross-retailer catalog product.
+| key                        | definition                             |
+|----------------------------+----------------------------------------|
+| `catalog_id` PK            | stable catalog product id              |
+| `catalog_name`             | human-reviewed product name            |
+| `product_type`             | generic product eg `apple`, `milk`     |
+| `category`                 | broad section eg `produce`, `dairy`    |
+| `brand`                    | canonical brand when applicable        |
+| `variant`                  | canonical variant                      |
+| `size_value`               | normalized size value                  |
+| `size_unit`                | normalized size unit                   |
+| `pack_qty`                 | normalized pack/count                  |
+| `measure_type`             | normalized measure type                |
+| `normalized_quantity`      | numeric comparison basis value         |
+| `normalized_quantity_unit` | basis unit such as `oz`, `lb`, `count` |
+| `notes`                    | optional human notes                   |
+| `created_at`               | creation timestamp or date             |
+| `updated_at`               | last update timestamp or date          |
 
-Primary key:
+Notes:
+- Do not auto-create new catalog rows from weak normalized names alone.
+- Do not encode packaging/count into `catalog_name` unless it is essential to product identity.
+- `catalog_name` should come from review-approved naming, not raw retailer strings.
 
-- (`review_id`)
+** `data/purchases.csv`
+One row per purchased item (i.e., `is_item`==true from normalized layer), with
+catalog attributes denormalized in and discounts already applied.
 
-** current giant mapping
+| key                        | definition                                                     |
+|----------------------------+----------------------------------------------------------------|
+| `purchase_date`            | date of purchase (from order)                                  |
+| `retailer`                 | retailer slug                                                  |
+| `order_id`                 | retailer order id                                              |
+| `line_no`                  | line number within order                                       |
+| `normalized_row_id`        | `<retailer>:<order_id>:<line_no>`                              |
+| `normalized_item_id`       | retailer-level normalized item identity                        |
+| `catalog_id`               | linked catalog product id                                      |
+| `catalog_name`             | catalog product name for analysis                              |
+| `catalog_product_type`     | broader product family (e.g., `egg`, `milk`)                   |
+| `catalog_category`         | category such as `produce`, `dairy`                            |
+| `catalog_brand`            | canonical brand when applicable                                |
+| `catalog_variant`          | canonical variant when applicable                              |
+| `raw_item_name`            | original retailer item name                                    |
+| `normalized_item_name`     | cleaned/normalized retailer item name                          |
+| `retailer_item_id`         | retailer-native item id                                        |
+| `upc`                      | UPC/PLU when available                                         |
+| `qty`                      | retailer quantity field                                        |
+| `unit`                     | retailer unit (e.g., `EA`, `LB`)                               |
+| `pack_qty`                 | parsed pack/count                                              |
+| `size_value`               | parsed size value                                              |
+| `size_unit`                | parsed size unit                                               |
+| `measure_type`             | `each`, `weight`, `volume`, `count`                            |
+| `normalized_quantity`      | normalized comparison quantity                                 |
+| `normalized_quantity_unit` | unit for normalized quantity                                   |
+| `unit_price`               | retailer unit price                                            |
+| `line_total`               | original retailer extended price (pre-discount)                |
+| `matched_discount_amount`  | discount amount matched from discount lines                    |
+| `net_line_total`           | effective price after discount (`line_total` + discounts)      |
+| `store_name`               | retailer store name                                            |
+| `store_city`               | store city                                                     |
+| `store_state`              | store state                                                    |
+| `price_per_each`           | derived per-each price                                         |
+| `price_per_each_basis`     | source basis for per-each calc                                 |
+| `price_per_count`          | derived per-count price                                        |
+| `price_per_count_basis`    | source basis for per-count calc                                |
+| `price_per_lb`             | derived per-pound price                                        |
+| `price_per_lb_basis`       | source basis for per-pound calc                                |
+| `price_per_oz`             | derived per-ounce price                                        |
+| `price_per_oz_basis`       | source basis for per-ounce calc                                |
+| `is_fee`                   | true if row represents non-product fee                         |
+| `raw_order_path`           | relative path to original order payload                        |
 
-Current scraper outputs map to the new layout as follows:
+Notes:
+- Only rows that represent purchased items should appear here.
+- `line_total` preserves retailer truth; `net_line_total` is what you actually paid.
+- catalog fields are denormalized in to make pivoting trivial.
+- no discount/coupon rows exist here; their effects are carried via `matched_discount_amount`.
+- review/link decisions should apply at the `normalized_item_id` level, then fan out to all purchase rows sharing that id.
 
-- `giant_output/raw/history.json` -> `data/giant/raw/history.json`
+* /
-- `giant_output/raw/<order_id>.json` -> `data/giant/raw/orders/<order_id>.json`
-- `giant_output/orders.csv` -> `data/giant/orders.csv`
-- `giant_output/items.csv` -> `data/giant/items_raw.csv`
-
-Current Giant raw order payloads already expose fields needed for future
-enrichment, including `image`, `itemName`, `primUpcCd`, `lbEachCd`,
-`unitPrice`, `groceryAmount`, and `totalPickedWeight`.

pm/notes.org

@@ -71,6 +71,12 @@ l l : open local reflog
 put point on the commit; highlighted remote gitea/cx
 X   : reset branch; prompts you, selected cx
 
+
+
+** merge branch
+b b : switch to branch to be merged into (cx)
+m m : pick branch to merge into current branch
+
 * giant requests
 ** item:
 get:
@@ -250,18 +256,247 @@ python build_observed_products.py
 python build_review_queue.py
 python build_canonical_layer.py
 python validate_cross_retailer_flow.py
-* t1.11 tasks [2026-03-17 Tue 13:49]
+* t1.13 tasks [2026-03-17 Tue 13:49]
 ok i ran a few. time to run some cleanups here - i'm wondering if we shouldn't be less aggressive with canonical names and encourage a better manual process to start. 
-1. auto-created canonical_names lack category, product_type - ok with filling these in manually in the catalog once the queue is empty
+** TODO fill in auto-created canonical category, product-type
-2. canonical_names feel too specific, e.g., "5DZ egg"
+auto-created canonical_names lack category, product_type - ok with filling these in manually in the catalog once the queue is empty
-3. some canonical_names need consolidation, eg "LIME" and "LIME  . / ." ; poss cleanup issue. there are 5 entries for ergg but but they are all regular large grade A white eggs, just different amounts in dozens.
+
+** TODO consolidation cleanup
+1. canonical_names feel too specific, e.g., "5DZ egg" - probably a problem with the enrich_* steps not adding appropraite normalizing data /and/ removing from observed product title?
+2.  some canonical_names need consolidation, eg "LIME" and "LIME  . / ." ; poss cleanup issue. there are 5 entries for ergg but but they are all regular large grade A white eggs, just different amounts in dozens.
   Eggs are actually a great candidate for the kind of analysis we want to do - the pipeline should have caught and properly sorted these into size/qty:
+  #+begin_example
   ```canonical_product_id	canonical_name	category	product_type	brand	variant	size_value	size_unit	pack_qty	measure_type	notes	created_at	updated_at
   gcan_0e350505fd22	5DZ EGG / /			KS					each	auto-linked via exact_name		
   gcan_47279a80f5f3	EGG 5 DOZ. BBS								each	auto-linked via exact_name		
   gcan_7d099130c1bf	LRG WHITE EGG			SB				30	count	auto-linked via exact_upc		
   gcan_849c2817e667	GDA LRG WHITE EGG			SB				18	count	auto-linked via exact_upc		
   gcan_cb0c6c8cf480	LG EGG CONVENTIONAL					18	count		count	auto-linked via exact_name_size		  ```
-4. Build costco mechanism for matching discount to line item.
+  #+end_example
+** TODO costco discount matching
+Build costco mechanism for matching discount to line item.
    1. Discounts appear as their own line items with a number like /123456, this matches the UPC of the discounted item
    2. must be date-matched to the UPC
+
+Data model might be missing shape:
+1. match discount rows like `item_name:/2303476` to `retailer_item_id:2303476`
+2. display this value on the item somehow? maybe update line_total? otherwise we lose fidelity. should be stored in items_enriched somehow
+#+begin_example
+```retailer	order_id	line_no	observed_item_key	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	item_name_norm	brand_guess	variant	size_value	size_unit	pack_qty	measure_type	is_store_brand	is_fee	is_discount_line	is_coupon_line	price_per_each	price_per_lb	price_per_oz	parse_version	parse_notes
+costco	2.11115E+22	3	costco:21111520101942404241753:3	4/24/2024	2303476		KA 6QT MIXER P16 KSM60SECXER/CU FY23		33	33	1	None	399.99	399.99							costco_output/raw/21111520101942404241753-2024-04-24T17-53-00.json	KA 6QT MIXER KSM60SECXER/CU						each	FALSE	FALSE	FALSE	FALSE	399.99			costco-enrich-v1	
+costco	2.11115E+22	4	costco:21111520101942404241753:4	4/24/2024	325173		/2303476		33	33	-1	None	0	-100				-100			costco_output/raw/21111520101942404241753-2024-04-24T17-53-00.json	/2303476						each	FALSE	FALSE	TRUE	TRUE	100			costco-enrich-v1	```
+#+end_example
+** TODO giant discount matching
+
+* prompt
+ do not add new abstractions unless they remove real duplication. prefer explicit retailer-specific logic over generic heuristics. do not auto-create new canonical products from weak normalized names.
+ and propose the smallest set of edits needed.
+* 1.13 fixes
+** 15x Costco discounts not caught
+- 15x, some with slash-space: `/ 1768123`and some without: `/2303476`
+** canonical names suck - tempted to force manual config from scratch?
+- maybe first-pass should be naming groups, starting with largest groups and going on down.
+- unfortunately not seeing many cross-retailer items? looks like costco-only; just taking Giant as gospel
+- could be as simple as changing canonical name in canonical_catalog.csv  
+- tough to figure out where the data is, leading to below:  
+** need to refactor whole flow and where data is stored
+group by browser or by site, or both? currently mixed. 
+1. Scrape
+   - Script:
+   - Output: /output/raw/orderN.json, history.json, orders.csv, history.csv
+2. Enrich
+   - Scripts:
+   - Output: /output/enrich/items.json
+3. Combined - /output/?
+   - Review step?
+
+** propsed fixes
+* 1.14 prep - OBE
+** [ ] t1.14.1 define and document the filesystem/data-layer layout (2-3 commits)
+make stage ownership and retailer ownership explicit so every artifact has one obvious home
+
+** AC
+1. define and document the canonical directory layout for the pipeline, separating retailer-specific artifacts from shared combined artifacts
+2. adopt an explicit layout of the form:
+   - `data/<retailer>/raw/`
+   - `data/<retailer>/orders.csv`
+   - `data/<retailer>/items.csv`
+   - `data/<retailer>/items_enriched.csv`
+   - `data/combined/products_observed.csv`
+   - `data/combined/review_queue.csv`
+   - `data/combined/item_aliases.csv`
+   - `data/combined/canonical_catalog.csv`
+   - `data/combined/product_links.csv`
+   - `data/combined/purchases.csv`
+   - `data/combined/pipeline_status.csv`
+   - `data/combined/pipeline_status.json`
+3. update docs/readme and pipeline docs so each script’s inputs and outputs point to the new layout
+4. remove or deprecate ambiguous stage outputs living under a retailer-specific output directory when they are actually shared artifacts
+- pm note: goal is “where does this file live?” should have one answer, not three
+
+** evidence
+- commit:
+- tests:
+- date:
+
+** notes
+
+** [ ] t1.14.2 define the row-level data model for raw, enriched, observed, canonical, and purchases layers (2-4 commits)
+lock the item model before further refactors so each stage has a clear grain and purpose
+
+** AC
+1. document the row grain for each layer:
+   - raw item row = one receipt line from one retailer order
+   - enriched item row = one retailer line with retailer-specific parsed fields
+   - observed product row = one grouped retailer-facing product concept
+   - canonical catalog row = one review-controlled product identity
+   - purchase row = one final pivot-ready purchased item line
+2. define the required fields for each layer, including stable ids and provenance fields
+3. explicitly document which fields are allowed to be blank at each layer (e.g. `upc`, `canonical_item_id`, category)
+4. document the relationship between:
+   - `raw_item_name`
+   - `normalized_item_name`
+   - `observed_product_id`
+   - `canonical_item_id`
+5. document how retailer-native ids (e.g. Costco `retailer_item_id`) fit into the shared model without being forced into `upc`
+- pm note: this is the schema contract task; code should follow it, not invent it ad hoc
+
+** evidence
+- commit:
+- tests:
+- date:
+
+** notes
+** [ ] t1.14.3 refactor pipeline outputs to the new layout without changing semantics (2-4 commits)
+move files and script defaults to the new structure while preserving current behavior
+
+** AC
+1. update scraper and enrich scripts to write retailer-specific outputs under `data/<retailer>/...`
+2. update combined/shared scripts to read from retailer-specific enriched outputs and write to `data/combined/...`
+3. preserve current content/meaning of outputs during the move; this is a location/structure refactor, not a behavior rewrite
+4. update tests, docs, and script defaults to use the new paths
+- pm note: do not mix data-layout cleanup with canonical/review logic changes in this task
+
+** evidence
+- commit:
+- tests:
+- date:
+
+** notes
+** [ ] t1.14.4 make the review and catalog layer explicit and authoritative (2-4 commits)
+treat review and canonical resolution as first-class data, not incidental byproducts
+
+** AC
+1. define `review_queue.csv`, `item_aliases.csv`, and `canonical_catalog.csv` as the authoritative review/catalog files in `data/combined/`
+2. document the intended purpose of each:
+   - `review_queue.csv` = unresolved observed items needing action
+   - `item_aliases.csv` = approved mapping from observed/normalized names to canonical ids
+   - `canonical_catalog.csv` = review-controlled canonical product definitions and display names
+3. ensure final purchase generation reads from these files as the source of truth for resolution
+4. stop relying on weak implicit canonical creation as a substitute for the explicit review/catalog layer
+- pm note: this is the control-plane task; observed products may be automatic, canonical products are review-controlled
+
+** evidence
+- commit:
+- tests:
+- date:
+
+** notes
+** [ ] t1.14.5 define and document the final pivot-ready purchases output (2-3 commits)
+make the final analysis artifact explicit so excel/pivot/chart use is a first-class target
+
+** AC
+1. define `data/combined/purchases.csv` as the final normalized purchase log
+2. ensure each purchase row retains:
+   - purchase date
+   - retailer
+   - order id
+   - raw item name
+   - normalized item name
+   - canonical item id when resolved
+   - quantity and unit
+   - original line total
+   - discount-adjusted fields when applicable
+   - store/location fields where available
+3. document that `purchases.csv` is the primary excel/pivot input and that earlier files are staging layers
+4. document expected pivot uses such as purchase frequency and cost over time by canonical item
+- pm note: this task is about making the final artifact explicit and stable, not about adding new metrics
+
+** evidence
+- commit:
+- tests:
+- date:
+
+** notes
+
+* pipeline prep [2026-03-17 Tue]
+
+data saved to /data
+1. "scrape_<retailer>" gathers data from a retailer and outputs:
+   1. raw list of items per visit          ./<retailer>/scraped/raw/order-<uid>.json
+   2. raw list of visits                   ./<retailer>/scraped_visits.csv
+   3. raw list of items from all visits    ./<retailer>/scraped_items.csv
+2. "enrich <retailer>" takes /scraped/ data and outputs:
+   1. normalized list of items             ./<retailer>/enriched_items.csv
+3. "combine" takes retailer 
+ input:
+   1. all enriched items                   ./<retailer>/enriched_items.csv
+   2. all retailer visits                  ./<retailer>/scraped_visits.csv
+ outputs:
+   1. observed product groups              ./combined/observed/products_observed.csv
+   2. unresolved products for review       ./combined/review/review_queue.csv
+   3. pipeline accounting/status           ./combined/status/pipeline_status.csv
+   4. pipeline accounting/status           ./combined/status/pipeline_status.json
+4. review resolves unknown or weakly identified products and maintains:
+   1. canonical product catalog            ./combined/review/canonical_catalog.csv
+   2. approved alias mappings              ./combined/review/item_aliases.csv
+   3. optional observed→canonical links    ./combined/review/product_links.csv
+5. build purchases takes combined observed data plus review/catalog data and outputs:
+   [1]. final normalized purchase log        ./combined/purchases/purchases.csv
+
+lets get this pipeline right before more refactoring.
+
+* Pipeline - moved to data-model.org [2026-03-18 Wed]
+Key: 
+- (1) input
+- [2] output
+
+Each step can be run alone if its dependents exist.
+
+** 1. Collect
+Get raw receipt/visit and item data from a retailer.  Scraping is unique to a Retailer and method (e.g., Giant-Web and Giant-Scan).  Preserve complete raw data and preserve fidelity.  Avoid interpretation beyond basic data flattening.
+ - (1) Source access (Varies, eg header data, auth for API access)
+ - [1] collected visits from each retailer
+ - [2] collected items from each retailer
+ - [3] any other raw data that supports [1] and [2]; explicit source (eventual receipt scan?)
+   
+** 2. Normalize
+Parse and extract structured facts from retailer-specific raw data to create a standardized item format.  Strictly dependent on Collect method and output.
+ - Extract quantity, size, pack, pricing, variant
+ - Consolidate discount with item using upc/retail_item_id and concurrence
+ - Cleanup naming to facilitate later matching
+ - (1) collected items from each retailer
+ - (2) collected visits from each retailer
+ - [1] normalized items from each retailer
+
+** 3. Review/Combine (Canonicalization)
+Decide whether two normalized retailer items are "the same product"; match items across retailers using algo/logic and human review.  Create catalog linked to normalized items.
+ - Grouping the same item from retailer
+ - Asking human to create a canonical/catalog item with:
+   - friendly/canonical_name: "bell pepper"; "milk"
+   - category: "produce"; "dairy"
+   - product_type: "pepper"; "milk"
+   - ? variant? "whole, "skim", "2pct"
+ - (1) normalized items from each retailer
+ - [1] review queue of items to be reviewed
+ - [2] catalog (lookup table) of confirmed retailer_item and canonical_name
+ - [3] canonical purchase list, pivot-ready
+   
+** Unresolved Issues
+2. Create tags: canonical_name (need better label), category, product_type is missing data like Variant, shouldn't this be part of the normalization step?
+3. need central script to orchestrate; metadata belongs here and nowhere else
+
+** Symptoms
+- `LIME` and `LIME . / .` appearing in canonical_catalog:
+  - names must come from review-approved names, not raw strings
+* 

pm/task-sample.org

@@ -0,0 +1,22 @@
+#+title: Task Log
+#+updated: [2026-03-18 Wed 14:19]
+
+Use the template below, which should be a top-level org-mode header.
+
+* [ ] M.m.m: Task Title (estimate # commits)
+replace the old observed/canonical workflow with a review-first pipeline that groups normalized rows only during review/combine and links them to catalog items
+
+** Acceptance Criteria
+1. Criterion
+   - expanded data
+2. Criterion 
+
+- pm note: amplifying information
+
+** evidence
+- commit: abc123, bcd234
+- tests: 
+- datetime: [2026-03-18 Wed 14:15]
+
+** notes    
+- explanation of work done, decisions made, reasoning

pm/tasks.org

@@ -1,3 +1,5 @@
+#+title: Scrape-Giant Task Log
+
 * [X] t1.1: harden giant receipt fetch cli (2-4 commits)
 ** acceptance criteria
 - giant scraper runs from cli with prompts or env-backed defaults for `user_id` and `loyalty`
@@ -416,10 +418,259 @@ Clearly show current state separate from proposed future state.
 - Numbered canonical selection plus confirmation worked better than free-text id entry and should reduce accidental links.
 - Deterministic suggestions remain intentionally conservative; they speed up common cases, but unresolved items still depend on human review by design.
 
-* [ ] t1.10: add optional llm-assisted suggestion workflow for unresolved products (2-4 commits)
+* [X] t1.13.1 pipeline accountability and stage visibility (1-2 commits)
+add simple accounting so we can see what survives or drops at each pipeline stage
+
+** AC
+1. emit counts for raw, enriched, combined/observed, review-queued, canonical-linked, and final purchase-log rows
+2. report unresolved and dropped item counts explicitly
+3. make it easy to verify that missing items were intentionally left in review rather than silently lost
+- pm note: simple text/json/csv summary is sufficient; trust and visibility matter more than presentation
+
+** evidence
+- commit: `967e19e`
+- tests: `./venv/bin/python -m unittest discover -s tests`; `./venv/bin/python report_pipeline_status.py --help`; `./venv/bin/python report_pipeline_status.py`; verified `combined_output/pipeline_status.csv` and `combined_output/pipeline_status.json`
+- date: 2026-03-17
+
+** notes
+- Added a single explicit status script instead of threading counters through every pipeline step; this keeps the pipeline simple while still making row survival visible.
+- The most useful check here is `unresolved_not_in_review_rows`; when it is non-zero, we know we have a real accounting bug rather than normal unresolved work.
+
+* [X] t1.13.2 costco discount matching and net pricing in enrich_costco (2-3 commits)
+refactor costco enrichment so discount lines are matched to purchased items and net pricing is preserved
+
+** AC
+1. detect costco discount/coupon rows like `/<retailer_item_id>` and match them to purchased items within the same order
+2. preserve raw discount rows for auditability while also carrying matched discount values onto the purchased item row
+3. add explicit fields for discount-adjusted pricing, e.g. `matched_discount_amount` and `net_line_total` (or equivalent)
+4. preserve original raw receipt amounts (`line_total`) without overwriting them
+- pm note: keep this retailer-specific and explicit; do not introduce generic discount heuristics
+
+** evidence
+- commit: `56a03bc`
+- tests: `./venv/bin/python -m unittest discover -s tests`; `./venv/bin/python enrich_costco.py`; verified matched Costco discount rows now populate `matched_discount_amount` and `net_line_total` while preserving raw `line_total`
+- date: 2026-03-17
+
+** notes
+- Kept this retailer-specific and literal: only discount rows with `/<retailer_item_id>` are matched, and only within the same order.
+- Raw discount rows are still preserved for auditability; the purchased row now carries the matched adjustment separately rather than overwriting the original amount.
+* [X] t1.13.3 canonical cleanup and review-first product identity (3-4 commits)
+refactor canonical generation so product identity is cleaner, duplicate canonicals are reduced, and unresolved items stay in review instead of spawning junk canonicals
+
+** AC
+1. stop auto-creating new canonical products from weak normalized names alone; unresolved items remain in `review_queue.csv`
+2. canonical names are based on stable product identity rather than noisy observed titles
+3. packaging/count/size tokens are removed from canonical names when they belong in structured fields (`pack_qty`, `size_value`, `size_unit`)
+4. consolidate obvious duplicate canonicals (e.g. egg/lime cases) and ensure final outputs retain raw item name, normalized item name, and canonical item id
+- pm note: prefer conservative canonical creation and a better manual review loop over aggressive auto-unification
+
+** evidence
+- commit: `08e2a86`
+- tests: `./venv/bin/python -m unittest discover -s tests`; `./venv/bin/python build_purchases.py`; `./venv/bin/python review_products.py --refresh-only`; verified weaker exact-name cases now remain unresolved in `combined_output/review_queue.csv` and canonical names are cleaned before auto-catalog creation
+- date: 2026-03-17
+
+** notes
+- Removed weak exact-name auto-canonical creation so ambiguous products stay in review instead of generating junk canonicals.
+- Canonical display names are now cleaned of obvious punctuation and packaging noise, but I kept the cleanup conservative rather than adding a broad fuzzy merge layer.
+* [X] t1.14: refactor retailer collection into the new data model (2-4 commits)
+move Giant and Costco collection into the new collect structure and make both retailers emit the same collected schemas
+
+** Acceptance Criteria
+1. create retailer-specific collect scripts in the target naming pattern, e.g.:
+  - collect_giant_web.py
+  - collect_costco_web.py
+2. collected outputs conform to pm/data-model.org:
+  - data/<retailer-method>/raw/...
+  - data/<retailer-method>/collected_orders.csv
+  - data/<retailer-method>/collected_items.csv
+3. current Giant and Costco raw acquisition behavior is preserved during the move
+4. collected schemas preserve retailer truth and provenance:
+  - no interpretation beyond basic flattening
+  - raw_order_path/raw_history_path remain usable
+  - unknown values remain blank rather than guessed
+5. old paths should be removed or deprecated
+6. collect_* scripts do not depend on any normalize/review files or scripts
+- pm note: this is a path/schema refactor, not a parsing rewrite
+
+** evidence
+- commit: `48c6eaf`
+- tests: `./venv/bin/python -m unittest tests.test_scraper tests.test_costco_pipeline tests.test_browser_session`; `./venv/bin/python collect_giant_web.py --help`; `./venv/bin/python collect_costco_web.py --help`; `./venv/bin/python scrape_giant.py --help`; `./venv/bin/python scrape_costco.py --help`
+- datetime: 2026-03-18
+
+** notes
+- Kept this as a path/schema move, not a parsing rewrite: the existing Giant and Costco collection behavior remains in place behind new `collect_*` entry points.
+- Added lightweight deprecation nudges on the legacy `scrape_*` commands rather than removing them immediately, so the move is inspectable and low-risk.
+- The main schema fix was on Giant collection, which was missing retailer/provenance/audit fields that Costco collection already carried.
+
+* [X] t1.14.1: refactor retailer normalization into the new normalized_items schema (3-5 commits)
+make Giant and Costco emit the shared normalized line-item schema without introducing cross-retailer identity logic
+
+** Acceptance Criteria
+1. create retailer-specific normalize scripts in the target naming pattern, e.g.:
+   - normalize_giant_web.py
+   - normalize_costco_web.py
+2. normalized outputs conform to pm/data-model.org:
+   - data/<retailer-method>/normalized_items.csv
+   - one row per collected line item
+   - normalized_row_id is stable and present
+   - normalized_item_id is stable, present, and represents retailer-level identity reused across repeated purchase rows when deterministic retailer evidence is sufficient
+   - normalized_quantity and normalized_quantity_unit
+   - repeated rows for the same retailer product resolve to the same normalized_item_id only when supported by deterministic retailer evidence, e.g. exact upc, exact retailer_item_id, exact cleaned name + same size/pack
+   - normalization_basis is explicit
+3. Giant normalization preserves current useful parsing:
+   - normalized item name
+   - size/unit/pack parsing
+   - fee/store-brand flags
+   - derived price fields
+4. Costco normalization preserves current useful parsing:
+   - normalized item name
+   - size/unit/pack parsing
+   - explicit discount matching using retailer-specific logic
+   - matched_discount_amount and net_line_total
+5. both normalizers preserve raw retailer truth:
+   - line_total is never overwritten
+   - unknown values remain blank rather than guessed
+6. no cross-retailer identity assignment occurs in normalization
+7. normalize never uses fuzzy or semantic matching to assign normalized_item_id
+
+- pm note: prefer explicit retailer-specific code paths over generic normalization helpers unless the duplication is truly mechanical
+- pm note: normalization may resolve retailer-level identity, but not catalog identity
+- pm note: normalized_item_id is the only retailer-level grouping identity; do not introduce observed_products or a second grouping artifact
+** evidence
+- commit: `9064de5`
+- tests: `./venv/bin/python -m unittest discover -s tests`; `./venv/bin/python -m unittest tests.test_enrich_giant tests.test_costco_pipeline tests.test_purchases`; `./venv/bin/python normalize_giant_web.py --help`; `./venv/bin/python normalize_costco_web.py --help`; `./venv/bin/python enrich_giant.py --help`; `./venv/bin/python enrich_costco.py --help`
+- datetime: 2026-03-18
+
+** notes
+- Kept the existing Giant and Costco parsing logic intact and added the new normalized schema fields in place, rather than rewriting the enrichers from scratch.
+- `normalized_item_id` is always present, but it only collapses repeated rows when the evidence is strong; otherwise it falls back to row-level identity via `normalized_row_id`.
+- Added `normalize_*` entry points for the new data-model layout while leaving the legacy `enrich_*` commands available during the transition.
+
+* [ ] t1.14.2: finalize filesystem and schema alignment for the refactor (2-4 commits)
+bring on-disk outputs fully into the target `data/` structure without changing retailer behavior
+
+** Acceptance Criteria
+1. retailer data directories conform to pm/data-model.org:
+   - `data/giant-web/raw/...`
+   - `data/giant-web/collected_orders.csv`
+   - `data/giant-web/collected_items.csv`
+   - `data/giant-web/normalized_items.csv`
+   - `data/costco-web/raw/...`
+   - `data/costco-web/collected_orders.csv`
+   - `data/costco-web/collected_items.csv`
+   - `data/costco-web/normalized_items.csv`
+2. review/combine outputs are moved or rewritten into the target review paths:
+   - `data/review/review_queue.csv`
+   - `data/review/product_links.csv`
+   - `data/review/review_resolutions.csv`
+   - `data/review/purchases.csv`
+   - `data/review/pipeline_status.csv`
+   - `data/review/pipeline_status.json`
+3. old transitional output paths are either:
+   - removed from active script defaults, or
+   - left as explicit compatibility shims with clear deprecation notes
+4. no recollection is required if existing raw files and collected csvs can be moved/copied losslessly into the new structure
+5. no schema information is lost during the move:
+   - raw paths still resolve
+   - collected/normalized csvs still open with the expected headers
+6. README and task/docs reflect the final active paths
+- pm note: prefer moving/adapting existing files over recollecting from retailers unless a real data loss or schema mismatch forces recollection
+- pm note: this is a structure-alignment task, not a retailer parsing task
+
+** evidence
+- commit:
+- tests:
+- datetime:
+
+** notes
+
+* [ ] t1.14.3: retailer-specific Costco normalization cleanup (2-4 commits)
+tighten Costco-specific normalization so normalized item names are cleaner and deterministic retailer grouping is less noisy
+
+** Acceptance Criteria
+1. improve Costco item-name cleanup for obvious non-identity noise, such as:
+   - trailing slash fragments
+   - code tokens and receipt-format artifacts
+   - duplicated measurement fragments already captured in structured fields
+2. preserve deterministic normalization rules only:
+   - exact retailer_item_id
+   - exact cleaned name + same size/pack when needed
+   - approved retailer alias
+   - no fuzzy or semantic matching
+3. normalized Costco names improve on known bad examples, e.g.:
+   - `MANDARIN /` -> cleaner normalized item name
+   - `LIFE 6'TABLE ... /` -> cleaner normalized item name
+4. cleanup does not overwrite retailer truth:
+   - raw `item_name` is unchanged
+   - parsed `size_value`, `size_unit`, `pack_qty`, and pricing fields remain intact
+5. discount-row behavior remains correct:
+   - matched discount rows still populate `matched_discount_amount`
+   - `net_line_total` remains correct
+   - discount rows remain auditable
+6. add regression tests for the cleaned Costco examples and any new parsing rules
+- pm note: keep this explicitly Costco-specific; do not introduce a generic cleanup framework
+- pm note: prefer a short allowlist/blocklist of known receipt artifacts over broad heuristics
+
+** evidence
+- commit:
+- tests:
+- datetime:
+
+** notes
+
+* [ ] t1.15: refactor review/combine pipeline around normalized_item_id and catalog links (4-8 commits)
+replace the old observed/canonical workflow with a review-first pipeline that uses normalized_item_id as the retailer-level review unit and links it to catalog items
+
+** Acceptance Criteria
+1. refactor review outputs to conform to pm/data-model.org:
+   - data/review/review_queue.csv
+   - data/review/product_links.csv
+   - data/catalog.csv
+   - data/purchases.csv
+2. review logic uses normalized_item_id as the upstream retailer-level review identity:
+   - no dependency on observed_product_id
+   - no dependency on products_observed.csv
+   - one review/link decision applies to all purchase rows sharing the same normalized_item_id
+3. product_links.csv stores review-approved links from normalized_item_id to catalog_id
+   - one row per approved retailer-level identity to catalog mapping
+4. catalog.csv entries are review-first and conservative:
+   - no auto-creation from weak normalized names alone
+   - names come from reviewed catalog naming, not raw retailer strings
+   - packaging/count is not embedded in catalog_name unless essential to identity
+   - catalog_name/product_type/category/brand/variant may be blank until reviewed; blank is preferred to guessed
+5. purchases.csv remains pivot-ready and retains:
+   - raw item name
+   - normalized item name
+   - normalized_row_id (not for review)
+   - normalized_item_id
+   - catalog_id
+   - catalog fields
+   - raw line_total
+   - matched_discount_amount and net_line_total when present
+   - derived price fields and their bases
+6. terminal review flow remains simple and usable:
+   - reviewer sees one grouped retailer item identity (normalized_item_id) with count and list of matches, not one prompt per purchase row; use existing pattern as a template
+   - link to existing catalog item
+   - create new catalog item
+   - exclude
+   - skip
+7. pipeline accounting remains valid after the refactor:
+   - unresolved items are visible
+   - missing items are not silently dropped
+8. pm note: prefer a better manual review loop over aggressive automatic grouping. initial manual data entry is expected, and should resolve over time
+9. pm note: keep review/combine auditable; each catalog link should be explainable from normalized rows and review state
+
+** evidence
+- commit:
+- tests:
+- datetime:
+
+** notes
+
+* [ ] 1t.10: add optional llm-assisted suggestion workflow for unresolved normalized retailer items (2-4 commits)
 
 ** acceptance criteria
-- llm suggestions are generated only for unresolved observed products
+- llm suggestions are generated only for unresolved normalized retailer items
 - llm outputs are stored as suggestions, not auto-applied truth
 - reviewer can approve/edit/reject suggestions
 - approved decisions are persisted into canonical/link files

report_pipeline_status.py

@@ -0,0 +1,119 @@
+import json
+from pathlib import Path
+
+import click
+
+import build_observed_products
+import build_purchases
+import review_products
+from layer_helpers import read_csv_rows, write_csv_rows
+
+
+SUMMARY_FIELDS = ["stage", "count"]
+
+
+def read_rows_if_exists(path):
+    path = Path(path)
+    if not path.exists():
+        return []
+    return read_csv_rows(path)
+
+
+def build_status_summary(
+    giant_orders,
+    giant_items,
+    giant_enriched,
+    costco_orders,
+    costco_items,
+    costco_enriched,
+    purchases,
+    resolutions,
+):
+    enriched_rows = giant_enriched + costco_enriched
+    observed_rows = build_observed_products.build_observed_products(enriched_rows)
+    queue_rows = review_products.build_review_queue(purchases, resolutions)
+
+    unresolved_purchase_rows = [
+        row
+        for row in purchases
+        if row.get("observed_product_id")
+        and not row.get("canonical_product_id")
+        and row.get("is_fee") != "true"
+        and row.get("is_discount_line") != "true"
+        and row.get("is_coupon_line") != "true"
+    ]
+    excluded_rows = [
+        row
+        for row in purchases
+        if row.get("resolution_action") == "exclude"
+    ]
+    linked_purchase_rows = [row for row in purchases if row.get("canonical_product_id")]
+
+    summary = [
+        {"stage": "raw_orders", "count": len(giant_orders) + len(costco_orders)},
+        {"stage": "raw_items", "count": len(giant_items) + len(costco_items)},
+        {"stage": "enriched_items", "count": len(enriched_rows)},
+        {"stage": "observed_products", "count": len(observed_rows)},
+        {"stage": "review_queue_observed_products", "count": len(queue_rows)},
+        {"stage": "canonical_linked_purchase_rows", "count": len(linked_purchase_rows)},
+        {"stage": "final_purchase_rows", "count": len(purchases)},
+        {"stage": "unresolved_purchase_rows", "count": len(unresolved_purchase_rows)},
+        {"stage": "excluded_purchase_rows", "count": len(excluded_rows)},
+        {
+            "stage": "unresolved_not_in_review_rows",
+            "count": len(
+                [
+                    row
+                    for row in unresolved_purchase_rows
+                    if row.get("observed_product_id")
+                    not in {queue_row["observed_product_id"] for queue_row in queue_rows}
+                ]
+            ),
+        },
+    ]
+    return summary
+
+
+@click.command()
+@click.option("--giant-orders-csv", default="giant_output/orders.csv", show_default=True)
+@click.option("--giant-items-csv", default="giant_output/items.csv", show_default=True)
+@click.option("--giant-enriched-csv", default="giant_output/items_enriched.csv", show_default=True)
+@click.option("--costco-orders-csv", default="costco_output/orders.csv", show_default=True)
+@click.option("--costco-items-csv", default="costco_output/items.csv", show_default=True)
+@click.option("--costco-enriched-csv", default="costco_output/items_enriched.csv", show_default=True)
+@click.option("--purchases-csv", default="combined_output/purchases.csv", show_default=True)
+@click.option("--resolutions-csv", default="combined_output/review_resolutions.csv", show_default=True)
+@click.option("--summary-csv", default="combined_output/pipeline_status.csv", show_default=True)
+@click.option("--summary-json", default="combined_output/pipeline_status.json", show_default=True)
+def main(
+    giant_orders_csv,
+    giant_items_csv,
+    giant_enriched_csv,
+    costco_orders_csv,
+    costco_items_csv,
+    costco_enriched_csv,
+    purchases_csv,
+    resolutions_csv,
+    summary_csv,
+    summary_json,
+):
+    summary_rows = build_status_summary(
+        read_rows_if_exists(giant_orders_csv),
+        read_rows_if_exists(giant_items_csv),
+        read_rows_if_exists(giant_enriched_csv),
+        read_rows_if_exists(costco_orders_csv),
+        read_rows_if_exists(costco_items_csv),
+        read_rows_if_exists(costco_enriched_csv),
+        read_rows_if_exists(purchases_csv),
+        read_rows_if_exists(resolutions_csv),
+    )
+    write_csv_rows(summary_csv, summary_rows, SUMMARY_FIELDS)
+    summary_json_path = Path(summary_json)
+    summary_json_path.parent.mkdir(parents=True, exist_ok=True)
+    summary_json_path.write_text(json.dumps(summary_rows, indent=2), encoding="utf-8")
+    for row in summary_rows:
+        click.echo(f"{row['stage']}: {row['count']}")
+
+
+if __name__ == "__main__":
+    main()

scrape_costco.py

@@ -648,6 +648,27 @@ def main(
     window_days,
     months_back,
     firefox_profile_dir,
+):
+    click.echo("legacy entrypoint: prefer collect_costco_web.py for data-model outputs")
+    run_collection(
+        outdir=outdir,
+        document_type=document_type,
+        document_sub_type=document_sub_type,
+        window_days=window_days,
+        months_back=months_back,
+        firefox_profile_dir=firefox_profile_dir,
+    )
+
+
+def run_collection(
+    outdir,
+    document_type,
+    document_sub_type,
+    window_days,
+    months_back,
+    firefox_profile_dir,
+    orders_filename="orders.csv",
+    items_filename="items.csv",
 ):
     outdir = Path(outdir)
     raw_dir = outdir / "raw"
@@ -706,8 +727,8 @@ def main(
         write_json(raw_dir / f"{safe_filename(receipt_id)}.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 / orders_filename, orders, ORDER_FIELDS)
-    write_csv(outdir / "items.csv", items, ITEM_FIELDS)
+    write_csv(outdir / items_filename, items, ITEM_FIELDS)
     click.echo(f"wrote {len(orders)} orders and {len(items)} item rows to {outdir}")
 
 

scrape_giant.py

@@ -13,8 +13,10 @@ from browser_session import find_firefox_profile_dir, load_firefox_cookies
 
 BASE = "https://giantfood.com"
 ACCOUNT_PAGE = f"{BASE}/account/history/invoice/in-store"
+RETAILER = "giant"
 
 ORDER_FIELDS = [
+    "retailer",
     "order_id",
     "order_date",
     "delivery_date",
@@ -33,12 +35,16 @@ ORDER_FIELDS = [
     "store_zipcode",
     "refund_order",
     "ebt_order",
+    "raw_history_path",
+    "raw_order_path",
 ]
 
 ITEM_FIELDS = [
+    "retailer",
     "order_id",
     "order_date",
     "line_no",
+    "retailer_item_id",
     "pod_id",
     "item_name",
     "upc",
@@ -53,6 +59,10 @@ ITEM_FIELDS = [
     "reward_savings",
     "coupon_savings",
     "coupon_price",
+    "image_url",
+    "raw_order_path",
+    "is_discount_line",
+    "is_coupon_line",
 ]
 
 
@@ -130,18 +140,21 @@ def get_order_detail(session, user_id, order_id):
     return response.json()
 
 
-def flatten_orders(history, details):
+def flatten_orders(history, details, history_path=None, raw_dir=None):
     orders = []
     items = []
     history_lookup = {record["orderId"]: record for record in history.get("records", [])}
+    history_path_value = history_path.as_posix() if history_path else ""
 
     for detail in details:
         order_id = str(detail["orderId"])
         history_row = history_lookup.get(detail["orderId"], {})
         pickup = detail.get("pup", {})
+        raw_order_path = (raw_dir / f"{order_id}.json").as_posix() if raw_dir else ""
 
         orders.append(
             {
+                "retailer": RETAILER,
                 "order_id": order_id,
                 "order_date": detail.get("orderDate"),
                 "delivery_date": detail.get("deliveryDate"),
@@ -160,15 +173,19 @@ def flatten_orders(history, details):
                 "store_zipcode": pickup.get("storeZipcode"),
                 "refund_order": detail.get("refundOrder"),
                 "ebt_order": detail.get("ebtOrder"),
+                "raw_history_path": history_path_value,
+                "raw_order_path": raw_order_path,
             }
         )
 
         for line_no, item in enumerate(detail.get("items", []), start=1):
             items.append(
                 {
+                    "retailer": RETAILER,
                     "order_id": order_id,
                     "order_date": detail.get("orderDate"),
                     "line_no": str(line_no),
+                    "retailer_item_id": "",
                     "pod_id": item.get("podId"),
                     "item_name": item.get("itemName"),
                     "upc": item.get("primUpcCd"),
@@ -183,6 +200,10 @@ def flatten_orders(history, details):
                     "reward_savings": item.get("rewardSavings"),
                     "coupon_savings": item.get("couponSavings"),
                     "coupon_price": item.get("couponPrice"),
+                    "image_url": "",
+                    "raw_order_path": raw_order_path,
+                    "is_discount_line": "false",
+                    "is_coupon_line": "false",
                 }
             )
 
@@ -269,6 +290,18 @@ def write_json(path, payload):
     help="Delay between order detail requests.",
 )
 def main(user_id, loyalty, outdir, sleep_seconds):
+    click.echo("legacy entrypoint: prefer collect_giant_web.py for data-model outputs")
+    run_collection(user_id, loyalty, outdir, sleep_seconds)
+
+
+def run_collection(
+    user_id,
+    loyalty,
+    outdir,
+    sleep_seconds,
+    orders_filename="orders.csv",
+    items_filename="items.csv",
+):
     config = load_config()
     user_id = user_id or config["user_id"] or click.prompt("Giant user id", type=str)
     loyalty = loyalty or config["loyalty"] or click.prompt(
@@ -279,13 +312,14 @@ def main(user_id, loyalty, outdir, sleep_seconds):
     rawdir = outdir / "raw"
     rawdir.mkdir(parents=True, exist_ok=True)
 
-    orders_csv = outdir / "orders.csv"
+    orders_csv = outdir / orders_filename
-    items_csv = outdir / "items.csv"
+    items_csv = outdir / items_filename
     existing_order_ids = read_existing_order_ids(orders_csv)
 
     session = build_session()
     history = get_history(session, user_id, loyalty)
-    write_json(rawdir / "history.json", history)
+    history_path = rawdir / "history.json"
+    write_json(history_path, history)
 
     records = history.get("records", [])
     click.echo(f"history returned {len(records)} visits; Giant exposes only the most recent 50")
@@ -310,7 +344,7 @@ def main(user_id, loyalty, outdir, sleep_seconds):
         if index < len(unseen_records):
             time.sleep(sleep_seconds)
 
-    orders, items = flatten_orders(history, details)
+    orders, items = flatten_orders(history, details, history_path=history_path, raw_dir=rawdir)
     merged_orders = append_dedup(
         orders_csv,
         orders,

tests/test_canonical_layer.py

@@ -4,7 +4,7 @@ import build_canonical_layer
 
 
 class CanonicalLayerTests(unittest.TestCase):
-    def test_build_canonical_layer_auto_links_exact_upc_and_name_size(self):
+    def test_build_canonical_layer_auto_links_exact_upc_and_name_size_only(self):
         observed_rows = [
             {
                 "observed_product_id": "gobs_1",
@@ -81,6 +81,21 @@ class CanonicalLayerTests(unittest.TestCase):
                 "is_discount_line": "false",
                 "is_coupon_line": "false",
             },
+            {
+                "observed_product_id": "gobs_6",
+                "representative_upc": "",
+                "representative_retailer_item_id": "",
+                "representative_name_norm": "LIME",
+                "representative_brand": "",
+                "representative_variant": "",
+                "representative_size_value": "",
+                "representative_size_unit": "",
+                "representative_pack_qty": "",
+                "representative_measure_type": "each",
+                "is_fee": "false",
+                "is_discount_line": "false",
+                "is_coupon_line": "false",
+            },
         ]
 
         canonicals, links = build_canonical_layer.build_canonical_layer(observed_rows)
@@ -93,6 +108,11 @@ class CanonicalLayerTests(unittest.TestCase):
         self.assertEqual("exact_name_size", methods["gobs_3"])
         self.assertEqual("exact_name_size", methods["gobs_4"])
         self.assertNotIn("gobs_5", methods)
+        self.assertNotIn("gobs_6", methods)
+
+    def test_clean_canonical_name_removes_packaging_noise(self):
+        self.assertEqual("LIME", build_canonical_layer.clean_canonical_name("LIME  . / ."))
+        self.assertEqual("EGG", build_canonical_layer.clean_canonical_name("5DZ EGG / /"))
 
 
 if __name__ == "__main__":

tests/test_costco_pipeline.py

@@ -258,6 +258,11 @@ class CostcoPipelineTests(unittest.TestCase):
         self.assertEqual("MIXED PEPPER", row["item_name_norm"])
         self.assertEqual("6", row["pack_qty"])
         self.assertEqual("count", row["measure_type"])
+        self.assertEqual("costco:abc:1", row["normalized_row_id"])
+        self.assertEqual("exact_retailer_item_id", row["normalization_basis"])
+        self.assertTrue(row["normalized_item_id"])
+        self.assertEqual("6", row["normalized_quantity"])
+        self.assertEqual("count", row["normalized_quantity_unit"])
 
         discount = enrich_costco.parse_costco_item(
             order_id="abc",
@@ -278,6 +283,58 @@ class CostcoPipelineTests(unittest.TestCase):
         )
         self.assertEqual("true", discount["is_discount_line"])
         self.assertEqual("true", discount["is_coupon_line"])
+        self.assertEqual("false", discount["is_item"])
+
+    def test_build_items_enriched_matches_discount_to_item(self):
+        with tempfile.TemporaryDirectory() as tmpdir:
+            raw_dir = Path(tmpdir) / "raw"
+            raw_dir.mkdir()
+            payload = {
+                "data": {
+                    "receiptsWithCounts": {
+                        "receipts": [
+                            {
+                                "transactionBarcode": "abc",
+                                "transactionDate": "2026-03-12",
+                                "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,
+                                    },
+                                ],
+                            }
+                        ]
+                    }
+                }
+            }
+            (raw_dir / "abc.json").write_text(json.dumps(payload), encoding="utf-8")
+
+            rows = enrich_costco.build_items_enriched(raw_dir)
+
+            purchase_row = next(row for row in rows if row["is_discount_line"] == "false")
+            discount_row = next(row for row in rows if row["is_discount_line"] == "true")
+            self.assertEqual("-5", purchase_row["matched_discount_amount"])
+            self.assertEqual("14.99", purchase_row["net_line_total"])
+            self.assertIn("matched_discount=4873222", purchase_row["parse_notes"])
+            self.assertIn("matched_to_item=4873222", discount_row["parse_notes"])
 
     def test_cross_retailer_validation_writes_proof_example(self):
         with tempfile.TemporaryDirectory() as tmpdir:

tests/test_enrich_giant.py

@@ -51,6 +51,11 @@ class EnrichGiantTests(unittest.TestCase):
         self.assertEqual("1.99", row["price_per_lb"])
         self.assertEqual("0.1244", row["price_per_oz"])
         self.assertEqual("https://example.test/apple.jpg", row["image_url"])
+        self.assertEqual("giant:abc123:1", row["normalized_row_id"])
+        self.assertEqual("exact_upc", row["normalization_basis"])
+        self.assertEqual("5", row["normalized_quantity"])
+        self.assertEqual("lb", row["normalized_quantity_unit"])
+        self.assertEqual("true", row["is_item"])
 
         fee_row = enrich_giant.parse_item(
             order_id="abc123",
@@ -77,6 +82,7 @@ class EnrichGiantTests(unittest.TestCase):
 
         self.assertEqual("true", fee_row["is_fee"])
         self.assertEqual("GL BAG CHARGE", fee_row["item_name_norm"])
+        self.assertEqual("false", fee_row["is_item"])
 
     def test_parse_item_derives_packaged_weight_prices_from_size_tokens(self):
         row = enrich_giant.parse_item(
@@ -179,6 +185,8 @@ class EnrichGiantTests(unittest.TestCase):
             self.assertEqual("7.5", rows[0]["size_value"])
             self.assertEqual("10", rows[0]["retailer_item_id"])
             self.assertEqual("true", rows[1]["is_store_brand"])
+            self.assertTrue(rows[0]["normalized_item_id"])
+            self.assertEqual("exact_upc", rows[0]["normalization_basis"])
 
             with output_csv.open(newline="", encoding="utf-8") as handle:
                 written_rows = list(csv.DictReader(handle))

tests/test_pipeline_status.py

@@ -0,0 +1,80 @@
+import unittest
+
+import report_pipeline_status
+
+
+class PipelineStatusTests(unittest.TestCase):
+    def test_build_status_summary_reports_unresolved_and_reviewed_counts(self):
+        summary = report_pipeline_status.build_status_summary(
+            giant_orders=[{"order_id": "g1"}],
+            giant_items=[{"order_id": "g1", "line_no": "1"}],
+            giant_enriched=[
+                {
+                    "retailer": "giant",
+                    "order_id": "g1",
+                    "line_no": "1",
+                    "item_name_norm": "BANANA",
+                    "item_name": "FRESH BANANA",
+                    "retailer_item_id": "1",
+                    "upc": "4011",
+                    "brand_guess": "",
+                    "variant": "",
+                    "size_value": "",
+                    "size_unit": "",
+                    "pack_qty": "",
+                    "measure_type": "weight",
+                    "image_url": "",
+                    "is_store_brand": "false",
+                    "is_fee": "false",
+                    "is_discount_line": "false",
+                    "is_coupon_line": "false",
+                    "order_date": "2026-03-01",
+                    "line_total": "1.29",
+                }
+            ],
+            costco_orders=[],
+            costco_items=[],
+            costco_enriched=[],
+            purchases=[
+                {
+                    "observed_product_id": "gobs_banana",
+                    "canonical_product_id": "gcan_banana",
+                    "resolution_action": "",
+                    "is_fee": "false",
+                    "is_discount_line": "false",
+                    "is_coupon_line": "false",
+                    "retailer": "giant",
+                    "raw_item_name": "FRESH BANANA",
+                    "normalized_item_name": "BANANA",
+                    "upc": "4011",
+                    "line_total": "1.29",
+                },
+                {
+                    "observed_product_id": "gobs_lime",
+                    "canonical_product_id": "",
+                    "resolution_action": "",
+                    "is_fee": "false",
+                    "is_discount_line": "false",
+                    "is_coupon_line": "false",
+                    "retailer": "costco",
+                    "raw_item_name": "LIME 5LB",
+                    "normalized_item_name": "LIME",
+                    "upc": "",
+                    "line_total": "4.99",
+                },
+            ],
+            resolutions=[],
+        )
+
+        counts = {row["stage"]: row["count"] for row in summary}
+        self.assertEqual(1, counts["raw_orders"])
+        self.assertEqual(1, counts["raw_items"])
+        self.assertEqual(1, counts["enriched_items"])
+        self.assertEqual(1, counts["canonical_linked_purchase_rows"])
+        self.assertEqual(1, counts["unresolved_purchase_rows"])
+        self.assertEqual(1, counts["review_queue_observed_products"])
+        self.assertEqual(0, counts["unresolved_not_in_review_rows"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_scraper.py

@@ -58,14 +58,25 @@ class ScraperTests(unittest.TestCase):
             }
         ]
 
-        orders, items = scraper.flatten_orders(history, details)
+        orders, items = scraper.flatten_orders(
+            history,
+            details,
+            history_path=Path("data/giant-web/raw/history.json"),
+            raw_dir=Path("data/giant-web/raw"),
+        )
 
         self.assertEqual(1, len(orders))
         self.assertEqual("abc123", orders[0]["order_id"])
+        self.assertEqual("giant", orders[0]["retailer"])
         self.assertEqual("PICKUP", orders[0]["service_type"])
+        self.assertEqual("data/giant-web/raw/history.json", orders[0]["raw_history_path"])
+        self.assertEqual("data/giant-web/raw/abc123.json", orders[0]["raw_order_path"])
         self.assertEqual(1, len(items))
         self.assertEqual("1", items[0]["line_no"])
         self.assertEqual("Bananas", items[0]["item_name"])
+        self.assertEqual("giant", items[0]["retailer"])
+        self.assertEqual("data/giant-web/raw/abc123.json", items[0]["raw_order_path"])
+        self.assertEqual("false", items[0]["is_discount_line"])
 
     def test_append_dedup_replaces_duplicate_rows_and_preserves_new_values(self):
         with tempfile.TemporaryDirectory() as tmpdir: