1

backend/apply_migration_005.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""
+Apply SQL migration 005 to existing database.
+Migration 005: Add OS/display/battery metadata columns to hardware_snapshots.
+Usage: python apply_migration_005.py
+"""
+
+import os
+import sqlite3
+from typing import Dict, List, Tuple
+
+DB_PATH = os.path.join(os.path.dirname(__file__), "data", "data.db")
+MIGRATION_PATH = os.path.join(
+    os.path.dirname(__file__), "migrations", "005_add_os_display_and_battery.sql"
+)
+
+COLUMNS_TO_ADD: List[Tuple[str, str]] = [
+    ("screen_resolution", "Résolution écran"),
+    ("display_server", "Serveur d'affichage"),
+    ("session_type", "Type de session"),
+    ("last_boot_time", "Dernier boot"),
+    ("uptime_seconds", "Uptime en secondes"),
+    ("battery_percentage", "Pourcentage batterie"),
+    ("battery_status", "Statut batterie"),
+    ("battery_health", "Santé batterie"),
+]
+
+
+def _load_statements() -> Dict[str, str]:
+    """Load ALTER statements from migration file keyed by column name."""
+    with open(MIGRATION_PATH, "r", encoding="utf-8") as fh:
+        filtered = []
+        for line in fh:
+            stripped = line.strip()
+            if not stripped or stripped.startswith("--"):
+                continue
+            filtered.append(line.rstrip("\n"))
+
+    statements: Dict[str, str] = {}
+    for statement in "\n".join(filtered).split(";"):
+        stmt = statement.strip()
+        if not stmt:
+            continue
+        for column, _ in COLUMNS_TO_ADD:
+            if column in stmt:
+                statements[column] = stmt
+                break
+    return statements
+
+
+def apply_migration():
+    """Apply migration 005 to the SQLite database."""
+    if not os.path.exists(DB_PATH):
+        print(f"❌ Database not found at {DB_PATH}")
+        print("   The database will be created automatically on first run.")
+        return
+
+    if not os.path.exists(MIGRATION_PATH):
+        print(f"❌ Migration file not found at {MIGRATION_PATH}")
+        return
+
+    print(f"📂 Database: {DB_PATH}")
+    print(f"📄 Migration: {MIGRATION_PATH}")
+    print()
+
+    conn = sqlite3.connect(DB_PATH)
+    cursor = conn.cursor()
+
+    try:
+        cursor.execute("PRAGMA table_info(hardware_snapshots)")
+        existing_columns = {row[1] for row in cursor.fetchall()}
+
+        missing = [col for col, _ in COLUMNS_TO_ADD if col not in existing_columns]
+        if not missing:
+            print("⚠️  Migration 005 already applied (columns exist)")
+            print("✅ Database is up to date")
+            return
+
+        statements = _load_statements()
+
+        print("🔄 Applying migration 005...")
+        for column, description in COLUMNS_TO_ADD:
+            if column not in missing:
+                print(f"⏩ Column {column} already present, skipping")
+                continue
+
+            statement = statements.get(column)
+            if not statement:
+                raise RuntimeError(
+                    f"No SQL statement found for column '{column}' in migration file"
+                )
+
+            print(f"➕ Adding {description} ({column})...")
+            cursor.execute(statement)
+
+        conn.commit()
+
+        print("✅ Migration 005 applied successfully!")
+        print("New columns added to hardware_snapshots:")
+        for column, description in COLUMNS_TO_ADD:
+            if column in missing:
+                print(f"  - {column}: {description}")
+
+    except (sqlite3.Error, RuntimeError) as exc:
+        print(f"❌ Error applying migration: {exc}")
+        conn.rollback()
+    finally:
+        conn.close()
+
+
+if __name__ == "__main__":
+    apply_migration()