108 lines
3.3 KiB
Python
Executable File
108 lines
3.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
Apply SQL migration 004 to existing database.
|
||
Migration 004: Add hostname/desktop environment/PCI/USB columns to hardware_snapshots.
|
||
Usage: python apply_migration_004.py
|
||
"""
|
||
|
||
import os
|
||
import sqlite3
|
||
from typing import Dict, List, Tuple
|
||
|
||
# Database path
|
||
DB_PATH = os.path.join(os.path.dirname(__file__), "data", "data.db")
|
||
MIGRATION_PATH = os.path.join(
|
||
os.path.dirname(__file__), "migrations", "004_add_snapshot_details.sql"
|
||
)
|
||
|
||
COLUMNS_TO_ADD: List[Tuple[str, str]] = [
|
||
("hostname", "Nom d'hôte du snapshot"),
|
||
("desktop_environment", "Environnement de bureau détecté"),
|
||
("pci_devices_json", "Liste PCI en JSON"),
|
||
("usb_devices_json", "Liste USB en JSON"),
|
||
]
|
||
|
||
|
||
def _load_statements() -> Dict[str, str]:
|
||
"""Return ALTER TABLE statements indexed by column name."""
|
||
with open(MIGRATION_PATH, "r", encoding="utf-8") as f:
|
||
filtered = []
|
||
for line in f:
|
||
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 the SQL migration 004."""
|
||
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 004 already applied (columns exist)")
|
||
print("✅ Database is up to date")
|
||
return
|
||
|
||
statements = _load_statements()
|
||
|
||
print("🔄 Applying migration 004...")
|
||
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 004 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()
|