Die Dateien, die in diesem Commit geändert wurden, sind:
1. backend/migrate_user_schema.py - Eine neue oder aktualisierte Skriptdatei zur Migration des Benutzerschemas. 2. backend/logs/* - Verschiedene Log-Dateien für verschiedene Aspekte der Backend-Infrastruktur, wie z.B. Admin, API, App, Datenmanagement, Druckersteuerung,
This commit is contained in:
205
backend/migrate_user_schema.py
Normal file
205
backend/migrate_user_schema.py
Normal file
@ -0,0 +1,205 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user