""" Cantaloupe Sync — staged import pipeline for Cantaloupe machine data. Parses Cantaloupe Excel exports, maps columns to canteen schema, computes diffs against live data, and provides approve/reject workflow. Depends on: - openpyxl for Excel parsing - The shared SQLite DB (same CANTEEN_DB_PATH as admin_server) Exports a FastAPI APIRouter for mounting into admin_server.py. """ import json as _json import logging import os import sqlite3 import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Optional import openpyxl from fastapi import APIRouter, HTTPException, Query, Request, UploadFile, File from pydantic import BaseModel # ─── Replacement models ────────────────────────────────────────────────────── class ReplacementActionItem(BaseModel): new_machine_id: str old_machine_id: str action: str # "approve" or "reject" class ReplacementActionRequest(BaseModel): replacements: list[ReplacementActionItem] logger = logging.getLogger("cantaloupe_sync") # ─── Shared DB access ─────────────────────────────────────────────────────── DB_PATH = os.environ.get("CANTEEN_DB_PATH", str(Path(__file__).parent / "assets.db")) def _get_db() -> sqlite3.Connection: conn = sqlite3.connect(DB_PATH) conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") conn.row_factory = sqlite3.Row return conn # ─── Router ────────────────────────────────────────────────────────────────── router = APIRouter(prefix="/api/admin/cantaloupe", tags=["cantaloupe"]) # ─── Table creation (called from admin_server init_db) ────────────────────── def create_sync_tables(conn: sqlite3.Connection): """Create cantaloupe_sync_batches table if not exists.""" conn.executescript(""" CREATE TABLE IF NOT EXISTS cantaloupe_sync_batches ( id INTEGER PRIMARY KEY AUTOINCREMENT, status TEXT NOT NULL DEFAULT 'pending', created_at TEXT NOT NULL DEFAULT (datetime('now')), approved_at TEXT, file_path TEXT, row_count INTEGER DEFAULT 0, diff_summary TEXT, -- JSON: {new_assets, removed_assets, changed_assets, ...} raw_data TEXT, -- JSON: full export rows error_message TEXT ); CREATE INDEX IF NOT EXISTS idx_csb_status ON cantaloupe_sync_batches(status); CREATE INDEX IF NOT EXISTS idx_csb_created ON cantaloupe_sync_batches(created_at); """) # ─── Pydantic models ──────────────────────────────────────────────────────── class BatchSummary(BaseModel): id: int status: str created_at: str approved_at: Optional[str] = None file_path: Optional[str] = None row_count: int diff_summary: Optional[dict] = None error_message: Optional[str] = None class BatchDetail(BatchSummary): raw_data: Optional[list] = None diff_rows: Optional[dict] = None # {new: [...], removed: [...], changed: [...]} # ─── Column mapping (heuristic) ───────────────────────────────────────────── # Canonical canteen asset columns (excluding auto-generated id/created_at/updated_at) ASSET_COLUMNS = [ "machine_id", "serial_number", "name", "description", "category", "status", "make", "model", "address", "building_name", "building_number", "floor", "room", "trailer_number", "walking_directions", "map_link", "parking_location", "photo_path", "latitude", "longitude", "geofence_radius_meters", "dex_report_date", "deployed", "pulled_date", ] # Heuristic keywords → canonical column. Order matters — first match wins. COLUMN_HEURISTICS = [ # machine_id is the primary key for upserts (["asset id", "assetid", "machine id", "machineid", "machine_id", "equipment id", "unit id", "asset number"], "machine_id"), # Multi-word / specific heuristics first to avoid greedy single-word matches (["location name", "location_name", "site name"], "_location_name"), (["building name", "building_name", "bldg name"], "building_name"), (["building number", "building_number", "building_no", "bldg number", "bldg #", "building #"], "building_number"), (["trailer number", "trailer_number", "trailer_no", "trailer #", "mobile unit"], "trailer_number"), (["walking directions", "walking_directions", "how to get there"], "walking_directions"), (["map link", "map_link", "google maps", "map url", "location url"], "map_link"), (["parking location", "parking_location", "parking spot"], "parking_location"), (["street address", "location address", "site address"], "address"), (["asset name", "equipment name"], "name"), (["model number", "model_no", "model #"], "model"), (["serial number", "serialnumber", "serial_no", "s/n"], "serial_number"), (["room number", "room #", "suite"], "room"), (["customer name", "client name", "account name", "company"], "_customer_name"), (["photo path", "photo_path", "image path"], "photo_path"), (["route number", "route #", "delivery route"], "_route"), (["sales rep", "salesperson", "account manager"], "_sales_rep"), (["qr code", "tag number", "tag #"], "_barcode"), (["geofence radius", "geofence_radius", "geo radius"], "geofence_radius_meters"), (["gps lat", "lat"], "latitude"), (["gps lng", "gps long", "long", "lng", "lon"], "longitude"), (["last dex report time", "dex report time", "last dex", "dex report", "dex time"], "dex_report_date"), (["deployed"], "deployed"), (["pulled date", "pulled", "removal date"], "pulled_date"), # Cantaloupe Machine List export overrides (before generic single-word) (["place"], "name"), # Place → name (breakroom/site name) (["state"], "_state"), # State → prefixed, NOT overwrite status (["class"], "category"), # Class → category (cleaner than Type values) (["type", "asset type", "equipment type"], "_type"), # Type → prefixed (dirty values) # Single-word/generic heuristics (last, after specific multi-word ones) (["serial", "s/n"], "serial_number"), (["category"], "category"), (["status", "condition"], "status"), # removed 'state' to prevent col 33 override (["make", "manufacturer", "brand", "vendor"], "make"), (["address", "street"], "address"), (["floor", "level", "story"], "floor"), (["latitude", "gps"], "latitude"), (["longitude", "geo"], "longitude"), (["geofence", "radius"], "geofence_radius_meters"), (["customer", "client", "account"], "_customer_name"), (["location", "site"], "_location_name"), (["name"], "name"), (["description", "notes", "detail", "comments"], "description"), (["model"], "model"), (["building"], "building_name"), (["room"], "room"), (["trailer"], "trailer_number"), (["walking", "directions"], "walking_directions"), (["map"], "map_link"), (["parking"], "parking_location"), (["photo", "image", "picture"], "photo_path"), (["route"], "_route"), (["sales", "rep"], "_sales_rep"), (["barcode", "tag", "qr"], "_barcode"), ] def _normalize(s: str) -> str: """Lowercase, strip, collapse whitespace. Replace underscores with spaces so 'machine_id' matches 'machine id'.""" return " ".join(str(s).lower().strip().replace("_", " ").replace("#", " ").split()) def map_columns(excel_headers: list[str]) -> dict[str, str]: """ Heuristically map Excel column headers to canonical canteen columns. Returns dict: {excel_column_name: canonical_column_name} """ mapping = {} discovered = [] unmatched = [] for header in excel_headers: norm = _normalize(header) if not norm: continue matched = False norm_words = set(norm.split()) for keywords, canonical in COLUMN_HEURISTICS: for kw in keywords: kw_norm = _normalize(kw) if not kw_norm: continue kw_words = set(kw_norm.split()) # Match: all keyword words must be present in the header. # This prevents short headers like "name" from matching # multi-word keywords like "location name". if kw_words.issubset(norm_words): mapping[header] = canonical discovered.append(f"{header} → {canonical}") matched = True break if matched: break if not matched: unmatched.append(header) # Log discoveries logger.info("Column mapping (%d headers):", len(excel_headers)) for d in discovered: logger.info(" ✓ %s", d) for u in unmatched: logger.info(" ✗ Unmapped: %s", u) return mapping def parse_excel(file_path: str) -> tuple[list[str], list[dict]]: """ Parse Cantaloupe Excel export. Returns (headers, rows) where rows are list of dicts (raw Excel values). """ wb = openpyxl.load_workbook(file_path, read_only=True, data_only=True) ws = wb.active rows_iter = ws.iter_rows(values_only=True) try: headers_raw = next(rows_iter) except StopIteration: wb.close() raise HTTPException(status_code=400, detail="Excel file is empty (no header row)") headers = [str(h).strip() if h is not None else f"__col_{i}" for i, h in enumerate(headers_raw)] data_rows = [] for row in rows_iter: if all(v is None or str(v).strip() == "" for v in row): continue # skip empty rows row_dict = {} for i, val in enumerate(row): if i < len(headers): # Convert to native Python types if isinstance(val, datetime): # Use space-separated format so SQLite comparison works # datetime('now') outputs '2026-05-18 22:04:00' (space) # .isoformat() outputs '2026-05-18T22:04:00' (T) which # breaks lexicographic comparison ('T' > ' ' in ASCII) val = val.strftime('%Y-%m-%d %H:%M:%S') elif val is not None and not isinstance(val, (str, int, float, bool)): val = str(val) row_dict[headers[i]] = val if val is not None else "" else: row_dict[f"__col_{i}"] = val if val is not None else "" data_rows.append(row_dict) wb.close() return headers, data_rows def apply_mapping(rows: list[dict], column_map: dict[str, str]) -> list[dict]: """ Apply column mapping to transform Excel rows into canteen-shaped dicts. Returns list of dicts with canonical column keys. Extra/custom fields are preserved with a `_raw_` prefix. """ mapped = [] for row in rows: item = {} for excel_col, value in row.items(): canonical = column_map.get(excel_col) if canonical: # If multiple Excel columns map to same canonical, first wins if canonical not in item: item[canonical] = value else: # Preserve unmapped columns item[f"_raw_{excel_col}"] = value mapped.append(item) return mapped # ─── Diff engine ──────────────────────────────────────────────────────────── def _fetch_live_data(conn: sqlite3.Connection) -> dict[str, list[dict]]: """Fetch all current assets, customers, locations from live DB.""" assets = [dict(r) for r in conn.execute("SELECT * FROM assets ORDER BY machine_id").fetchall()] customers = [dict(r) for r in conn.execute("SELECT * FROM customers ORDER BY name").fetchall()] locations = [dict(r) for r in conn.execute("SELECT * FROM locations ORDER BY id").fetchall()] return {"assets": assets, "customers": customers, "locations": locations} def detect_replacements( imported_rows: list[dict], live_assets: list[dict], live_customers: list[dict], ) -> dict: """ Detect potential machine replacements where a new Asset ID appears at the same customer + address as an existing asset not in the import. Confidence levels: - "high": exact address match (same street address string) - "medium": same customer + same building_name, or same street name """ import re # Build lookups live_assets_by_mid = {str(a.get("machine_id", "")).strip(): a for a in live_assets} imported_mids = set() # customer name (lower) → id cust_name_to_id: dict[str, int] = {} for c in live_customers: cname = str(c.get("name", "")).strip() if cname: cust_name_to_id[cname.lower()] = c["id"] # Collect all imported machine IDs for row in imported_rows: mid = str(row.get("machine_id", "")).strip() if mid: imported_mids.add(mid) replacements: list[dict] = [] seen_pairs: set[tuple[str, str]] = set() for row in imported_rows: new_mid = str(row.get("machine_id", "")).strip() if not new_mid: continue if new_mid in live_assets_by_mid: continue # already exists, not a replacement cust_name = str(row.get("_customer_name", "")).strip() addr = str(row.get("address", "")).strip() bldg = str(row.get("building_name", "")).strip() if not cust_name: continue cust_id = cust_name_to_id.get(cust_name.lower()) if cust_id is None: continue # unknown customer, can't match # Scan live assets at the same customer, not in the import for old_mid, live in live_assets_by_mid.items(): if old_mid == new_mid: continue if live.get("customer_id") != cust_id: continue if old_mid in imported_mids: continue # this asset IS in the import, not a leftover # Avoid duplicate pairs pair = (new_mid, old_mid) if pair in seen_pairs: continue live_addr = str(live.get("address", "")).strip() live_bldg = str(live.get("building_name", "")).strip() confidence = None # High: exact address match if addr and live_addr and addr.lower() == live_addr.lower(): confidence = "high" # Medium: same building_name elif bldg and live_bldg and bldg.lower() == live_bldg.lower(): confidence = "medium" # Medium: same street name (numbers and suffixes stripped) elif addr and live_addr and _same_street(addr, live_addr): confidence = "medium" if confidence: seen_pairs.add(pair) replacements.append({ "id": len(replacements) + 1, "new_machine_id": new_mid, "old_machine_id": old_mid, "location_address": addr or live_addr or "", "customer_name": cust_name, "old_asset_name": str(live.get("name", "")).strip(), "new_asset_name": str(row.get("name", "")).strip(), "confidence": confidence, }) return {"replacements": replacements} def _same_street(addr1: str, addr2: str) -> bool: """Check if two addresses share the same street name, ignoring numbers and suffixes.""" import re suffixes = { "street", "st", "drive", "dr", "road", "rd", "avenue", "ave", "blvd", "boulevard", "lane", "ln", "way", "court", "ct", "circle", "cir", "place", "pl", "highway", "hwy", "suite", "ste", "unit", "floor", "fl", "build", "bldg", } a1 = re.sub(r"\d+", "", addr1.lower()) a2 = re.sub(r"\d+", "", addr2.lower()) for suffix in suffixes: a1 = re.sub(rf"\b{suffix}\b", "", a1) a2 = re.sub(rf"\b{suffix}\b", "", a2) a1 = re.sub(r"\s+", " ", a1).strip().strip(",").strip("#").strip() a2 = re.sub(r"\s+", " ", a2).strip().strip(",").strip("#").strip() return bool(a1 and a2 and a1 == a2) def compute_diff(imported_rows: list[dict], live_data: dict) -> dict: """ Compare imported Cantaloupe rows against live data. Returns diff dict: { "new_assets": [...], # machine_ids not in live DB "removed_assets": [...], # in live DB but not in import "changed_assets": [...], # both exist but fields differ "unchanged_assets": [...], # match exactly "new_customers": [...], "new_locations": [...], "import_count": N, "live_asset_count": N, } Each entry in changed_assets: {machine_id, field, old_value, new_value} """ live_assets = {a["machine_id"]: a for a in live_data["assets"]} live_customers = {c["name"].lower(): c for c in live_data["customers"]} imported_machine_ids = set() new_assets = [] changed_assets = [] unchanged_assets = [] new_customers = [] new_locations = [] # Fields to compare for assets (skip id, created_at, updated_at, customer_id, location_id) COMPARE_FIELDS = [ "serial_number", "name", "description", "category", "status", "make", "model", "address", "building_name", "building_number", "floor", "room", "trailer_number", "walking_directions", "map_link", "parking_location", "photo_path", "latitude", "longitude", "geofence_radius_meters", "dex_report_date", "deployed", "pulled_date", ] for row in imported_rows: mid = str(row.get("machine_id", "")).strip() if not mid: continue # skip rows without a machine_id imported_machine_ids.add(mid) # Check for new customer cust_name = str(row.get("_customer_name", "")).strip() if cust_name and cust_name.lower() not in live_customers: if cust_name not in new_customers: new_customers.append(cust_name) # Check for new location loc_name = str(row.get("_location_name", "")).strip() if loc_name: # locations are matched by address+name later in approve step; # just flag for now new_locations.append(loc_name) if mid in live_assets: live = live_assets[mid] changes = [] for field in COMPARE_FIELDS: old_val = live.get(field) new_val = row.get(field) # Normalize comparison old_str = str(old_val).strip() if old_val is not None else "" new_str = str(new_val).strip() if new_val is not None else "" if old_str != new_str: changes.append({ "field": field, "old_value": old_val, "new_value": new_val, }) if changes: changed_assets.append({ "machine_id": mid, "name": live.get("name", ""), "changes": changes, }) else: unchanged_assets.append(mid) else: new_assets.append({ "machine_id": mid, "name": str(row.get("name", "")).strip(), "customer": str(row.get("_customer_name", "")).strip(), "location": str(row.get("_location_name", "")).strip(), }) # Removed: in live but not in import removed_assets = [ {"machine_id": mid, "name": a.get("name", "")} for mid, a in live_assets.items() if mid not in imported_machine_ids ] return { "new_assets": new_assets, "removed_assets": removed_assets, "changed_assets": changed_assets, "unchanged_assets": unchanged_assets, "new_customers": list(set(new_customers)), "new_locations": list(set(new_locations)), "import_count": len(imported_rows), "live_asset_count": len(live_assets), } # ─── Route: Run sync ──────────────────────────────────────────────────────── @router.post("/sync") async def run_sync( request: Request, file: Optional[UploadFile] = File(None), file_path: Optional[str] = Query(None, description="Server-side path to Excel file"), ): """ Import an Excel file into a staged batch for review. Accepts either: - An uploaded Excel file (multipart form) - A server-side file path (query param) Parses the Excel, maps columns to canteen schema, computes diffs, stores everything in a new cantaloupe_sync_batches row with status='pending'. """ if file and file.filename: # Save uploaded file to temp suffix = Path(file.filename).suffix or ".xlsx" tmp = tempfile.NamedTemporaryFile(delete=False, suffix=suffix) try: content = await file.read() tmp.write(content) tmp.close() excel_path = tmp.name display_path = f"upload:{file.filename}" except Exception as e: tmp.close() Path(tmp.name).unlink(missing_ok=True) raise HTTPException(status_code=400, detail=f"Failed to read uploaded file: {e}") elif file_path: p = Path(file_path) if not p.exists(): raise HTTPException(status_code=400, detail=f"File not found: {file_path}") excel_path = str(p) display_path = file_path else: raise HTTPException( status_code=400, detail="No file provided. Upload an Excel file to import asset data." ) conn = _get_db() try: # Parse Excel try: headers, raw_rows = parse_excel(excel_path) except Exception as e: raise HTTPException(status_code=400, detail=f"Failed to parse Excel: {e}") if not raw_rows: raise HTTPException(status_code=400, detail="Excel file contains no data rows") # Map columns column_map = map_columns(headers) mapped_rows = apply_mapping(raw_rows, column_map) # Normalize categories: strip "GF " prefix so "GF Food" → "Food", etc. for row in mapped_rows: cat = row.get("category", "") if cat and cat.startswith("GF "): row["category"] = cat[3:] # Compute diff live_data = _fetch_live_data(conn) diff = compute_diff(mapped_rows, live_data) # Detect machine replacements replacements = detect_replacements( mapped_rows, live_data["assets"], live_data["customers"] ) diff["replacements"] = replacements.get("replacements", []) # Store batch raw_json = _json.dumps(mapped_rows, default=str, ensure_ascii=False) diff_json = _json.dumps(diff, default=str, ensure_ascii=False) cursor = conn.execute( """INSERT INTO cantaloupe_sync_batches (status, file_path, row_count, diff_summary, raw_data) VALUES ('pending', ?, ?, ?, ?)""", (display_path, len(raw_rows), diff_json, raw_json), ) conn.commit() batch_id = cursor.lastrowid # Fetch back row = conn.execute( "SELECT * FROM cantaloupe_sync_batches WHERE id = ?", (batch_id,) ).fetchone() result = _batch_to_dict(row) logger.info( "Sync batch %d created: %d rows, %d new, %d changed, %d removed, %d unchanged", batch_id, diff["import_count"], len(diff["new_assets"]), len(diff["changed_assets"]), len(diff["removed_assets"]), len(diff["unchanged_assets"]), ) return result finally: conn.close() # Clean up temp file if we created one if file and file.filename and excel_path: Path(excel_path).unlink(missing_ok=True) # ─── Helper: serialize batch row ──────────────────────────────────────────── def _batch_to_dict(row: sqlite3.Row) -> dict: d = dict(row) # Parse JSON fields for field in ("diff_summary", "raw_data"): if isinstance(d.get(field), str): try: d[field] = _json.loads(d[field]) except (_json.JSONDecodeError, TypeError): pass return d # ─── Route: List batches ──────────────────────────────────────────────────── @router.get("/batches") def list_batches( status: Optional[str] = Query(None, description="Filter by status"), limit: int = Query(50, ge=1, le=500), offset: int = Query(0, ge=0), ): """List all sync batches, newest first.""" conn = _get_db() try: where = "" params = [] if status: where = "WHERE status = ?" params.append(status) rows = conn.execute( f"SELECT id, status, created_at, approved_at, file_path, row_count, diff_summary, error_message " f"FROM cantaloupe_sync_batches {where} " f"ORDER BY created_at DESC, id DESC LIMIT ? OFFSET ?", params + [limit, offset], ).fetchall() return [_batch_to_dict(r) for r in rows] finally: conn.close() # ─── Route: Get batch detail ──────────────────────────────────────────────── @router.get("/batches/{batch_id}") def get_batch(batch_id: int): """Get full batch detail including row-level diff and raw imported data.""" conn = _get_db() try: row = conn.execute( "SELECT * FROM cantaloupe_sync_batches WHERE id = ?", (batch_id,) ).fetchone() if row is None: raise HTTPException(status_code=404, detail="Batch not found") batch = _batch_to_dict(row) # Compute row-level diff display diff = batch.get("diff_summary", {}) if isinstance(batch.get("diff_summary"), dict) else {} batch["diff_rows"] = { "new": diff.get("new_assets", []), "removed": diff.get("removed_assets", []), "changed": diff.get("changed_assets", []), } # Replacement stats replacements = diff.get("replacements", []) if isinstance(diff, dict) else [] batch["replacement_count"] = len(replacements) batch["unresolved_replacements"] = len([r for r in replacements if not r.get("status")]) # Include raw data (imported rows) raw = batch.get("raw_data") if isinstance(raw, str): try: raw = _json.loads(raw) except _json.JSONDecodeError: pass batch["raw_data"] = raw return batch finally: conn.close() # ─── Route: Approve batch ─────────────────────────────────────────────────── @router.post("/batches/{batch_id}/approve") def approve_batch( batch_id: int, request: Request, force: bool = Query(False, description="Force approval even with unresolved replacements"), ): """ Apply staged data to live tables: - Upsert assets (match on machine_id) - Upsert customers (match on name) - Upsert locations (match on address + building_name + building_number) Checks for unresolved replacement candidates and blocks unless forced. Sets status='approved' and records approved_at. """ conn = _get_db() try: row = conn.execute( "SELECT * FROM cantaloupe_sync_batches WHERE id = ?", (batch_id,) ).fetchone() if row is None: raise HTTPException(status_code=404, detail="Batch not found") if row["status"] != "pending": raise HTTPException( status_code=400, detail=f"Cannot approve batch in status '{row['status']}'. Only 'pending' batches can be approved.", ) # Check for unresolved replacements diff = row["diff_summary"] if isinstance(diff, str): try: diff = _json.loads(diff) except _json.JSONDecodeError: diff = {} if isinstance(diff, dict): replacements = diff.get("replacements", []) unresolved = [r for r in replacements if not r.get("status")] if unresolved and not force: raise HTTPException( status_code=409, detail=f"Cannot approve: {len(unresolved)} replacement(s) are unresolved. Resolve them first or use ?force=true.", ) # Parse raw data raw_data = row["raw_data"] if isinstance(raw_data, str): raw_data = _json.loads(raw_data) if not raw_data: raise HTTPException(status_code=400, detail="Batch has no raw data to apply") imported_rows = raw_data if isinstance(raw_data, list) else [] # Collect stats stats = { "assets_upserted": 0, "customers_created": 0, "locations_created": 0, } # Track customer name → id mapping customer_cache = {} for item in imported_rows: machine_id = str(item.get("machine_id", "")).strip() if not machine_id: continue # 1. Upsert customer (match on name) cust_name = str(item.get("_customer_name", "")).strip() cust_id = None if cust_name: if cust_name in customer_cache: cust_id = customer_cache[cust_name] else: cust_row = conn.execute( "SELECT id FROM customers WHERE LOWER(name) = LOWER(?)", (cust_name,), ).fetchone() if cust_row: cust_id = cust_row["id"] else: cursor = conn.execute( "INSERT INTO customers (name) VALUES (?)", (cust_name,) ) cust_id = cursor.lastrowid stats["customers_created"] += 1 customer_cache[cust_name] = cust_id # 2. Upsert location (match on address + building_name + building_number) loc_name = str(item.get("_location_name", "")).strip() loc_id = None if loc_name: loc_row = conn.execute( "SELECT id FROM locations WHERE LOWER(name) = LOWER(?)", (loc_name,), ).fetchone() if loc_row: loc_id = loc_row["id"] else: cursor = conn.execute( "INSERT INTO locations (name, customer_id) VALUES (?, ?)", (loc_name, cust_id), ) loc_id = cursor.lastrowid stats["locations_created"] += 1 # Also try matching on address addr = str(item.get("address", "")).strip() if addr and not loc_id: loc_row = conn.execute( "SELECT id FROM locations WHERE LOWER(address) = LOWER(?) AND address != ''", (addr,), ).fetchone() if loc_row: loc_id = loc_row["id"] # 3. Upsert asset (match on machine_id) existing = conn.execute( "SELECT id FROM assets WHERE machine_id = ?", (machine_id,) ).fetchone() if existing: # Update existing asset updates = {} field_map = { "serial_number": "serial_number", "name": "name", "description": "description", "category": "category", "status": "status", "make": "make", "model": "model", "address": "address", "building_name": "building_name", "building_number": "building_number", "floor": "floor", "room": "room", "trailer_number": "trailer_number", "walking_directions": "walking_directions", "map_link": "map_link", "parking_location": "parking_location", "photo_path": "photo_path", "dex_report_date": "dex_report_date", "deployed": "deployed", "pulled_date": "pulled_date", } for import_key, db_col in field_map.items(): val = item.get(import_key) if val is not None and str(val).strip() != "": updates[db_col] = str(val).strip() # Handle numeric fields (exclude GPS — comes only from real check-ins) for num_field in ("geofence_radius_meters",): val = item.get(num_field) if val is not None and str(val).strip() != "": try: updates[num_field] = float(val) except (ValueError, TypeError): pass if cust_id: updates["customer_id"] = cust_id if loc_id: updates["location_id"] = loc_id # Set Disney flag based on customer name if cust_name: updates["is_disney"] = 1 if cust_name.startswith("D-") else 0 if updates: updates["updated_at"] = "datetime('now')" set_clause = ", ".join( f"{k} = {v}" if k == "updated_at" else f"{k} = ?" for k, v in updates.items() ) values = [v for k, v in updates.items() if k != "updated_at"] conn.execute( f"UPDATE assets SET {set_clause} WHERE id = ?", values + [existing["id"]], ) else: # Insert new asset name = str(item.get("name", machine_id)).strip() category = str(item.get("category", "Other")).strip() status = str(item.get("status", "active")).strip() # Validate enums against DB categories table VALID_STATUSES = {"active", "maintenance", "retired"} try: valid_cats = {r[0] for r in conn.execute( "SELECT name FROM categories").fetchall()} except Exception: valid_cats = {"Other"} if category not in valid_cats: category = "Other" if status not in VALID_STATUSES: status = "active" conn.execute( """INSERT INTO assets (machine_id, serial_number, name, description, category, status, make, model, address, building_name, building_number, floor, room, trailer_number, walking_directions, map_link, parking_location, photo_path, customer_id, location_id, is_disney, latitude, longitude, geofence_radius_meters, dex_report_date, deployed, pulled_date) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", ( machine_id, str(item.get("serial_number", "")).strip() or "", name, str(item.get("description", "")).strip() or "", category, status, str(item.get("make", "")).strip() or "", str(item.get("model", "")).strip() or "", str(item.get("address", "")).strip() or "", str(item.get("building_name", "")).strip() or "", str(item.get("building_number", "")).strip() or "", str(item.get("floor", "")).strip() or "", str(item.get("room", "")).strip() or "", str(item.get("trailer_number", "")).strip() or "", str(item.get("walking_directions", "")).strip() or "", str(item.get("map_link", "")).strip() or "", str(item.get("parking_location", "")).strip() or "", str(item.get("photo_path", "")).strip() or "", cust_id, loc_id, 1 if cust_name and cust_name.startswith("D-") else 0, None, # GPS only from real check-ins None, # GPS only from real check-ins _safe_int(item.get("geofence_radius_meters"), 50), str(item.get("dex_report_date", "")).strip() or "", str(item.get("deployed", "")).strip() or "", str(item.get("pulled_date", "")).strip() or "", ), ) stats["assets_upserted"] += 1 # Mark batch as approved approved_at = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") conn.execute( "UPDATE cantaloupe_sync_batches SET status = 'approved', approved_at = ? WHERE id = ?", (approved_at, batch_id), ) conn.commit() # Log activity user_id = getattr(request.state, "user_id", None) conn.execute( "INSERT INTO activity_log (user_id, action, entity_type, entity_id, details) " "VALUES (?, 'approved', 'cantaloupe_batch', ?, ?)", (user_id, batch_id, _json.dumps(stats)), ) # Fetch updated batch updated = conn.execute( "SELECT * FROM cantaloupe_sync_batches WHERE id = ?", (batch_id,) ).fetchone() result = _batch_to_dict(updated) result["applied_stats"] = stats return result except HTTPException: raise except Exception as e: conn.rollback() raise HTTPException(status_code=500, detail=f"Approve failed: {e}") finally: conn.close() def _safe_float(val) -> Optional[float]: """Convert to float, return None on failure.""" if val is None: return None try: return float(val) except (ValueError, TypeError): return None def _safe_int(val, default: Optional[int] = None) -> Optional[int]: """Convert to int, return default on failure.""" if val is None: return default try: return int(float(val)) except (ValueError, TypeError): return default # ─── Route: Reject batch ──────────────────────────────────────────────────── @router.post("/batches/{batch_id}/reject") def reject_batch(batch_id: int, request: Request): """ Reject a pending batch: - Sets status='rejected' - Clears raw_data to save space """ conn = _get_db() try: row = conn.execute( "SELECT * FROM cantaloupe_sync_batches WHERE id = ?", (batch_id,) ).fetchone() if row is None: raise HTTPException(status_code=404, detail="Batch not found") if row["status"] != "pending": raise HTTPException( status_code=400, detail=f"Cannot reject batch in status '{row['status']}'. Only 'pending' batches can be rejected.", ) # Clear raw_data, keep diff_summary for audit trail conn.execute( "UPDATE cantaloupe_sync_batches SET status = 'rejected', raw_data = NULL WHERE id = ?", (batch_id,), ) # Log activity user_id = getattr(request.state, "user_id", None) conn.execute( "INSERT INTO activity_log (user_id, action, entity_type, entity_id, details) " "VALUES (?, 'rejected', 'cantaloupe_batch', ?, 'Batch rejected')", (user_id, batch_id), ) conn.commit() updated = conn.execute( "SELECT * FROM cantaloupe_sync_batches WHERE id = ?", (batch_id,) ).fetchone() return _batch_to_dict(updated) except HTTPException: raise except Exception as e: conn.rollback() raise HTTPException(status_code=500, detail=f"Reject failed: {e}") finally: conn.close() # ─── Route: Get replacements ──────────────────────────────────────────────── @router.get("/batches/{batch_id}/replacements") def get_replacements(batch_id: int): """Return replacement candidates for a sync batch.""" conn = _get_db() try: row = conn.execute( "SELECT diff_summary FROM cantaloupe_sync_batches WHERE id = ?", (batch_id,), ).fetchone() if row is None: raise HTTPException(status_code=404, detail="Batch not found") diff = row["diff_summary"] if isinstance(diff, str): try: diff = _json.loads(diff) except _json.JSONDecodeError: diff = {} if not isinstance(diff, dict): diff = {} replacements = diff.get("replacements", []) return {"batch_id": batch_id, "replacements": replacements} finally: conn.close() # ─── Route: Resolve replacements ──────────────────────────────────────────── @router.post("/batches/{batch_id}/replacements") def resolve_replacements(batch_id: int, body: ReplacementActionRequest): """Approve or reject individual replacement candidates. Approve: retires the old asset (status='retired') and sets install_date = today on the new asset. Reject: records the decision without modifying assets. """ conn = _get_db() try: batch_row = conn.execute( "SELECT * FROM cantaloupe_sync_batches WHERE id = ?", (batch_id,) ).fetchone() if batch_row is None: raise HTTPException(status_code=404, detail="Batch not found") if batch_row["status"] != "pending": raise HTTPException( status_code=400, detail=f"Cannot resolve replacements in status '{batch_row['status']}'. Only 'pending' batches can be modified.", ) # Parse diff_summary diff = batch_row["diff_summary"] if isinstance(diff, str): try: diff = _json.loads(diff) except _json.JSONDecodeError: diff = {} if not isinstance(diff, dict): diff = {} replacements = diff.get("replacements", []) today = datetime.now(timezone.utc).strftime("%Y-%m-%d") resolved = [] # Build index of replacements by (new_mid, old_mid) repl_index: dict[tuple[str, str], dict] = {} for r in replacements: key = (str(r.get("new_machine_id", "")).strip(), str(r.get("old_machine_id", "")).strip()) repl_index[key] = r for item in body.replacements: key = (str(item.new_machine_id).strip(), str(item.old_machine_id).strip()) repl = repl_index.get(key) if repl is None: raise HTTPException( status_code=404, detail=f"Replacement not found: new={item.new_machine_id} old={item.old_machine_id}", ) if item.action not in ("approve", "reject"): raise HTTPException( status_code=400, detail=f"Invalid action '{item.action}'. Must be 'approve' or 'reject'.", ) if item.action == "approve": # Retire the old asset old_asset = conn.execute( "SELECT id, name, description FROM assets WHERE machine_id = ?", (item.old_machine_id,), ).fetchone() if old_asset: old_desc = (old_asset["description"] or "").strip() link_note = f"[Replaced by {item.new_machine_id} on {today}]" new_desc = (old_desc + "\n" + link_note).strip() if old_desc else link_note conn.execute( "UPDATE assets SET status = 'retired', description = ?, updated_at = datetime('now') WHERE id = ?", (new_desc, old_asset["id"]), ) # Set install_date on the new asset new_asset = conn.execute( "SELECT id, name, description FROM assets WHERE machine_id = ?", (item.new_machine_id,), ).fetchone() if new_asset: new_desc = (new_asset["description"] or "").strip() link_note = f"[Replaced {item.old_machine_id} on {today}]" new_desc = (new_desc + "\n" + link_note).strip() if new_desc else link_note conn.execute( "UPDATE assets SET install_date = ?, description = ?, updated_at = datetime('now') WHERE id = ?", (today, new_desc, new_asset["id"]), ) repl["status"] = "approved" else: repl["status"] = "rejected" repl["resolved_at"] = today resolved.append(repl) # Persist updated replacements in diff_summary diff["replacements"] = replacements conn.execute( "UPDATE cantaloupe_sync_batches SET diff_summary = ? WHERE id = ?", (_json.dumps(diff, default=str, ensure_ascii=False), batch_id), ) conn.commit() return {"batch_id": batch_id, "replacements": replacements, "resolved_count": len(body.replacements)} except HTTPException: raise except Exception as e: conn.rollback() raise HTTPException(status_code=500, detail=f"Resolve replacements failed: {e}") finally: conn.close() # ─── Per-field granular approval ────────────────────────────────────────────── class FieldChangeItem(BaseModel): machine_id: str field: str class PerFieldApprovalRequest(BaseModel): changes: list[FieldChangeItem] @router.get("/batches/{batch_id}/field-changes") def get_field_changes(batch_id: int): """Return flat list of per-field changes for UI rendering. Each entry: {machine_id, name, field, old_value, new_value, is_blank_fill} """ conn = _get_db() try: row = conn.execute( "SELECT diff_summary FROM cantaloupe_sync_batches WHERE id = ?", (batch_id,), ).fetchone() if row is None: raise HTTPException(status_code=404, detail="Batch not found") diff = row["diff_summary"] if isinstance(diff, str): try: diff = _json.loads(diff) except _json.JSONDecodeError: diff = {} if not isinstance(diff, dict): diff = {} changed = diff.get("changed_assets", []) result = [] for asset in changed: mid = asset.get("machine_id", "") name = asset.get("name", "") for ch in asset.get("changes", []): old_val = ch.get("old_value") new_val = ch.get("new_value") is_blank = new_val is None or str(new_val).strip() == "" result.append({ "machine_id": mid, "name": name, "field": ch.get("field", ""), "old_value": old_val, "new_value": new_val, "is_blank_fill": is_blank, }) # Also add new assets (field-level doesn't apply to inserts) new_assets = diff.get("new_assets", []) replacements = diff.get("replacements", []) return { "batch_id": batch_id, "field_changes": result, "new_asset_count": len(new_assets), "replacement_count": len(replacements), "import_count": diff.get("import_count", 0), "live_asset_count": diff.get("live_asset_count", 0), } finally: conn.close() @router.post("/batches/{batch_id}/apply-field-changes") def apply_field_changes(batch_id: int, body: PerFieldApprovalRequest, request: Request): """Apply only the selected field changes to existing assets. New assets and replacements are NOT handled by this endpoint — use the existing approve/replacement endpoints for those. Body: {changes: [{machine_id: str, field: str}, ...]} Only fields listed are applied (old_value→new_value for existing assets). """ conn = _get_db() try: row = conn.execute( "SELECT * FROM cantaloupe_sync_batches WHERE id = ?", (batch_id,), ).fetchone() if row is None: raise HTTPException(status_code=404, detail="Batch not found") if row["status"] != "pending": raise HTTPException( status_code=400, detail=f"Cannot apply changes in status '{row['status']}'. Only 'pending' batches can be modified.", ) # Parse diff and raw data diff = row["diff_summary"] if isinstance(diff, str): try: diff = _json.loads(diff) except _json.JSONDecodeError: diff = {} if not isinstance(diff, dict): diff = {} raw_data = row["raw_data"] if isinstance(raw_data, str): raw_data = _json.loads(raw_data) if not raw_data: raise HTTPException(status_code=400, detail="Batch has no raw data") # Index imported rows by machine_id imported_by_mid: dict[str, dict] = {} for item in raw_data: mid = str(item.get("machine_id", "")).strip() if mid: imported_by_mid[mid] = item stats = {"assets_updated": 0, "fields_applied": 0} # Group approved changes by machine_id approved_by_mid: dict[str, set[str]] = {} for change in body.changes: mid = change.machine_id.strip() if mid not in approved_by_mid: approved_by_mid[mid] = set() approved_by_mid[mid].add(change.field) for mid, approved_fields in approved_by_mid.items(): existing = conn.execute( "SELECT id FROM assets WHERE machine_id = ?", (mid,), ).fetchone() if not existing: continue # skip new assets (handled by approve endpoint) imported = imported_by_mid.get(mid) if not imported: continue updates = {} for field in approved_fields: val = imported.get(field) if val is not None and str(val).strip() != "": updates[field] = str(val).strip() if not updates: continue updates["updated_at"] = "datetime('now')" set_clause = ", ".join( f"{k} = {v}" if k == "updated_at" else f"{k} = ?" for k, v in updates.items() ) values = [v for k, v in updates.items() if k != "updated_at"] conn.execute( f"UPDATE assets SET {set_clause} WHERE id = ?", values + [existing["id"]], ) stats["assets_updated"] += 1 stats["fields_applied"] += len(approved_fields) conn.commit() # Log activity user_id = getattr(request.state, "user_id", None) conn.execute( "INSERT INTO activity_log (user_id, action, entity_type, entity_id, details) " "VALUES (?, 'apply_field_changes', 'cantaloupe_batch', ?, ?)", (user_id, batch_id, _json.dumps(stats)), ) conn.commit() return { "batch_id": batch_id, "applied_stats": stats, "message": f"Applied {stats['fields_applied']} field change(s) across {stats['assets_updated']} asset(s).", } except HTTPException: raise except Exception as e: conn.rollback() raise HTTPException(status_code=500, detail=f"Apply field changes failed: {e}") finally: conn.close()