canteen-seed-import/backup.py

#!/usr/bin/env python3
"""
📦 Backup GPS & Check-ins Module

Before each import run, backs up existing GPS (lat, lng) and photo_path
fields from the target database to a compressed JSON backup file.

After import completes, compares backed up values with what was written
and flags any GPS/photo values that were incorrectly overwritten.

CLI Usage:
    python3 backup.py --backup seed-data/assets.db
        → Creates backup_YYYY-MM-DD.json.gz

    python3 backup.py --restore backup_2026-05-24.json.gz [target.db]
        → Restores preserved fields from backup (uses embedded path if
          target.db not provided)

    python3 backup.py --compare backup_2026-05-24.json.gz [target.db]
        → Compares backup vs current DB, flags unexpected changes
"""

import argparse
import gzip
import json
import os
import sqlite3
import sys
from datetime import date
from typing import Any, Optional

# ─── Fields to preserve ───────────────────────────────────────────────────────

PRESERVED_FIELDS = [
    "lat",
    "lng",
    "photo_path",
]

BACKUP_VERSION = 1


# ─── Helpers ──────────────────────────────────────────────────────────────────

def _resolve_path(path: str) -> str:
    """Resolve ~ and relative paths to absolute."""
    return os.path.abspath(os.path.expanduser(path))


def _backup_filename() -> str:
    """Generate a timestamped backup filename."""
    today = date.today().isoformat()  # YYYY-MM-DD
    return f"backup_{today}.json.gz"


def _conn(db_path: str) -> sqlite3.Connection:
    """Open a read-write connection to the SQLite database."""
    path = _resolve_path(db_path)
    if not os.path.isfile(path):
        print(f"❌ Database not found: {path}", file=sys.stderr)
        sys.exit(1)
    return sqlite3.connect(path)


# ─── Schema introspection ─────────────────────────────────────────────────────

def _get_table_info(conn: sqlite3.Connection) -> list[dict[str, Any]]:
    """Return list of {name, columns, id_column} for candidate tables."""
    cursor = conn.execute(
        "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name"
    )
    tables = []
    for (tname,) in cursor.fetchall():
        cols = [
            row[1]
            for row in conn.execute(f"PRAGMA table_info({tname!r})").fetchall()
        ]
        # Find the most likely ID column
        id_col = _pick_id_column(cols)
        tables.append({"name": tname, "columns": cols, "id_column": id_col})
    return tables


def _pick_id_column(cols: list[str]) -> Optional[str]:
    """Pick the most probable identity column from a list of column names."""
    lower = [c.lower() for c in cols]
    for candidate in ["machine_id", "asset_id", "id", "uid"]:
        if candidate in lower:
            idx = lower.index(candidate)
            return cols[idx]
    # Fallback: first column (usually a PK)
    return cols[0] if cols else None


def _discover_asset_table(
    conn: sqlite3.Connection,
) -> tuple[str, str, list[str]]:
    """
    Find the 'assets' table (or best guess) and return
    (table_name, id_column, preserved_columns_that_exist).
    """
    tables = _get_table_info(conn)
    if not tables:
        print("❌ No tables found in database.", file=sys.stderr)
        sys.exit(1)

    # Prefer a table named 'assets' or similar
    preferred = ["assets", "machines", "machines", "asset"]
    asset_table = None
    for pref in preferred:
        for t in tables:
            if t["name"].lower() == pref:
                asset_table = t
                break
        if asset_table:
            break

    if not asset_table:
        # Use the first table that has at least one preserved field
        for t in tables:
            cols_lower = [c.lower() for c in t["columns"]]
            if any(f.lower() in cols_lower for f in PRESERVED_FIELDS):
                asset_table = t
                break

    if not asset_table:
        print(
            "❌ Could not find a suitable table with GPS/photo columns.",
            file=sys.stderr,
        )
        print(f"   Tables found: {[t['name'] for t in tables]}", file=sys.stderr)
        sys.exit(1)

    cols_lower = [c.lower() for c in asset_table["columns"]]
    existing = [f for f in PRESERVED_FIELDS if f.lower() in cols_lower]

    return asset_table["name"], asset_table["id_column"] or "rowid", existing


# ─── Core operations ──────────────────────────────────────────────────────────

def backup(db_path: str) -> str:
    """
    Back up GPS and photo_path fields from the database.
    Returns the path to the created backup file.
    """
    db_path = _resolve_path(db_path)
    conn = _conn(db_path)
    table, id_col, fields = _discover_asset_table(conn)

    if not fields:
        print(
            f"⚠️  No preserved fields ({', '.join(PRESERVED_FIELDS)}) "
            f"found in table '{table}'.",
            file=sys.stderr,
        )
        conn.close()
        sys.exit(1)

    # Build query
    col_exprs = ", ".join(f'"{c}"' for c in [id_col] + fields)
    rows = conn.execute(f'SELECT {col_exprs} FROM "{table}"').fetchall()

    backup_data: dict[str, Any] = {
        "version": BACKUP_VERSION,
        "source_db": db_path,
        "source_table": table,
        "id_column": id_col,
        "fields": fields,
        "created": date.today().isoformat(),
        "records": [],
    }

    for row in rows:
        rec: dict[str, Any] = {"id": row[0]}
        for i, field in enumerate(fields):
            val = row[i + 1]
            # Convert None to null (JSON-friendly)
            rec[field] = val
        backup_data["records"].append(rec)

    conn.close()

    # Write compressed backup
    out_path = _backup_filename()
    json_bytes = json.dumps(backup_data, indent=2, default=str).encode("utf-8")
    with gzip.open(out_path, "wb") as f:
        f.write(json_bytes)

    record_count = len(backup_data["records"])
    non_null = sum(
        1
        for r in backup_data["records"]
        if any(r.get(f) is not None for f in fields)
    )
    print(
        f"✅ Backup saved: {out_path}  "
        f"({record_count} records, {non_null} with non-null preserved fields)"
    )
    return out_path


def restore(backup_path: str, target_db: Optional[str] = None) -> None:
    """
    Restore GPS and photo_path fields from a backup file into the database.
    """
    backup_path = _resolve_path(backup_path)
    if not os.path.isfile(backup_path):
        print(f"❌ Backup file not found: {backup_path}", file=sys.stderr)
        sys.exit(1)

    # Load backup
    with gzip.open(backup_path, "rb") as f:
        backup_data = json.loads(f.read().decode("utf-8"))

    # Determine target DB
    db_path = target_db or backup_data.get("source_db")
    if not db_path:
        print(
            "❌ No target database specified and no source_db in backup.",
            file=sys.stderr,
        )
        sys.exit(1)

    db_path = _resolve_path(db_path)
    if not os.path.isfile(db_path):
        print(f"❌ Target database not found: {db_path}", file=sys.stderr)
        sys.exit(1)

    table = backup_data["source_table"]
    id_col = backup_data["id_column"]
    fields = backup_data["fields"]
    records = backup_data["records"]

    conn = sqlite3.connect(db_path)

    # Verify table and columns still exist
    existing_cols = [
        r[1]
        for r in conn.execute(f"PRAGMA table_info({table!r})").fetchall()
    ]
    existing_lower = [c.lower() for c in existing_cols]

    missing_fields = [f for f in fields if f.lower() not in existing_lower]
    if missing_fields:
        print(
            f"⚠️  Columns missing in target table '{table}': "
            f"{', '.join(missing_fields)}",
            file=sys.stderr,
        )
        fields = [f for f in fields if f.lower() in existing_lower]
        if not fields:
            print("❌ No preserved columns exist in target.", file=sys.stderr)
            conn.close()
            sys.exit(1)

    # Build UPDATE statements
    set_clause = ", ".join(f'"{c}" = ?' for c in fields)
    sql = f'UPDATE "{table}" SET {set_clause} WHERE "{id_col}" = ?'

    restored = 0
    skipped = 0
    for rec in records:
        values = [rec.get(f) for f in fields] + [rec["id"]]
        cursor = conn.execute(sql, values)
        if cursor.rowcount > 0:
            restored += 1
        else:
            skipped += 1

    conn.commit()
    conn.close()

    print(
        f"✅ Restored {restored} records from {backup_path}\n"
        f"   Fields: {', '.join(fields)}\n"
        f"   Skipped (no matching row): {skipped}"
    )


def compare(backup_path: str, target_db: Optional[str] = None) -> int:
    """
    Compare backed-up values against current database values.
    Flags any GPS/photo values that were incorrectly overwritten.

    Returns count of flags.
    """
    backup_path = _resolve_path(backup_path)
    if not os.path.isfile(backup_path):
        print(f"❌ Backup file not found: {backup_path}", file=sys.stderr)
        sys.exit(1)

    # Load backup
    with gzip.open(backup_path, "rb") as f:
        backup_data = json.loads(f.read().decode("utf-8"))

    # Determine target DB
    db_path = target_db or backup_data.get("source_db")
    if not db_path:
        print(
            "❌ No target database specified and no source_db in backup.",
            file=sys.stderr,
        )
        sys.exit(1)

    db_path = _resolve_path(db_path)
    if not os.path.isfile(db_path):
        print(f"❌ Target database not found: {db_path}", file=sys.stderr)
        sys.exit(1)

    table = backup_data["source_table"]
    id_col = backup_data["id_column"]
    fields = backup_data["fields"]
    records = backup_data["records"]

    conn = sqlite3.connect(db_path)

    # Build a lookup of current values
    col_exprs = ", ".join(f'"{c}"' for c in [id_col] + fields)
    try:
        current_rows = conn.execute(
            f'SELECT {col_exprs} FROM "{table}"'
        ).fetchall()
    except sqlite3.OperationalError as e:
        print(f"❌ Error querying table '{table}': {e}", file=sys.stderr)
        conn.close()
        sys.exit(1)

    current_lookup: dict[str, dict[str, Any]] = {}
    for row in current_rows:
        rec: dict[str, Any] = {"id": row[0]}
        for i, field in enumerate(fields):
            rec[field] = row[i + 1]
        current_lookup[str(row[0])] = rec

    # Compare
    flags = []
    for rec in records:
        rec_id = str(rec["id"])
        current = current_lookup.get(rec_id)
        if current is None:
            # Record was deleted — not a concern
            continue

        for field in fields:
            backup_val = rec.get(field)
            current_val = current.get(field)

            if backup_val is None:
                # Backup had no value — import may have set one, that's fine
                continue

            if str(backup_val) != str(current_val):
                # Previously non-null value was changed
                flags.append(
                    {
                        "id": rec_id,
                        "field": field,
                        "backup_value": backup_val,
                        "current_value": current_val,
                    }
                )

    conn.close()

    if not flags:
        print("✅ No unexpected GPS/photo overwrites detected.")
        return 0

    print(
        f"⚠️  Found {len(flags)} potentially overwritten field(s):\n",
        file=sys.stderr,
    )
    for f in flags:
        print(
            f"   ID {f['id']:>10s} | {f['field']:15s} "
            f"| was: {str(f['backup_value']):>25s} "
            f"| now: {str(f['current_value']):>25s}",
            file=sys.stderr,
        )
    print(
        "\n💡 Use --restore to revert if these changes are unintended.",
        file=sys.stderr,
    )
    return len(flags)


# ─── CLI ──────────────────────────────────────────────────────────────────────

def main():
    parser = argparse.ArgumentParser(
        description="📦 Backup GPS & Check-ins Module — Preserve GPS/photo data "
        "across import runs.",
    )
    parser.add_argument(
        "--backup",
        metavar="DB_PATH",
        help="Backup GPS (lat, lng) and photo_path from database.",
    )
    parser.add_argument(
        "--restore",
        metavar="BACKUP_FILE",
        help="Restore preserved fields from a backup .json.gz file.",
    )
    parser.add_argument(
        "--compare",
        metavar="BACKUP_FILE",
        help="Compare backup against current DB and flag overwrites.",
    )
    parser.add_argument(
        "target",
        nargs="?",
        default=None,
        help="Target database path (optional for --restore and --compare "
        "if source_db is embedded in the backup file).",
    )

    args = parser.parse_args()

    # Exactly one mode
    modes = sum(
        1 for m in [args.backup, args.restore, args.compare] if m
    )
    if modes != 1:
        parser.print_help()
        print(
            "\n❌ Specify exactly one of: --backup, --restore, --compare",
            file=sys.stderr,
        )
        sys.exit(1)

    if args.backup:
        backup(args.backup)
    elif args.restore:
        restore(args.restore, args.target)
    elif args.compare:
        compare(args.compare, args.target)


if __name__ == "__main__":
    main()