""" Unified catalogue + cross-retailer price matcher. Pulls the flat product tables from each retailer's SQLite DB into one baskit.db, normalises barcodes, and joins on barcode to produce a cross-retailer price comparison. Sources (each optional — skipped if the DB is missing): pnp.db products(code, name, brand, barcode, price_value, ...) checkers.db products(id, name, barcode, price_value, ...) woolworths.db products(id, name, brand, barcode, price_value, ...) Join key: normalised barcode (digits only, leading zeros stripped, length >= 8). Variable-weight / in-store barcodes (EAN starting with '2') are flagged and excluded from cross-retailer matching because they are store-specific. Commands: build : (re)build baskit.db catalogue from all retailer DBs compare : write a cross-retailer price-comparison CSV status : coverage + match stats """ from __future__ import annotations import argparse import re import sqlite3 from datetime import datetime from pathlib import Path import config ROOT = config.ROOT BASKIT_DB = config.BASKIT_DB # retailer -> (db path, select sql producing the unified columns) SOURCES = { "pnp": ( config.PNP_DB, """SELECT code AS product_key, barcode, name, brand, price_value, NULL AS was_price, primary_image_url AS image_url, url, category_path AS bucket, scraped_at FROM products""", ), "checkers": ( config.CHECKERS_DB, """SELECT id AS product_key, barcode, name, brand, price_value, was_price, primary_image_url AS image_url, url, buckets AS bucket, scraped_at FROM products""", ), "woolworths": ( config.WOOLWORTHS_DB, """SELECT id AS product_key, barcode, name, brand, price_value, was_price, primary_image_url AS image_url, url, buckets AS bucket, scraped_at FROM products""", ), } # MVP catalogue caps per the SKU-scope brief (~825 SKUs total, within 600-1,200). # Order matters: a product is assigned to the first bucket it qualifies for. BUCKET_CAPS = [ ("staples", 250), ("fresh_produce", 150), ("dairy_bakery_chilled", 100), ("household_cleaning", 100), ("pantry", 100), ("premium", 75), ("basket_builders", 50), ] MVP_BUCKETS = {b for b, _ in BUCKET_CAPS} SCHEMA = """ CREATE TABLE IF NOT EXISTS catalogue ( retailer TEXT NOT NULL, product_key TEXT NOT NULL, barcode_raw TEXT, barcode_norm TEXT, is_instore_bc INTEGER DEFAULT 0, -- 1 = variable-weight/store-specific barcode name TEXT, brand TEXT, price REAL, was_price REAL, image_url TEXT, url TEXT, bucket TEXT, scraped_at TEXT, PRIMARY KEY (retailer, product_key) ); CREATE INDEX IF NOT EXISTS idx_cat_bcnorm ON catalogue(barcode_norm); CREATE INDEX IF NOT EXISTS idx_cat_retailer ON catalogue(retailer); """ def normalise_barcode(raw: str | None) -> tuple[str | None, int]: """Return (normalised_barcode, is_instore_flag). None if unusable.""" if not raw: return None, 0 digits = re.sub(r"\D", "", str(raw)) instore = 1 if digits[:1] == "2" else 0 # GS1 prefix 2 = in-store/variable weight stripped = digits.lstrip("0") if len(stripped) < 8: # too short to be a real GTIN/EAN/UPC return None, instore return stripped, instore # ------------------------------------------------------------------ build def cmd_build(_args: argparse.Namespace) -> None: conn = sqlite3.connect(BASKIT_DB) conn.executescript(SCHEMA) conn.execute("DELETE FROM catalogue") total = 0 for retailer, (db_path, sql) in SOURCES.items(): path = Path(db_path) if not path.exists(): if retailer == "pnp": from db_init import init_pnp_db init_pnp_db() print(f"[build] {retailer}: created {path}") else: print(f"[build] {retailer}: {path} not found — skipped") continue src = sqlite3.connect(f"file:{path}?mode=ro", uri=True) rows = src.execute(sql).fetchall() src.close() batch = [] for (pkey, barcode, name, brand, price, was, image, url, bucket, scraped_at) in rows: norm, instore = normalise_barcode(barcode) batch.append((retailer, str(pkey), barcode, norm, instore, name, brand, price, was, image, url, bucket, scraped_at)) with conn: conn.executemany( "INSERT OR REPLACE INTO catalogue(retailer,product_key,barcode_raw," "barcode_norm,is_instore_bc,name,brand,price,was_price,image_url,url,bucket,scraped_at)" " VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", batch) total += len(batch) print(f"[build] {retailer}: {len(batch)} products loaded") print(f"[build] catalogue total: {total}") _print_match_summary(conn) def _print_match_summary(conn: sqlite3.Connection) -> None: retailers = [r[0] for r in conn.execute( "SELECT DISTINCT retailer FROM catalogue ORDER BY retailer")] print("\n--- barcode matchable products (branded, non-instore, valid barcode) ---") for r in retailers: n = conn.execute( "SELECT COUNT(*) FROM catalogue WHERE retailer=? AND barcode_norm IS NOT NULL " "AND is_instore_bc=0", (r,)).fetchone()[0] print(f" {r:<12} {n}") # how many barcodes appear in >=2 retailers rows = conn.execute(""" SELECT n_retailers, COUNT(*) FROM ( SELECT barcode_norm, COUNT(DISTINCT retailer) AS n_retailers FROM catalogue WHERE barcode_norm IS NOT NULL AND is_instore_bc=0 GROUP BY barcode_norm ) GROUP BY n_retailers ORDER BY n_retailers""").fetchall() print("--- products by # of retailers sharing the barcode ---") for n_ret, cnt in rows: print(f" in {n_ret} retailer(s): {cnt}") # ------------------------------------------------------------------ compare def cmd_compare(args: argparse.Namespace) -> None: conn = sqlite3.connect(BASKIT_DB) retailers = [r[0] for r in conn.execute( "SELECT DISTINCT retailer FROM catalogue ORDER BY retailer")] if not retailers: print("[compare] empty catalogue — run build first.") return # one row per barcode, with each retailer's price + a representative name price_cols = ",\n".join( f"MAX(CASE WHEN retailer='{r}' THEN price END) AS {r}_price" for r in retailers) sql = f""" SELECT barcode_norm, MAX(name) AS name, MAX(brand) AS brand, COUNT(DISTINCT retailer) AS n_retailers, {price_cols}, MIN(price) AS min_price, MAX(price) AS max_price FROM catalogue WHERE barcode_norm IS NOT NULL AND is_instore_bc=0 AND price IS NOT NULL GROUP BY barcode_norm HAVING n_retailers >= ? ORDER BY (MAX(price)-MIN(price)) DESC """ rows = conn.execute(sql, (args.min_retailers,)).fetchall() headers = (["barcode", "name", "brand", "n_retailers"] + [f"{r}_price" for r in retailers] + ["min_price", "max_price", "spread", "spread_pct"]) out_dir = config.EXPORTS_DIR out_dir.mkdir(exist_ok=True) stamp = datetime.now().strftime("%Y%m%d_%H%M%S") import csv csv_path = out_dir / f"baskit_price_comparison_{stamp}.csv" n = 0 with open(csv_path, "w", encoding="utf-8", newline="") as f: w = csv.writer(f) w.writerow(headers) for row in rows: mn, mx = row[-2], row[-1] spread = round(mx - mn, 2) if (mn is not None and mx is not None) else None spread_pct = round((spread / mn) * 100, 1) if (spread and mn) else None w.writerow(list(row) + [spread, spread_pct]) n += 1 print(f"[compare] {n} matched products (in >= {args.min_retailers} retailers) -> {csv_path}") # show the biggest price gaps print(f"\n--- top price gaps (matched across {args.min_retailers}+ retailers) ---") for row in rows[:12]: d = dict(zip(["barcode", "name", "brand", "n"] + [f"{r}" for r in retailers] + ["mn", "mx"], row)) prices = " ".join(f"{r}=R{d[r]:.2f}" for r in retailers if d.get(r) is not None) print(f" R{d['mx']-d['mn']:>6.2f} {(d['name'] or '')[:42]:<42} {prices}") # ------------------------------------------------------------------ mvp subset MVP_SCHEMA = """ CREATE TABLE IF NOT EXISTS mvp_catalogue ( barcode TEXT PRIMARY KEY, bucket TEXT, name TEXT, brand TEXT, n_retailers INTEGER, pnp_price REAL, checkers_price REAL, woolworths_price REAL, min_price REAL, max_price REAL ); """ def _pick_bucket(bucket_blob: str | None) -> str | None: """Assign one MVP bucket from the (comma-joined, multi-retailer) bucket text.""" if not bucket_blob: return None tokens = {t.strip() for t in bucket_blob.split(",")} & MVP_BUCKETS for b, _ in BUCKET_CAPS: # priority order if b in tokens: return b return None def cmd_mvp(args: argparse.Namespace) -> None: conn = sqlite3.connect(BASKIT_DB) conn.executescript(MVP_SCHEMA) conn.execute("DELETE FROM mvp_catalogue") # one row per barcode, with prices + concatenated bucket text (from checkers/woolies) rows = conn.execute(""" SELECT barcode_norm, group_concat(bucket, ',') AS buckets, MAX(name) AS name, MAX(brand) AS brand, COUNT(DISTINCT retailer) AS n_retailers, MAX(CASE WHEN retailer='pnp' THEN price END) AS pnp, MAX(CASE WHEN retailer='checkers' THEN price END) AS ck, MAX(CASE WHEN retailer='woolworths' THEN price END) AS ww, MIN(price) AS mn, MAX(price) AS mx FROM catalogue WHERE barcode_norm IS NOT NULL AND is_instore_bc=0 AND price IS NOT NULL GROUP BY barcode_norm """).fetchall() # bucket -> candidate rows, ranked: more retailers first, then cheaper (proxy for staple) buckets: dict[str, list] = {b: [] for b, _ in BUCKET_CAPS} for r in rows: b = _pick_bucket(r[1]) if b: buckets[b].append(r) selected = [] for bucket, cap in BUCKET_CAPS: cands = sorted(buckets[bucket], key=lambda r: (-r[4], r[8] or 1e9)) for r in cands[:cap]: selected.append((r[0], bucket, r[2], r[3], r[4], r[5], r[6], r[7], r[8], r[9])) with conn: conn.executemany( "INSERT OR REPLACE INTO mvp_catalogue(barcode,bucket,name,brand,n_retailers," "pnp_price,checkers_price,woolworths_price,min_price,max_price) " "VALUES (?,?,?,?,?,?,?,?,?,?)", selected) # report print("[mvp] bucket selected / cap (avail)") for bucket, cap in BUCKET_CAPS: avail = len(buckets[bucket]) got = conn.execute("SELECT COUNT(*) FROM mvp_catalogue WHERE bucket=?", (bucket,)).fetchone()[0] flag = "" if avail >= cap else " <- short" print(f" {bucket:<22} {got:>4} / {cap:<4} ({avail}){flag}") total = conn.execute("SELECT COUNT(*) FROM mvp_catalogue").fetchone()[0] multi = conn.execute("SELECT COUNT(*) FROM mvp_catalogue WHERE n_retailers>=2").fetchone()[0] print(f"[mvp] total {total} SKUs, {multi} price-comparable across 2+ retailers") out_dir = config.EXPORTS_DIR; out_dir.mkdir(exist_ok=True) stamp = datetime.now().strftime("%Y%m%d_%H%M%S") import csv path = out_dir / f"baskit_mvp_catalogue_{stamp}.csv" cols = ["barcode", "bucket", "name", "brand", "n_retailers", "pnp_price", "checkers_price", "woolworths_price", "min_price", "max_price"] with open(path, "w", encoding="utf-8", newline="") as f: w = csv.writer(f); w.writerow(cols) w.writerows(conn.execute(f"SELECT {','.join(cols)} FROM mvp_catalogue " "ORDER BY bucket, n_retailers DESC, name")) print(f"[mvp] wrote {path}") # ------------------------------------------------------------------ status def cmd_status(_args: argparse.Namespace) -> None: if not BASKIT_DB.exists(): print("[status] no baskit.db — run build first.") return conn = sqlite3.connect(BASKIT_DB) total = conn.execute("SELECT COUNT(*) FROM catalogue").fetchone()[0] print(f"DB: {BASKIT_DB} ({BASKIT_DB.stat().st_size/1024/1024:.1f} MB) rows={total}") _print_match_summary(conn) def main() -> None: p = argparse.ArgumentParser(description="Baskit unified catalogue + price matcher") sub = p.add_subparsers(dest="cmd", required=True) sub.add_parser("build", help="rebuild baskit.db from retailer DBs").set_defaults(func=cmd_build) sp = sub.add_parser("compare", help="export cross-retailer price comparison") sp.add_argument("--min-retailers", type=int, default=2) sp.set_defaults(func=cmd_compare) sub.add_parser("mvp", help="select the ~600-1,200 SKU MVP catalogue").set_defaults(func=cmd_mvp) sub.add_parser("status", help="coverage + match stats").set_defaults(func=cmd_status) args = p.parse_args() args.func(args) if __name__ == "__main__": main()