cantaloupe-downloader/cantaloupe/diff.py

"""
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