📝 'feat': Renamed and moved various documentation files to '/docs/' directory

backend/app/utils/migrate_db.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""
+Datenbank-Migrationsskript für Guest-Requests, UserPermissions und Notifications
+"""
+
+import os
+import sys
+import sqlite3
+from datetime import datetime
+
+# Pfad zur App hinzufügen
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from models import init_db, get_cached_session, GuestRequest, UserPermission, Notification, User
+from utils.logging_config import get_logger
+from config.settings import DATABASE_PATH
+
+logger = get_logger("migrate")
+
+def 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 = [row[1] for row in cursor.fetchall()]
+    return column_name in columns
+
+def get_database_path():
+    """Ermittelt den Pfad zur Datenbankdatei."""
+    # Verwende den korrekten Datenbankpfad aus der Konfiguration
+    if os.path.exists(DATABASE_PATH):
+        return DATABASE_PATH
+    
+    # Fallback für alternative Pfade mit korrektem Dateinamen
+    alternative_paths = [
+        os.path.join('database', 'myp.db'),
+        'myp.db',
+        '../database/myp.db',
+        './database/myp.db',
+        # Legacy-Pfade für Rückwärtskompatibilität
+        os.path.join('database', 'app.db'),
+        'app.db',
+        '../database/app.db',
+        './database/app.db'
+    ]
+    
+    for path in alternative_paths:
+        if os.path.exists(path):
+            return path
+    
+    # Falls keine Datei gefunden wird, verwende den konfigurierten Pfad
+    return DATABASE_PATH
+
+def migrate_guest_requests_table():
+    """Migriert die guest_requests Tabelle für neue Spalten."""
+    db_path = get_database_path()
+    
+    if not os.path.exists(db_path):
+        logger.warning(f"Datenbankdatei nicht gefunden: {db_path}")
+        return False
+    
+    try:
+        conn = sqlite3.connect(db_path)
+        cursor = conn.cursor()
+        
+        # Prüfen ob otp_used_at Spalte bereits existiert
+        if not column_exists(cursor, 'guest_requests', 'otp_used_at'):
+            cursor.execute("""
+                ALTER TABLE guest_requests 
+                ADD COLUMN otp_used_at DATETIME
+            """)
+            logger.info("Spalte 'otp_used_at' zur guest_requests Tabelle hinzugefügt")
+        else:
+            logger.info("Spalte 'otp_used_at' existiert bereits")
+        
+        conn.commit()
+        conn.close()
+        return True
+        
+    except Exception as e:
+        logger.error(f"Fehler bei der Migration der guest_requests Tabelle: {str(e)}")
+        if 'conn' in locals():
+            conn.rollback()
+            conn.close()
+        return False
+
+def main():
+    """Führt die Datenbank-Migration aus."""
+    try:
+        logger.info("Starte Datenbank-Migration...")
+        
+        # Datenbank initialisieren (erstellt neue Tabellen)
+        init_db()
+        
+        # Spezifische Spalten-Migrationen
+        logger.info("Führe spezifische Tabellen-Migrationen aus...")
+        migrate_guest_requests_table()
+        
+        logger.info("Datenbank-Migration erfolgreich abgeschlossen")
+        
+        # Testen, ob die neuen Tabellen funktionieren
+        test_new_tables()
+        
+    except Exception as e:
+        logger.error(f"Fehler bei der Datenbank-Migration: {str(e)}")
+        sys.exit(1)
+
+def test_new_tables():
+    """Testet, ob die neuen Tabellen korrekt erstellt wurden."""
+    try:
+        with get_cached_session() as session:
+            # Test der GuestRequest-Tabelle
+            test_request = GuestRequest(
+                name="Test User",
+                email="test@example.com",
+                reason="Test migration",
+                duration_min=60
+            )
+            session.add(test_request)
+            session.flush()
+            
+            # Test der UserPermission-Tabelle (mit Admin-User falls vorhanden)
+            admin_user = session.query(User).filter_by(role="admin").first()
+            if admin_user:
+                # Prüfen, ob bereits Permissions für diesen User existieren
+                existing_permission = session.query(UserPermission).filter_by(user_id=admin_user.id).first()
+                
+                if not existing_permission:
+                    permission = UserPermission(
+                        user_id=admin_user.id,
+                        can_start_jobs=True,
+                        needs_approval=False,
+                        can_approve_jobs=True
+                    )
+                    session.add(permission)
+                    session.flush()
+                    logger.info(f"UserPermission für Admin-User {admin_user.id} erstellt")
+                else:
+                    logger.info(f"UserPermission für Admin-User {admin_user.id} existiert bereits")
+                
+                # Test der Notification-Tabelle
+                notification = Notification(
+                    user_id=admin_user.id,
+                    type="test",
+                    payload='{"message": "Test notification"}'
+                )
+                session.add(notification)
+                session.flush()
+            
+            # Test-Daten wieder löschen
+            session.rollback()
+            
+            logger.info("Alle neuen Tabellen wurden erfolgreich getestet")
+            
+    except Exception as e:
+        logger.error(f"Fehler beim Testen der neuen Tabellen: {str(e)}")
+        raise
+
+if __name__ == "__main__":
+    main()