#!/usr/bin/env python3 """ Datenbank-Migration Script für User-Schema-Update ================================================ Dieses Script behebt das kritische Problem mit fehlenden User-Spalten in der SQLite-Datenbank. Fehler: (sqlite3.OperationalError) no such column: users.theme_preference Lösung: Fügt alle fehlenden User-Spalten zur bestehenden Tabelle hinzu. Autor: Till Tomczak - Mercedes-Benz TBA Marienfelde Datum: 2025-06-19 """ import os import sys import sqlite3 from datetime import datetime # Pfad zur Datenbank DATABASE_PATH = os.path.join(os.path.dirname(__file__), "database", "myp.db") def check_column_exists(cursor, table_name, column_name): """Prüft ob eine Spalte in einer Tabelle existiert""" cursor.execute(f"PRAGMA table_info({table_name})") columns = [column[1] for column in cursor.fetchall()] return column_name in columns def migrate_user_schema(): """ Migriert das User-Schema zur neuesten Version. Fügt alle fehlenden Spalten hinzu die im SQLAlchemy Model definiert sind. """ print("🔧 Starte User-Schema-Migration...") if not os.path.exists(DATABASE_PATH): print(f"❌ Datenbank nicht gefunden: {DATABASE_PATH}") return False # Backup der Datenbank erstellen backup_path = DATABASE_PATH + f".backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}" try: import shutil shutil.copy2(DATABASE_PATH, backup_path) print(f"✅ Backup erstellt: {backup_path}") except Exception as e: print(f"⚠️ Backup-Warnung: {e}") # Neue Spalten die hinzugefügt werden müssen new_columns = [ ("theme_preference", "VARCHAR(20)", "auto"), ("language_preference", "VARCHAR(10)", "de"), ("email_notifications", "BOOLEAN", "1"), ("browser_notifications", "BOOLEAN", "1"), ("dashboard_layout", "VARCHAR(20)", "default"), ("compact_mode", "BOOLEAN", "0"), ("show_completed_jobs", "BOOLEAN", "1"), ("auto_refresh_interval", "INTEGER", "30"), ("auto_logout_timeout", "INTEGER", "0") ] try: # Verbindung zur Datenbank conn = sqlite3.connect(DATABASE_PATH) cursor = conn.cursor() print("📊 Prüfe vorhandene Spalten...") # Prüfe welche Spalten bereits existieren existing_columns = [] missing_columns = [] for column_name, column_type, default_value in new_columns: if check_column_exists(cursor, "users", column_name): existing_columns.append(column_name) print(f" ✅ {column_name} - bereits vorhanden") else: missing_columns.append((column_name, column_type, default_value)) print(f" ❌ {column_name} - fehlt") if not missing_columns: print("🎉 Alle Spalten sind bereits vorhanden! Schema ist aktuell.") conn.close() return True print(f"🔨 Füge {len(missing_columns)} fehlende Spalten hinzu...") # Füge fehlende Spalten hinzu for column_name, column_type, default_value in missing_columns: alter_sql = f"ALTER TABLE users ADD COLUMN {column_name} {column_type}" # Default-Value hinzufügen wenn SQLite es unterstützt if default_value: alter_sql += f" DEFAULT '{default_value}'" try: cursor.execute(alter_sql) print(f" ✅ Spalte {column_name} hinzugefügt") except Exception as e: print(f" ❌ Fehler bei {column_name}: {e}") # Bei Fehlern trotzdem weitermachen # Für existierende Benutzer Default-Werte setzen print("🔄 Setze Default-Werte für existierende Benutzer...") for column_name, column_type, default_value in missing_columns: if default_value: try: update_sql = f"UPDATE users SET {column_name} = ? WHERE {column_name} IS NULL" cursor.execute(update_sql, (default_value,)) print(f" ✅ Default-Werte für {column_name} gesetzt") except Exception as e: print(f" ⚠️ Default-Werte für {column_name}: {e}") # Änderungen speichern conn.commit() # Validierung: Alle Spalten nochmal prüfen print("🔍 Validiere Migration...") all_present = True for column_name, _, _ in new_columns: if not check_column_exists(cursor, "users", column_name): print(f" ❌ {column_name} immer noch nicht vorhanden!") all_present = False else: print(f" ✅ {column_name} erfolgreich migriert") conn.close() if all_present: print("🎉 Migration erfolgreich abgeschlossen!") print(" Alle neuen User-Spalten wurden hinzugefügt.") print(" Die SQLAlchemy-Fehler sollten jetzt behoben sein.") return True else: print("⚠️ Migration teilweise fehlgeschlagen - siehe Details oben") return False except Exception as e: print(f"❌ Kritischer Fehler bei der Migration: {e}") print(f"💾 Backup verfügbar unter: {backup_path}") return False def verify_migration(): """Verifiziert dass die Migration erfolgreich war""" try: # Test-Import des User-Models sys.path.append(os.path.dirname(__file__)) from models import User, get_cached_session print("🧪 Teste User-Model-Import...") # Teste eine einfache Query with get_cached_session() as session: user_count = session.query(User).count() print(f"✅ User-Query erfolgreich: {user_count} Benutzer gefunden") # Teste Zugriff auf neue Spalten test_user = session.query(User).first() if test_user: # Teste alle neuen Attribute _ = test_user.theme_preference _ = test_user.language_preference _ = test_user.email_notifications _ = test_user.browser_notifications _ = test_user.dashboard_layout _ = test_user.compact_mode _ = test_user.show_completed_jobs _ = test_user.auto_refresh_interval _ = test_user.auto_logout_timeout print("✅ Alle neuen User-Attribute funktionieren") return True except Exception as e: print(f"❌ Verifikation fehlgeschlagen: {e}") return False if __name__ == "__main__": print("=" * 60) print("MYP Datenbank-Migration: User-Schema-Update") print("=" * 60) success = migrate_user_schema() if success: print("\n🔍 Führe Verifikation durch...") verify_success = verify_migration() if verify_success: print("\n🎉 MIGRATION ERFOLGREICH!") print(" Die User-Schema-Probleme wurden behoben.") print(" Der Server kann jetzt ohne Datenbankfehler neu gestartet werden.") else: print("\n⚠️ Migration abgeschlossen, aber Verifikation fehlgeschlagen.") print(" Bitte prüfe die Logs manuell.") else: print("\n❌ MIGRATION FEHLGESCHLAGEN!") print(" Siehe Details oben. Backup-Datei wurde erstellt.") print("=" * 60)