#!/usr/bin/env python3 """ Datenbank-Migrationsskript für MYP - Drucker-Schema-Update Dieses Skript aktualisiert das Drucker-Schema, um die Tapo-Felder und MAC-Adresse als nullable zu machen. """ import os import sys import sqlite3 from datetime import datetime # Pfad zu backend hinzufügen sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from utils.logging_config import get_logger from utils.utilities_collection import DATABASE_PATH logger = get_logger("migration") def backup_database(): """Erstellt ein Backup der Datenbank vor der Migration""" try: backup_path = DATABASE_PATH.replace('.db', f'_backup_{datetime.now().strftime("%Y%m%d_%H%M%S")}.db') # Kopiere die Datenbank import shutil shutil.copy2(DATABASE_PATH, backup_path) logger.info(f"✅ Datenbank-Backup erstellt: {backup_path}") return backup_path except Exception as e: logger.error(f"❌ Fehler beim Erstellen des Backups: {e}") raise def check_column_nullable(conn, table_name, column_name): """Überprüft, ob eine Spalte nullable ist""" cursor = conn.cursor() cursor.execute(f"PRAGMA table_info({table_name})") columns = cursor.fetchall() for column in columns: if column[1] == column_name: # column[3] ist 1 wenn NOT NULL, 0 wenn nullable return column[3] == 0 return None def migrate_printers_table(): """Führt die Migration der Drucker-Tabelle durch""" logger.info("🚀 Starte Drucker-Tabellen-Migration...") try: # Verbindung zur Datenbank conn = sqlite3.connect(DATABASE_PATH) cursor = conn.cursor() # Prüfe, ob die Spalten existieren cursor.execute("PRAGMA table_info(printers)") columns = {col[1]: col for col in cursor.fetchall()} # Prüfe, ob updated_at Spalte existiert if 'updated_at' not in columns: logger.info("➕ Füge updated_at Spalte hinzu...") cursor.execute(""" ALTER TABLE printers ADD COLUMN updated_at TIMESTAMP """) # Setze initialen Wert cursor.execute(""" UPDATE printers SET updated_at = created_at WHERE updated_at IS NULL """) logger.info("✅ updated_at Spalte hinzugefügt") # SQLite unterstützt kein direktes ALTER COLUMN für NOT NULL Änderungen # Wir müssen die Tabelle neu erstellen # Prüfe ob Migration notwendig ist mac_nullable = check_column_nullable(conn, 'printers', 'mac_address') plug_ip_nullable = check_column_nullable(conn, 'printers', 'plug_ip') if mac_nullable and plug_ip_nullable: logger.info("ℹ️ Migration bereits durchgeführt - keine Änderungen notwendig") conn.close() return logger.info("🔄 Erstelle neue Drucker-Tabelle mit aktualisierten Schema...") # Temporäre Tabelle mit neuem Schema erstellen cursor.execute(""" CREATE TABLE IF NOT EXISTS printers_new ( id INTEGER PRIMARY KEY, name VARCHAR(100) NOT NULL, model VARCHAR(100), location VARCHAR(100), ip_address VARCHAR(50), mac_address VARCHAR(50) UNIQUE, plug_ip VARCHAR(50), plug_username VARCHAR(100), plug_password VARCHAR(100), status VARCHAR(20) DEFAULT 'offline', active BOOLEAN DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, last_checked TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) """) # Daten kopieren logger.info("📋 Kopiere Daten in neue Tabelle...") cursor.execute(""" INSERT INTO printers_new SELECT id, name, model, location, ip_address, mac_address, plug_ip, plug_username, plug_password, status, active, created_at, last_checked, COALESCE(updated_at, created_at) as updated_at FROM printers """) # Alte Tabelle löschen und neue umbenennen cursor.execute("DROP TABLE printers") cursor.execute("ALTER TABLE printers_new RENAME TO printers") # Generiere MAC-Adressen für Drucker ohne MAC cursor.execute("SELECT id, name FROM printers WHERE mac_address IS NULL OR mac_address = ''") printers_without_mac = cursor.fetchall() if printers_without_mac: logger.info(f"🔧 Generiere MAC-Adressen für {len(printers_without_mac)} Drucker...") import uuid for printer_id, printer_name in printers_without_mac: # Generiere eindeutige MAC-Adresse mac = "00:50:56:" + ":".join([f"{uuid.uuid4().hex[:2]}" for _ in range(3)]) cursor.execute( "UPDATE printers SET mac_address = ? WHERE id = ?", (mac, printer_id) ) logger.info(f" ✅ {printer_name}: {mac}") # Änderungen committen conn.commit() logger.info("✅ Migration erfolgreich abgeschlossen!") # Statistiken anzeigen cursor.execute("SELECT COUNT(*) FROM printers") total_printers = cursor.fetchone()[0] cursor.execute("SELECT COUNT(*) FROM printers WHERE plug_ip IS NOT NULL") printers_with_plug = cursor.fetchone()[0] logger.info(f"📊 Statistiken:") logger.info(f" - Gesamt Drucker: {total_printers}") logger.info(f" - Drucker mit Steckdose: {printers_with_plug}") conn.close() except Exception as e: logger.error(f"❌ Fehler bei der Migration: {e}") raise def main(): """Hauptfunktion für die Migration""" logger.info("=" * 60) logger.info("MYP Datenbank-Migration - Drucker-Schema-Update") logger.info("=" * 60) try: # Backup erstellen backup_path = backup_database() # Migration durchführen migrate_printers_table() logger.info("✅ Migration erfolgreich abgeschlossen!") logger.info(f"ℹ️ Backup verfügbar unter: {backup_path}") except Exception as e: logger.error(f"❌ Migration fehlgeschlagen: {e}") logger.error("ℹ️ Bitte Backup wiederherstellen und Fehler beheben") sys.exit(1) if __name__ == "__main__": main()