""" Diff engine for comparing Cantaloupe machine list Excel exports. Reports: - 🟢 New machines added (present in new, not in old) - 🔴 Machines removed (present in old, not in new) - 🟡 Machines with changed fields (same ID, different values) Also manages a JSON manifest at ~/.cantaloupe-exports/manifest.json that tracks export history for auto-diff functionality. """ import json import os from datetime import datetime from pathlib import Path from typing import Optional import openpyxl MANIFEST_DIR = Path.home() / ".cantaloupe-exports" MANIFEST_FILE = MANIFEST_DIR / "manifest.json" # ── xlsx reading ──────────────────────────────────────────────────────────── def _read_xlsx(path: Path) -> tuple[list[str], list[dict]]: """Read an xlsx file and return (headers, list of row dicts). Each row dict maps header name → cell value. Returns empty lists if the file is empty or unreadable. """ wb = openpyxl.load_workbook(path, read_only=True) try: ws = wb.active if ws is None: return [], [] rows = list(ws.iter_rows(values_only=True)) finally: wb.close() if not rows: return [], [] headers = [str(h) if h is not None else "" for h in rows[0]] data = [] for row in rows[1:]: row_dict = {} for i, cell in enumerate(row): if i < len(headers): row_dict[headers[i]] = cell data.append(row_dict) return headers, data def _find_id_column(headers: list[str]) -> str: """Find the best unique-identifier column from headers. Priority: 1. Column whose name contains 'asset' AND 'id' (case-insensitive) 2. Column whose name equals 'id' (case-insensitive) 3. First column (fallback) """ for h in headers: hl = h.lower() if "asset" in hl and "id" in hl: return h for h in headers: if h.lower() == "id": return h return headers[0] if headers else "" # ── diff engine ───────────────────────────────────────────────────────────── def _normalize(val) -> str: """Convert a cell value to a string for comparison and keying.""" if val is None: return "" return str(val).strip() def diff_excels( from_path: Path, to_path: Path, id_column: Optional[str] = None, ) -> dict: """Compare two xlsx files and return a structured diff result. Args: from_path: Path to the older ("from") xlsx file. to_path: Path to the newer ("to") xlsx file. id_column: Column name to use as the unique machine identifier. Auto-detected if not provided (looks for 'Asset ID'). Returns: dict with keys: - id_column: str — column used as the key - from_path, to_path: str — file paths - from_count, to_count: int — data row counts - added: list[dict] — rows present in 'to' but not 'from' - removed: list[dict] — rows present in 'from' but not 'to' - changed: list[dict] — {id, changes: {col: {old, new}}} - added_count, removed_count, changed_count: int Raises: ValueError: if either file has no data / no rows. FileNotFoundError: if either file doesn't exist. """ from_headers, from_data = _read_xlsx(from_path) to_headers, to_data = _read_xlsx(to_path) if not to_headers: raise ValueError(f"No data found in {to_path}") # Determine id column (prefer the 'to' file's headers) id_col = id_column or _find_id_column(to_headers) # Build lookup dicts keyed on the id column from_by_id: dict[str, dict] = {} for row in from_data: key = _normalize(row.get(id_col, "")) from_by_id[key] = row to_by_id: dict[str, dict] = {} for row in to_data: key = _normalize(row.get(id_col, "")) to_by_id[key] = row from_ids = set(from_by_id.keys()) to_ids = set(to_by_id.keys()) added_ids = to_ids - from_ids removed_ids = from_ids - to_ids common_ids = from_ids & to_ids # Collect changed rows changed: list[dict] = [] # Use headers from 'to' file as canonical for cid in sorted(common_ids): old_row = from_by_id[cid] new_row = to_by_id[cid] field_changes: dict[str, dict[str, object]] = {} for col in to_headers: old_val = _normalize(old_row.get(col)) new_val = _normalize(new_row.get(col)) if old_val != new_val: field_changes[col] = {"old": old_row.get(col), "new": new_row.get(col)} if field_changes: changed.append({"id": cid, "changes": field_changes}) return { "id_column": id_col, "from_path": str(from_path), "to_path": str(to_path), "from_count": len(from_data), "to_count": len(to_data), "added": [to_by_id[aid] for aid in sorted(added_ids)], "removed": [from_by_id[rid] for rid in sorted(removed_ids)], "changed": changed, "added_count": len(added_ids), "removed_count": len(removed_ids), "changed_count": len(changed), } # ── manifest management ───────────────────────────────────────────────────── def load_manifest() -> dict: """Load the export history manifest from ~/.cantaloupe-exports/manifest.json. Returns a dict with an 'exports' list. Creates an empty manifest if the file doesn't exist. """ MANIFEST_DIR.mkdir(parents=True, exist_ok=True) if MANIFEST_FILE.exists(): try: return json.loads(MANIFEST_FILE.read_text()) except (json.JSONDecodeError, OSError): return {"exports": []} return {"exports": []} def save_manifest(manifest: dict) -> None: """Save the manifest to disk.""" MANIFEST_DIR.mkdir(parents=True, exist_ok=True) MANIFEST_FILE.write_text(json.dumps(manifest, indent=2, default=str) + "\n") def record_export( filepath: Path, row_count: int, file_size: int, ) -> None: """Record a completed export in the manifest.""" manifest = load_manifest() manifest.setdefault("exports", []).append( { "path": str(filepath), "filename": filepath.name, "timestamp": datetime.now().isoformat(), "row_count": row_count, "file_size": file_size, } ) save_manifest(manifest) def get_previous_export() -> Optional[Path]: """Return the path of the most recent export, or None if no history.""" manifest = load_manifest() exports = manifest.get("exports", []) if not exports: return None previous = exports[-1]["path"] p = Path(previous) if p.exists(): return p # File was moved/deleted — return None so we don't diff against a ghost return None def get_latest_exports(n: int = 2) -> list[Path]: """Return paths of the N most recent existing exports, newest first.""" manifest = load_manifest() exports = manifest.get("exports", []) result: list[Path] = [] for entry in reversed(exports): p = Path(entry["path"]) if p.exists(): result.append(p) if len(result) >= n: break return result