616 lines
23 KiB
Python
616 lines
23 KiB
Python
"""
|
|
Drucker-Blueprint für MYP Platform
|
|
Enthält alle Routen und Funktionen zur Druckerverwaltung, Statusüberwachung und Steuerung.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from flask import Blueprint, request, jsonify, current_app, abort, Response
|
|
from flask_login import login_required, current_user
|
|
from werkzeug.utils import secure_filename
|
|
from werkzeug.exceptions import NotFound, BadRequest
|
|
from sqlalchemy import func, desc, asc
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from typing import Dict, List, Tuple, Any, Optional
|
|
|
|
from models import Printer, User, Job, get_db_session
|
|
from utils.logging_config import get_logger, measure_execution_time
|
|
from utils.permissions import require_permission, Permission, check_permission
|
|
from utils.printer_monitor import printer_monitor
|
|
|
|
# Logger initialisieren
|
|
printers_logger = get_logger("printers")
|
|
|
|
# Blueprint erstellen
|
|
printers_blueprint = Blueprint("printers", __name__, url_prefix="/api/printers")
|
|
|
|
@printers_blueprint.route("/monitor/live-status", methods=["GET"])
|
|
@login_required
|
|
@measure_execution_time(logger=printers_logger, task_name="API-Live-Drucker-Status-Abfrage")
|
|
def get_live_printer_status():
|
|
"""
|
|
Liefert den aktuellen Live-Status aller Drucker.
|
|
|
|
Query-Parameter:
|
|
- use_cache: ob Cache verwendet werden soll (default: true)
|
|
|
|
Returns:
|
|
JSON mit Live-Status aller Drucker
|
|
"""
|
|
printers_logger.info(f"🔄 Live-Status-Abfrage von Benutzer {current_user.name} (ID: {current_user.id})")
|
|
|
|
# Parameter auslesen
|
|
use_cache_param = request.args.get("use_cache", "true").lower()
|
|
use_cache = use_cache_param == "true"
|
|
|
|
try:
|
|
# Live-Status über den PrinterMonitor abrufen
|
|
status_data = printer_monitor.get_live_printer_status(use_session_cache=use_cache)
|
|
|
|
# Zusammenfassung der Druckerstatus erstellen
|
|
summary = printer_monitor.get_printer_summary()
|
|
|
|
# Antwort mit Status und Zusammenfassung
|
|
response = {
|
|
"success": True,
|
|
"status": status_data,
|
|
"summary": summary,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"cache_used": use_cache
|
|
}
|
|
|
|
printers_logger.info(f"✅ Live-Status-Abfrage erfolgreich: {len(status_data)} Drucker")
|
|
return jsonify(response)
|
|
|
|
except Exception as e:
|
|
printers_logger.error(f"❌ Fehler bei Live-Status-Abfrage: {str(e)}")
|
|
return jsonify({
|
|
"success": False,
|
|
"error": "Fehler bei Abfrage des Druckerstatus",
|
|
"message": str(e)
|
|
}), 500
|
|
|
|
@printers_blueprint.route("/control/<int:printer_id>/power", methods=["POST"])
|
|
@login_required
|
|
@require_permission(Permission.CONTROL_PRINTER) # Verwende die bereits vorhandene Berechtigung
|
|
@measure_execution_time(logger=printers_logger, task_name="API-Drucker-Stromversorgung-Steuerung")
|
|
def control_printer_power(printer_id):
|
|
"""
|
|
Steuert die Stromversorgung eines Druckers (ein-/ausschalten).
|
|
|
|
Args:
|
|
printer_id: ID des zu steuernden Druckers
|
|
|
|
JSON-Parameter:
|
|
- action: "on" oder "off"
|
|
|
|
Returns:
|
|
JSON mit Ergebnis der Steuerungsaktion
|
|
"""
|
|
printers_logger.info(f"🔌 Stromsteuerung für Drucker {printer_id} von Benutzer {current_user.name}")
|
|
|
|
# Parameter validieren
|
|
data = request.get_json()
|
|
if not data or "action" not in data:
|
|
return jsonify({
|
|
"success": False,
|
|
"error": "Parameter 'action' fehlt"
|
|
}), 400
|
|
|
|
action = data["action"]
|
|
if action not in ["on", "off"]:
|
|
return jsonify({
|
|
"success": False,
|
|
"error": "Ungültige Aktion. Erlaubt sind 'on' oder 'off'."
|
|
}), 400
|
|
|
|
try:
|
|
# Drucker aus Datenbank holen
|
|
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": f"Drucker mit ID {printer_id} nicht gefunden"
|
|
}), 404
|
|
|
|
# Prüfen, ob Drucker eine Steckdose konfiguriert hat
|
|
if not printer.plug_ip or not printer.plug_username or not printer.plug_password:
|
|
db_session.close()
|
|
return jsonify({
|
|
"success": False,
|
|
"error": f"Drucker {printer.name} hat keine Steckdose konfiguriert"
|
|
}), 400
|
|
|
|
# Steckdose steuern
|
|
from PyP100 import PyP110
|
|
try:
|
|
# TP-Link Tapo P110 Verbindung herstellen
|
|
p110 = PyP110.P110(printer.plug_ip, printer.plug_username, printer.plug_password)
|
|
p110.handshake() # Authentifizierung
|
|
p110.login() # Login
|
|
|
|
# Steckdose ein- oder ausschalten
|
|
if action == "on":
|
|
p110.turnOn()
|
|
success = True
|
|
message = "Steckdose erfolgreich eingeschaltet"
|
|
printer.status = "starting" # Status aktualisieren
|
|
else:
|
|
p110.turnOff()
|
|
success = True
|
|
message = "Steckdose erfolgreich ausgeschaltet"
|
|
printer.status = "offline" # Status aktualisieren
|
|
|
|
# Zeitpunkt der letzten Prüfung aktualisieren
|
|
printer.last_checked = datetime.now()
|
|
db_session.commit()
|
|
|
|
# Cache leeren, damit neue Status-Abfragen aktuell sind
|
|
printer_monitor.clear_all_caches()
|
|
|
|
printers_logger.info(f"✅ {action.upper()}: Drucker {printer.name} erfolgreich {message}")
|
|
|
|
except Exception as e:
|
|
printers_logger.error(f"❌ Fehler bei Steckdosensteuerung für {printer.name}: {str(e)}")
|
|
db_session.close()
|
|
return jsonify({
|
|
"success": False,
|
|
"error": f"Fehler bei Steckdosensteuerung: {str(e)}"
|
|
}), 500
|
|
|
|
db_session.close()
|
|
return jsonify({
|
|
"success": True,
|
|
"message": message,
|
|
"printer_id": printer_id,
|
|
"printer_name": printer.name,
|
|
"action": action,
|
|
"timestamp": datetime.now().isoformat()
|
|
})
|
|
|
|
except Exception as e:
|
|
printers_logger.error(f"❌ Allgemeiner Fehler bei Stromsteuerung: {str(e)}")
|
|
return jsonify({
|
|
"success": False,
|
|
"error": f"Allgemeiner Fehler: {str(e)}"
|
|
}), 500
|
|
|
|
@printers_blueprint.route("/test/socket/<int:printer_id>", methods=["GET"])
|
|
@login_required
|
|
@require_permission(Permission.ADMIN)
|
|
@measure_execution_time(logger=printers_logger, task_name="API-Steckdosen-Test-Status")
|
|
def test_socket_status(printer_id):
|
|
"""
|
|
Prüft den aktuellen Status einer Steckdose für Testzwecke (nur für Ausbilder/Administratoren).
|
|
|
|
Args:
|
|
printer_id: ID des Druckers dessen Steckdose getestet werden soll
|
|
|
|
Returns:
|
|
JSON mit detailliertem Status der Steckdose und Warnungen
|
|
"""
|
|
printers_logger.info(f"🔍 Steckdosen-Test-Status für Drucker {printer_id} von Admin {current_user.name}")
|
|
|
|
try:
|
|
# Drucker aus Datenbank holen
|
|
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": f"Drucker mit ID {printer_id} nicht gefunden"
|
|
}), 404
|
|
|
|
# Prüfen, ob Drucker eine Steckdose konfiguriert hat
|
|
if not printer.plug_ip or not printer.plug_username or not printer.plug_password:
|
|
db_session.close()
|
|
return jsonify({
|
|
"success": False,
|
|
"error": f"Drucker {printer.name} hat keine Steckdose konfiguriert",
|
|
"warning": "Steckdose kann nicht getestet werden - Konfiguration fehlt"
|
|
}), 400
|
|
|
|
# Prüfen, ob der Drucker gerade aktive Jobs hat
|
|
active_jobs = db_session.query(Job).filter(
|
|
Job.printer_id == printer_id,
|
|
Job.status.in_(["running", "printing", "active"])
|
|
).all()
|
|
|
|
db_session.close()
|
|
|
|
# Steckdosen-Status prüfen
|
|
from PyP100 import PyP110
|
|
socket_status = None
|
|
socket_info = None
|
|
error_message = None
|
|
|
|
try:
|
|
# TP-Link Tapo P110 Verbindung herstellen
|
|
p110 = PyP110.P110(printer.plug_ip, printer.plug_username, printer.plug_password)
|
|
p110.handshake() # Authentifizierung
|
|
p110.login() # Login
|
|
|
|
# Geräteinformationen abrufen
|
|
device_info = p110.getDeviceInfo()
|
|
socket_status = "online" if device_info["result"]["device_on"] else "offline"
|
|
|
|
# Energieverbrauch abrufen (falls verfügbar)
|
|
try:
|
|
energy_info = p110.getEnergyUsage()
|
|
current_power = energy_info.get("result", {}).get("current_power", 0)
|
|
except:
|
|
current_power = None
|
|
|
|
socket_info = {
|
|
"device_on": device_info["result"]["device_on"],
|
|
"signal_level": device_info["result"].get("signal_level", 0),
|
|
"current_power": current_power,
|
|
"device_id": device_info["result"].get("device_id", "Unbekannt"),
|
|
"model": device_info["result"].get("model", "Unbekannt"),
|
|
"hw_ver": device_info["result"].get("hw_ver", "Unbekannt"),
|
|
"fw_ver": device_info["result"].get("fw_ver", "Unbekannt")
|
|
}
|
|
|
|
except Exception as e:
|
|
printers_logger.warning(f"⚠️ Fehler bei Steckdosen-Status-Abfrage für {printer.name}: {str(e)}")
|
|
socket_status = "error"
|
|
error_message = str(e)
|
|
|
|
# Warnungen und Empfehlungen zusammenstellen
|
|
warnings = []
|
|
recommendations = []
|
|
risk_level = "low"
|
|
|
|
if active_jobs:
|
|
warnings.append(f"ACHTUNG: Drucker hat {len(active_jobs)} aktive(n) Job(s)!")
|
|
risk_level = "high"
|
|
recommendations.append("Warten Sie bis alle Jobs abgeschlossen sind bevor Sie die Steckdose ausschalten")
|
|
|
|
if socket_status == "online" and socket_info and socket_info.get("device_on"):
|
|
if socket_info.get("current_power", 0) > 10: # Mehr als 10W Verbrauch
|
|
warnings.append(f"Drucker verbraucht aktuell {socket_info['current_power']}W - vermutlich aktiv")
|
|
risk_level = "medium" if risk_level == "low" else risk_level
|
|
recommendations.append("Prüfen Sie den Druckerstatus bevor Sie die Steckdose ausschalten")
|
|
else:
|
|
recommendations.append("Drucker scheint im Standby-Modus zu sein - Test sollte sicher möglich sein")
|
|
|
|
if socket_status == "error":
|
|
warnings.append("Steckdose nicht erreichbar - Netzwerk oder Konfigurationsproblem")
|
|
recommendations.append("Prüfen Sie die Netzwerkverbindung und Steckdosen-Konfiguration")
|
|
|
|
if not warnings and socket_status == "offline":
|
|
recommendations.append("Steckdose ist ausgeschaltet - Test kann sicher durchgeführt werden")
|
|
|
|
printers_logger.info(f"✅ Steckdosen-Test-Status erfolgreich abgerufen für {printer.name}")
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"printer": {
|
|
"id": printer.id,
|
|
"name": printer.name,
|
|
"model": printer.model,
|
|
"location": printer.location,
|
|
"status": printer.status
|
|
},
|
|
"socket": {
|
|
"status": socket_status,
|
|
"info": socket_info,
|
|
"error": error_message,
|
|
"ip_address": printer.plug_ip
|
|
},
|
|
"safety": {
|
|
"risk_level": risk_level,
|
|
"warnings": warnings,
|
|
"recommendations": recommendations,
|
|
"active_jobs_count": len(active_jobs),
|
|
"safe_to_test": len(warnings) == 0
|
|
},
|
|
"timestamp": datetime.now().isoformat()
|
|
})
|
|
|
|
except Exception as e:
|
|
printers_logger.error(f"❌ Allgemeiner Fehler bei Steckdosen-Test-Status: {str(e)}")
|
|
return jsonify({
|
|
"success": False,
|
|
"error": f"Allgemeiner Fehler: {str(e)}"
|
|
}), 500
|
|
|
|
@printers_blueprint.route("/test/socket/<int:printer_id>/control", methods=["POST"])
|
|
@login_required
|
|
@require_permission(Permission.ADMIN)
|
|
@measure_execution_time(logger=printers_logger, task_name="API-Steckdosen-Test-Steuerung")
|
|
def test_socket_control(printer_id):
|
|
"""
|
|
Steuert eine Steckdose für Testzwecke (nur für Ausbilder/Administratoren).
|
|
Diese Funktion zeigt Warnungen an, erlaubt aber trotzdem die Steuerung für Tests.
|
|
|
|
Args:
|
|
printer_id: ID des Druckers dessen Steckdose gesteuert werden soll
|
|
|
|
JSON-Parameter:
|
|
- action: "on" oder "off"
|
|
- force: boolean - überschreibt Sicherheitswarnungen (default: false)
|
|
- test_reason: string - Grund für den Test (optional)
|
|
|
|
Returns:
|
|
JSON mit Ergebnis der Steuerungsaktion und Warnungen
|
|
"""
|
|
printers_logger.info(f"🧪 Steckdosen-Test-Steuerung für Drucker {printer_id} von Admin {current_user.name}")
|
|
|
|
# Parameter validieren
|
|
data = request.get_json()
|
|
if not data or "action" not in data:
|
|
return jsonify({
|
|
"success": False,
|
|
"error": "Parameter 'action' fehlt"
|
|
}), 400
|
|
|
|
action = data["action"]
|
|
if action not in ["on", "off"]:
|
|
return jsonify({
|
|
"success": False,
|
|
"error": "Ungültige Aktion. Erlaubt sind 'on' oder 'off'."
|
|
}), 400
|
|
|
|
force = data.get("force", False)
|
|
test_reason = data.get("test_reason", "Routinetest")
|
|
|
|
try:
|
|
# Drucker aus Datenbank holen
|
|
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": f"Drucker mit ID {printer_id} nicht gefunden"
|
|
}), 404
|
|
|
|
# Prüfen, ob Drucker eine Steckdose konfiguriert hat
|
|
if not printer.plug_ip or not printer.plug_username or not printer.plug_password:
|
|
db_session.close()
|
|
return jsonify({
|
|
"success": False,
|
|
"error": f"Drucker {printer.name} hat keine Steckdose konfiguriert"
|
|
}), 400
|
|
|
|
# Aktive Jobs prüfen
|
|
active_jobs = db_session.query(Job).filter(
|
|
Job.printer_id == printer_id,
|
|
Job.status.in_(["running", "printing", "active"])
|
|
).all()
|
|
|
|
# Sicherheitsprüfungen
|
|
warnings = []
|
|
should_block = False
|
|
|
|
if active_jobs and action == "off":
|
|
warnings.append(f"WARNUNG: {len(active_jobs)} aktive Job(s) würden abgebrochen!")
|
|
if not force:
|
|
should_block = True
|
|
|
|
if should_block:
|
|
db_session.close()
|
|
return jsonify({
|
|
"success": False,
|
|
"error": "Aktion blockiert aufgrund von Sicherheitsbedenken",
|
|
"warnings": warnings,
|
|
"hint": "Verwenden Sie 'force': true um die Aktion trotzdem auszuführen",
|
|
"requires_force": True
|
|
}), 409 # Conflict
|
|
|
|
# Steckdose steuern
|
|
from PyP100 import PyP110
|
|
try:
|
|
# TP-Link Tapo P110 Verbindung herstellen
|
|
p110 = PyP110.P110(printer.plug_ip, printer.plug_username, printer.plug_password)
|
|
p110.handshake() # Authentifizierung
|
|
p110.login() # Login
|
|
|
|
# Aktuellen Status vor der Änderung abrufen
|
|
device_info_before = p110.getDeviceInfo()
|
|
status_before = device_info_before["result"]["device_on"]
|
|
|
|
# Steckdose ein- oder ausschalten
|
|
if action == "on":
|
|
p110.turnOn()
|
|
success = True
|
|
message = "Steckdose für Test erfolgreich eingeschaltet"
|
|
new_printer_status = "starting"
|
|
else:
|
|
p110.turnOff()
|
|
success = True
|
|
message = "Steckdose für Test erfolgreich ausgeschaltet"
|
|
new_printer_status = "offline"
|
|
|
|
# Kurz warten und neuen Status prüfen
|
|
time.sleep(2)
|
|
device_info_after = p110.getDeviceInfo()
|
|
status_after = device_info_after["result"]["device_on"]
|
|
|
|
# Drucker-Status aktualisieren
|
|
printer.status = new_printer_status
|
|
printer.last_checked = datetime.now()
|
|
db_session.commit()
|
|
|
|
# Cache leeren, damit neue Status-Abfragen aktuell sind
|
|
printer_monitor.clear_all_caches()
|
|
|
|
# Test-Eintrag für Audit-Log
|
|
printers_logger.info(f"🧪 TEST DURCHGEFÜHRT: {action.upper()} für {printer.name} | "
|
|
f"Admin: {current_user.name} | Grund: {test_reason} | "
|
|
f"Force: {force} | Status: {status_before} → {status_after}")
|
|
|
|
except Exception as e:
|
|
printers_logger.error(f"❌ Fehler bei Test-Steckdosensteuerung für {printer.name}: {str(e)}")
|
|
db_session.close()
|
|
return jsonify({
|
|
"success": False,
|
|
"error": f"Fehler bei Steckdosensteuerung: {str(e)}"
|
|
}), 500
|
|
|
|
db_session.close()
|
|
return jsonify({
|
|
"success": True,
|
|
"message": message,
|
|
"test_info": {
|
|
"admin": current_user.name,
|
|
"reason": test_reason,
|
|
"forced": force,
|
|
"status_before": status_before,
|
|
"status_after": status_after
|
|
},
|
|
"printer": {
|
|
"id": printer_id,
|
|
"name": printer.name,
|
|
"status": new_printer_status
|
|
},
|
|
"action": action,
|
|
"warnings": warnings,
|
|
"timestamp": datetime.now().isoformat()
|
|
})
|
|
|
|
except Exception as e:
|
|
printers_logger.error(f"❌ Allgemeiner Fehler bei Test-Steckdosensteuerung: {str(e)}")
|
|
return jsonify({
|
|
"success": False,
|
|
"error": f"Allgemeiner Fehler: {str(e)}"
|
|
}), 500
|
|
|
|
@printers_blueprint.route("/test/all-sockets", methods=["GET"])
|
|
@login_required
|
|
@require_permission(Permission.ADMIN)
|
|
@measure_execution_time(logger=printers_logger, task_name="API-Alle-Steckdosen-Test-Status")
|
|
def test_all_sockets_status():
|
|
"""
|
|
Liefert den Test-Status aller konfigurierten Steckdosen (nur für Ausbilder/Administratoren).
|
|
|
|
Returns:
|
|
JSON mit Status aller Steckdosen und Gesamtübersicht
|
|
"""
|
|
printers_logger.info(f"🔍 Alle-Steckdosen-Test-Status von Admin {current_user.name}")
|
|
|
|
try:
|
|
# Alle Drucker mit Steckdosen-Konfiguration holen
|
|
db_session = get_db_session()
|
|
printers = db_session.query(Printer).filter(
|
|
Printer.plug_ip.isnot(None),
|
|
Printer.plug_username.isnot(None),
|
|
Printer.plug_password.isnot(None)
|
|
).all()
|
|
|
|
results = []
|
|
total_online = 0
|
|
total_offline = 0
|
|
total_error = 0
|
|
total_warnings = 0
|
|
|
|
from PyP100 import PyP110
|
|
|
|
for printer in printers:
|
|
# Aktive 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"])
|
|
).count()
|
|
|
|
# Steckdosen-Status prüfen
|
|
socket_status = "unknown"
|
|
device_on = False
|
|
current_power = None
|
|
error_message = None
|
|
warnings = []
|
|
|
|
try:
|
|
p110 = PyP110.P110(printer.plug_ip, printer.plug_username, printer.plug_password)
|
|
p110.handshake()
|
|
p110.login()
|
|
|
|
device_info = p110.getDeviceInfo()
|
|
device_on = device_info["result"]["device_on"]
|
|
socket_status = "online" if device_on else "offline"
|
|
|
|
# Energieverbrauch abrufen
|
|
try:
|
|
energy_info = p110.getEnergyUsage()
|
|
current_power = energy_info.get("result", {}).get("current_power", 0)
|
|
except:
|
|
current_power = None
|
|
|
|
# Warnungen generieren
|
|
if active_jobs > 0:
|
|
warnings.append(f"{active_jobs} aktive Job(s)")
|
|
|
|
if device_on and current_power and current_power > 10:
|
|
warnings.append(f"Hoher Verbrauch: {current_power}W")
|
|
|
|
except Exception as e:
|
|
socket_status = "error"
|
|
error_message = str(e)
|
|
warnings.append(f"Verbindungsfehler: {str(e)[:50]}")
|
|
|
|
# Statistiken aktualisieren
|
|
if socket_status == "online":
|
|
total_online += 1
|
|
elif socket_status == "offline":
|
|
total_offline += 1
|
|
else:
|
|
total_error += 1
|
|
|
|
if warnings:
|
|
total_warnings += 1
|
|
|
|
results.append({
|
|
"printer": {
|
|
"id": printer.id,
|
|
"name": printer.name,
|
|
"model": printer.model,
|
|
"location": printer.location
|
|
},
|
|
"socket": {
|
|
"status": socket_status,
|
|
"device_on": device_on,
|
|
"current_power": current_power,
|
|
"ip_address": printer.plug_ip,
|
|
"error": error_message
|
|
},
|
|
"warnings": warnings,
|
|
"active_jobs": active_jobs,
|
|
"safe_to_test": len(warnings) == 0
|
|
})
|
|
|
|
db_session.close()
|
|
|
|
# Gesamtübersicht erstellen
|
|
summary = {
|
|
"total_sockets": len(results),
|
|
"online": total_online,
|
|
"offline": total_offline,
|
|
"error": total_error,
|
|
"with_warnings": total_warnings,
|
|
"safe_to_test": len(results) - total_warnings
|
|
}
|
|
|
|
printers_logger.info(f"✅ Alle-Steckdosen-Status erfolgreich abgerufen: {len(results)} Steckdosen")
|
|
|
|
return jsonify({
|
|
"success": True,
|
|
"sockets": results,
|
|
"summary": summary,
|
|
"timestamp": datetime.now().isoformat()
|
|
})
|
|
|
|
except Exception as e:
|
|
printers_logger.error(f"❌ Fehler bei Alle-Steckdosen-Test-Status: {str(e)}")
|
|
return jsonify({
|
|
"success": False,
|
|
"error": f"Allgemeiner Fehler: {str(e)}"
|
|
}), 500 |