"""Load erp_transactions from data/sales_detailed.csv. This is the largest migration in the project — ~250K rows from a ~65 MB CSV. Loaded via COPY ... FROM STDIN (text format) for speed; chunked at 10K rows so progress ticks frequently and per-chunk auto-discovery of missing SKUs / partners / stores keeps memory bounded. Auto-discovery: if a row references a sku / partner code / unit code that isn't in the corresponding dim_* table, INSERT a minimal row (sku+name+active=true, code+name+country, unit_code+name+country) and use the new id. Keeps the load self-healing when the source CSV drifts ahead of the dim tables (new SKUs appear in sales before anyone updates sku_category_map.csv). After loading, REFRESH MATERIALIZED VIEW v_sales_weekly. CSV → erp_transactions column mapping dokument → document date → transaction_date (DATE) partner → partner_id (FK lookup dim_partners.code) sku → product_id (FK lookup dim_products.sku) jedinica → store_id (FK lookup dim_stores.unit_code) tip_dok → channel_map_id (FK lookup lookup_channel_map.doc_type) komercijalist → sales_rep kolicina → quantity nabavna_vrijednost_eur → purchase_value ruc_eur → ruc_eur ruc_pct → ruc_pct porezna_osnovica_eur → tax_base pdv_eur → vat vrijednost_eur → total_value odobreni_rabat_eur → approved_discount CSV columns dropped: source_country (used only for store autodiscover country mapping), source_file, year, week, naziv_partnera (used only for partner autodiscover), naziv (product autodiscover), proizvodjac, mj_troska, naziv_mj_troska, naziv_jedinice (store autodiscover), kategorija_artikla, grupacija_artikla, naziv_grupacije, podkat, podkategorija, drzava (partner autodiscover). Idempotent: TRUNCATEs erp_transactions. dim_* tables are NOT truncated — autodiscovered rows accumulate across runs. Re-run after a fresh schema rebuild by running migrate_dimensions.py first, then this. Usage from project root: python -m db.migrate_sales """ from __future__ import annotations import io from pathlib import Path import pandas as pd from psycopg2.extras import execute_values from db.connection import get_connection PROJECT_ROOT = Path(__file__).resolve().parent.parent SOURCE_CSV = PROJECT_ROOT / "data" / "sales_detailed.csv" CHUNK_SIZE = 10_000 # Keep FK-lookup columns as strings (we use them as dict keys); let # numeric columns parse naturally so to_csv writes them as numbers. _DTYPE = { "source_country": "string", "dokument": "string", "partner": "string", "naziv_partnera": "string", "sku": "string", "naziv": "string", "jedinica": "string", "naziv_jedinice": "string", "tip_dok": "string", "komercijalist": "string", "drzava": "string", } # Column order written to erp_transactions — must match the COPY column list. TARGET_COLS = [ "document", "transaction_date", "partner_id", "product_id", "store_id", "channel_map_id", "sales_rep", "quantity", "purchase_value", "ruc_eur", "ruc_pct", "tax_base", "vat", "total_value", "approved_discount", "has_loyalty", ] _COUNTRY_MAP = {"cro": "HR", "slo": "SI", "at": "AT"} # Pseudo-SKU prefixes — non-product line items (coupons, postage, marketing # rebates, gift cards, packaging labels, marketing materials, services). # Dropped at chunk-load time so they never enter erp_transactions or # autodiscover into dim_products. See hard-delete cleanup history. import re _PSEUDO_SKU_RE = re.compile(r"^(OST|MKT|CARD|WOO|AMB|USL|MSM|WOLTD)", re.IGNORECASE) def _drop_pseudo_skus(chunk: pd.DataFrame) -> tuple[pd.DataFrame, int]: """Filter pseudo-SKU rows out of a chunk. Returns (filtered_chunk, n_dropped).""" skus = chunk["sku"].fillna("").str.strip() mask = skus.str.match(_PSEUDO_SKU_RE) n_dropped = int(mask.sum()) if n_dropped: chunk = chunk[~mask] return chunk, n_dropped # ---------- dimension lookup maps ------------------------------------ def _load_lookup_maps(conn) -> dict: out: dict[str, dict] = {} with conn.cursor() as cur: cur.execute("SELECT id, sku FROM dim_products") out["product"] = {r[1]: r[0] for r in cur.fetchall()} cur.execute("SELECT id, code FROM dim_partners") out["partner"] = {r[1]: r[0] for r in cur.fetchall()} cur.execute("SELECT id, unit_code FROM dim_stores") out["store"] = {r[1]: r[0] for r in cur.fetchall()} cur.execute("SELECT id, doc_type FROM lookup_channel_map") out["channel_map"] = {r[1]: r[0] for r in cur.fetchall()} return out # ---------- auto-discovery ------------------------------------------- def _autodiscover_products(conn, chunk: pd.DataFrame, maps: dict) -> int: skus = chunk["sku"].fillna("").str.strip() unknown = skus[(skus != "") & (~skus.isin(maps["product"]))].unique() if len(unknown) == 0: return 0 name_lookup = ( chunk.assign(_sku=skus, _name=chunk["naziv"].fillna("").str.strip()) .groupby("_sku")["_name"] .first() .to_dict() ) rows = [(sku, (name_lookup.get(sku) or None), True) for sku in unknown] with conn.cursor() as cur: result = execute_values( cur, "INSERT INTO dim_products (sku, name, active) VALUES %s " "RETURNING id, sku", rows, fetch=True, ) for new_id, sku in result: maps["product"][sku] = new_id return len(rows) def _autodiscover_partners(conn, chunk: pd.DataFrame, maps: dict) -> int: codes = chunk["partner"].fillna("").str.strip() unknown = codes[(codes != "") & (~codes.isin(maps["partner"]))].unique() if len(unknown) == 0: return 0 lookup = ( chunk.assign( _code=codes, _name=chunk["naziv_partnera"].fillna("").str.strip(), _country=chunk["drzava"].fillna("").str.strip(), ) .groupby("_code") .agg(_name=("_name", "first"), _country=("_country", "first")) .to_dict("index") ) rows = [] for code in unknown: info = lookup.get(code, {}) rows.append(( code, (info.get("_name") or None), None, # partner_type (info.get("_country") or None), )) with conn.cursor() as cur: result = execute_values( cur, "INSERT INTO dim_partners (code, name, partner_type, country) " "VALUES %s RETURNING id, code", rows, fetch=True, ) for new_id, code in result: maps["partner"][code] = new_id return len(rows) def _autodiscover_stores(conn, chunk: pd.DataFrame, maps: dict) -> int: units = chunk["jedinica"].fillna("").str.strip() unknown = units[(units != "") & (~units.isin(maps["store"]))].unique() if len(unknown) == 0: return 0 src = chunk["source_country"].fillna("").str.strip().str.lower() lookup = ( chunk.assign( _unit=units, _name=chunk["naziv_jedinice"].fillna("").str.strip(), _src=src, ) .groupby("_unit") .agg(_name=("_name", "first"), _src=("_src", "first")) .to_dict("index") ) rows = [] for unit in unknown: info = lookup.get(unit, {}) rows.append(( unit, (info.get("_name") or None), _COUNTRY_MAP.get((info.get("_src") or "")), unit == "01", # unit '01' is the warehouse )) with conn.cursor() as cur: result = execute_values( cur, "INSERT INTO dim_stores (unit_code, name, country, is_warehouse) " "VALUES %s RETURNING id, unit_code", rows, fetch=True, ) for new_id, unit in result: maps["store"][unit] = new_id return len(rows) # ---------- chunk prep + COPY ---------------------------------------- def _prepare_chunk(chunk: pd.DataFrame, maps: dict) -> pd.DataFrame: out = pd.DataFrame() doc = chunk["dokument"].fillna("").str.strip() out["document"] = doc.where(doc != "", None) # Datetime → date; NaT will be written as the NULL marker by to_csv. out["transaction_date"] = pd.to_datetime(chunk["date"], errors="coerce") skus = chunk["sku"].fillna("").str.strip() out["product_id"] = skus.map(maps["product"]).astype("Int64") codes = chunk["partner"].fillna("").str.strip() out["partner_id"] = codes.map(maps["partner"]).astype("Int64") units = chunk["jedinica"].fillna("").str.strip() out["store_id"] = units.map(maps["store"]).astype("Int64") dt = chunk["tip_dok"].fillna("").str.strip() out["channel_map_id"] = dt.map(maps["channel_map"]).astype("Int64") sr = chunk["komercijalist"].fillna("").astype(str).str.strip() out["sales_rep"] = sr.where(sr != "", None) out["quantity"] = pd.to_numeric(chunk["kolicina"], errors="coerce") out["purchase_value"] = pd.to_numeric(chunk["nabavna_vrijednost_eur"], errors="coerce") out["ruc_eur"] = pd.to_numeric(chunk["ruc_eur"], errors="coerce") out["ruc_pct"] = pd.to_numeric(chunk["ruc_pct"], errors="coerce") out["tax_base"] = pd.to_numeric(chunk["porezna_osnovica_eur"], errors="coerce") out["vat"] = pd.to_numeric(chunk["pdv_eur"], errors="coerce") out["total_value"] = pd.to_numeric(chunk["vrijednost_eur"], errors="coerce") out["approved_discount"] = pd.to_numeric(chunk["odobreni_rabat_eur"], errors="coerce") # Loyalty flag — "Loyalty kartica" is a long numeric card-ID string when # a card was used, "nonLoyalty" or empty otherwise. We store just the # boolean since the bridge only needs to know "was loyalty used", not # which card. The discount amount is inferred from list vs realized # price at compute time (Polleo bridge spec, May 2026). if "loyalty_kartica" in chunk.columns: lk = chunk["loyalty_kartica"].fillna("").astype(str).str.strip() out["has_loyalty"] = lk.str.match(r"^\d{6,}$") else: out["has_loyalty"] = False return out[TARGET_COLS] def _copy_chunk(conn, df: pd.DataFrame) -> None: """Write the chunk to a tab-separated buffer and stream it via COPY. Text format defaults: DELIMITER tab, NULL '\\N' — match pandas to_csv(sep='\\t', na_rep='\\\\N', date_format='%Y-%m-%d').""" buf = io.StringIO() df.to_csv( buf, index=False, header=False, sep="\t", na_rep=r"\N", date_format="%Y-%m-%d", ) buf.seek(0) cols = ", ".join(TARGET_COLS) with conn.cursor() as cur: cur.copy_expert( f"COPY erp_transactions ({cols}) FROM STDIN WITH (FORMAT text)", buf, ) # ---------- verification --------------------------------------------- def _verify(conn) -> None: with conn.cursor() as cur: cur.execute("SELECT COUNT(*) FROM erp_transactions") n_tx = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM v_sales_weekly") n_view = cur.fetchone()[0] print() print(f" erp_transactions rows: {n_tx:>10,}") print(f" v_sales_weekly rows: {n_view:>10,}") # Cross-check qty totals for the 3 SKUs with the most transactions. with conn.cursor() as cur: cur.execute(""" SELECT product_id, COUNT(*) AS n FROM erp_transactions WHERE product_id IS NOT NULL GROUP BY product_id ORDER BY n DESC LIMIT 3 """) top = [r[0] for r in cur.fetchall()] if top: print() print(" Cross-check (top 3 SKUs): SUM(erp_transactions.quantity)" " vs SUM(v_sales_weekly.qty_total)") for pid in top: with conn.cursor() as cur: cur.execute(""" SELECT p.sku, (SELECT SUM(quantity) FROM erp_transactions WHERE product_id = %s) AS tx_sum, (SELECT SUM(qty_total) FROM v_sales_weekly WHERE product_id = %s) AS view_sum FROM dim_products p WHERE p.id = %s """, (pid, pid, pid)) sku, tx_sum, view_sum = cur.fetchone() ok = "OK" if (tx_sum or 0) == (view_sum or 0) else "MISMATCH" print(f" {sku:12s} tx={float(tx_sum or 0):>12,.2f} " f"view={float(view_sum or 0):>12,.2f} {ok}") with conn.cursor() as cur: cur.execute(""" SELECT p.sku, v.year, v.week, v.qty_retail, v.qty_webshop, v.qty_wholesale, v.qty_total FROM v_sales_weekly v JOIN dim_products p ON v.product_id = p.id ORDER BY v.qty_total DESC NULLS LAST LIMIT 5 """) rows = cur.fetchall() print() print(" Sample v_sales_weekly (top 5 by qty_total):") print(f" {'SKU':12s} {'YR':>4s} {'WK':>3s} " f"{'retail':>10s} {'webshop':>10s} {'wholesale':>10s} {'total':>10s}") for r in rows: sku, yr, wk, qr, qw, qws, qt = r print(f" {sku:12s} {yr:>4d} {wk:>3d} " f"{float(qr or 0):>10,.0f} {float(qw or 0):>10,.0f} " f"{float(qws or 0):>10,.0f} {float(qt or 0):>10,.0f}") # ---------- entry point ---------------------------------------------- def migrate_sales() -> None: print("Connecting to Postgres...") conn = get_connection() conn.autocommit = False try: print("Truncating erp_transactions...") with conn.cursor() as cur: cur.execute("TRUNCATE TABLE erp_transactions RESTART IDENTITY") conn.commit() print("Loading dimension lookup maps...") maps = _load_lookup_maps(conn) print(f" products={len(maps['product']):>5,} " f"partners={len(maps['partner']):>4,} " f"stores={len(maps['store']):>3,} " f"doctypes={len(maps['channel_map']):>2,}") print(f"Streaming {SOURCE_CSV.name} in {CHUNK_SIZE:,}-row chunks...") total = 0 added_prod = added_part = added_store = 0 n_pseudo_dropped = 0 chunk_n = 0 for chunk in pd.read_csv( SOURCE_CSV, chunksize=CHUNK_SIZE, dtype=_DTYPE, encoding="utf-8", low_memory=False, ): chunk_n += 1 # Drop pseudo-SKU rows before any autodiscover or load — keeps # coupons, postage, marketing rebates, gift cards, packaging # labels etc. out of dim_products and erp_transactions. chunk, dropped = _drop_pseudo_skus(chunk) n_pseudo_dropped += dropped if len(chunk) == 0: continue added_prod += _autodiscover_products(conn, chunk, maps) added_part += _autodiscover_partners(conn, chunk, maps) added_store += _autodiscover_stores(conn, chunk, maps) prepared = _prepare_chunk(chunk, maps) _copy_chunk(conn, prepared) conn.commit() total += len(prepared) print(f" chunk {chunk_n:>3d}: +{len(prepared):>6,} rows " f"(total: {total:>7,})") if n_pseudo_dropped: print(f"\n Pseudo-SKU rows dropped: {n_pseudo_dropped:,}") print() print(f"Auto-added: products={added_prod}, " f"partners={added_part}, stores={added_store}") print() print("Refreshing materialized view v_sales_weekly...") with conn.cursor() as cur: cur.execute("REFRESH MATERIALIZED VIEW v_sales_weekly") conn.commit() _verify(conn) finally: conn.close() print() print("Sales loaded.") if __name__ == "__main__": migrate_sales()