shawn/canteen-seed-import
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
canteen-seed-import/sync.py
T
shawn 6fd9238f79 fix: GPS never from seed — only from real check-ins 
GPS (latitude/longitude) and photo_path are now NEVER set by seed
import pipelines, even if empty. GPS comes only from field tech
check-ins via the main app.

- db_writer.py: FIELDS_NEVER_FROM_SEED set, GPS always None in
  _extract_row_values, skipped in update mode
- sync.py: latitude/longitude set to None in asset_data
- FIELD_MAP.md: updated policy table
2026-06-02 22:56:04 -04:00

335 lines
13 KiB
Python
Raw Permalink Blame History

"""
🔄 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()
Reference in New Issue View Git Blame Copy Permalink