Spaces:
Sleeping
Sleeping
File size: 2,994 Bytes
924a755 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 | """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()
|