253 lines
9.5 KiB
Python
253 lines
9.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Sofortige Datenbank-Reparatur für fehlende updated_at Spalte
|
|
"""
|
|
|
|
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 config.settings import DATABASE_PATH
|
|
|
|
def fix_users_table_immediate():
|
|
"""Repariert die users Tabelle sofort."""
|
|
print(f"Repariere Datenbank: {DATABASE_PATH}")
|
|
|
|
if not os.path.exists(DATABASE_PATH):
|
|
print(f"Datenbankdatei nicht gefunden: {DATABASE_PATH}")
|
|
return False
|
|
|
|
try:
|
|
conn = sqlite3.connect(DATABASE_PATH)
|
|
cursor = conn.cursor()
|
|
|
|
# Prüfen, welche Spalten existieren
|
|
cursor.execute("PRAGMA table_info(users)")
|
|
existing_columns = [row[1] for row in cursor.fetchall()]
|
|
print(f"Vorhandene Spalten in users: {existing_columns}")
|
|
|
|
# Fehlende Spalten hinzufügen
|
|
required_columns = [
|
|
('updated_at', 'DATETIME'),
|
|
('settings', 'TEXT'),
|
|
('department', 'VARCHAR(100)'),
|
|
('position', 'VARCHAR(100)'),
|
|
('phone', 'VARCHAR(50)'),
|
|
('bio', 'TEXT')
|
|
]
|
|
|
|
for column_name, column_type in required_columns:
|
|
if column_name not in existing_columns:
|
|
try:
|
|
if column_name == 'updated_at':
|
|
# Einfacher Ansatz: NULL erlauben und später updaten
|
|
cursor.execute(f"ALTER TABLE users ADD COLUMN {column_name} {column_type}")
|
|
print(f"✓ Spalte '{column_name}' hinzugefügt")
|
|
|
|
# Alle vorhandenen Benutzer mit aktuellem Timestamp updaten
|
|
cursor.execute(f"UPDATE users SET {column_name} = CURRENT_TIMESTAMP WHERE {column_name} IS NULL")
|
|
print(f"✓ Vorhandene Benutzer mit {column_name} aktualisiert")
|
|
|
|
# Trigger für automatische Updates erstellen
|
|
cursor.execute("""
|
|
CREATE TRIGGER IF NOT EXISTS update_users_updated_at
|
|
AFTER UPDATE ON users
|
|
FOR EACH ROW
|
|
BEGIN
|
|
UPDATE users SET updated_at = CURRENT_TIMESTAMP WHERE id = NEW.id;
|
|
END
|
|
""")
|
|
print(f"✓ Auto-Update-Trigger für {column_name} erstellt")
|
|
else:
|
|
cursor.execute(f"ALTER TABLE users ADD COLUMN {column_name} {column_type}")
|
|
print(f"✓ Spalte '{column_name}' hinzugefügt")
|
|
|
|
except Exception as e:
|
|
print(f"✗ Fehler bei Spalte '{column_name}': {str(e)}")
|
|
else:
|
|
print(f"○ Spalte '{column_name}' bereits vorhanden")
|
|
|
|
# Weitere fehlende Tabellen prüfen und erstellen
|
|
create_missing_tables(cursor)
|
|
|
|
# Optimierungsindizes erstellen
|
|
create_performance_indexes(cursor)
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
print("✓ Datenbank-Reparatur erfolgreich abgeschlossen")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"✗ Fehler bei der Datenbank-Reparatur: {str(e)}")
|
|
if 'conn' in locals():
|
|
conn.rollback()
|
|
conn.close()
|
|
return False
|
|
|
|
def create_missing_tables(cursor):
|
|
"""Erstellt fehlende Tabellen."""
|
|
|
|
# Prüfen, welche Tabellen existieren
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
|
|
existing_tables = [row[0] for row in cursor.fetchall()]
|
|
print(f"Vorhandene Tabellen: {existing_tables}")
|
|
|
|
# user_permissions Tabelle
|
|
if 'user_permissions' not in existing_tables:
|
|
cursor.execute("""
|
|
CREATE TABLE user_permissions (
|
|
user_id INTEGER PRIMARY KEY,
|
|
can_start_jobs BOOLEAN DEFAULT 0,
|
|
needs_approval BOOLEAN DEFAULT 1,
|
|
can_approve_jobs BOOLEAN DEFAULT 0,
|
|
FOREIGN KEY (user_id) REFERENCES users (id)
|
|
)
|
|
""")
|
|
print("✓ Tabelle 'user_permissions' erstellt")
|
|
|
|
# notifications Tabelle
|
|
if 'notifications' not in existing_tables:
|
|
cursor.execute("""
|
|
CREATE TABLE notifications (
|
|
id INTEGER PRIMARY KEY,
|
|
user_id INTEGER NOT NULL,
|
|
type VARCHAR(50) NOT NULL,
|
|
payload TEXT,
|
|
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
read BOOLEAN DEFAULT 0,
|
|
FOREIGN KEY (user_id) REFERENCES users (id)
|
|
)
|
|
""")
|
|
print("✓ Tabelle 'notifications' erstellt")
|
|
|
|
# stats Tabelle
|
|
if 'stats' not in existing_tables:
|
|
cursor.execute("""
|
|
CREATE TABLE stats (
|
|
id INTEGER PRIMARY KEY,
|
|
total_print_time INTEGER DEFAULT 0,
|
|
total_jobs_completed INTEGER DEFAULT 0,
|
|
total_material_used REAL DEFAULT 0.0,
|
|
last_updated DATETIME DEFAULT CURRENT_TIMESTAMP
|
|
)
|
|
""")
|
|
print("✓ Tabelle 'stats' erstellt")
|
|
|
|
# Initial stats record erstellen
|
|
cursor.execute("""
|
|
INSERT INTO stats (total_print_time, total_jobs_completed, total_material_used, last_updated)
|
|
VALUES (0, 0, 0.0, CURRENT_TIMESTAMP)
|
|
""")
|
|
print("✓ Initial-Statistiken erstellt")
|
|
|
|
# system_logs Tabelle
|
|
if 'system_logs' not in existing_tables:
|
|
cursor.execute("""
|
|
CREATE TABLE system_logs (
|
|
id INTEGER PRIMARY KEY,
|
|
timestamp DATETIME DEFAULT CURRENT_TIMESTAMP NOT NULL,
|
|
level VARCHAR(20) NOT NULL,
|
|
message VARCHAR(1000) NOT NULL,
|
|
module VARCHAR(100),
|
|
user_id INTEGER,
|
|
ip_address VARCHAR(50),
|
|
user_agent VARCHAR(500),
|
|
FOREIGN KEY (user_id) REFERENCES users (id)
|
|
)
|
|
""")
|
|
print("✓ Tabelle 'system_logs' erstellt")
|
|
|
|
def create_performance_indexes(cursor):
|
|
"""Erstellt Performance-Indices."""
|
|
print("Erstelle Performance-Indices...")
|
|
|
|
indexes = [
|
|
("idx_users_email", "users(email)"),
|
|
("idx_users_username", "users(username)"),
|
|
("idx_users_role", "users(role)"),
|
|
("idx_jobs_user_id", "jobs(user_id)"),
|
|
("idx_jobs_printer_id", "jobs(printer_id)"),
|
|
("idx_jobs_status", "jobs(status)"),
|
|
("idx_jobs_start_at", "jobs(start_at)"),
|
|
("idx_notifications_user_id", "notifications(user_id)"),
|
|
("idx_notifications_read", "notifications(read)"),
|
|
("idx_system_logs_timestamp", "system_logs(timestamp)"),
|
|
("idx_system_logs_level", "system_logs(level)"),
|
|
("idx_guest_requests_status", "guest_requests(status)"),
|
|
("idx_printers_status", "printers(status)"),
|
|
("idx_printers_active", "printers(active)")
|
|
]
|
|
|
|
for index_name, index_def in indexes:
|
|
try:
|
|
cursor.execute(f"CREATE INDEX IF NOT EXISTS {index_name} ON {index_def}")
|
|
print(f"✓ Index '{index_name}' erstellt")
|
|
except Exception as e:
|
|
print(f"○ Index '{index_name}': {str(e)}")
|
|
|
|
def test_database_access():
|
|
"""Testet den Datenbankzugriff nach der Reparatur."""
|
|
print("\nTeste Datenbankzugriff...")
|
|
|
|
try:
|
|
# Models importieren und testen
|
|
from models import get_cached_session, User, Printer, Job
|
|
|
|
with get_cached_session() as session:
|
|
# Test User-Query
|
|
users = session.query(User).limit(5).all()
|
|
print(f"✓ User-Abfrage erfolgreich - {len(users)} Benutzer gefunden")
|
|
|
|
# Test Printer-Query
|
|
printers = session.query(Printer).limit(5).all()
|
|
print(f"✓ Printer-Abfrage erfolgreich - {len(printers)} Drucker gefunden")
|
|
|
|
# Test Job-Query
|
|
jobs = session.query(Job).limit(5).all()
|
|
print(f"✓ Job-Abfrage erfolgreich - {len(jobs)} Jobs gefunden")
|
|
|
|
print("✓ Alle Datenbank-Tests erfolgreich!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"✗ Datenbank-Test fehlgeschlagen: {str(e)}")
|
|
return False
|
|
|
|
def main():
|
|
"""Hauptfunktion für die sofortige Datenbank-Reparatur."""
|
|
print("=== SOFORTIGE DATENBANK-REPARATUR ===")
|
|
print(f"Zeitstempel: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|
print(f"Datenbank: {DATABASE_PATH}")
|
|
print()
|
|
|
|
# Backup erstellen
|
|
if os.path.exists(DATABASE_PATH):
|
|
backup_path = f"{DATABASE_PATH}.backup_immediate_{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-Erstellung fehlgeschlagen: {str(e)}")
|
|
|
|
# Reparatur durchführen
|
|
if fix_users_table_immediate():
|
|
print("\n=== DATENBANK-TEST ===")
|
|
if test_database_access():
|
|
print("\n🎉 DATENBANK-REPARATUR ERFOLGREICH!")
|
|
print("Die Anwendung sollte jetzt funktionieren.")
|
|
else:
|
|
print("\n❌ DATENBANK-TEST FEHLGESCHLAGEN!")
|
|
print("Weitere Diagnose erforderlich.")
|
|
else:
|
|
print("\n❌ DATENBANK-REPARATUR FEHLGESCHLAGEN!")
|
|
print("Manuelle Intervention erforderlich.")
|
|
|
|
if __name__ == "__main__":
|
|
main() |