""" 🔄 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 * 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": None, # GPS comes only from real check-ins "longitude": None, # GPS comes only from real check-ins "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: # MSFS is primary — seed fills gaps only # Build filtered update dict: only set fields that are currently empty existing_dict = dict(existing) filtered_data = {} for col, val in asset_data.items(): current = existing_dict.get(col) if current is None or str(current).strip() == "": filtered_data[col] = val report["assets_updated"] += 1 if len(report["first_10"]) < 10: skipped_fields = len(asset_data) - len(filtered_data) report["first_10"].append(f"UPDATE {mid}: {name} ({cat}) — {len(filtered_data)} fields, {skipped_fields} skipped (MSFS/cleanup data preserved)") if not dry_run and filtered_data: sets = ", ".join(f"{col} = ?" for col in filtered_data) vals = list(filtered_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()