Spaces:
Sleeping
Sleeping
| """Build data/warehouse.db from schema.sql + seed CSVs. Idempotent (drops & rebuilds). | |
| Run: python -m scripts.init_db | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| import sqlite3 | |
| import sys | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) | |
| from core.config import PROJECT_ROOT, db_path # noqa: E402 | |
| DATA = PROJECT_ROOT / "data" | |
| SEED = DATA / "seed" | |
| def load_csv(path: Path) -> list[dict]: | |
| with open(path, newline="", encoding="utf-8") as f: | |
| return list(csv.DictReader(f)) | |
| TABLES = ("run_trace", "purchase_orders", "vendor_memory", "supplier_skus", | |
| "suppliers", "inventory") | |
| def main() -> None: | |
| db = db_path() | |
| conn = sqlite3.connect(db, timeout=10) | |
| # Rebuild in-place (drop + recreate) rather than deleting the file — on | |
| # Windows, unlink fails if any reader (e.g. the dashboard) holds the file. | |
| for t in TABLES: | |
| conn.execute(f"DROP TABLE IF EXISTS {t}") | |
| conn.commit() | |
| conn.executescript((DATA / "schema.sql").read_text(encoding="utf-8")) | |
| inv = load_csv(SEED / "inventory.csv") | |
| for r in inv: | |
| qty, rp = int(r["qty_on_hand"]), int(r["reorder_point"]) | |
| status = "low" if qty < rp else "ok" | |
| conn.execute( | |
| "INSERT INTO inventory VALUES (?,?,?,?,?,?,?,?)", | |
| (r["sku"], r["name"], r["unit"], qty, rp, | |
| int(r["reorder_qty"]), float(r["list_price"]), status), | |
| ) | |
| sup = load_csv(SEED / "suppliers.csv") | |
| for r in sup: | |
| conn.execute( | |
| "INSERT INTO suppliers VALUES (?,?,?,?,?,?,?)", | |
| (r["vendor_id"], r["name"], r["persona"], float(r["base_price"]), | |
| float(r["cost_floor"]), int(r["lead_time_days"]), float(r["reliability"])), | |
| ) | |
| links = load_csv(SEED / "supplier_skus.csv") | |
| for r in links: | |
| conn.execute( | |
| "INSERT INTO supplier_skus VALUES (?,?,?,?)", | |
| (r["vendor_id"], r["sku"], float(r["base_price"]), int(r["in_stock"])), | |
| ) | |
| conn.commit() | |
| # --- self-check (CP1 DoD) --- | |
| n_sku = conn.execute("SELECT COUNT(*) FROM inventory").fetchone()[0] | |
| n_sup = conn.execute("SELECT COUNT(*) FROM suppliers").fetchone()[0] | |
| low = conn.execute( | |
| "SELECT sku, qty_on_hand, reorder_point FROM inventory WHERE status='low'" | |
| ).fetchall() | |
| demo_vendors = conn.execute( | |
| "SELECT COUNT(*) FROM supplier_skus WHERE sku='SEED-MAIZE-01' AND in_stock=1" | |
| ).fetchone()[0] | |
| conn.close() | |
| assert n_sku >= 20, f"need >=20 SKUs, got {n_sku}" | |
| assert n_sup >= 8, f"need >=8 suppliers, got {n_sup}" | |
| assert any(r[0] == "SEED-MAIZE-01" for r in low), "demo item must be low-stock" | |
| assert demo_vendors >= 3, f"need >=3 vendors on demo item, got {demo_vendors}" | |
| print(f"[init_db] OK: {n_sku} SKUs, {n_sup} suppliers, " | |
| f"{len(low)} low-stock items {[r[0] for r in low]}, " | |
| f"{demo_vendors} vendors stock the demo item.") | |
| if __name__ == "__main__": | |
| main() | |