"""Migrate data gaps identified in the April 2026 audit into PostgreSQL. Sections (run in order — FK dependencies): 1. migrate_on_top_inputs ← vp_input_detail.csv + mp_input_detail.csv 2. migrate_consensus_snapshots← data/consensus/snapshot_*.json 3. migrate_webshop_orders ← webshop_coupon_orders.csv 4. migrate_sales_history ← sales_clean.csv → sales_clean_import table + v_sales_weekly_full UNION matview 5. migrate_promo_calendar ← erp_promo_weeks (gaps-and-islands) 6. migrate_erp_promo_campaigns← data/rabatne.xlsx 7. migrate_stock_history ← stock_stores_at.csv + stock_stores_slo.csv Idempotency: - on_top_inputs, consensus_snapshots: UPSERT (app may have written rows) - all others: TRUNCATE then INSERT Usage from project root: py -3.12 -m db.migrate_remaining_gaps """ from __future__ import annotations import io import json import re from datetime import date, datetime 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 DATA_DIR = PROJECT_ROOT / "data" # ------------------------------------------------------------------- # KAM → user_id mapping (confirmed from users table) # VP file: 'Patrik' → id=2 (Patrik_VP), 'Selma' → id=1 # MP file: 'Patrik' → id=4 (Patrik_MP), 'Ivan' → id=3 # ------------------------------------------------------------------- KAM_VP_MAP: dict[str, int] = {"Patrik": 2, "Selma": 1} KAM_MP_MAP: dict[str, int] = {"Patrik": 4, "Ivan": 3} # All CW columns in these files cover calendar year 2026 CW_YEAR = 2026 # Only one active S&OP cycle in this DB CYCLE_ID = 1 # ------------------------------------------------------------------- # Helpers # ------------------------------------------------------------------- def _load_product_map(conn) -> dict[str, int]: with conn.cursor() as cur: cur.execute("SELECT sku, id FROM dim_products WHERE sku IS NOT NULL") return {r[0]: r[1] for r in cur.fetchall()} def _truncate(conn, table: str) -> None: with conn.cursor() as cur: cur.execute(f"TRUNCATE TABLE {table} RESTART IDENTITY CASCADE") def _cw_to_year_week(cw_label: str, year: int) -> int | None: """'CW19' → 202619, 'CW08' → 202608.""" m = re.fullmatch(r"CW(\d{1,2})", cw_label.strip(), re.IGNORECASE) if not m: return None return year * 100 + int(m.group(1)) # ------------------------------------------------------------------- # Section 1 — on_top_inputs # ------------------------------------------------------------------- def migrate_on_top_inputs(conn) -> int: """Melt VP and MP input detail CSVs into on_top_inputs. VP: channel='wholesale', KAM map KAM_VP_MAP, has 'buyer' column. MP: channel='retail', KAM map KAM_MP_MAP, no buyer column. UPSERT on (cycle_id, submitted_by_id, channel, LOWER(buyer), product_id, year_week) — avoids stomping rows the React app may have written. """ product_map = _load_product_map(conn) rows: list[tuple] = [] # --- VP --- vp_fp = DATA_DIR / "vp_input_detail.csv" if not vp_fp.exists(): print(" [warn] vp_input_detail.csv missing — skipping VP section") else: vp = pd.read_csv(vp_fp, dtype=str, encoding="utf-8") cw_cols_vp = [c for c in vp.columns if re.fullmatch(r"CW\d{1,2}", c, re.IGNORECASE)] for _, row in vp.iterrows(): sku = str(row["sku"]).strip() if pd.notna(row["sku"]) else "" if not sku: continue pid = product_map.get(sku) if pid is None: continue kam = str(row["kam"]).strip() if pd.notna(row["kam"]) else "" uid = KAM_VP_MAP.get(kam) if uid is None: print(f" [warn] VP KAM '{kam}' not in KAM_VP_MAP — skipping row") continue buyer = str(row.get("buyer", "") or "").strip() or None for cw in cw_cols_vp: raw = row[cw] if pd.isna(raw): continue try: qty = float(raw) except (ValueError, TypeError): continue if qty == 0: continue yw = _cw_to_year_week(cw, CW_YEAR) if yw is None: continue rows.append((CYCLE_ID, pid, yw, qty, "wholesale", buyer, uid)) # --- MP --- mp_fp = DATA_DIR / "mp_input_detail.csv" if not mp_fp.exists(): print(" [warn] mp_input_detail.csv missing — skipping MP section") else: mp = pd.read_csv(mp_fp, dtype=str, encoding="utf-8") # Skip metadata cols: name, cat, oznaka (not CW patterns) cw_cols_mp = [c for c in mp.columns if re.fullmatch(r"CW\d{1,2}", c, re.IGNORECASE)] for _, row in mp.iterrows(): sku = str(row["sku"]).strip() if pd.notna(row["sku"]) else "" if not sku: continue pid = product_map.get(sku) if pid is None: continue kam = str(row["kam"]).strip() if pd.notna(row["kam"]) else "" uid = KAM_MP_MAP.get(kam) if uid is None: print(f" [warn] MP KAM '{kam}' not in KAM_MP_MAP — skipping row") continue for cw in cw_cols_mp: raw = row[cw] if pd.isna(raw): continue try: qty = float(raw) except (ValueError, TypeError): continue if qty == 0: continue yw = _cw_to_year_week(cw, CW_YEAR) if yw is None: continue rows.append((CYCLE_ID, pid, yw, qty, "retail", None, uid)) if not rows: print(" [info] No non-zero on-top rows found in either file") return 0 # Aggregate duplicate keys from the CSV before inserting # (mp_input_detail can have the same SKU×KAM×CW in multiple rows) agg: dict[tuple, float] = {} for cycle_id, pid, yw, qty, channel, buyer, uid in rows: key = (cycle_id, pid, yw, channel, (buyer or "").lower(), uid) agg[key] = agg.get(key, 0.0) + (qty or 0.0) rows = [ (cycle_id, pid, yw, qty, channel, buyer if buyer else None, uid) for (cycle_id, pid, yw, channel, buyer, uid), qty in agg.items() if qty != 0 ] # schema.sql has no unique constraint on on_top_inputs. # Deduplicate any existing rows (e.g. from a prior run of this script), # then create a functional unique index for future idempotency. with conn.cursor() as cur: cur.execute(""" DELETE FROM on_top_inputs a USING on_top_inputs b WHERE a.id < b.id AND a.cycle_id = b.cycle_id AND a.submitted_by_id = b.submitted_by_id AND a.channel = b.channel AND COALESCE(LOWER(a.buyer), '') = COALESCE(LOWER(b.buyer), '') AND a.product_id = b.product_id AND a.year_week = b.year_week """) cur.execute(""" CREATE UNIQUE INDEX IF NOT EXISTS idx_on_top_csv_unique ON on_top_inputs( cycle_id, submitted_by_id, channel, COALESCE(LOWER(buyer), ''), product_id, year_week ) """) conn.commit() upsert_sql = """ INSERT INTO on_top_inputs (cycle_id, product_id, year_week, quantity, channel, buyer, submitted_by_id) VALUES %s ON CONFLICT (cycle_id, submitted_by_id, channel, COALESCE(LOWER(buyer), ''), product_id, year_week) DO UPDATE SET quantity = EXCLUDED.quantity """ with conn.cursor() as cur: execute_values(cur, upsert_sql, rows, page_size=2000) return len(rows) # ------------------------------------------------------------------- # Section 2 — consensus_snapshots # ------------------------------------------------------------------- def migrate_consensus_snapshots(conn) -> int: """Parse data/consensus/snapshot_*.json → consensus_snapshots. UPSERT on label — label is unique per snapshot (e.g. 'CW20-2026'). Older snapshot files may lack vp_inputs / mp_inputs keys; those are stored as NULL in the respective JSONB columns. """ snap_dir = DATA_DIR / "consensus" files = sorted(snap_dir.glob("snapshot_*.json")) if snap_dir.exists() else [] if not files: print(" [warn] No snapshot_*.json files found in data/consensus/ — skipping") return 0 inserted = 0 for fp in files: try: snap = json.loads(fp.read_text(encoding="utf-8")) except Exception as e: print(f" [warn] Could not parse {fp.name}: {e}") continue label = snap.get("label") if not label: print(f" [warn] {fp.name} has no 'label' key — skipping") continue # Derive cycle_id from snap_year + cw if possible; default to 1 snap_year = snap.get("snap_year") snap_cw = snap.get("cw") cycle_id = CYCLE_ID # all these snapshots belong to the one active cycle n_skus = snap.get("n_skus") total_rev_raw = snap.get("total_rev") try: total_rev = float(total_rev_raw) if total_rev_raw is not None else None except (TypeError, ValueError): total_rev = None rows_data = snap.get("rows") vp_inputs = snap.get("vp_inputs") mp_inputs = snap.get("mp_inputs") with conn.cursor() as cur: # No UNIQUE constraint on label — DELETE + INSERT is idempotent cur.execute("DELETE FROM consensus_snapshots WHERE label = %s", (label,)) cur.execute(""" INSERT INTO consensus_snapshots (cycle_id, label, n_skus, total_rev, rows, wholesale_inputs, retail_inputs) VALUES (%s, %s, %s, %s, %s, %s, %s) """, ( cycle_id, label, n_skus, total_rev, json.dumps(rows_data) if rows_data is not None else None, json.dumps(vp_inputs) if vp_inputs is not None else None, json.dumps(mp_inputs) if mp_inputs is not None else None, )) inserted += 1 return inserted # ------------------------------------------------------------------- # Section 3 — webshop_coupon_orders # ------------------------------------------------------------------- def migrate_webshop_orders(conn) -> int: """Load webshop_coupon_orders.csv. Campaigns are extracted from (coupon_id, coupon_name, coupon_code, coupon_config_discount) and upserted into webshop_campaigns first. product_id is already present in the CSV — no SKU lookup needed. Only orders with order_status NOT matching '08 - Otkazano' are imported. """ fp = DATA_DIR / "webshop_coupon_orders.csv" if not fp.exists(): print(" [warn] webshop_coupon_orders.csv missing — skipping") return 0 df = pd.read_csv(fp, dtype={"product_id": "Int64", "coupon_id": "Int64"}, encoding="utf-8", low_memory=False) df.columns = [c.strip() for c in df.columns] # Filter out cancelled orders df = df[~df["order_status"].str.contains("Otkazano", na=False)] # --- Campaigns --- # One campaign per coupon_id (distinct source coupon). Use coupon_id as # the external key for deduplication. camp_df = ( df[["coupon_id", "coupon_name", "coupon_code", "coupon_config_discount"]] .dropna(subset=["coupon_id"]) .drop_duplicates(subset=["coupon_id"]) ) _truncate(conn, "webshop_coupon_orders") _truncate(conn, "webshop_campaigns") # coupon_id → db campaign id campaign_map: dict[int, int] = {} for _, c in camp_df.iterrows(): cid_raw = c["coupon_id"] try: cid = int(cid_raw) except (ValueError, TypeError): continue name = str(c["coupon_name"]) if pd.notna(c["coupon_name"]) else str(c["coupon_code"]) label = str(c["coupon_code"]) if pd.notna(c["coupon_code"]) else None # Derive date range from orders for this coupon mask = df["coupon_id"] == cid_raw dates = pd.to_datetime(df.loc[mask, "date_added"], errors="coerce").dropna() start_d = dates.min().date() if len(dates) else None end_d = dates.max().date() if len(dates) else None with conn.cursor() as cur: cur.execute(""" INSERT INTO webshop_campaigns (name, label, start_date, end_date, channel) VALUES (%s, %s, %s, %s, 'webshop') RETURNING id """, (name, label, start_d, end_d)) db_id = cur.fetchone()[0] campaign_map[cid] = db_id # --- Orders --- order_rows: list[tuple] = [] for _, r in df.iterrows(): try: order_id = str(r["order_id"]).strip() order_date_raw = r.get("date_added") order_date = ( pd.to_datetime(order_date_raw, errors="coerce").date() if pd.notna(order_date_raw) else None ) pid = int(r["product_id"]) if pd.notna(r["product_id"]) else None if pid is None: continue cid_raw = r.get("coupon_id") db_campaign_id = None if pd.notna(cid_raw): try: db_campaign_id = campaign_map.get(int(cid_raw)) except (ValueError, TypeError): pass coupon_code = str(r["coupon_code"]) if pd.notna(r.get("coupon_code")) else None def _num(col): v = r.get(col) try: return float(v) if pd.notna(v) else None except (ValueError, TypeError): return None order_rows.append(( order_id, order_date, pid, coupon_code, db_campaign_id, _num("quantity"), _num("unit_price_before_coupon"), _num("coupon_discount_per_unit"), _num("coupon_discount_percent"), _num("unit_price_after_coupon"), str(r["order_status"]) if pd.notna(r.get("order_status")) else None, )) except Exception as e: print(f" [warn] skipping order row: {e}") continue if order_rows: with conn.cursor() as cur: execute_values( cur, """INSERT INTO webshop_coupon_orders (order_id, order_date, product_id, coupon_code, campaign_id, quantity, original_price, discount_amount, discount_pct, final_price, order_status) VALUES %s""", order_rows, page_size=2000, ) return len(order_rows) # ------------------------------------------------------------------- # Section 4 — sales_clean_import + v_sales_weekly_full # ------------------------------------------------------------------- def migrate_sales_history(conn) -> int: """Create sales_clean_import table (if not exists) and load sales_clean.csv into it, then create/replace v_sales_weekly_full as a UNION of v_sales_weekly (erp_transactions window) and the historical data from sales_clean_import. sales_clean.csv columns used: sku, year, week, qty_retail, qty_webshop, qty_wholesale (is_any_promo etc. are GBR model features — not stored in this table) """ fp = DATA_DIR / "sales_clean.csv" if not fp.exists(): print(" [warn] sales_clean.csv missing — skipping") return 0 product_map = _load_product_map(conn) # Create table if not exists with conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS sales_clean_import ( id BIGSERIAL PRIMARY KEY, product_id INT REFERENCES dim_products(id), year INT NOT NULL, week INT NOT NULL, qty_retail DECIMAL(12,2), qty_webshop DECIMAL(12,2), qty_wholesale DECIMAL(12,2), imported_at TIMESTAMPTZ DEFAULT now() ) """) cur.execute(""" CREATE UNIQUE INDEX IF NOT EXISTS idx_sci_product_yw ON sales_clean_import(product_id, year, week) """) conn.commit() # Truncate and reload with conn.cursor() as cur: cur.execute("TRUNCATE TABLE sales_clean_import RESTART IDENTITY") CHUNK = 20_000 total = 0 for chunk in pd.read_csv(fp, dtype={"sku": "string"}, encoding="utf-8", chunksize=CHUNK, low_memory=False): chunk["sku"] = chunk["sku"].fillna("").str.strip() rows: list[tuple] = [] for _, r in chunk.iterrows(): sku = r["sku"] if not sku: continue pid = product_map.get(sku) if pid is None: continue def _num(col): v = r.get(col) try: return float(v) if pd.notna(v) else None except (ValueError, TypeError): return None rows.append(( pid, int(r["year"]) if pd.notna(r.get("year")) else None, int(r["week"]) if pd.notna(r.get("week")) else None, _num("qty_retail"), _num("qty_webshop"), _num("qty_wholesale"), )) with conn.cursor() as cur: execute_values( cur, """INSERT INTO sales_clean_import (product_id, year, week, qty_retail, qty_webshop, qty_wholesale) VALUES %s""", rows, page_size=2000, ) conn.commit() total += len(rows) # Create / replace UNION view # v_sales_weekly covers the recent erp_transactions window. # v_sales_weekly_full adds the historical weekly totals from sales_clean. with conn.cursor() as cur: # Drop and recreate so the definition is always current on re-runs cur.execute("DROP MATERIALIZED VIEW IF EXISTS v_sales_weekly_full") cur.execute(""" CREATE MATERIALIZED VIEW v_sales_weekly_full AS -- Historical weeks from sales_clean.csv that are NOT already -- covered by erp_transactions (avoids double-counting the -- ~9-week overlap where both sources have data). -- GREATEST(...,0) defensively clamps any negative weekly -- totals to zero so returns can't push downstream calcs negative. SELECT sci.product_id, sci.year, sci.week, GREATEST(COALESCE(sci.qty_retail, 0), 0) AS qty_retail, GREATEST(COALESCE(sci.qty_webshop, 0), 0) AS qty_webshop, GREATEST(COALESCE(sci.qty_wholesale, 0), 0) AS qty_wholesale, GREATEST(COALESCE(sci.qty_retail, 0), 0) + GREATEST(COALESCE(sci.qty_webshop, 0), 0) + GREATEST(COALESCE(sci.qty_wholesale, 0), 0) AS qty_total FROM sales_clean_import sci WHERE NOT EXISTS ( SELECT 1 FROM v_sales_weekly vsw WHERE vsw.product_id = sci.product_id AND vsw.year = sci.year AND vsw.week = sci.week ) UNION ALL -- Recent window from erp_transactions (already non-negative — -- v_sales_weekly clamps with GREATEST at the SUM level). SELECT product_id, year, week, qty_retail, qty_webshop, qty_wholesale, qty_total FROM v_sales_weekly WITH NO DATA """) # Refresh immediately so queries work cur.execute("REFRESH MATERIALIZED VIEW v_sales_weekly_full") return total # ------------------------------------------------------------------- # Section 5 — promo_calendar_entries (gaps-and-islands) # ------------------------------------------------------------------- def migrate_promo_calendar(conn) -> int: """Derive promo_calendar_entries from erp_promo_weeks using a gaps-and-islands approach: consecutive weeks with is_erp_promo=TRUE for the same product become a single entry. Also merges approved promo_proposals if any exist (status='approved'). """ _truncate(conn, "promo_calendar_entries") # Gaps-and-islands in pure SQL — faster than pulling rows into Python with conn.cursor() as cur: cur.execute(""" WITH numbered AS ( SELECT product_id, year, week, -- row_number to identify gaps: if consecutive, -- (year*100+week) increments by 1 (within same year) -- Use a simpler grp: row_number minus a dense rank -- over (year*100+week) gives stable group ids. ROW_NUMBER() OVER (PARTITION BY product_id ORDER BY year, week) AS rn, (year * 100 + week) AS yw FROM erp_promo_weeks WHERE is_erp_promo = TRUE ), grp_calc AS ( SELECT product_id, year, week, yw, rn - DENSE_RANK() OVER ( PARTITION BY product_id ORDER BY yw ) AS grp FROM numbered ), islands AS ( SELECT product_id, MIN(year) AS start_year, MIN(week) AS start_week, MAX(year) AS end_year, MAX(week) AS end_week FROM grp_calc GROUP BY product_id, grp ) INSERT INTO promo_calendar_entries (product_id, department, promo_name, mechanic, discount_pct, start_date, end_date, start_year, start_week, end_year, end_week, source, status) SELECT i.product_id, 'ERP' AS department, 'ERP promo' AS promo_name, NULL AS mechanic, NULL AS discount_pct, -- ISO week → exact Monday / Sunday dates via to_date() to_date(i.start_year::text || ' ' || i.start_week::text || ' 1', 'IYYY IW ID') AS start_date, to_date(i.end_year::text || ' ' || i.end_week::text || ' 7', 'IYYY IW ID') AS end_date, i.start_year, i.start_week, i.end_year, i.end_week, 'erp_promo_weeks' AS source, 'active' AS status FROM islands i """) n_erp = cur.rowcount # Merge approved promo_proposals if the table exists. # promo_proposals.skus is a JSONB array of SKU strings; no product_id col. cur.execute(""" SELECT EXISTS ( SELECT 1 FROM information_schema.tables WHERE table_schema='public' AND table_name='promo_proposals' ) """) if cur.fetchone()[0]: cur.execute(""" INSERT INTO promo_calendar_entries (product_id, department, promo_name, mechanic, discount_pct, start_date, end_date, start_year, start_week, end_year, end_week, source, status, created_by_id, created_at) SELECT dp.id, pp.source AS department, pp.name AS promo_name, pp.mechanic, pp.discount_pct, to_date(pp.start_year::text || ' ' || pp.start_week::text || ' 1', 'IYYY IW ID'), to_date(pp.end_year::text || ' ' || pp.end_week::text || ' 7', 'IYYY IW ID'), pp.start_year, pp.start_week, pp.end_year, pp.end_week, 'promo_proposals', pp.status, pp.proposed_by_id, pp.created_at FROM promo_proposals pp CROSS JOIN LATERAL jsonb_array_elements_text(pp.skus) AS sku_val JOIN dim_products dp ON dp.sku = sku_val WHERE pp.status = 'approved' """) return n_erp # ------------------------------------------------------------------- # Section 6 — erp_promo_campaigns + erp_promo_items (rabatne.xlsx) # ------------------------------------------------------------------- def migrate_erp_promo_campaigns(conn) -> int: """Load data/rabatne.xlsx into erp_promo_campaigns + erp_promo_items. File structure: - Rows where CodeArt is NaN are campaign header rows (Vrsta = campaign name). - Item rows have CodeArt = SKU string. - Rabat column: large integer (e.g. 28579594) → divide by 1_000_000 → pct. - Dates: VrijediOd / VrijediDo as 'DD.MM.YYYY' strings. """ fp = DATA_DIR / "rabatne.xlsx" if not fp.exists(): print(" [warn] rabatne.xlsx missing — skipping") return 0 product_map = _load_product_map(conn) _truncate(conn, "erp_promo_items") _truncate(conn, "erp_promo_campaigns") df = pd.read_excel(fp, dtype=str, engine="openpyxl") df.columns = [c.strip() for c in df.columns] def _parse_date(s) -> date | None: if not s or (isinstance(s, float) and pd.isna(s)): return None s = str(s).strip() for fmt in ("%d.%m.%Y", "%Y-%m-%d"): try: return datetime.strptime(s, fmt).date() except ValueError: continue return None total_items = 0 current_campaign_id: int | None = None current_valid_from: date | None = None current_valid_to: date | None = None item_batch: list[tuple] = [] def _flush_items(): nonlocal item_batch if not item_batch: return with conn.cursor() as cur: execute_values( cur, """INSERT INTO erp_promo_items (campaign_id, product_id, description, discount_raw, discount_pct, valid_from, valid_to) VALUES %s""", item_batch, page_size=2000, ) item_batch = [] for _, row in df.iterrows(): code_art = row.get("CodeArt", "") vrsta = str(row.get("Vrsta", "") or "").strip() is_header = ( pd.isna(row.get("CodeArt")) or str(code_art).strip() == "" or str(code_art).strip().lower() == "nan" ) if is_header: # Flush previous campaign's items _flush_items() if not vrsta: current_campaign_id = None continue valid_from = _parse_date(row.get("VrijediOd")) valid_to = _parse_date(row.get("VrijediDo")) current_valid_from = valid_from current_valid_to = valid_to opis = str(row.get("Opis", "") or "").strip() or None with conn.cursor() as cur: cur.execute( """INSERT INTO erp_promo_campaigns (type, description, valid_from, valid_to) VALUES (%s, %s, %s, %s) RETURNING id""", (vrsta, opis, valid_from, valid_to), ) current_campaign_id = cur.fetchone()[0] else: if current_campaign_id is None: continue sku = str(code_art).strip() pid = product_map.get(sku) if pid is None: continue rabat_raw = row.get("Rabat") try: rabat_int = int(float(str(rabat_raw).strip())) if rabat_raw and str(rabat_raw).strip() not in ("", "nan") else None except (ValueError, TypeError): rabat_int = None rabat_pct = round(rabat_int / 1_000_000, 6) if rabat_int is not None else None # Item-level dates (may differ from campaign dates) item_from = _parse_date(row.get("VrijediOd")) or current_valid_from item_to = _parse_date(row.get("VrijediDo")) or current_valid_to opis = str(row.get("Opis", "") or "").strip() or None item_batch.append(( current_campaign_id, pid, opis, rabat_int, rabat_pct, item_from, item_to, )) total_items += 1 _flush_items() conn.commit() return total_items # ------------------------------------------------------------------- # Section 7 — erp_stock_history (country aggregate snapshots) # ------------------------------------------------------------------- def migrate_stock_history(conn) -> int: """Load stock_stores_at.csv + stock_stores_slo.csv into erp_stock_history using synthetic country-aggregate store IDs. snapshot_date is taken from the file's mtime (best available proxy). Only AT and SLO files — HR stores are already in erp_stock_current via migrate_remaining.py. """ # Synthetic store unit_codes created by migrate_remaining.py COUNTRY_STORES = { "AT": "STORES_AT", "SLO": "STORES_SLO", } sources = [ ("stock_stores_at.csv", "AT"), ("stock_stores_slo.csv", "SLO"), ] # Load store_id map with conn.cursor() as cur: cur.execute("SELECT unit_code, id FROM dim_stores") store_map = {r[0]: r[1] for r in cur.fetchall()} product_map = _load_product_map(conn) _truncate(conn, "erp_stock_history") total = 0 rows: list[tuple] = [] for fname, country in sources: fp = DATA_DIR / fname if not fp.exists(): print(f" [warn] {fname} missing — skipping") continue store_code = COUNTRY_STORES[country] store_id = store_map.get(store_code) if store_id is None: print(f" [warn] Synthetic store '{store_code}' not in dim_stores.") print(f" Run db.migrate_remaining first to create it.") continue # File mtime as snapshot_date snap_date = date.fromtimestamp(fp.stat().st_mtime) df = pd.read_csv(fp, dtype=str, encoding="utf-8") df["sku"] = df["sku"].fillna("").str.strip() df = df[df["sku"] != ""].drop_duplicates(subset=["sku"], keep="first") for _, r in df.iterrows(): pid = product_map.get(r["sku"]) if pid is None: continue try: qty = float(r["on_hand"]) if r["on_hand"] not in (None, "", "nan") else None except (ValueError, TypeError): qty = None rows.append((pid, store_id, qty, snap_date)) total += 1 if rows: with conn.cursor() as cur: execute_values( cur, """INSERT INTO erp_stock_history (product_id, store_id, stock_qty, snapshot_date) VALUES %s""", rows, page_size=2000, ) return total # ------------------------------------------------------------------- # Entry point # ------------------------------------------------------------------- _STEPS = [ ("on_top_inputs", migrate_on_top_inputs), ("consensus_snapshots", migrate_consensus_snapshots), ("webshop_coupon_orders", migrate_webshop_orders), ("sales_clean_import", migrate_sales_history), ("promo_calendar_entries", migrate_promo_calendar), ("erp_promo_campaigns", migrate_erp_promo_campaigns), ("erp_stock_history", migrate_stock_history), ] VERIFY_TABLES = [ "on_top_inputs", "consensus_snapshots", "webshop_campaigns", "webshop_coupon_orders", "sales_clean_import", "promo_calendar_entries", "erp_promo_campaigns", "erp_promo_items", "erp_stock_history", ] def main() -> None: print("Connecting to Postgres...") conn = get_connection() conn.autocommit = False summary: list[tuple[str, str]] = [] try: for label, fn in _STEPS: print() print(f"=== {label} ===") try: n = fn(conn) conn.commit() summary.append((label, f"{n:>9,} rows")) print(f" -> {n:,} rows") except Exception as e: conn.rollback() summary.append((label, f"FAILED: {type(e).__name__}: {e}")) print(f" -> FAILED: {type(e).__name__}: {e}") finally: # Refresh the full sales view if it exists print() print("Refreshing v_sales_weekly_full (if exists)...") try: with conn.cursor() as cur: cur.execute(""" SELECT 1 FROM pg_matviews WHERE schemaname='public' AND matviewname='v_sales_weekly_full' """) if cur.fetchone(): cur.execute("REFRESH MATERIALIZED VIEW v_sales_weekly_full") conn.commit() print(" -> refreshed") else: print(" -> not yet created (sales section may have failed)") except Exception as e: conn.rollback() print(f" -> {e}") print() print("=== Final row counts ===") with conn.cursor() as cur: for t in VERIFY_TABLES: try: cur.execute(f"SELECT COUNT(*) FROM {t}") n = cur.fetchone()[0] print(f" {t:28s} {n:>9,}") except Exception as e: print(f" {t:28s} ERROR ({e})") print() print("=== Step summary ===") for label, result in summary: print(f" {label:28s} {result}") conn.close() if __name__ == "__main__": main()