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

# ─── 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
                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

                # 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,
                        _safe_float(item.get("latitude")),
                        _safe_float(item.get("longitude")),
                        _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()