""" Woolworths (woolworths.co.za) product scraper. Woolworths' site search is powered by Constructor.io, hosted on cnstrc.com — *outside* the woolworths.co.za Cloudflare perimeter. So unlike Checkers, no browser is needed: plain HTTP works. GET https://wpkmgeuco-zone.cnstrc.com/v1/search/{term} ?key=key_tw9hKe0fkfgEf36D&i={clientId}&s=1&page=N&num_results_per_page=24... Each result's `data.id` is the product barcode/EAN (e.g. 6009171091446) — our cross-retailer join key. `p10`/`p30`/`p60` are price tiers (standard / variant fulfilment); `*_wp` are was-prices for promos. Seeding: shared MVP staples list (mvp_terms.py). Resumable via woolworths.db. Commands: scrape | export | status """ from __future__ import annotations import argparse import json import random import sqlite3 import time import uuid from datetime import datetime, timezone from pathlib import Path import requests import config import mvp_terms # ------------------------------------------------------------------ config ROOT = config.ROOT DB_PATH = config.WOOLWORTHS_DB CNSTRC_HOST = "https://wpkmgeuco-zone.cnstrc.com" CNSTRC_KEY = "key_tw9hKe0fkfgEf36D" UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36") PAGE_SIZE = 50 PER_TERM_CAP = config.PER_TERM_CAP DEFAULT_DELAY = config.WOOLWORTHS_DELAY RETAILER = "woolworths" # ------------------------------------------------------------------ db SCHEMA = """ CREATE TABLE IF NOT EXISTS terms ( bucket TEXT NOT NULL, term TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'pending', total_count INTEGER, n_collected INTEGER DEFAULT 0, attempts INTEGER NOT NULL DEFAULT 0, last_error TEXT, fetched_at TEXT, PRIMARY KEY (bucket, term) ); CREATE TABLE IF NOT EXISTS products ( id TEXT PRIMARY KEY, -- constructor item id (== barcode) barcode TEXT, name TEXT, brand TEXT, price_value REAL, was_price REAL, on_promotion INTEGER, price_p10 REAL, price_p30 REAL, price_p60 REAL, department TEXT, product_type TEXT, sku_count INTEGER, average_rating REAL, number_of_reviews INTEGER, primary_image_url TEXT, buckets TEXT, matched_terms TEXT, url TEXT, raw_json TEXT NOT NULL, scraped_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_ww_barcode ON products(barcode); CREATE INDEX IF NOT EXISTS idx_ww_name ON products(name); """ PRODUCT_COLS = ( "id,barcode,name,brand,price_value,was_price,on_promotion,price_p10,price_p30," "price_p60,department,product_type,sku_count,average_rating,number_of_reviews," "primary_image_url,buckets,matched_terms,url,raw_json,scraped_at" ) PRODUCT_QMARKS = ",".join(["?"] * len(PRODUCT_COLS.split(","))) def db_connect() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA synchronous=NORMAL") conn.executescript(SCHEMA) return conn def now_iso() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds") def seed_terms(conn: sqlite3.Connection) -> None: with conn: conn.executemany("INSERT OR IGNORE INTO terms(bucket, term) VALUES (?, ?)", mvp_terms.all_terms()) # ------------------------------------------------------------------ http def make_session() -> requests.Session: s = requests.Session() s.headers.update({ "User-Agent": UA, "Accept": "application/json", "Referer": "https://www.woolworths.co.za/", "Origin": "https://www.woolworths.co.za", }) s.params = {"key": CNSTRC_KEY, "i": str(uuid.uuid4()), "s": "1", "c": "ciojs"} return s def search(session: requests.Session, term: str, page: int) -> tuple[list[dict], int]: url = f"{CNSTRC_HOST}/v1/search/{requests.utils.quote(term)}" params = { "num_results_per_page": str(PAGE_SIZE), "page": str(page), "sort_by": "relevance", "sort_order": "descending", "filters[visibility]": "all", } r = session.get(url, params=params, timeout=30) r.raise_for_status() resp = r.json().get("response", {}) return resp.get("results") or [], int(resp.get("total_num_results") or 0) # ------------------------------------------------------------------ extraction def _num(v): return v if isinstance(v, (int, float)) and v else None def extract_row(res: dict, bucket: str, term: str) -> tuple: d = res.get("data") or {} bid = str(d.get("id")) if d.get("id") is not None else res.get("value") p10, p30, p60 = _num(d.get("p10")), _num(d.get("p30")), _num(d.get("p60")) price = p10 or p30 or p60 wp = _num(d.get("p10_wp")) or _num(d.get("p30_wp")) or _num(d.get("p60_wp")) was = wp if (wp and price and wp > price) else None url = d.get("url") full_url = f"https://www.woolworths.co.za/{url.lstrip('/')}" if url else None return ( bid, bid if (bid and bid.isdigit()) else None, # numeric id == barcode res.get("value"), d.get("brand"), price, was, 1 if was else 0, p10, p30, p60, str(d.get("dept")) if d.get("dept") is not None else None, d.get("prodtype"), d.get("skucnt"), _num(d.get("ratings")), d.get("reviews"), d.get("image_url"), bucket, term, full_url, json.dumps(res, separators=(",", ":")), now_iso(), ) def upsert_products(conn: sqlite3.Connection, results: list[dict], bucket: str, term: str) -> int: new = 0 with conn: for res in results: d = res.get("data") or {} pid = str(d.get("id")) if d.get("id") is not None else res.get("value") if not pid: continue existing = conn.execute( "SELECT buckets, matched_terms FROM products WHERE id=?", (pid,) ).fetchone() if existing: buckets = set(filter(None, (existing[0] or "").split(","))) terms = set(filter(None, (existing[1] or "").split("|"))) buckets.add(bucket); terms.add(term) conn.execute("UPDATE products SET buckets=?, matched_terms=? WHERE id=?", (",".join(sorted(buckets)), "|".join(sorted(terms)), pid)) else: conn.execute( f"INSERT INTO products({PRODUCT_COLS}) VALUES ({PRODUCT_QMARKS})", extract_row(res, bucket, term)) new += 1 return new # ------------------------------------------------------------------ scrape def cmd_scrape(args: argparse.Namespace) -> None: conn = db_connect() seed_terms(conn) if args.refresh: with conn: n = conn.execute( "UPDATE terms SET status='pending' WHERE status='done'" ).rowcount print(f"[scrape] refresh: re-queued {n} done terms") statuses = "('pending','error')" if args.retry_errors else "('pending')" pending = conn.execute( f"SELECT bucket, term FROM terms WHERE status IN {statuses} " f"ORDER BY status DESC, bucket, term").fetchall() if not pending: print("[scrape] nothing pending. (use --retry-errors to retry failures)") return if args.limit: pending = pending[: args.limit] print(f"[scrape] {len(pending)} terms pending delay={args.delay}s " f"page_size={PAGE_SIZE} cap={PER_TERM_CAP}") session = make_session() total_new = 0 try: for i, (bucket, term) in enumerate(pending, 1): try: collected = new_here = page = 0 total = None while collected < PER_TERM_CAP: results, total = search(session, term, page + 1) # constructor pages are 1-based if not results: break new_here += upsert_products(conn, results, bucket, term) collected += len(results) page += 1 if collected >= total or len(results) < PAGE_SIZE: break time.sleep(args.delay * (1 + random.uniform(-0.3, 0.3))) with conn: conn.execute( "UPDATE terms SET status='done', total_count=?, n_collected=?, " "attempts=attempts+1, fetched_at=?, last_error=NULL " "WHERE bucket=? AND term=?", (total, collected, now_iso(), bucket, term)) total_new += new_here print(f"[scrape] {i}/{len(pending)} {bucket}/{term!r}: +{new_here} new " f"(saw {collected}/{total}), total_new={total_new}") except Exception as e: with conn: conn.execute( "UPDATE terms SET status='error', attempts=attempts+1, " "fetched_at=?, last_error=? WHERE bucket=? AND term=?", (now_iso(), repr(e)[:300], bucket, term)) print(f"[scrape] {i}/{len(pending)} {bucket}/{term!r}: ERROR {e}") time.sleep(args.delay * (1 + random.uniform(-0.3, 0.3))) except KeyboardInterrupt: print("\n[scrape] interrupted — progress saved.") n = conn.execute("SELECT COUNT(*) FROM products").fetchone()[0] bc = conn.execute("SELECT COUNT(*) FROM products WHERE barcode IS NOT NULL").fetchone()[0] print(f"[scrape] done. products={n} with_barcode={bc} new_this_run={total_new}") # ------------------------------------------------------------------ export / status def cmd_export(args: argparse.Namespace) -> None: conn = db_connect() out_dir = config.EXPORTS_DIR out_dir.mkdir(exist_ok=True) stamp = datetime.now().strftime("%Y%m%d_%H%M%S") cols = [c for c in PRODUCT_COLS.split(",") if c != "raw_json"] sql = f"SELECT {','.join(cols)} FROM products" jsonl_path = out_dir / f"woolworths_flat_{stamp}.jsonl" n = 0 with open(jsonl_path, "w", encoding="utf-8") as f: for row in conn.execute(sql): f.write(json.dumps(dict(zip(cols, row)), ensure_ascii=False) + "\n"); n += 1 print(f"[export] wrote {n} rows to {jsonl_path}") if args.csv: import csv csv_path = out_dir / f"woolworths_flat_{stamp}.csv" with open(csv_path, "w", encoding="utf-8", newline="") as f: w = csv.writer(f); w.writerow(cols); w.writerows(conn.execute(sql)) print(f"[export] wrote CSV to {csv_path}") def cmd_status(_args: argparse.Namespace) -> None: if not DB_PATH.exists(): print(f"[status] no database at {DB_PATH}. Run scrape first.") return conn = db_connect() seed_terms(conn) print(f"DB: {DB_PATH} ({DB_PATH.stat().st_size/1024/1024:.1f} MB)") for status, count in conn.execute( "SELECT status, COUNT(*) FROM terms GROUP BY status ORDER BY 2 DESC"): print(f" {status:<10} {count:>6}") n = conn.execute("SELECT COUNT(*) FROM products").fetchone()[0] bc = conn.execute("SELECT COUNT(*) FROM products WHERE barcode IS NOT NULL").fetchone()[0] print(f"--- products: {n} with_barcode: {bc} ---") def main() -> None: p = argparse.ArgumentParser(description="Woolworths product scraper") sub = p.add_subparsers(dest="cmd", required=True) sp = sub.add_parser("scrape"); sp.add_argument("--delay", type=float, default=DEFAULT_DELAY) sp.add_argument("--limit", type=int, default=0) sp.add_argument("--retry-errors", action="store_true") sp.add_argument("--refresh", action="store_true", help="re-queue done terms so prices are re-fetched") sp.set_defaults(func=cmd_scrape) sp = sub.add_parser("export"); sp.add_argument("--csv", action="store_true"); sp.set_defaults(func=cmd_export) sp = sub.add_parser("status"); sp.set_defaults(func=cmd_status) args = p.parse_args() args.func(args) if __name__ == "__main__": main()