canteen-admin-server/cantaloupe_sync.py

"""
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"),
    # 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"),
    # Single-word/generic heuristics (last, after specific multi-word ones)
    (["serial", "s/n"], "serial_number"),
    (["category", "type", "asset type", "equipment type", "class"], "category"),
    (["status", "state", "condition"], "status"),
    (["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):
                    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
            cantaloupe_dir = os.path.expanduser("~/projects/cantaloupe-downloader")
            result = subprocess.run(
                ["python", "-m", "cantaloupe", "export", "--scheduled"],
                capture_output=True, text=True, timeout=120,
                cwd=cantaloupe_dir,
            )
            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()