Add sync module: push seed data to main app DB (dev/prod)
- New sync.py: reads seed import's 39-col schema, maps to main app's customers/locations/assets schema with foreign keys - New POST /api/sync endpoint (target=dev|prod, dry_run=true/false) - New GET /api/sync/targets endpoint - Category mapping: Bev→Beverage, Snack→Snack, Food→Food, etc. - Automatically creates customers and locations from company/place - Idempotent: updates existing assets by machine_id - Dry-run mode: preview changes without writing
This commit is contained in:
@@ -0,0 +1,324 @@
|
||||
"""
|
||||
🔄 Canteen Seed Import -> Main App Database Syncer
|
||||
|
||||
Pushes seed-import data (39-column schema) into the main canteen
|
||||
asset tracker's DB schema (customers/locations/assets with foreign keys).
|
||||
|
||||
Usage:
|
||||
python3 sync.py # Push to dev (default)
|
||||
python3 sync.py --target prod # Push to production
|
||||
python3 sync.py --dry-run # Preview without writing
|
||||
"""
|
||||
import argparse
|
||||
import sqlite3
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
# ─── Default paths ────────────────────────────────────────────────────────────
|
||||
|
||||
SEED_DB = str(Path(__file__).parent / "seed-data" / "assets.db")
|
||||
|
||||
TARGETS = {
|
||||
"dev": {
|
||||
"path": str(Path.home() / "projects/canteen-asset-tracker-dev" / "assets.dev.db"),
|
||||
},
|
||||
"prod": {
|
||||
"path": os.environ.get("PROD_DB_PATH", str(Path.home() / "projects/canteen-asset-tracker" / "assets.db")),
|
||||
},
|
||||
}
|
||||
|
||||
# ─── Category mapping ──────────────────────────────────────────────────────────
|
||||
|
||||
CATEGORY_MAP = {
|
||||
"Bev": "Beverage",
|
||||
"Snack": "Snack",
|
||||
"Food": "Food",
|
||||
"Snack/Bev": "Snack & Beverage",
|
||||
"Coffee": "Coffee",
|
||||
"Cold Drink": "Cold Drink",
|
||||
"Frozen": "Frozen",
|
||||
"Unknown": "Other",
|
||||
}
|
||||
|
||||
|
||||
# ─── Schema helpers ────────────────────────────────────────────────────────────
|
||||
|
||||
def _val(row, col, default=""):
|
||||
"""Safely get a value from sqlite3.Row by column name."""
|
||||
try:
|
||||
v = row[col]
|
||||
return v if v is not None else default
|
||||
except (KeyError, IndexError, AttributeError):
|
||||
return default
|
||||
|
||||
|
||||
def _get_or_create_customer(main, company):
|
||||
cur = main.execute("SELECT id FROM customers WHERE name = ?", (company,))
|
||||
row = cur.fetchone()
|
||||
if row:
|
||||
return row[0]
|
||||
main.execute("INSERT INTO customers (name) VALUES (?)", (company,))
|
||||
return main.execute("SELECT id FROM customers WHERE name = ?", (company,)).fetchone()[0]
|
||||
|
||||
|
||||
def _get_or_create_location(main, place, customer_id, row):
|
||||
cur = main.execute(
|
||||
"SELECT id FROM locations WHERE name = ? AND customer_id = ?",
|
||||
(place, customer_id),
|
||||
)
|
||||
loc = cur.fetchone()
|
||||
if loc:
|
||||
return loc[0]
|
||||
main.execute(
|
||||
"INSERT INTO locations (name, customer_id, address, building_name, floor) VALUES (?, ?, ?, ?, ?)",
|
||||
(place, customer_id,
|
||||
_val(row, "address"),
|
||||
_val(row, "building_name"),
|
||||
_val(row, "floor")),
|
||||
)
|
||||
return main.execute(
|
||||
"SELECT id FROM locations WHERE name = ? AND customer_id = ?",
|
||||
(place, customer_id),
|
||||
).fetchone()[0]
|
||||
|
||||
|
||||
# ─── Main push logic ───────────────────────────────────────────────────────────
|
||||
|
||||
def push(source_db: str, target_db: str, dry_run: bool = False) -> dict:
|
||||
if not os.path.exists(source_db):
|
||||
return {"error": f"Source DB not found: {source_db}"}
|
||||
if not os.path.exists(target_db):
|
||||
return {"error": f"Target DB not found: {target_db}"}
|
||||
|
||||
src = sqlite3.connect(source_db)
|
||||
src.row_factory = sqlite3.Row
|
||||
tgt = sqlite3.connect(target_db)
|
||||
tgt.execute("PRAGMA foreign_keys = OFF")
|
||||
|
||||
report = {
|
||||
"source": source_db,
|
||||
"target": target_db,
|
||||
"dry_run": dry_run,
|
||||
"customers_created": 0,
|
||||
"locations_created": 0,
|
||||
"assets_created": 0,
|
||||
"assets_updated": 0,
|
||||
"assets_skipped": 0,
|
||||
"errors": [],
|
||||
"first_10": [],
|
||||
}
|
||||
|
||||
try:
|
||||
rows = src.execute("SELECT * FROM assets ORDER BY machine_id").fetchall()
|
||||
report["total_source"] = len(rows)
|
||||
|
||||
# Check what columns seed DB actually has
|
||||
src_cols = {row[1] for row in src.execute("PRAGMA table_info(assets)").fetchall()}
|
||||
|
||||
customers_before = tgt.execute("SELECT COUNT(*) FROM customers").fetchone()[0]
|
||||
locations_before = tgt.execute("SELECT COUNT(*) FROM locations").fetchone()[0]
|
||||
assets_before = tgt.execute("SELECT COUNT(*) FROM assets").fetchone()[0]
|
||||
report["target_before"] = {
|
||||
"customers": customers_before,
|
||||
"locations": locations_before,
|
||||
"assets": assets_before,
|
||||
}
|
||||
|
||||
for r in rows:
|
||||
mid = _val(r, "machine_id", "")
|
||||
company = (_val(r, "company") or "Unknown").strip()
|
||||
place = (_val(r, "place") or "Unknown").strip()
|
||||
|
||||
if not dry_run:
|
||||
customer_id = _get_or_create_customer(tgt, company)
|
||||
location_id = _get_or_create_location(tgt, place, customer_id, r)
|
||||
else:
|
||||
customer_id = 0
|
||||
location_id = 0
|
||||
|
||||
existing = tgt.execute(
|
||||
"SELECT id FROM assets WHERE machine_id = ?", (mid,)
|
||||
).fetchone()
|
||||
|
||||
# Category mapping
|
||||
raw_class = _val(r, "class", "Unknown")
|
||||
cat = CATEGORY_MAP.get(raw_class, "Other")
|
||||
|
||||
# Build description from seed-only fields
|
||||
desc_parts = [
|
||||
f"{label}: {_val(r, col)}"
|
||||
for label, col in [
|
||||
("Priority", "priority"),
|
||||
("Disney Area", "disney_area"),
|
||||
("Pricing", "remote_pricing_status"),
|
||||
("Telemetry", "telemetry_provider"),
|
||||
("Reader", "card_reader_brand"),
|
||||
("Alerts", "alerts"),
|
||||
("Zone", "zone"),
|
||||
("Area", "location_area"),
|
||||
]
|
||||
if _val(r, col)
|
||||
]
|
||||
description = " | ".join(desc_parts)
|
||||
|
||||
name = _val(r, "name") or f"{_val(r, 'make', 'Unknown')} @ {place}"
|
||||
|
||||
# deployed is TEXT in main app, may be INTEGER in seed
|
||||
deployed_raw = r["deployed"] if "deployed" in src_cols else None
|
||||
if deployed_raw is not None:
|
||||
if isinstance(deployed_raw, (int, float)):
|
||||
deployed_raw = "1" if deployed_raw else "0"
|
||||
else:
|
||||
deployed_raw = str(deployed_raw) if deployed_raw else ""
|
||||
else:
|
||||
deployed_raw = ""
|
||||
|
||||
has_disney = _val(r, "disney_area")
|
||||
|
||||
is_disney = 1 if has_disney else 0
|
||||
|
||||
asset_data = {
|
||||
"machine_id": mid,
|
||||
"serial_number": _val(r, "serial_number"),
|
||||
"name": name,
|
||||
"description": description,
|
||||
"category": cat,
|
||||
"status": "active",
|
||||
"make": _val(r, "make"),
|
||||
"model": _val(r, "model"),
|
||||
"address": _val(r, "address"),
|
||||
"building_name": _val(r, "building_name"),
|
||||
"building_number": "",
|
||||
"floor": _val(r, "floor"),
|
||||
"room": _val(r, "room"),
|
||||
"trailer_number": _val(r, "trailer"),
|
||||
"walking_directions": "",
|
||||
"map_link": "",
|
||||
"parking_location": "",
|
||||
"photo_path": _val(r, "photo_path"),
|
||||
"customer_id": customer_id,
|
||||
"location_id": location_id,
|
||||
"assigned_to": None,
|
||||
"latitude": _val(r, "latitude"),
|
||||
"longitude": _val(r, "longitude"),
|
||||
"geofence_radius_meters": 50,
|
||||
"disney_park": _val(r, "disney_park"),
|
||||
"is_disney": is_disney,
|
||||
"created_at": _val(r, "created_at", ""),
|
||||
"updated_at": _val(r, "updated_at", ""),
|
||||
"dex_report_date": _val(r, "dex_report_date"),
|
||||
"install_date": _val(r, "install_date"),
|
||||
"deployed": deployed_raw,
|
||||
"pulled_date": _val(r, "pulled_date"),
|
||||
}
|
||||
|
||||
if existing:
|
||||
report["assets_updated"] += 1
|
||||
if len(report["first_10"]) < 10:
|
||||
report["first_10"].append(f"UPDATE {mid}: {name} ({cat})")
|
||||
if not dry_run:
|
||||
sets = ", ".join(f"{col} = ?" for col in asset_data)
|
||||
vals = list(asset_data.values()) + [mid]
|
||||
tgt.execute(
|
||||
f"UPDATE assets SET {sets} WHERE machine_id = ?",
|
||||
vals,
|
||||
)
|
||||
else:
|
||||
report["assets_created"] += 1
|
||||
if len(report["first_10"]) < 10:
|
||||
report["first_10"].append(f"CREATE {mid}: {name} ({cat})")
|
||||
if not dry_run:
|
||||
cols = list(asset_data.keys())
|
||||
qmarks = ", ".join("?" for _ in cols)
|
||||
vals = [asset_data[c] for c in cols]
|
||||
tgt.execute(
|
||||
f"INSERT INTO assets ({', '.join(cols)}) VALUES ({qmarks})",
|
||||
vals,
|
||||
)
|
||||
|
||||
if not dry_run:
|
||||
tgt.commit()
|
||||
report["customers_created"] = (
|
||||
tgt.execute("SELECT COUNT(*) FROM customers").fetchone()[0]
|
||||
- customers_before
|
||||
)
|
||||
report["locations_created"] = (
|
||||
tgt.execute("SELECT COUNT(*) FROM locations").fetchone()[0]
|
||||
- locations_before
|
||||
)
|
||||
report["assets_after"] = tgt.execute(
|
||||
"SELECT COUNT(*) FROM assets"
|
||||
).fetchone()[0]
|
||||
|
||||
except Exception as e:
|
||||
import traceback
|
||||
traceback.print_exc()
|
||||
report["errors"].append(str(e))
|
||||
if not dry_run:
|
||||
tgt.rollback()
|
||||
finally:
|
||||
src.close()
|
||||
if not dry_run:
|
||||
tgt.execute("PRAGMA foreign_keys = ON")
|
||||
tgt.close()
|
||||
|
||||
report["total_actions"] = report["assets_created"] + report["assets_updated"]
|
||||
return report
|
||||
|
||||
|
||||
# ─── CLI ────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Sync seed import data -> main app DB")
|
||||
parser.add_argument("--target", choices=list(TARGETS.keys()), default="dev",
|
||||
help="Which environment to push to (default: dev)")
|
||||
parser.add_argument("--source", default=SEED_DB, help="Source DB path")
|
||||
parser.add_argument("--target-db", help="Override target DB path instead of --target preset")
|
||||
parser.add_argument("--dry-run", action="store_true", help="Preview without writing")
|
||||
args = parser.parse_args()
|
||||
|
||||
target_db = args.target_db or TARGETS[args.target]["path"]
|
||||
|
||||
print(f"📤 Seed Import > Main App Sync")
|
||||
print(f" Source: {args.source}")
|
||||
print(f" Target: {target_db}")
|
||||
print(f" Mode: {'DRY RUN' if args.dry_run else 'LIVE'}")
|
||||
print()
|
||||
|
||||
report = push(args.source, target_db, dry_run=args.dry_run)
|
||||
|
||||
if "error" in report:
|
||||
print(f"ERROR: {report['error']}")
|
||||
sys.exit(1)
|
||||
|
||||
print(f" Source records: {report.get('total_source', '?')}")
|
||||
print(f" Assets created: {report['assets_created']}")
|
||||
print(f" Assets updated: {report['assets_updated']}")
|
||||
print(f" Total actions: {report['total_actions']}")
|
||||
print()
|
||||
|
||||
if report.get("first_10"):
|
||||
print(" Sample actions:")
|
||||
for a in report["first_10"]:
|
||||
print(f" {a}")
|
||||
|
||||
if report.get("errors"):
|
||||
print(f"\nWarnings ({len(report['errors'])}):")
|
||||
for e in report["errors"]:
|
||||
print(f" {e}")
|
||||
sys.exit(1)
|
||||
|
||||
if args.dry_run:
|
||||
print("Dry run complete. Run without --dry-run to apply.")
|
||||
else:
|
||||
b = report.get("target_before", {})
|
||||
print("Sync complete!")
|
||||
if "target_before" in report:
|
||||
print(f" Customers: {b['customers']} > +{report['customers_created']}")
|
||||
print(f" Locations: {b['locations']} > +{report['locations_created']}")
|
||||
print(f" Assets: {b['assets']} > +{report['assets_created']} / upd {report['assets_updated']}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -37,6 +37,7 @@ sys.path.insert(0, str(PROJECT_ROOT))
|
||||
from parser import parse_excel
|
||||
from db_writer import write_seed, write_update, ALL_COLUMNS, PRESERVE_FIELDS, AUTO_FIELDS
|
||||
from backup import backup as do_backup, compare as compare_backup
|
||||
from sync import push as sync_push, TARGETS as SYNC_TARGETS
|
||||
|
||||
|
||||
# ─── Helpers ──────────────────────────────────────────────────────────────────
|
||||
@@ -550,6 +551,43 @@ async def api_db_wipe(db_path: str = Query(DEFAULT_DB_PATH)):
|
||||
raise HTTPException(500, f"Wipe error: {e}")
|
||||
|
||||
|
||||
# ─── Sync / Push to Main App ──────────────────────────────────────────────────
|
||||
|
||||
|
||||
@app.post("/api/sync")
|
||||
async def api_sync(
|
||||
target: str = Query("dev"),
|
||||
dry_run: bool = Query(False),
|
||||
):
|
||||
"""Push seed import data to the main app database."""
|
||||
if target not in SYNC_TARGETS:
|
||||
raise HTTPException(400, f"Unknown target: {target}. Options: {list(SYNC_TARGETS.keys())}")
|
||||
target_db = SYNC_TARGETS[target]["path"]
|
||||
source_db = DEFAULT_DB_PATH
|
||||
report = sync_push(source_db, target_db, dry_run=dry_run)
|
||||
if "error" in report:
|
||||
raise HTTPException(500, report["error"])
|
||||
return report
|
||||
|
||||
|
||||
@app.get("/api/sync/targets")
|
||||
async def api_sync_targets():
|
||||
"""Return available sync targets."""
|
||||
return {
|
||||
"targets": {
|
||||
name: {
|
||||
"path": info["path"],
|
||||
"exists": os.path.exists(info["path"]),
|
||||
}
|
||||
for name, info in SYNC_TARGETS.items()
|
||||
},
|
||||
"source": {
|
||||
"path": DEFAULT_DB_PATH,
|
||||
"exists": os.path.exists(DEFAULT_DB_PATH),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
# ─── Static files ─────────────────────────────────────────────────────────────
|
||||
|
||||
STATIC_DIR = Path(__file__).parent / "static"
|
||||
|
||||
Reference in New Issue
Block a user