"""CFO S&OP audit — Step 0: enumerate every data source. Lists every file under data/ (recursively) AND every relevant DB table / view. For each, prints row count + columns + 3 sample rows + flags columns matching: selling price, revenue, margin, actuals, plan, lead time, promo, COGS. No outputs written. Print-only — Lovro reads this and approves before the main script runs. """ from __future__ import annotations import os import re from pathlib import Path from typing import Iterable import pandas as pd from sqlalchemy import text from backend.models.database import SessionLocal # Tokens we want to spotlight in column names FLAGS = { "selling_price": [r"sell", r"retail", r"msrp", r"mpc", r"rrp", r"price", r"vpc"], "margin": [r"margin", r"marž", r"ruc", r"markup"], "cost": [r"cost", r"nabav", r"nc"], "lead_time": [r"lead", r"\blt\b", r"delivery_day", r"production_day"], "promo": [r"promo", r"contest", r"uplift", r"campaign", r"activation", r"akcij"], "actuals": [r"actual", r"realiz", r"realized", r"history", r"transaction", r"sales", r"sold"], "plan": [r"plan", r"budget", r"forecast", r"target", r"baseline"], } def find_flags(cols: Iterable[str]) -> dict[str, list[str]]: cols_lower = [(c, c.lower()) for c in cols] out: dict[str, list[str]] = {} for tag, patterns in FLAGS.items(): hits: list[str] = [] for orig, lo in cols_lower: for p in patterns: if re.search(p, lo): hits.append(orig) break if hits: out[tag] = hits return out # ============================================================ # Part A — files on disk # ============================================================ def audit_data_folder() -> None: root = Path(__file__).resolve().parents[1] / "data" print(f"\n{'='*70}\nPART A — data/ folder ({root})\n{'='*70}") if not root.exists(): print(" (missing)") return files: list[Path] = [] for p in sorted(root.rglob("*")): # skip dirs, temp lock files, archive trees if not p.is_file(): continue if p.name.startswith("~$") or p.name.startswith("_") or p.name.startswith("."): continue if any(part.startswith("_backup") or part in ("_shadow", "archive") for part in p.relative_to(root).parts[:-1]): continue files.append(p) print(f" Found {len(files)} candidate files\n") for p in files: rel = p.relative_to(root) ext = p.suffix.lower() size_kb = p.stat().st_size / 1024 print(f"\n-- {rel} ({size_kb:,.1f} KB)") try: if ext == ".csv": df = pd.read_csv(p, nrows=200, low_memory=False) # Get true row count via cheap pass with open(p, "r", encoding="utf-8", errors="replace") as fh: n_rows = sum(1 for _ in fh) - 1 _dump_df(df, n_rows) elif ext in (".xlsx", ".xlsm"): xl = pd.ExcelFile(p) for sn in xl.sheet_names: try: df = pd.read_excel(p, sheet_name=sn, nrows=200) except Exception as e: print(f" [sheet {sn}] read error: {e}") continue full = pd.read_excel(p, sheet_name=sn) print(f" [sheet {sn}]") _dump_df(df, len(full)) elif ext == ".json": import json try: j = json.loads(p.read_text(encoding="utf-8")) except Exception as e: print(f" read error: {e}") continue if isinstance(j, list) and j and isinstance(j[0], dict): df = pd.DataFrame(j[:10]) _dump_df(df, len(j)) elif isinstance(j, dict): print(f" top-level keys: {list(j.keys())[:20]}") else: print(f" {type(j).__name__}") else: print(f" skipped (unsupported ext)") except Exception as e: print(f" ERROR reading: {e}") def _dump_df(df: pd.DataFrame, n_rows: int) -> None: cols = list(df.columns) print(f" rows: {n_rows} cols: {len(cols)}") print(f" columns: {cols}") flags = find_flags(cols) if flags: print(f" FLAGS:") for tag, hits in flags.items(): print(f" {tag}: {hits}") if not df.empty: # Show 3 sample rows print(" sample:") for _, row in df.head(3).iterrows(): preview = {k: (str(v)[:40] if v is not None else None) for k, v in row.to_dict().items()} print(f" {preview}") # ============================================================ # Part B — DB tables / views # ============================================================ PRIORITY_TABLES = [ # core dims "dim_products", "dim_categories", "dim_suppliers", "dim_stores", "users", # planning / pricing "sku_planning", "erp_costs", "erp_prices", "erp_nc30", # actuals / forecasts "v_sales_weekly_full", "forecasts", "forecast_runs", "erp_transactions", "erp_stock_current", "incoming_supply", # promo / on-top "erp_promo_weeks", "erp_promo_calendar", "promo_proposals", "on_top_inputs", # supply ops "supply_master", "order_proposals", # NPL / NPD "npd_products", "npl_products", # lookups "lookup_channel_map", ] def audit_db() -> None: db = SessionLocal() try: print(f"\n{'='*70}\nPART B — DB tables\n{'='*70}") existing = {r["table_name"] for r in db.execute(text(""" SELECT table_name FROM information_schema.tables WHERE table_schema='public' """)).mappings()} for t in PRIORITY_TABLES: if t not in existing: print(f"\n-- {t}: MISSING") continue cols = [(r["column_name"], r["data_type"]) for r in db.execute(text(""" SELECT column_name, data_type FROM information_schema.columns WHERE table_schema='public' AND table_name=:t ORDER BY ordinal_position """), {"t": t}).mappings()] try: n = db.execute(text(f"SELECT COUNT(*) FROM {t}")).scalar() except Exception: n = "?" print(f"\n-- {t} rows: {n} cols: {len(cols)}") cn = [c for c, _ in cols] print(f" columns: {cn}") flags = find_flags(cn) if flags: print(f" FLAGS:") for tag, hits in flags.items(): print(f" {tag}: {hits}") try: sample = db.execute(text(f"SELECT * FROM {t} LIMIT 3")).mappings().all() for row in sample: preview = {k: (str(v)[:40] if v is not None else None) for k, v in dict(row).items()} print(f" {preview}") except Exception as e: print(f" sample read error: {e}") # Also list any other public tables we missed rest = sorted(existing - set(PRIORITY_TABLES)) if rest: print(f"\n-- Other public tables ({len(rest)}):") for t in rest: try: n = db.execute(text(f"SELECT COUNT(*) FROM {t}")).scalar() except Exception: n = "?" print(f" {t} ({n} rows)") finally: db.close() if __name__ == "__main__": audit_data_folder() audit_db()