more robust import script

scripts/import_excel.py

@@ -1,5 +1,4 @@
 import argparse
-import shutil
 import sqlite3
 from datetime import date, datetime, timedelta
 from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
@@ -13,6 +12,9 @@ YELLOW = "FFFFFF00"
 ORANGE_THEME = 5
 MONTH_HEADER_ROWS = (2, 11, 20)
 MONTH_START_COLUMNS = (2, 10, 18, 26)
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_DATABASE = PROJECT_ROOT / "instance" / "calendar.sqlite"
+REQUIRED_TABLES = {"date_choices", "date_statuses"}
 
 
 def cell_color(cell):
@@ -77,8 +79,6 @@ def read_workbook(path):
 
 
 def existing_conflicts(database_path, amounts, statuses):
-    if not database_path.exists():
-        return 0, 0
     with sqlite3.connect(database_path) as database:
         existing_amounts = dict(database.execute("SELECT choice_date, amount_cents FROM date_choices"))
         existing_statuses = dict(database.execute("SELECT status_date, day_status FROM date_statuses"))
@@ -87,11 +87,59 @@ def existing_conflicts(database_path, amounts, statuses):
     return amount_conflicts, status_conflicts
 
 
+def inspect_database(database_path):
+    if not database_path.is_file():
+        raise RuntimeError(
+            f"Database does not exist: {database_path}\n"
+            "Start the app once to initialize it, or pass the correct path with --database."
+        )
+
+    with sqlite3.connect(database_path) as database:
+        tables = {
+            row[0]
+            for row in database.execute("SELECT name FROM sqlite_master WHERE type = 'table'")
+        }
+        missing = REQUIRED_TABLES - tables
+        if missing:
+            raise RuntimeError(
+                f"Database has the wrong schema; missing tables: {', '.join(sorted(missing))}"
+            )
+        amount_count = database.execute("SELECT COUNT(*) FROM date_choices").fetchone()[0]
+        status_count = database.execute("SELECT COUNT(*) FROM date_statuses").fetchone()[0]
+        dates = database.execute(
+            """
+            SELECT MIN(recorded_date), MAX(recorded_date) FROM (
+                SELECT choice_date AS recorded_date FROM date_choices
+                UNION ALL
+                SELECT status_date AS recorded_date FROM date_statuses
+            )
+            """
+        ).fetchone()
+    return amount_count, status_count, dates[0], dates[1]
+
+
+def verify_import(database_path, amounts, statuses):
+    with sqlite3.connect(database_path) as database:
+        stored_amounts = dict(
+            database.execute("SELECT choice_date, amount_cents FROM date_choices")
+        )
+        stored_statuses = dict(
+            database.execute("SELECT status_date, day_status FROM date_statuses")
+        )
+    bad_amounts = [key for key, value in amounts.items() if stored_amounts.get(key) != value]
+    bad_statuses = [key for key, value in statuses.items() if stored_statuses.get(key) != value]
+    if bad_amounts or bad_statuses:
+        raise RuntimeError(
+            "Post-import verification failed: "
+            f"{len(bad_amounts)} amounts and {len(bad_statuses)} statuses do not match."
+        )
+
+
 def import_data(database_path, amounts, statuses):
-    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f")
     backup_path = database_path.with_name(f"{database_path.name}.backup-{timestamp}")
-    if database_path.exists():
-        shutil.copy2(database_path, backup_path)
+    with sqlite3.connect(database_path) as source, sqlite3.connect(backup_path) as backup:
+        source.backup(backup)
 
     with sqlite3.connect(database_path) as database:
         database.executemany(
@@ -112,26 +160,56 @@ def import_data(database_path, amounts, statuses):
             """,
             statuses.items(),
         )
-    return backup_path if backup_path.exists() else None
+    verify_import(database_path, amounts, statuses)
+    return backup_path
 
 
 def main():
     parser = argparse.ArgumentParser(description="Import Fatima calendar data from Excel.")
     parser.add_argument("workbook", type=Path)
-    parser.add_argument("--database", type=Path, default=Path("instance/calendar.sqlite"))
+    parser.add_argument(
+        "--database",
+        type=Path,
+        default=DEFAULT_DATABASE,
+        help=f"SQLite database used by the app (default: {DEFAULT_DATABASE})",
+    )
     parser.add_argument("--apply", action="store_true", help="Write the data; otherwise perform a dry run.")
     args = parser.parse_args()
+    database_path = args.database.expanduser().resolve()
+    workbook_path = args.workbook.expanduser().resolve()
 
-    amounts, statuses = read_workbook(args.workbook)
-    amount_conflicts, status_conflicts = existing_conflicts(args.database, amounts, statuses)
+    if not workbook_path.is_file():
+        parser.error(f"Workbook does not exist: {workbook_path}")
+    try:
+        before_amounts, before_statuses, first_date, last_date = inspect_database(database_path)
+    except RuntimeError as error:
+        parser.error(str(error))
+
+    print(f"Workbook: {workbook_path}")
+    print(f"Database: {database_path}")
+    print(
+        f"Database before import: {before_amounts} amounts, {before_statuses} statuses"
+        + (f", date range {first_date} through {last_date}." if first_date else ", no recorded dates.")
+    )
+
+    amounts, statuses = read_workbook(workbook_path)
+    if not amounts and not statuses:
+        parser.error("Workbook produced no importable amounts or statuses; refusing to continue.")
+    amount_conflicts, status_conflicts = existing_conflicts(database_path, amounts, statuses)
     print(f"Found {len(amounts)} amounts, {sum(v == 'vacation' for v in statuses.values())} vacation days, and {sum(v == 'sick' for v in statuses.values())} sick days.")
     print(f"Existing differing values to overwrite: {amount_conflicts} amounts, {status_conflicts} statuses.")
     if not args.apply:
         print("Dry run only; no database changes made.")
         return
 
-    backup = import_data(args.database, amounts, statuses)
-    print(f"Import complete. Backup: {backup}" if backup else "Import complete; no previous database existed.")
+    backup = import_data(database_path, amounts, statuses)
+    after_amounts, after_statuses, first_date, last_date = inspect_database(database_path)
+    print(f"Import complete and verified: all {len(amounts)} amounts and {len(statuses)} statuses match.")
+    print(f"Backup: {backup}")
+    print(
+        f"Database after import: {after_amounts} amounts, {after_statuses} statuses, "
+        f"date range {first_date} through {last_date}."
+    )
 
 
 if __name__ == "__main__":