Die Dateien wurden geändert oder hinzugefügt:

This commit is contained in:
2025-06-13 07:17:00 +02:00
parent ae95d82afc
commit 691a4f2d41
137 changed files with 2354 additions and 29 deletions

View File

@ -691,6 +691,16 @@ def inject_now():
"""Injiziert die aktuelle Zeit in alle Templates""" """Injiziert die aktuelle Zeit in alle Templates"""
return {'now': datetime.now} return {'now': datetime.now}
@app.context_processor
def inject_current_route():
"""
Stellt current_route für alle Templates bereit.
Verhindert Template-Fehler wenn request.endpoint None ist (z.B. bei 404-Fehlern).
"""
current_route = getattr(request, 'endpoint', None) or ''
return {'current_route': current_route}
@app.template_filter('format_datetime') @app.template_filter('format_datetime')
def format_datetime_filter(value, format='%d.%m.%Y %H:%M'): def format_datetime_filter(value, format='%d.%m.%Y %H:%M'):
"""Template-Filter für Datums-Formatierung""" """Template-Filter für Datums-Formatierung"""
@ -1093,7 +1103,8 @@ def api_stats():
try: try:
from models import get_db_session, User, Printer, Job from models import get_db_session, User, Printer, Job
with get_cached_session() as db_session: db_session = get_db_session()
try:
# Grundlegende Counts # Grundlegende Counts
total_users = db_session.query(User).count() total_users = db_session.query(User).count()
total_printers = db_session.query(Printer).count() total_printers = db_session.query(Printer).count()
@ -1116,6 +1127,8 @@ def api_stats():
online_printers = db_session.query(Printer).filter( online_printers = db_session.query(Printer).filter(
Printer.active == True Printer.active == True
).count() ).count()
finally:
db_session.close()
stats = { stats = {
'total_users': total_users, 'total_users': total_users,
@ -1199,7 +1212,13 @@ def forbidden_error(error):
"message": "Sie haben keine Berechtigung für diese Aktion", "message": "Sie haben keine Berechtigung für diese Aktion",
"status_code": 403 "status_code": 403
}), 403 }), 403
try:
return render_template('errors/403.html'), 403 return render_template('errors/403.html'), 403
except Exception as template_error:
# Fallback bei Template-Fehlern
app_logger.error(f"Template-Fehler in 403-Handler: {str(template_error)}")
return f"<h1>403 - Zugriff verweigert</h1><p>Sie haben keine Berechtigung für diese Aktion.</p>", 403
@app.errorhandler(404) @app.errorhandler(404)
def not_found_error(error): def not_found_error(error):
@ -1211,7 +1230,13 @@ def not_found_error(error):
"message": "Die angeforderte Ressource wurde nicht gefunden", "message": "Die angeforderte Ressource wurde nicht gefunden",
"status_code": 404 "status_code": 404
}), 404 }), 404
try:
return render_template('errors/404.html'), 404 return render_template('errors/404.html'), 404
except Exception as template_error:
# Fallback bei Template-Fehlern
app_logger.error(f"Template-Fehler in 404-Handler: {str(template_error)}")
return f"<h1>404 - Nicht gefunden</h1><p>Die angeforderte Seite wurde nicht gefunden.</p>", 404
@app.errorhandler(405) @app.errorhandler(405)
def method_not_allowed_error(error): def method_not_allowed_error(error):
@ -1271,7 +1296,12 @@ def internal_error(error):
"status_code": 500 "status_code": 500
}), 500 }), 500
try:
return render_template('errors/500.html', error_id=error_id), 500 return render_template('errors/500.html', error_id=error_id), 500
except Exception as template_error:
# Fallback bei Template-Fehlern
app_logger.error(f"Template-Fehler in 500-Handler: {str(template_error)}")
return f"<h1>500 - Interner Serverfehler</h1><p>Ein unerwarteter Fehler ist aufgetreten. Fehler-ID: {error_id}</p>", 500
@app.errorhandler(502) @app.errorhandler(502)
def bad_gateway_error(error): def bad_gateway_error(error):
@ -1338,7 +1368,12 @@ def handle_exception(error):
"status_code": 500 "status_code": 500
}), 500 }), 500
try:
return render_template('errors/500.html', error_id=error_id), 500 return render_template('errors/500.html', error_id=error_id), 500
except Exception as template_error:
# Fallback bei Template-Fehlern
app_logger.error(f"Template-Fehler im Exception-Handler: {str(template_error)}")
return f"<h1>500 - Unerwarteter Fehler</h1><p>Ein unerwarteter Fehler ist aufgetreten. Fehler-ID: {error_id}</p>", 500
# ===== HAUPTFUNKTION ===== # ===== HAUPTFUNKTION =====
def main(): def main():
@ -1506,5 +1541,17 @@ def production_info():
"""Stellt Production-Informationen für Templates bereit""" """Stellt Production-Informationen für Templates bereit"""
return get_production_info() return get_production_info()
# Nach der Initialisierung der Blueprints und vor dem App-Start
try:
# Admin-Berechtigungen beim Start korrigieren
from utils.permissions import fix_all_admin_permissions
result = fix_all_admin_permissions()
if result['success']:
app_logger.info(f"Admin-Berechtigungen beim Start korrigiert: {result['created']} erstellt, {result['corrected']} aktualisiert")
else:
app_logger.warning(f"Fehler beim Korrigieren der Admin-Berechtigungen: {result.get('error', 'Unbekannter Fehler')}")
except Exception as e:
app_logger.error(f"Fehler beim Korrigieren der Admin-Berechtigungen beim Start: {str(e)}")
if __name__ == "__main__": if __name__ == "__main__":
main() main()

Binary file not shown.

View File

@ -1824,3 +1824,655 @@ def get_status_color(status):
'unknown': '#6b7280' # Grau 'unknown': '#6b7280' # Grau
} }
return status_colors.get(status, '#6b7280') return status_colors.get(status, '#6b7280')
# ===== FEHLENDE API-ROUTEN HINZUFÜGEN =====
@admin_api_blueprint.route('/system-health', methods=['GET'])
@admin_required
def api_admin_system_health_alias():
"""
Alias-Route für system-health (Kompatibilität mit Frontend).
Leitet Anfragen an die bestehende system/health Route weiter.
"""
return api_admin_system_health()
@admin_api_blueprint.route('/error-recovery/status', methods=['GET'])
@admin_required
def api_admin_error_recovery_status():
"""
API-Endpunkt für Error-Recovery-Status.
Gibt Informationen über das Error-Recovery-System zurück,
einschließlich Status, Statistiken und letzter Aktionen.
"""
try:
admin_api_logger.info(f"Error-Recovery-Status angefordert von {current_user.username}")
# Error-Recovery-Basis-Status sammeln
recovery_status = {
'enabled': True, # Error-Recovery ist standardmäßig aktiviert
'last_check': datetime.now().isoformat(),
'status': 'active',
'errors_detected': 0,
'errors_recovered': 0,
'last_recovery_action': None,
'monitoring_active': True,
'recovery_methods': [
'automatic_restart',
'service_health_check',
'database_recovery',
'cache_cleanup'
]
}
# Versuche Log-Informationen zu sammeln
try:
# Prüfe auf kürzliche Fehler in System-Logs
with get_cached_session() as db_session:
# Letzte Stunde nach Error-Logs suchen
last_hour = datetime.now() - timedelta(hours=1)
error_logs = db_session.query(SystemLog).filter(
SystemLog.level == 'ERROR',
SystemLog.timestamp >= last_hour
).count()
recovery_logs = db_session.query(SystemLog).filter(
SystemLog.message.like('%Recovery%'),
SystemLog.timestamp >= last_hour
).count()
recovery_status['errors_detected'] = error_logs
recovery_status['errors_recovered'] = recovery_logs
# Letzten Recovery-Eintrag finden
last_recovery = db_session.query(SystemLog).filter(
SystemLog.message.like('%Recovery%')
).order_by(SystemLog.timestamp.desc()).first()
if last_recovery:
recovery_status['last_recovery_action'] = {
'timestamp': last_recovery.timestamp.isoformat(),
'action': 'system_log_recovery',
'message': last_recovery.message,
'module': last_recovery.module
}
except Exception as log_error:
admin_api_logger.warning(f"Log-Analyse für Error-Recovery fehlgeschlagen: {str(log_error)}")
recovery_status['errors_detected'] = 0
recovery_status['errors_recovered'] = 0
# System-Load als Indikator für potenzielle Probleme
try:
import psutil
cpu_percent = psutil.cpu_percent(interval=1)
memory_percent = psutil.virtual_memory().percent
# Hohe System-Last kann auf Probleme hindeuten
if cpu_percent > 80 or memory_percent > 85:
recovery_status['status'] = 'warning'
recovery_status['last_recovery_action'] = {
'timestamp': datetime.now().isoformat(),
'action': 'system_load_warning',
'details': {
'cpu_percent': cpu_percent,
'memory_percent': memory_percent
}
}
# System-Performance-Daten hinzufügen
recovery_status['system_performance'] = {
'cpu_percent': cpu_percent,
'memory_percent': memory_percent,
'status': 'normal' if cpu_percent < 80 and memory_percent < 85 else 'high_load'
}
except ImportError:
admin_api_logger.info("psutil nicht verfügbar für Error-Recovery-Monitoring")
recovery_status['system_performance'] = {
'available': False,
'message': 'psutil-Bibliothek nicht installiert'
}
except Exception as system_error:
admin_api_logger.warning(f"System-Load-Check für Error-Recovery fehlgeschlagen: {str(system_error)}")
recovery_status['system_performance'] = {
'available': False,
'error': str(system_error)
}
# Datenbank-Gesundheit als Recovery-Indikator
try:
with get_cached_session() as db_session:
# Einfacher DB-Test
db_session.execute("SELECT 1")
recovery_status['database_health'] = 'healthy'
except Exception as db_error:
recovery_status['database_health'] = 'unhealthy'
recovery_status['status'] = 'critical'
admin_api_logger.error(f"Datenbank-Health-Check für Error-Recovery fehlgeschlagen: {str(db_error)}")
admin_api_logger.info(f"Error-Recovery-Status abgerufen: {recovery_status['status']}")
return jsonify({
'success': True,
'error_recovery': recovery_status,
'message': f"Error-Recovery-Status: {recovery_status['status']}"
})
except Exception as e:
admin_api_logger.error(f"Fehler beim Abrufen des Error-Recovery-Status: {str(e)}")
return jsonify({
'success': False,
'error': 'Error-Recovery-Status nicht verfügbar',
'details': str(e),
'error_recovery': {
'status': 'error',
'enabled': False,
'last_check': datetime.now().isoformat()
}
}), 500
# ===== ERWEITERTE TAPO-STECKDOSEN-VERWALTUNG =====
@admin_blueprint.route("/tapo-monitoring")
@admin_required
def tapo_monitoring():
"""
Erweiterte Tapo-Steckdosen-Überwachung für Administratoren.
Bietet Real-Time-Monitoring aller Drucker-Steckdosen mit automatischer Überprüfung.
"""
admin_logger.info(f"Tapo-Monitoring aufgerufen von {current_user.username}")
try:
with get_cached_session() as db_session:
# Alle Drucker mit konfigurierten Steckdosen laden
printers_with_plugs = db_session.query(Printer).filter(
Printer.plug_ip.isnot(None),
Printer.active == True
).all()
# Grundlegende Statistiken
total_printers = db_session.query(Printer).count()
printers_with_tapo = len(printers_with_plugs)
# Aktueller Status aller Tapo-Steckdosen abrufen
try:
from utils.hardware_integration import tapo_controller
tapo_available = True
# Status für jeden Drucker mit Tapo-Steckdose abrufen
printer_status = []
online_count = 0
offline_count = 0
error_count = 0
for printer in printers_with_plugs:
try:
reachable, status = tapo_controller.check_outlet_status(
printer.plug_ip,
printer_id=printer.id
)
if reachable:
if status == 'on':
online_count += 1
status_class = 'success'
else:
offline_count += 1
status_class = 'secondary'
else:
error_count += 1
status_class = 'danger'
status = 'unreachable'
# Aktuelle Jobs für diesen Drucker prüfen
active_jobs = db_session.query(Job).filter(
Job.printer_id == printer.id,
Job.status.in_(['running', 'printing', 'active', 'scheduled'])
).count()
printer_info = {
'id': printer.id,
'name': printer.name,
'model': printer.model,
'location': printer.location,
'plug_ip': printer.plug_ip,
'plug_status': status,
'plug_reachable': reachable,
'status_class': status_class,
'active_jobs': active_jobs,
'last_checked': datetime.now(),
'has_issues': not reachable or active_jobs > 0
}
printer_status.append(printer_info)
except Exception as e:
admin_logger.error(f"Fehler beim Status-Check für {printer.name}: {str(e)}")
error_count += 1
printer_status.append({
'id': printer.id,
'name': printer.name,
'model': printer.model,
'location': printer.location,
'plug_ip': printer.plug_ip,
'plug_status': 'error',
'plug_reachable': False,
'status_class': 'danger',
'active_jobs': 0,
'last_checked': datetime.now(),
'has_issues': True,
'error': str(e)
})
except Exception as e:
admin_logger.error(f"Tapo-Controller nicht verfügbar: {str(e)}")
tapo_available = False
printer_status = []
online_count = offline_count = error_count = 0
# Statistiken zusammenstellen
monitoring_stats = {
'total_printers': total_printers,
'printers_with_tapo': printers_with_tapo,
'tapo_available': tapo_available,
'online_count': online_count,
'offline_count': offline_count,
'error_count': error_count,
'coverage_percentage': round((printers_with_tapo / total_printers * 100), 1) if total_printers > 0 else 0
}
admin_logger.info(f"Tapo-Monitoring geladen: {printers_with_tapo} Steckdosen, {online_count} online")
return render_template('admin_tapo_monitoring.html',
printer_status=printer_status,
stats=monitoring_stats,
page_title="Tapo-Steckdosen-Monitoring",
breadcrumb=[
{"name": "Admin-Dashboard", "url": url_for("admin.admin_dashboard")},
{"name": "Tapo-Monitoring", "url": "#"}
])
except Exception as e:
admin_logger.error(f"Fehler beim Laden des Tapo-Monitorings: {str(e)}")
flash("Fehler beim Laden der Tapo-Monitoring-Daten.", "error")
return redirect(url_for("admin.admin_dashboard"))
@admin_api_blueprint.route('/tapo/bulk-control', methods=['POST'])
@admin_required
def api_admin_bulk_tapo_control():
"""
API-Endpunkt für Massensteuerung von Tapo-Steckdosen.
Ermöglicht das gleichzeitige Ein-/Ausschalten mehrerer Steckdosen.
"""
admin_api_logger.info(f"Bulk-Tapo-Steuerung von {current_user.username}")
try:
data = request.get_json()
action = data.get('action') # 'on', 'off', 'status'
printer_ids = data.get('printer_ids', [])
if not action or not printer_ids:
return jsonify({
'success': False,
'error': 'Aktion und Drucker-IDs sind erforderlich'
}), 400
if action not in ['on', 'off', 'status']:
return jsonify({
'success': False,
'error': 'Ungültige Aktion. Erlaubt: on, off, status'
}), 400
# Tapo-Controller laden
try:
from utils.hardware_integration import tapo_controller
except Exception as e:
return jsonify({
'success': False,
'error': f'Tapo-Controller nicht verfügbar: {str(e)}'
}), 500
results = []
success_count = 0
error_count = 0
with get_cached_session() as db_session:
for printer_id in printer_ids:
try:
printer = db_session.query(Printer).filter(Printer.id == printer_id).first()
if not printer:
results.append({
'printer_id': printer_id,
'success': False,
'error': 'Drucker nicht gefunden'
})
error_count += 1
continue
if not printer.plug_ip:
results.append({
'printer_id': printer_id,
'printer_name': printer.name,
'success': False,
'error': 'Keine Steckdose konfiguriert'
})
error_count += 1
continue
# Aktion ausführen
if action == 'status':
reachable, status = tapo_controller.check_outlet_status(
printer.plug_ip,
printer_id=printer_id
)
results.append({
'printer_id': printer_id,
'printer_name': printer.name,
'success': True,
'status': status,
'reachable': reachable
})
success_count += 1
else:
# Ein- oder Ausschalten
state = action == 'on'
success = tapo_controller.toggle_plug(printer.plug_ip, state)
if success:
# Drucker-Status in DB aktualisieren
printer.status = 'starting' if state else 'offline'
printer.last_checked = datetime.now()
results.append({
'printer_id': printer_id,
'printer_name': printer.name,
'success': True,
'action': action,
'message': f'Steckdose erfolgreich {"ein" if state else "aus"}geschaltet'
})
success_count += 1
else:
results.append({
'printer_id': printer_id,
'printer_name': printer.name,
'success': False,
'error': f'Steckdose konnte nicht {"ein" if state else "aus"}geschaltet werden'
})
error_count += 1
except Exception as e:
admin_api_logger.error(f"Fehler bei Bulk-Steuerung für Drucker {printer_id}: {str(e)}")
results.append({
'printer_id': printer_id,
'success': False,
'error': str(e)
})
error_count += 1
# Änderungen speichern
if action in ['on', 'off']:
db_session.commit()
admin_api_logger.info(f"Bulk-Tapo-Steuerung abgeschlossen: {success_count} erfolgreich, {error_count} Fehler")
return jsonify({
'success': True,
'results': results,
'summary': {
'total': len(printer_ids),
'success': success_count,
'errors': error_count
},
'timestamp': datetime.now().isoformat()
})
except Exception as e:
admin_api_logger.error(f"Unerwarteter Fehler bei Bulk-Tapo-Steuerung: {str(e)}")
return jsonify({
'success': False,
'error': f'Systemfehler: {str(e)}'
}), 500
@admin_api_blueprint.route('/tapo/health-check', methods=['POST'])
@admin_required
def api_admin_tapo_health_check():
"""
Führt eine umfassende Gesundheitsüberprüfung aller Tapo-Steckdosen durch.
Testet Konnektivität, Authentifizierung und Funktionsfähigkeit.
"""
admin_api_logger.info(f"Tapo-Gesundheitscheck von {current_user.username}")
try:
# Tapo-Controller laden
try:
from utils.hardware_integration import tapo_controller
tapo_available = True
except Exception as e:
return jsonify({
'success': False,
'error': f'Tapo-Controller nicht verfügbar: {str(e)}',
'tapo_available': False
}), 500
health_results = {
'overall_status': 'healthy',
'tapo_available': tapo_available,
'timestamp': datetime.now().isoformat(),
'printers': [],
'summary': {
'total': 0,
'healthy': 0,
'warning': 0,
'critical': 0
},
'recommendations': []
}
with get_cached_session() as db_session:
# Alle Drucker mit Steckdosen laden
printers_with_plugs = db_session.query(Printer).filter(
Printer.plug_ip.isnot(None)
).all()
health_results['summary']['total'] = len(printers_with_plugs)
for printer in printers_with_plugs:
printer_health = {
'id': printer.id,
'name': printer.name,
'plug_ip': printer.plug_ip,
'status': 'unknown',
'issues': [],
'checks': {
'connectivity': False,
'authentication': False,
'functionality': False
}
}
try:
# Check 1: Konnektivität (Ping)
ping_success = tapo_controller.ping_address(printer.plug_ip, timeout=3)
printer_health['checks']['connectivity'] = ping_success
if not ping_success:
printer_health['issues'].append('Netzwerkverbindung fehlgeschlagen')
# Check 2: Authentifizierung und Geräteinformationen
if ping_success:
try:
test_result = tapo_controller.test_connection(printer.plug_ip)
printer_health['checks']['authentication'] = test_result['success']
if not test_result['success']:
printer_health['issues'].append(f'Authentifizierung fehlgeschlagen: {test_result.get("error", "Unbekannt")}')
except Exception as auth_error:
printer_health['issues'].append(f'Authentifizierungstest fehlgeschlagen: {str(auth_error)}')
# Check 3: Funktionalität (Status abrufen)
if printer_health['checks']['authentication']:
try:
reachable, status = tapo_controller.check_outlet_status(
printer.plug_ip,
printer_id=printer.id
)
printer_health['checks']['functionality'] = reachable
printer_health['current_status'] = status
if not reachable:
printer_health['issues'].append('Status-Abfrage fehlgeschlagen')
except Exception as func_error:
printer_health['issues'].append(f'Funktionstest fehlgeschlagen: {str(func_error)}')
# Gesamtstatus bewerten
if len(printer_health['issues']) == 0:
printer_health['status'] = 'healthy'
health_results['summary']['healthy'] += 1
elif len(printer_health['issues']) <= 1:
printer_health['status'] = 'warning'
health_results['summary']['warning'] += 1
else:
printer_health['status'] = 'critical'
health_results['summary']['critical'] += 1
# Aktuelle Jobs prüfen (für Sicherheitswarnungen)
active_jobs = db_session.query(Job).filter(
Job.printer_id == printer.id,
Job.status.in_(['running', 'printing', 'active'])
).count()
if active_jobs > 0:
printer_health['active_jobs'] = active_jobs
printer_health['issues'].append(f'{active_jobs} aktive(r) Job(s) - Vorsicht bei Steckdosen-Änderungen')
except Exception as e:
admin_api_logger.error(f"Fehler beim Gesundheitscheck für {printer.name}: {str(e)}")
printer_health['status'] = 'critical'
printer_health['issues'].append(f'Systemfehler: {str(e)}')
health_results['summary']['critical'] += 1
health_results['printers'].append(printer_health)
# Gesamtstatus und Empfehlungen bestimmen
if health_results['summary']['critical'] > 0:
health_results['overall_status'] = 'critical'
health_results['recommendations'].append('Kritische Probleme bei Tapo-Steckdosen beheben')
elif health_results['summary']['warning'] > 0:
health_results['overall_status'] = 'warning'
health_results['recommendations'].append('Warnungen bei Tapo-Steckdosen überprüfen')
# Zusätzliche Empfehlungen
coverage = (len(printers_with_plugs) / db_session.query(Printer).count()) * 100 if db_session.query(Printer).count() > 0 else 0
if coverage < 80:
health_results['recommendations'].append(f'Tapo-Abdeckung nur {coverage:.1f}% - weitere Steckdosen konfigurieren')
admin_api_logger.info(f"Tapo-Gesundheitscheck abgeschlossen: {health_results['summary']}")
return jsonify(health_results)
except Exception as e:
admin_api_logger.error(f"Unerwarteter Fehler beim Tapo-Gesundheitscheck: {str(e)}")
return jsonify({
'success': False,
'error': 'Fehler beim Health-Check',
'message': str(e),
'health': {
'overall': 'error',
'timestamp': datetime.now().isoformat()
}
}), 500
@admin_api_blueprint.route('/printers/tapo-configure', methods=['POST'])
@admin_required
def api_admin_configure_printer_tapo():
"""
Konfiguriert oder aktualisiert die Tapo-Steckdosen-Einstellungen für einen Drucker.
"""
admin_api_logger.info(f"Tapo-Konfiguration von {current_user.username}")
try:
data = request.get_json()
printer_id = data.get('printer_id')
plug_ip = data.get('plug_ip')
plug_username = data.get('plug_username')
plug_password = data.get('plug_password')
test_connection = data.get('test_connection', True)
if not printer_id:
return jsonify({
'success': False,
'error': 'Drucker-ID ist erforderlich'
}), 400
with get_cached_session() as db_session:
printer = db_session.query(Printer).filter(Printer.id == printer_id).first()
if not printer:
return jsonify({
'success': False,
'error': 'Drucker nicht gefunden'
}), 404
# Tapo-Einstellungen aktualisieren
if plug_ip:
try:
import ipaddress
ipaddress.ip_address(plug_ip)
printer.plug_ip = plug_ip
except ValueError:
return jsonify({
'success': False,
'error': 'Ungültige IP-Adresse'
}), 400
if plug_username:
printer.plug_username = plug_username
if plug_password:
printer.plug_password = plug_password
# Verbindung testen falls gewünscht
test_result = None
if test_connection and printer.plug_ip:
try:
from utils.hardware_integration import tapo_controller
test_result = tapo_controller.test_connection(
printer.plug_ip,
username=printer.plug_username,
password=printer.plug_password
)
if test_result['success']:
printer.last_checked = datetime.now()
printer.status = 'online'
else:
admin_api_logger.warning(f"Tapo-Test für {printer.name} fehlgeschlagen: {test_result.get('error')}")
except Exception as e:
test_result = {
'success': False,
'error': f'Test fehlgeschlagen: {str(e)}'
}
db_session.commit()
admin_api_logger.info(f"Tapo-Konfiguration für {printer.name} aktualisiert")
return jsonify({
'success': True,
'message': f'Tapo-Einstellungen für {printer.name} erfolgreich aktualisiert',
'printer_id': printer_id,
'test_result': test_result,
'timestamp': datetime.now().isoformat()
})
except Exception as e:
admin_api_logger.error(f"Fehler bei Tapo-Konfiguration: {str(e)}")
return jsonify({
'success': False,
'error': f'Systemfehler: {str(e)}'
}), 500

View File

@ -520,3 +520,39 @@ def get_error_recovery_status():
'error': 'Fehler beim Laden des Wiederherstellungs-Status', 'error': 'Fehler beim Laden des Wiederherstellungs-Status',
'message': str(e) 'message': str(e)
}), 500 }), 500
@api_blueprint.route('/admin/fix-permissions', methods=['POST'])
@admin_required
def fix_admin_permissions():
"""
Korrigiert die Admin-Berechtigungen im System.
Nur für Administratoren zugänglich.
"""
try:
from utils.permissions import fix_all_admin_permissions
result = fix_all_admin_permissions()
if result['success']:
api_logger.info(f"Admin-Berechtigungen korrigiert von {current_user.username}: {result['created']} erstellt, {result['corrected']} aktualisiert")
return jsonify({
'success': True,
'message': 'Admin-Berechtigungen erfolgreich korrigiert',
'details': result
})
else:
api_logger.error(f"Fehler beim Korrigieren der Admin-Berechtigungen: {result['error']}")
return jsonify({
'success': False,
'error': 'Fehler beim Korrigieren der Berechtigungen',
'message': result['error']
}), 500
except Exception as e:
api_logger.error(f"Fehler beim Korrigieren der Admin-Berechtigungen: {str(e)}")
return jsonify({
'success': False,
'error': 'Fehler beim Korrigieren der Berechtigungen',
'message': str(e)
}), 500

View File

@ -32,23 +32,8 @@ class GuestRequestForm(FlaskForm):
'3D-Dateien sind erlaubt!')]) '3D-Dateien sind erlaubt!')])
# Hilfsfunktionen # Hilfsfunktionen
def can_approve_jobs(user_id): # Importiere Berechtigungsfunktionen aus utils.permissions
"""Prüft, ob ein Benutzer Anfragen genehmigen darf.""" from utils.permissions import can_approve_jobs, approver_required
with get_cached_session() as db_session:
permission = db_session.query(UserPermission).filter_by(user_id=user_id).first()
if not permission:
return False
return permission.can_approve_jobs
def approver_required(f):
"""Decorator zur Prüfung der Genehmigungsberechtigung."""
@wraps(f)
@login_required
def decorated_function(*args, **kwargs):
if not can_approve_jobs(current_user.id):
abort(403, "Keine Berechtigung zum Genehmigen von Anfragen")
return f(*args, **kwargs)
return decorated_function
# Gast-Routen # Gast-Routen
@guest_blueprint.route('/request', methods=['GET', 'POST']) @guest_blueprint.route('/request', methods=['GET', 'POST'])

View File

@ -18,7 +18,7 @@ from typing import Dict, List, Tuple, Any, Optional
from models import Printer, User, Job, get_db_session from models import Printer, User, Job, get_db_session
from utils.logging_config import get_logger, measure_execution_time from utils.logging_config import get_logger, measure_execution_time
from utils.security_suite import require_permission, Permission, check_permission from utils.security_suite import require_permission, Permission, check_permission
from utils.hardware_integration import printer_monitor from utils.hardware_integration import printer_monitor, tapo_controller
from utils.drag_drop_system import drag_drop_manager from utils.drag_drop_system import drag_drop_manager
# Logger initialisieren # Logger initialisieren
@ -1048,3 +1048,526 @@ def get_drag_drop_config():
# ============================================================================= # =============================================================================
# ENDE DRAG & DROP API # ENDE DRAG & DROP API
# ============================================================================= # =============================================================================
@printers_blueprint.route("/tapo/status-check", methods=["POST"])
@login_required
@require_permission(Permission.CONTROL_PRINTER)
@measure_execution_time(logger=printers_logger, task_name="API-Massenhafte-Tapo-Status-Prüfung")
def mass_tapo_status_check():
"""
Führt eine vollständige Tapo-Status-Überprüfung für alle Drucker durch.
Returns:
JSON mit detailliertem Status aller Tapo-Steckdosen
"""
printers_logger.info(f"Massenhafte Tapo-Status-Prüfung von Benutzer {current_user.name}")
try:
db_session = get_db_session()
# Alle Drucker laden
all_printers = db_session.query(Printer).all()
# Tapo-Controller laden
try:
from utils.hardware_integration import tapo_controller
tapo_available = True
except Exception as e:
db_session.close()
return jsonify({
"success": False,
"error": f"Tapo-Controller nicht verfügbar: {str(e)}",
"tapo_available": False
}), 500
printer_status = []
summary = {
"total_printers": len(all_printers),
"printers_with_tapo": 0,
"printers_without_tapo": 0,
"tapo_online": 0,
"tapo_offline": 0,
"tapo_unreachable": 0,
"configuration_issues": 0
}
for printer in all_printers:
printer_info = {
"id": printer.id,
"name": printer.name,
"model": printer.model,
"location": printer.location,
"active": printer.active,
"has_tapo_config": bool(printer.plug_ip),
"plug_ip": printer.plug_ip,
"last_checked": datetime.now()
}
if not printer.plug_ip:
# Drucker ohne Tapo-Konfiguration
summary["printers_without_tapo"] += 1
printer_info.update({
"tapo_status": "not_configured",
"tapo_reachable": False,
"power_status": None,
"recommendations": ["Tapo-Steckdose konfigurieren für automatische Steuerung"]
})
else:
# Drucker mit Tapo-Konfiguration
summary["printers_with_tapo"] += 1
# Konfigurationsprüfung
config_issues = []
if not printer.plug_username:
config_issues.append("Tapo-Benutzername fehlt")
if not printer.plug_password:
config_issues.append("Tapo-Passwort fehlt")
if config_issues:
summary["configuration_issues"] += 1
printer_info.update({
"tapo_status": "configuration_error",
"tapo_reachable": False,
"power_status": None,
"config_issues": config_issues,
"recommendations": ["Tapo-Anmeldedaten vervollständigen"]
})
else:
# Vollständige Konfiguration - Status prüfen
try:
reachable, status = tapo_controller.check_outlet_status(
printer.plug_ip,
printer_id=printer.id
)
if reachable:
if status == "on":
summary["tapo_online"] += 1
status_type = "online"
recommendations = []
else:
summary["tapo_offline"] += 1
status_type = "offline"
recommendations = ["Steckdose kann bei Bedarf eingeschaltet werden"]
else:
summary["tapo_unreachable"] += 1
status_type = "unreachable"
recommendations = ["Netzwerkverbindung prüfen", "IP-Adresse überprüfen"]
printer_info.update({
"tapo_status": status_type,
"tapo_reachable": reachable,
"power_status": status,
"recommendations": recommendations
})
# Drucker-Status in DB aktualisieren
if reachable:
printer.last_checked = datetime.now()
if status == "on":
printer.status = "online"
else:
printer.status = "offline"
else:
printer.status = "unreachable"
except Exception as tapo_error:
summary["tapo_unreachable"] += 1
printer_info.update({
"tapo_status": "error",
"tapo_reachable": False,
"power_status": None,
"error": str(tapo_error),
"recommendations": ["Tapo-Verbindung prüfen", "Anmeldedaten überprüfen"]
})
printers_logger.warning(f"Tapo-Fehler für {printer.name}: {str(tapo_error)}")
# Aktuelle Jobs für zusätzliche Info
active_jobs = db_session.query(Job).filter(
Job.printer_id == printer.id,
Job.status.in_(["running", "printing", "active", "scheduled"])
).count()
printer_info["active_jobs"] = active_jobs
if active_jobs > 0:
printer_info.setdefault("recommendations", []).append(
f"Vorsicht: {active_jobs} aktive Job(s) bei Steckdosen-Änderungen"
)
printer_status.append(printer_info)
# Änderungen in DB speichern
db_session.commit()
db_session.close()
# Übersicht der Ergebnisse
coverage_percentage = (summary["printers_with_tapo"] / summary["total_printers"] * 100) if summary["total_printers"] > 0 else 0
health_score = ((summary["tapo_online"] + summary["tapo_offline"]) / summary["printers_with_tapo"] * 100) if summary["printers_with_tapo"] > 0 else 0
printers_logger.info(f"Tapo-Status-Check abgeschlossen: {summary['printers_with_tapo']} konfiguriert, "
f"{summary['tapo_online']} online, {summary['tapo_unreachable']} nicht erreichbar")
return jsonify({
"success": True,
"tapo_available": tapo_available,
"printers": printer_status,
"summary": summary,
"metrics": {
"coverage_percentage": round(coverage_percentage, 1),
"health_score": round(health_score, 1),
"needs_attention": summary["configuration_issues"] + summary["tapo_unreachable"]
},
"timestamp": datetime.now().isoformat()
})
except Exception as e:
printers_logger.error(f"Unerwarteter Fehler bei Massenhafte-Tapo-Status-Prüfung: {str(e)}")
if 'db_session' in locals():
db_session.close()
return jsonify({
"success": False,
"error": f"Systemfehler: {str(e)}"
}), 500
@printers_blueprint.route("/tapo/configuration-wizard", methods=["POST"])
@login_required
@require_permission(Permission.ADMIN)
@measure_execution_time(logger=printers_logger, task_name="API-Tapo-Konfigurationsassistent")
def tapo_configuration_wizard():
"""
Automatischer Konfigurationsassistent für Tapo-Steckdosen.
Versucht automatisch verfügbare Steckdosen zu erkennen und zu konfigurieren.
"""
printers_logger.info(f"Tapo-Konfigurationsassistent von Admin {current_user.name}")
try:
data = request.get_json()
auto_configure = data.get('auto_configure', True)
test_ips = data.get('test_ips', [])
# Tapo-Controller laden
try:
from utils.hardware_integration import tapo_controller
except Exception as e:
return jsonify({
"success": False,
"error": f"Tapo-Controller nicht verfügbar: {str(e)}"
}), 500
db_session = get_db_session()
# Standard-IP-Bereich für Mercedes-Benz TBA (normalerweise 192.168.1.201-206)
if not test_ips:
test_ips = [f"192.168.1.{i}" for i in range(201, 207)] # 6 Standard-Arbeitsplätze
discovery_results = {
"tested_ips": test_ips,
"discovered_devices": [],
"configured_printers": [],
"errors": []
}
printers_logger.info(f"Teste {len(test_ips)} IP-Adressen auf Tapo-Geräte...")
# Discovery für jede IP-Adresse
for ip in test_ips:
try:
printers_logger.debug(f"Teste IP: {ip}")
# Ping-Test
if not tapo_controller.ping_address(ip, timeout=3):
discovery_results["errors"].append(f"{ip}: Nicht erreichbar (Ping fehlgeschlagen)")
continue
# Tapo-Verbindungstest
test_result = tapo_controller.test_connection(ip)
if test_result["success"]:
device_info = test_result.get("device_info", {})
discovered_device = {
"ip": ip,
"device_info": device_info,
"nickname": device_info.get("nickname", f"Tapo Device {ip}"),
"model": device_info.get("model", "Unknown"),
"device_on": device_info.get("device_on", False)
}
discovery_results["discovered_devices"].append(discovered_device)
printers_logger.info(f"✅ Tapo-Gerät gefunden: {ip} - {discovered_device['nickname']}")
# Auto-Konfiguration wenn gewünscht
if auto_configure:
# Suche nach Drucker ohne Tapo-Konfiguration
unconfigured_printer = db_session.query(Printer).filter(
Printer.plug_ip.is_(None),
Printer.active == True
).first()
if unconfigured_printer:
# Konfiguriere den ersten verfügbaren Drucker
unconfigured_printer.plug_ip = ip
unconfigured_printer.plug_username = "admin" # Standard für Tapo
unconfigured_printer.plug_password = "admin" # Standard für Tapo
unconfigured_printer.last_checked = datetime.now()
configured_info = {
"printer_id": unconfigured_printer.id,
"printer_name": unconfigured_printer.name,
"tapo_ip": ip,
"tapo_nickname": discovered_device['nickname']
}
discovery_results["configured_printers"].append(configured_info)
printers_logger.info(f"✅ Drucker '{unconfigured_printer.name}' automatisch mit {ip} verknüpft")
else:
discovery_results["errors"].append(f"{ip}: Tapo-Gerät gefunden, aber kein unkonfigurierter Drucker verfügbar")
else:
discovery_results["errors"].append(f"{ip}: Erreichbar, aber kein Tapo-Gerät oder Authentifizierung fehlgeschlagen")
except Exception as ip_error:
discovery_results["errors"].append(f"{ip}: Fehler beim Test - {str(ip_error)}")
printers_logger.warning(f"Fehler beim Testen von {ip}: {str(ip_error)}")
# Änderungen speichern
db_session.commit()
db_session.close()
# Zusammenfassung
summary = {
"tested_ips": len(test_ips),
"discovered_devices": len(discovery_results["discovered_devices"]),
"configured_printers": len(discovery_results["configured_printers"]),
"errors": len(discovery_results["errors"])
}
printers_logger.info(f"Tapo-Konfigurationsassistent abgeschlossen: {summary}")
return jsonify({
"success": True,
"message": f"Discovery abgeschlossen: {summary['discovered_devices']} Geräte gefunden, "
f"{summary['configured_printers']} Drucker konfiguriert",
"results": discovery_results,
"summary": summary,
"timestamp": datetime.now().isoformat()
})
except Exception as e:
printers_logger.error(f"Fehler beim Tapo-Konfigurationsassistent: {str(e)}")
if 'db_session' in locals():
db_session.close()
return jsonify({
"success": False,
"error": f"Systemfehler: {str(e)}"
}), 500
@printers_blueprint.route("/tapo/validate-configuration/<int:printer_id>", methods=["POST"])
@login_required
@require_permission(Permission.ADMIN)
@measure_execution_time(logger=printers_logger, task_name="API-Tapo-Konfigurationsvalidierung")
def validate_tapo_configuration(printer_id):
"""
Validiert die Tapo-Konfiguration eines spezifischen Druckers.
Führt umfassende Tests durch: Ping, Authentifizierung, Funktionalität.
"""
printers_logger.info(f"Tapo-Konfigurationsvalidierung für Drucker {printer_id} von Admin {current_user.name}")
try:
db_session = get_db_session()
printer = db_session.query(Printer).filter(Printer.id == printer_id).first()
if not printer:
db_session.close()
return jsonify({
"success": False,
"error": "Drucker nicht gefunden"
}), 404
# Tapo-Controller laden
try:
from utils.hardware_integration import tapo_controller
except Exception as e:
db_session.close()
return jsonify({
"success": False,
"error": f"Tapo-Controller nicht verfügbar: {str(e)}"
}), 500
validation_results = {
"printer": {
"id": printer.id,
"name": printer.name,
"model": printer.model,
"location": printer.location
},
"configuration": {
"has_ip": bool(printer.plug_ip),
"has_username": bool(printer.plug_username),
"has_password": bool(printer.plug_password),
"ip_address": printer.plug_ip
},
"tests": {
"ping": {"status": "not_run", "message": ""},
"authentication": {"status": "not_run", "message": ""},
"functionality": {"status": "not_run", "message": ""},
"device_info": {"status": "not_run", "message": ""}
},
"overall_status": "unknown",
"recommendations": []
}
# Konfigurationsprüfung
if not printer.plug_ip:
validation_results["overall_status"] = "not_configured"
validation_results["recommendations"].append("IP-Adresse der Tapo-Steckdose eintragen")
elif not printer.plug_username or not printer.plug_password:
validation_results["overall_status"] = "incomplete_config"
validation_results["recommendations"].append("Benutzername und Passwort für Tapo-Steckdose eintragen")
else:
# Umfassende Tests durchführen
all_tests_passed = True
# Test 1: Ping/Erreichbarkeit
try:
ping_success = tapo_controller.ping_address(printer.plug_ip, timeout=5)
if ping_success:
validation_results["tests"]["ping"] = {
"status": "passed",
"message": "Steckdose ist im Netzwerk erreichbar"
}
else:
validation_results["tests"]["ping"] = {
"status": "failed",
"message": "Steckdose nicht erreichbar - Netzwerkproblem oder falsche IP"
}
all_tests_passed = False
validation_results["recommendations"].append("IP-Adresse überprüfen")
validation_results["recommendations"].append("Netzwerkverbindung der Steckdose prüfen")
except Exception as ping_error:
validation_results["tests"]["ping"] = {
"status": "error",
"message": f"Ping-Test fehlgeschlagen: {str(ping_error)}"
}
all_tests_passed = False
# Test 2: Authentifizierung (nur wenn Ping erfolgreich)
if validation_results["tests"]["ping"]["status"] == "passed":
try:
auth_result = tapo_controller.test_connection(
printer.plug_ip,
username=printer.plug_username,
password=printer.plug_password
)
if auth_result["success"]:
validation_results["tests"]["authentication"] = {
"status": "passed",
"message": "Authentifizierung erfolgreich"
}
# Geräteinformationen extrahieren
device_info = auth_result.get("device_info", {})
validation_results["tests"]["device_info"] = {
"status": "passed",
"message": "Geräteinformationen abgerufen",
"data": {
"nickname": device_info.get("nickname", "Unbekannt"),
"model": device_info.get("model", "Unbekannt"),
"device_on": device_info.get("device_on", False),
"signal_level": device_info.get("signal_level", 0)
}
}
else:
validation_results["tests"]["authentication"] = {
"status": "failed",
"message": f"Authentifizierung fehlgeschlagen: {auth_result.get('error', 'Unbekannt')}"
}
all_tests_passed = False
validation_results["recommendations"].append("Benutzername und Passwort überprüfen")
except Exception as auth_error:
validation_results["tests"]["authentication"] = {
"status": "error",
"message": f"Authentifizierungstest fehlgeschlagen: {str(auth_error)}"
}
all_tests_passed = False
# Test 3: Funktionalität (nur wenn Authentifizierung erfolgreich)
if validation_results["tests"]["authentication"]["status"] == "passed":
try:
reachable, status = tapo_controller.check_outlet_status(
printer.plug_ip,
printer_id=printer_id
)
if reachable:
validation_results["tests"]["functionality"] = {
"status": "passed",
"message": f"Status erfolgreich abgerufen: {status}",
"current_status": status
}
# Drucker-Status in DB aktualisieren
printer.last_checked = datetime.now()
printer.status = "online" if status == "on" else "offline"
else:
validation_results["tests"]["functionality"] = {
"status": "failed",
"message": "Status konnte nicht abgerufen werden"
}
all_tests_passed = False
except Exception as func_error:
validation_results["tests"]["functionality"] = {
"status": "error",
"message": f"Funktionalitätstest fehlgeschlagen: {str(func_error)}"
}
all_tests_passed = False
# Gesamtstatus bestimmen
if all_tests_passed:
validation_results["overall_status"] = "fully_functional"
validation_results["recommendations"].append("Konfiguration ist vollständig und funktional")
else:
failed_tests = [test for test, result in validation_results["tests"].items()
if result["status"] in ["failed", "error"]]
if "ping" in failed_tests:
validation_results["overall_status"] = "network_issue"
elif "authentication" in failed_tests:
validation_results["overall_status"] = "auth_issue"
elif "functionality" in failed_tests:
validation_results["overall_status"] = "functionality_issue"
else:
validation_results["overall_status"] = "partial_failure"
# Aktuelle Jobs als Sicherheitshinweis
active_jobs = db_session.query(Job).filter(
Job.printer_id == printer_id,
Job.status.in_(["running", "printing", "active"])
).count()
if active_jobs > 0:
validation_results["safety_warning"] = f"{active_jobs} aktive Job(s) - Vorsicht bei Steckdosen-Tests"
db_session.commit()
db_session.close()
printers_logger.info(f"Tapo-Validierung für {printer.name} abgeschlossen: {validation_results['overall_status']}")
return jsonify({
"success": True,
"validation": validation_results,
"timestamp": datetime.now().isoformat()
})
except Exception as e:
printers_logger.error(f"Fehler bei Tapo-Konfigurationsvalidierung: {str(e)}")
if 'db_session' in locals():
db_session.close()
return jsonify({
"success": False,
"error": f"Systemfehler: {str(e)}"
}), 500

Some files were not shown because too many files have changed in this diff Show More