405 lines
15 KiB
Python
405 lines
15 KiB
Python
#!/usr/bin/env python3.11
|
||
"""
|
||
Development Utilities - Konsolidierte Entwicklungs- und Test-Hilfsfunktionen
|
||
Zusammenfassung von Datenbank-Tests, Session-Fixes und anderen Entwicklungstools
|
||
"""
|
||
|
||
import re
|
||
import os
|
||
import sys
|
||
from utils.logging_config import get_logger
|
||
|
||
# Logger initialisieren
|
||
logger = get_logger("development_utilities")
|
||
|
||
# ===== DATENBANK-TESTS =====
|
||
|
||
def test_database_connectivity():
|
||
"""
|
||
Testet die grundlegende Datenbank-Konnektivität.
|
||
|
||
Returns:
|
||
dict: Test-Ergebnisse
|
||
"""
|
||
try:
|
||
from models import get_cached_session, User, Printer, Job
|
||
|
||
logger.info("=== DATENBANK-KONNEKTIVITÄTS-TEST ===")
|
||
|
||
results = {
|
||
'success': True,
|
||
'tests': {},
|
||
'errors': []
|
||
}
|
||
|
||
with get_cached_session() as session:
|
||
# Test User-Query
|
||
try:
|
||
users = session.query(User).limit(5).all()
|
||
results['tests']['users'] = {
|
||
'success': True,
|
||
'count': len(users),
|
||
'sample': users[0].username if users else None
|
||
}
|
||
logger.info(f"✓ User-Abfrage erfolgreich - {len(users)} Benutzer gefunden")
|
||
|
||
if users:
|
||
user = users[0]
|
||
logger.info(f"✓ Test-User: {user.username} ({user.email})")
|
||
logger.info(f"✓ updated_at-Feld: {user.updated_at}")
|
||
|
||
except Exception as e:
|
||
results['tests']['users'] = {'success': False, 'error': str(e)}
|
||
results['errors'].append(f"User-Test: {str(e)}")
|
||
logger.error(f"❌ User-Test fehlgeschlagen: {str(e)}")
|
||
|
||
# Test Printer-Query
|
||
try:
|
||
printers = session.query(Printer).limit(5).all()
|
||
results['tests']['printers'] = {
|
||
'success': True,
|
||
'count': len(printers),
|
||
'sample': printers[0].name if printers else None
|
||
}
|
||
logger.info(f"✓ Printer-Abfrage erfolgreich - {len(printers)} Drucker gefunden")
|
||
|
||
except Exception as e:
|
||
results['tests']['printers'] = {'success': False, 'error': str(e)}
|
||
results['errors'].append(f"Printer-Test: {str(e)}")
|
||
logger.error(f"❌ Printer-Test fehlgeschlagen: {str(e)}")
|
||
|
||
# Test Job-Query
|
||
try:
|
||
jobs = session.query(Job).limit(5).all()
|
||
results['tests']['jobs'] = {
|
||
'success': True,
|
||
'count': len(jobs),
|
||
'sample': jobs[0].title if jobs else None
|
||
}
|
||
logger.info(f"✓ Job-Abfrage erfolgreich - {len(jobs)} Jobs gefunden")
|
||
|
||
except Exception as e:
|
||
results['tests']['jobs'] = {'success': False, 'error': str(e)}
|
||
results['errors'].append(f"Job-Test: {str(e)}")
|
||
logger.error(f"❌ Job-Test fehlgeschlagen: {str(e)}")
|
||
|
||
if results['errors']:
|
||
results['success'] = False
|
||
logger.error("❌ DATENBANK-TESTS TEILWEISE FEHLGESCHLAGEN")
|
||
else:
|
||
logger.info("🎉 ALLE DATENBANK-TESTS ERFOLGREICH!")
|
||
logger.info("Die Anwendung sollte jetzt ohne Fehler starten.")
|
||
|
||
return results
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ KRITISCHER DATENBANK-TEST-FEHLER: {str(e)}")
|
||
return {
|
||
'success': False,
|
||
'tests': {},
|
||
'errors': [f"Kritischer Fehler: {str(e)}"]
|
||
}
|
||
|
||
def test_database_schema():
|
||
"""
|
||
Testet die Datenbank-Schema-Integrität.
|
||
|
||
Returns:
|
||
dict: Schema-Test-Ergebnisse
|
||
"""
|
||
try:
|
||
from models import get_cached_session, User, Printer, Job, GuestRequest
|
||
from sqlalchemy import inspect
|
||
|
||
logger.info("=== DATENBANK-SCHEMA-TEST ===")
|
||
|
||
results = {
|
||
'success': True,
|
||
'tables': {},
|
||
'errors': []
|
||
}
|
||
|
||
with get_cached_session() as session:
|
||
inspector = inspect(session.bind)
|
||
|
||
# Erwartete Tabellen
|
||
expected_tables = ['users', 'printers', 'jobs', 'guest_requests', 'system_logs']
|
||
|
||
for table_name in expected_tables:
|
||
try:
|
||
if inspector.has_table(table_name):
|
||
columns = inspector.get_columns(table_name)
|
||
results['tables'][table_name] = {
|
||
'exists': True,
|
||
'columns': len(columns),
|
||
'column_names': [col['name'] for col in columns]
|
||
}
|
||
logger.info(f"✓ Tabelle '{table_name}': {len(columns)} Spalten")
|
||
else:
|
||
results['tables'][table_name] = {'exists': False}
|
||
results['errors'].append(f"Tabelle '{table_name}' nicht gefunden")
|
||
logger.error(f"❌ Tabelle '{table_name}' nicht gefunden")
|
||
|
||
except Exception as e:
|
||
results['tables'][table_name] = {'exists': False, 'error': str(e)}
|
||
results['errors'].append(f"Tabelle '{table_name}': {str(e)}")
|
||
logger.error(f"❌ Fehler bei Tabelle '{table_name}': {str(e)}")
|
||
|
||
if results['errors']:
|
||
results['success'] = False
|
||
logger.error("❌ SCHEMA-TESTS TEILWEISE FEHLGESCHLAGEN")
|
||
else:
|
||
logger.info("🎉 ALLE SCHEMA-TESTS ERFOLGREICH!")
|
||
|
||
return results
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ KRITISCHER SCHEMA-TEST-FEHLER: {str(e)}")
|
||
return {
|
||
'success': False,
|
||
'tables': {},
|
||
'errors': [f"Kritischer Fehler: {str(e)}"]
|
||
}
|
||
|
||
# ===== SESSION-FIXES =====
|
||
|
||
def fix_session_usage_in_file(file_path):
|
||
"""
|
||
Behebt Session-Usage in einer Datei durch Konvertierung zu Context Manager Pattern.
|
||
|
||
Args:
|
||
file_path (str): Pfad zur zu reparierenden Datei
|
||
|
||
Returns:
|
||
dict: Reparatur-Ergebnisse
|
||
"""
|
||
try:
|
||
if not os.path.exists(file_path):
|
||
return {
|
||
'success': False,
|
||
'message': f'Datei nicht gefunden: {file_path}',
|
||
'changes_made': False
|
||
}
|
||
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
|
||
# Pattern für direkte Session-Aufrufe
|
||
patterns = [
|
||
# session = get_cached_session() -> with get_cached_session() as session:
|
||
(r'(\s+)session = get_cached_session\(\)', r'\1with get_cached_session() as session:'),
|
||
|
||
# session.close() entfernen (wird automatisch durch Context Manager gemacht)
|
||
(r'\s+session\.close\(\)\s*\n', '\n'),
|
||
]
|
||
|
||
original_content = content
|
||
changes_count = 0
|
||
|
||
for pattern, replacement in patterns:
|
||
new_content = re.sub(pattern, replacement, content, flags=re.MULTILINE)
|
||
|
||
if new_content != content:
|
||
changes_count += 1
|
||
content = new_content
|
||
|
||
# Nur schreiben wenn sich etwas geändert hat
|
||
if content != original_content:
|
||
with open(file_path, 'w', encoding='utf-8') as f:
|
||
f.write(content)
|
||
|
||
logger.info(f"✅ {file_path} wurde aktualisiert ({changes_count} Änderungen)")
|
||
return {
|
||
'success': True,
|
||
'message': f'Datei erfolgreich aktualisiert',
|
||
'changes_made': True,
|
||
'changes_count': changes_count
|
||
}
|
||
else:
|
||
logger.info(f"ℹ️ {file_path} benötigt keine Änderungen")
|
||
return {
|
||
'success': True,
|
||
'message': 'Keine Änderungen erforderlich',
|
||
'changes_made': False,
|
||
'changes_count': 0
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ Fehler beim Reparieren von {file_path}: {str(e)}")
|
||
return {
|
||
'success': False,
|
||
'message': f'Fehler: {str(e)}',
|
||
'changes_made': False
|
||
}
|
||
|
||
def fix_session_usage_bulk(directory_path=None):
|
||
"""
|
||
Repariert Session-Usage in mehreren Dateien.
|
||
|
||
Args:
|
||
directory_path (str): Verzeichnis zum Durchsuchen (Standard: blueprints/)
|
||
|
||
Returns:
|
||
dict: Bulk-Reparatur-Ergebnisse
|
||
"""
|
||
if directory_path is None:
|
||
backend_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
directory_path = os.path.join(backend_dir, 'blueprints')
|
||
|
||
results = {
|
||
'success': True,
|
||
'files_processed': 0,
|
||
'files_changed': 0,
|
||
'total_changes': 0,
|
||
'errors': []
|
||
}
|
||
|
||
try:
|
||
for root, dirs, files in os.walk(directory_path):
|
||
for file in files:
|
||
if file.endswith('.py'):
|
||
file_path = os.path.join(root, file)
|
||
result = fix_session_usage_in_file(file_path)
|
||
|
||
results['files_processed'] += 1
|
||
|
||
if result['success']:
|
||
if result['changes_made']:
|
||
results['files_changed'] += 1
|
||
results['total_changes'] += result.get('changes_count', 0)
|
||
else:
|
||
results['errors'].append(f"{file_path}: {result['message']}")
|
||
results['success'] = False
|
||
|
||
logger.info(f"Bulk-Reparatur abgeschlossen: {results['files_processed']} Dateien verarbeitet, {results['files_changed']} geändert")
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ Fehler bei Bulk-Reparatur: {str(e)}")
|
||
results['success'] = False
|
||
results['errors'].append(f"Bulk-Fehler: {str(e)}")
|
||
|
||
return results
|
||
|
||
# ===== CODE-QUALITÄT =====
|
||
|
||
def analyze_code_quality(file_path):
|
||
"""
|
||
Analysiert die Code-Qualität einer Python-Datei.
|
||
|
||
Args:
|
||
file_path (str): Pfad zur zu analysierenden Datei
|
||
|
||
Returns:
|
||
dict: Code-Qualitäts-Analyse
|
||
"""
|
||
try:
|
||
if not os.path.exists(file_path):
|
||
return {
|
||
'success': False,
|
||
'message': f'Datei nicht gefunden: {file_path}'
|
||
}
|
||
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
content = f.read()
|
||
|
||
lines = content.split('\n')
|
||
|
||
analysis = {
|
||
'success': True,
|
||
'file_path': file_path,
|
||
'metrics': {
|
||
'total_lines': len(lines),
|
||
'code_lines': len([line for line in lines if line.strip() and not line.strip().startswith('#')]),
|
||
'comment_lines': len([line for line in lines if line.strip().startswith('#')]),
|
||
'empty_lines': len([line for line in lines if not line.strip()]),
|
||
'functions': len(re.findall(r'^\s*def\s+\w+', content, re.MULTILINE)),
|
||
'classes': len(re.findall(r'^\s*class\s+\w+', content, re.MULTILINE)),
|
||
'imports': len(re.findall(r'^\s*(import|from)\s+', content, re.MULTILINE))
|
||
},
|
||
'issues': []
|
||
}
|
||
|
||
# Potentielle Probleme identifizieren
|
||
if analysis['metrics']['total_lines'] > 1000:
|
||
analysis['issues'].append('Datei sehr groß (>1000 Zeilen) - Aufteilung empfohlen')
|
||
|
||
if analysis['metrics']['functions'] > 50:
|
||
analysis['issues'].append('Viele Funktionen (>50) - Modularisierung empfohlen')
|
||
|
||
# TODO-Kommentare finden
|
||
todos = re.findall(r'#.*TODO.*', content, re.IGNORECASE)
|
||
if todos:
|
||
analysis['metrics']['todos'] = len(todos)
|
||
analysis['issues'].append(f'{len(todos)} TODO-Kommentare gefunden')
|
||
|
||
return analysis
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ Fehler bei Code-Qualitäts-Analyse von {file_path}: {str(e)}")
|
||
return {
|
||
'success': False,
|
||
'message': f'Fehler: {str(e)}'
|
||
}
|
||
|
||
# ===== CLI INTERFACE =====
|
||
|
||
if __name__ == "__main__":
|
||
if len(sys.argv) > 1:
|
||
command = sys.argv[1]
|
||
|
||
if command == "test-db":
|
||
result = test_database_connectivity()
|
||
if result['success']:
|
||
print("🎉 Datenbank-Tests erfolgreich!")
|
||
else:
|
||
print("❌ Datenbank-Tests fehlgeschlagen:")
|
||
for error in result['errors']:
|
||
print(f" - {error}")
|
||
|
||
elif command == "test-schema":
|
||
result = test_database_schema()
|
||
if result['success']:
|
||
print("🎉 Schema-Tests erfolgreich!")
|
||
else:
|
||
print("❌ Schema-Tests fehlgeschlagen:")
|
||
for error in result['errors']:
|
||
print(f" - {error}")
|
||
|
||
elif command == "fix-sessions":
|
||
if len(sys.argv) > 2:
|
||
file_path = sys.argv[2]
|
||
result = fix_session_usage_in_file(file_path)
|
||
print(f"{'✅' if result['success'] else '❌'} {result['message']}")
|
||
else:
|
||
result = fix_session_usage_bulk()
|
||
print(f"Bulk-Reparatur: {result['files_changed']}/{result['files_processed']} Dateien geändert")
|
||
|
||
elif command == "analyze":
|
||
if len(sys.argv) > 2:
|
||
file_path = sys.argv[2]
|
||
result = analyze_code_quality(file_path)
|
||
if result['success']:
|
||
metrics = result['metrics']
|
||
print(f"=== Code-Analyse: {os.path.basename(file_path)} ===")
|
||
print(f"Zeilen gesamt: {metrics['total_lines']}")
|
||
print(f"Code-Zeilen: {metrics['code_lines']}")
|
||
print(f"Funktionen: {metrics['functions']}")
|
||
print(f"Klassen: {metrics['classes']}")
|
||
if result['issues']:
|
||
print("Probleme:")
|
||
for issue in result['issues']:
|
||
print(f" ⚠️ {issue}")
|
||
else:
|
||
print(f"❌ {result['message']}")
|
||
else:
|
||
print("Verwendung: python3.11 development_utilities.py analyze <datei>")
|
||
|
||
else:
|
||
print("Verfügbare Kommandos:")
|
||
print(" test-db - Testet Datenbank-Konnektivität")
|
||
print(" test-schema - Testet Datenbank-Schema")
|
||
print(" fix-sessions [datei] - Repariert Session-Usage")
|
||
print(" analyze <datei> - Analysiert Code-Qualität")
|
||
else:
|
||
print("Verwendung: python3.11 development_utilities.py <command>")
|
||
print("Verfügbare Kommandos: test-db, test-schema, fix-sessions, analyze") |