feat: detect machine replacements in cantaloupe sync

Add detect_replacements() function that identifies when a new Asset ID
appears at the same customer + address as an existing asset not in the
current import. Uses two confidence levels:
- high: exact address match
- medium: same building_name or same street name

Stored in diff_summary under 'replacements' key for admin review UI.

cantaloupe_sync.py

@@ -275,6 +275,123 @@ def _fetch_live_data(conn: sqlite3.Connection) -> dict[str, list[dict]]:
     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.
@@ -445,6 +562,12 @@ async def run_sync(
         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)