387 lines
14 KiB
Python
387 lines
14 KiB
Python
"""
|
|
Tapo-Steckdosen-Steuerung Blueprint
|
|
Eigenständige Web-Interface für direkte Tapo-Steckdosen-Kontrolle
|
|
"""
|
|
|
|
from flask import Blueprint, render_template, request, jsonify, flash, redirect, url_for
|
|
from flask_login import login_required, current_user
|
|
from datetime import datetime
|
|
import ipaddress
|
|
import time
|
|
|
|
from blueprints.admin_unified import admin_required
|
|
from utils.tapo_controller import tapo_controller
|
|
from utils.logging_config import get_logger
|
|
from utils.performance_tracker import measure_execution_time
|
|
from utils.permissions import require_permission, Permission
|
|
from models import get_db_session, Printer
|
|
|
|
# Blueprint initialisieren
|
|
tapo_blueprint = Blueprint('tapo', __name__, url_prefix='/tapo')
|
|
|
|
# Logger konfigurieren
|
|
tapo_logger = get_logger("tapo_control")
|
|
|
|
@tapo_blueprint.route("/")
|
|
@login_required
|
|
@require_permission(Permission.CONTROL_PRINTER)
|
|
def tapo_dashboard():
|
|
"""Haupt-Dashboard für Tapo-Steckdosen-Steuerung."""
|
|
try:
|
|
tapo_logger.info(f"Tapo Dashboard aufgerufen von Benutzer: {current_user.name}")
|
|
|
|
# Alle konfigurierten Tapo-Steckdosen aus der Datenbank laden
|
|
db_session = get_db_session()
|
|
printers_with_tapo = db_session.query(Printer).filter(
|
|
Printer.active == True,
|
|
Printer.plug_ip.isnot(None)
|
|
).all()
|
|
|
|
# Status aller Steckdosen abrufen - auch wenn sie nicht erreichbar sind
|
|
outlets_status = {}
|
|
online_count = 0
|
|
|
|
for printer in printers_with_tapo:
|
|
try:
|
|
tapo_logger.debug(f"Prüfe Tapo-Steckdose für Drucker {printer.name} ({printer.plug_ip})")
|
|
reachable, status = tapo_controller.check_outlet_status(
|
|
printer.plug_ip,
|
|
printer_id=printer.id
|
|
)
|
|
|
|
if reachable:
|
|
online_count += 1
|
|
tapo_logger.info(f"✅ Tapo-Steckdose {printer.plug_ip} erreichbar - Status: {status}")
|
|
else:
|
|
tapo_logger.warning(f"⚠️ Tapo-Steckdose {printer.plug_ip} nicht erreichbar")
|
|
|
|
outlets_status[printer.plug_ip] = {
|
|
'printer_name': printer.name,
|
|
'printer_id': printer.id,
|
|
'ip': printer.plug_ip,
|
|
'status': status,
|
|
'reachable': reachable,
|
|
'location': printer.location or "Unbekannt"
|
|
}
|
|
|
|
except Exception as e:
|
|
tapo_logger.error(f"❌ Fehler beim Status-Check für {printer.plug_ip}: {e}")
|
|
outlets_status[printer.plug_ip] = {
|
|
'printer_name': printer.name,
|
|
'printer_id': printer.id,
|
|
'ip': printer.plug_ip,
|
|
'status': 'error',
|
|
'reachable': False,
|
|
'location': printer.location or "Unbekannt",
|
|
'error': str(e)
|
|
}
|
|
|
|
db_session.close()
|
|
|
|
tapo_logger.info(f"Dashboard geladen: {len(outlets_status)} Steckdosen, {online_count} online")
|
|
|
|
return render_template('tapo_control.html',
|
|
outlets=outlets_status,
|
|
total_outlets=len(outlets_status),
|
|
online_outlets=online_count)
|
|
|
|
except Exception as e:
|
|
tapo_logger.error(f"Fehler beim Laden des Tapo-Dashboards: {e}")
|
|
flash(f"Fehler beim Laden der Tapo-Steckdosen: {str(e)}", "error")
|
|
return render_template('tapo_control.html', outlets={}, total_outlets=0, online_outlets=0)
|
|
|
|
@tapo_blueprint.route("/control", methods=["POST"])
|
|
@login_required
|
|
@require_permission(Permission.CONTROL_PRINTER)
|
|
@measure_execution_time(logger=tapo_logger, task_name="Tapo-Steckdosen-Steuerung")
|
|
def control_outlet():
|
|
"""Schaltet eine Tapo-Steckdose direkt ein oder aus."""
|
|
try:
|
|
data = request.get_json()
|
|
ip = data.get('ip')
|
|
action = data.get('action') # 'on' oder 'off'
|
|
|
|
if not ip or not action:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'IP-Adresse und Aktion sind erforderlich'
|
|
}), 400
|
|
|
|
# IP-Adresse validieren
|
|
try:
|
|
ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Ungültige IP-Adresse'
|
|
}), 400
|
|
|
|
if action not in ['on', 'off']:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Ungültige Aktion. Nur "on" oder "off" erlaubt.'
|
|
}), 400
|
|
|
|
# Steckdose schalten
|
|
state = action == 'on'
|
|
success = tapo_controller.toggle_plug(ip, state)
|
|
|
|
if success:
|
|
action_text = "eingeschaltet" if state else "ausgeschaltet"
|
|
tapo_logger.info(f"✅ Tapo-Steckdose {ip} erfolgreich {action_text} durch {current_user.name}")
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'Steckdose {ip} erfolgreich {action_text}',
|
|
'ip': ip,
|
|
'action': action,
|
|
'timestamp': datetime.now().isoformat(),
|
|
'user': current_user.name
|
|
})
|
|
else:
|
|
action_text = "einschalten" if state else "ausschalten"
|
|
tapo_logger.error(f"❌ Fehler beim {action_text} der Steckdose {ip}")
|
|
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'Fehler beim {action_text} der Steckdose {ip}'
|
|
}), 500
|
|
|
|
except Exception as e:
|
|
tapo_logger.error(f"Unerwarteter Fehler bei Steckdosen-Steuerung: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'Interner Serverfehler: {str(e)}'
|
|
}), 500
|
|
|
|
@tapo_blueprint.route("/status/<ip>", methods=["GET"])
|
|
@login_required
|
|
@require_permission(Permission.CONTROL_PRINTER)
|
|
def get_outlet_status(ip):
|
|
"""Holt den aktuellen Status einer spezifischen Tapo-Steckdose."""
|
|
try:
|
|
# IP-Adresse validieren
|
|
try:
|
|
ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Ungültige IP-Adresse'
|
|
}), 400
|
|
|
|
# Status prüfen
|
|
reachable, status = tapo_controller.check_outlet_status(ip)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'ip': ip,
|
|
'status': status,
|
|
'reachable': reachable,
|
|
'timestamp': datetime.now().isoformat()
|
|
})
|
|
|
|
except Exception as e:
|
|
tapo_logger.error(f"Fehler beim Abrufen des Status für {ip}: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'Fehler beim Status-Check: {str(e)}'
|
|
}), 500
|
|
|
|
@tapo_blueprint.route("/discover", methods=["POST"])
|
|
@login_required
|
|
@admin_required
|
|
@measure_execution_time(logger=tapo_logger, task_name="Tapo-Steckdosen-Erkennung")
|
|
def discover_outlets():
|
|
"""Startet die automatische Erkennung von Tapo-Steckdosen."""
|
|
try:
|
|
tapo_logger.info(f"🔍 Starte Tapo-Steckdosen-Erkennung auf Anfrage von {current_user.name}")
|
|
|
|
# Discovery starten
|
|
results = tapo_controller.auto_discover_outlets()
|
|
|
|
success_count = sum(1 for success in results.values() if success)
|
|
total_count = len(results)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'message': f'Erkennung abgeschlossen: {success_count}/{total_count} Steckdosen gefunden',
|
|
'results': results,
|
|
'discovered_count': success_count,
|
|
'total_tested': total_count,
|
|
'timestamp': datetime.now().isoformat()
|
|
})
|
|
|
|
except Exception as e:
|
|
tapo_logger.error(f"Fehler bei der Steckdosen-Erkennung: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'Fehler bei der Erkennung: {str(e)}'
|
|
}), 500
|
|
|
|
@tapo_blueprint.route("/test/<ip>", methods=["POST"])
|
|
@login_required
|
|
@require_permission(Permission.CONTROL_PRINTER)
|
|
def test_connection(ip):
|
|
"""Testet die Verbindung zu einer spezifischen Tapo-Steckdose."""
|
|
try:
|
|
# IP-Adresse validieren
|
|
try:
|
|
ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
return jsonify({
|
|
'success': False,
|
|
'error': 'Ungültige IP-Adresse'
|
|
}), 400
|
|
|
|
# Verbindung testen
|
|
test_result = tapo_controller.test_connection(ip)
|
|
|
|
return jsonify({
|
|
'success': True,
|
|
'ip': ip,
|
|
'test_result': test_result,
|
|
'timestamp': datetime.now().isoformat()
|
|
})
|
|
|
|
except Exception as e:
|
|
tapo_logger.error(f"Fehler beim Verbindungstest für {ip}: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': f'Fehler beim Verbindungstest: {str(e)}'
|
|
}), 500
|
|
|
|
@tapo_blueprint.route("/all-status", methods=["GET"])
|
|
@login_required
|
|
@require_permission(Permission.CONTROL_PRINTER)
|
|
def get_all_status():
|
|
"""Holt den Status aller konfigurierten Tapo-Steckdosen."""
|
|
try:
|
|
# Alle Tapo-Steckdosen aus der Datenbank laden
|
|
db_session = get_db_session()
|
|
printers_with_tapo = db_session.query(Printer).filter(
|
|
Printer.active == True,
|
|
Printer.plug_ip.isnot(None)
|
|
).all()
|
|
|
|
all_status = {}
|
|
online_count = 0
|
|
total_count = len(printers_with_tapo)
|
|
|
|
tapo_logger.info(f"Status-Abfrage für {total_count} Tapo-Steckdosen gestartet")
|
|
|
|
for printer in printers_with_tapo:
|
|
try:
|
|
tapo_logger.debug(f"Prüfe Status für {printer.plug_ip} ({printer.name})")
|
|
reachable, status = tapo_controller.check_outlet_status(
|
|
printer.plug_ip,
|
|
printer_id=printer.id
|
|
)
|
|
|
|
if reachable:
|
|
online_count += 1
|
|
tapo_logger.debug(f"✅ {printer.plug_ip} erreichbar - Status: {status}")
|
|
else:
|
|
tapo_logger.debug(f"⚠️ {printer.plug_ip} nicht erreichbar")
|
|
|
|
all_status[printer.plug_ip] = {
|
|
'printer_name': printer.name,
|
|
'printer_id': printer.id,
|
|
'status': status,
|
|
'reachable': reachable,
|
|
'location': printer.location or "Unbekannt",
|
|
'last_checked': time.time()
|
|
}
|
|
|
|
except Exception as e:
|
|
tapo_logger.warning(f"❌ Status-Check für {printer.plug_ip} fehlgeschlagen: {e}")
|
|
all_status[printer.plug_ip] = {
|
|
'printer_name': printer.name,
|
|
'printer_id': printer.id,
|
|
'status': 'error',
|
|
'reachable': False,
|
|
'location': printer.location or "Unbekannt",
|
|
'error': str(e),
|
|
'last_checked': time.time()
|
|
}
|
|
|
|
db_session.close()
|
|
|
|
# Zusammenfassung loggen
|
|
tapo_logger.info(f"Status-Abfrage abgeschlossen: {online_count}/{total_count} Steckdosen erreichbar")
|
|
|
|
response_data = {
|
|
'success': True,
|
|
'outlets': all_status,
|
|
'summary': {
|
|
'total': total_count,
|
|
'online': online_count,
|
|
'offline': total_count - online_count,
|
|
'last_update': time.time()
|
|
},
|
|
'timestamp': datetime.now().isoformat()
|
|
}
|
|
|
|
return jsonify(response_data)
|
|
|
|
except Exception as e:
|
|
tapo_logger.error(f"Fehler beim Abrufen aller Tapo-Status: {e}")
|
|
return jsonify({
|
|
'success': False,
|
|
'error': str(e),
|
|
'outlets': {},
|
|
'summary': {'total': 0, 'online': 0, 'offline': 0}
|
|
}), 500
|
|
|
|
@tapo_blueprint.route("/manual-control", methods=["GET", "POST"])
|
|
@login_required
|
|
@admin_required
|
|
def manual_control():
|
|
"""Manuelle Steuerung für beliebige IP-Adressen (Admin-only)."""
|
|
if request.method == 'GET':
|
|
return render_template('tapo_manual_control.html')
|
|
|
|
try:
|
|
ip = request.form.get('ip')
|
|
action = request.form.get('action')
|
|
|
|
if not ip or not action:
|
|
flash('IP-Adresse und Aktion sind erforderlich', 'error')
|
|
return redirect(url_for('tapo.manual_control'))
|
|
|
|
# IP-Adresse validieren
|
|
try:
|
|
ipaddress.ip_address(ip)
|
|
except ValueError:
|
|
flash('Ungültige IP-Adresse', 'error')
|
|
return redirect(url_for('tapo.manual_control'))
|
|
|
|
if action not in ['on', 'off', 'status']:
|
|
flash('Ungültige Aktion', 'error')
|
|
return redirect(url_for('tapo.manual_control'))
|
|
|
|
if action == 'status':
|
|
# Status abfragen
|
|
reachable, status = tapo_controller.check_outlet_status(ip)
|
|
if reachable:
|
|
flash(f'Steckdose {ip} ist {status.upper()}', 'success')
|
|
else:
|
|
flash(f'Steckdose {ip} ist nicht erreichbar', 'error')
|
|
else:
|
|
# Ein/Ausschalten
|
|
state = action == 'on'
|
|
success = tapo_controller.toggle_plug(ip, state)
|
|
|
|
if success:
|
|
action_text = "eingeschaltet" if state else "ausgeschaltet"
|
|
flash(f'Steckdose {ip} erfolgreich {action_text}', 'success')
|
|
tapo_logger.info(f"✅ Manuelle Steuerung: {ip} {action_text} durch {current_user.name}")
|
|
else:
|
|
action_text = "einschalten" if state else "ausschalten"
|
|
flash(f'Fehler beim {action_text} der Steckdose {ip}', 'error')
|
|
|
|
return redirect(url_for('tapo.manual_control'))
|
|
|
|
except Exception as e:
|
|
tapo_logger.error(f"Fehler bei manueller Steuerung: {e}")
|
|
flash(f'Fehler: {str(e)}', 'error')
|
|
return redirect(url_for('tapo.manual_control')) |