feat: Add Cantaloupe sync API with staged import workflow
- New cantaloupe_sync.py module with complete pipeline - Heuristic Excel column mapping (20+ keyword patterns) - Diff engine: detects new/changed/removed/unchanged assets - 5 API endpoints: sync, list batches, view batch, approve, reject - cantaloupe_sync_batches table with status workflow - Admin server port changed to 8090 - openpyxl added to requirements.txt
This commit is contained in:
Leo
@@ -0,0 +1,870 @@
|
||||
"""
|
||||
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
|
||||
|
||||
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",
|
||||
]
|
||||
|
||||
# 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"),
|
||||
(["serial", "serial number", "serialnumber", "serial_no", "s/n"], "serial_number"),
|
||||
(["name", "asset name", "equipment name", "description"], "name"),
|
||||
(["description", "notes", "detail", "comments"], "description"),
|
||||
(["category", "type", "asset type", "equipment type", "class"], "category"),
|
||||
(["status", "state", "condition"], "status"),
|
||||
(["make", "manufacturer", "brand", "vendor"], "make"),
|
||||
(["model", "model number", "model_no", "model #"], "model"),
|
||||
(["address", "street", "street address", "location address", "site address"], "address"),
|
||||
(["building name", "building_name", "bldg name", "site name"], "building_name"),
|
||||
(["building number", "building_number", "building_no", "bldg number", "bldg #", "building #"], "building_number"),
|
||||
(["floor", "level", "story"], "floor"),
|
||||
(["room", "room number", "room #", "suite"], "room"),
|
||||
(["trailer", "trailer number", "trailer_no", "trailer #", "mobile unit"], "trailer_number"),
|
||||
(["walking directions", "walking_directions", "directions", "how to get there"], "walking_directions"),
|
||||
(["map link", "map_link", "google maps", "map url", "location url"], "map_link"),
|
||||
(["parking", "parking location", "parking_location", "parking spot"], "parking_location"),
|
||||
(["latitude", "lat", "gps lat"], "latitude"),
|
||||
(["longitude", "long", "lng", "lon", "gps lng", "gps long"], "longitude"),
|
||||
(["geofence", "geo radius", "radius", "geofence radius", "geofence_radius"], "geofence_radius_meters"),
|
||||
(["customer", "client", "account", "company", "customer name"], "_customer_name"),
|
||||
(["location name", "location_name", "site", "site name", "location"], "_location_name"),
|
||||
(["photo", "photo path", "photo_path", "image", "picture"], "photo_path"),
|
||||
(["route", "route number", "route #", "delivery route"], "_route"),
|
||||
(["sales", "sales rep", "salesperson", "account manager"], "_sales_rep"),
|
||||
(["barcode", "qr code", "tag", "tag number", "tag #"], "_barcode"),
|
||||
]
|
||||
|
||||
|
||||
def _normalize(s: str) -> str:
|
||||
"""Lowercase, strip, collapse whitespace."""
|
||||
return " ".join(str(s).lower().strip().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
|
||||
for keywords, canonical in COLUMN_HEURISTICS:
|
||||
for kw in keywords:
|
||||
# Match if the header contains the keyword or vice versa
|
||||
if kw in norm or norm in kw:
|
||||
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):
|
||||
val = val.isoformat()
|
||||
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 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",
|
||||
]
|
||||
|
||||
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"),
|
||||
):
|
||||
"""
|
||||
Run a full Cantaloupe sync from an Excel file.
|
||||
|
||||
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:
|
||||
# Try subprocess call to cantaloupe export
|
||||
try:
|
||||
import subprocess
|
||||
result = subprocess.run(
|
||||
["python", "-m", "cantaloupe", "export", "--scheduled"],
|
||||
capture_output=True, text=True, timeout=120,
|
||||
cwd=str(Path(__file__).parent),
|
||||
)
|
||||
if result.returncode != 0:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail=f"Cantaloupe export failed: {result.stderr[:500]}"
|
||||
)
|
||||
# Parse output to find the exported file path
|
||||
for line in result.stdout.splitlines() + result.stderr.splitlines():
|
||||
line = line.strip()
|
||||
if line and Path(line).exists() and line.endswith((".xlsx", ".xls")):
|
||||
excel_path = line
|
||||
display_path = line
|
||||
break
|
||||
else:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="Cantaloupe export ran but produced no recognizable file path. Provide file upload or file_path parameter."
|
||||
)
|
||||
except FileNotFoundError:
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
detail="No file provided and cantaloupe module not found. Upload an Excel file or provide file_path."
|
||||
)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
raise HTTPException(status_code=500, detail=f"Cantaloupe export error: {e}")
|
||||
|
||||
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)
|
||||
|
||||
# Compute diff
|
||||
live_data = _fetch_live_data(conn)
|
||||
diff = compute_diff(mapped_rows, live_data)
|
||||
|
||||
# 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", []),
|
||||
}
|
||||
|
||||
# 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):
|
||||
"""
|
||||
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)
|
||||
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.",
|
||||
)
|
||||
|
||||
# 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",
|
||||
}
|
||||
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
|
||||
for num_field in ("latitude", "longitude", "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
|
||||
|
||||
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
|
||||
VALID_CATEGORIES = {"Furniture", "Appliances", "Utensils & Serveware", "Equipment", "Other"}
|
||||
VALID_STATUSES = {"active", "maintenance", "retired"}
|
||||
if category not in VALID_CATEGORIES:
|
||||
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,
|
||||
latitude, longitude, geofence_radius_meters)
|
||||
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,
|
||||
_safe_float(item.get("latitude")),
|
||||
_safe_float(item.get("longitude")),
|
||||
_safe_int(item.get("geofence_radius_meters"), 50),
|
||||
),
|
||||
)
|
||||
|
||||
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()
|
||||
Reference in New Issue
Block a user