"feat: Update admin and printer templates with new design"

This commit is contained in:
2025-05-26 12:50:13 +02:00
parent e122fe0cc9
commit 2406aedf38
3 changed files with 427 additions and 70 deletions

View File

@@ -1,15 +1,21 @@
import os
import sys
import logging
import threading
import time
import subprocess
import socket
import json
import secrets
import subprocess
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple, Any, Union
from functools import wraps
from typing import Optional, Dict, List, Tuple, Any, Union
from flask import Flask, request, jsonify, session, render_template, redirect, url_for, flash, Response
from flask_login import LoginManager, UserMixin, login_user, logout_user, current_user, login_required
from flask import Flask, render_template, request, redirect, url_for, flash, jsonify, session, send_file, Response
from flask_login import LoginManager, login_user, logout_user, login_required, current_user, UserMixin
from werkzeug.security import check_password_hash, generate_password_hash
from werkzeug.utils import secure_filename
import sqlalchemy.exc
import sqlalchemy
from PyP100 import PyP110
@@ -119,6 +125,82 @@ def job_owner_required(f):
return f(job_id, *args, **kwargs)
return decorated_function
def check_printer_status(ip_address: str, timeout: int = 7) -> Tuple[str, bool]:
"""
Überprüft den Status eines Druckers über Ping mit Timeout.
Args:
ip_address: IP-Adresse des Druckers
timeout: Timeout in Sekunden (Standard: 7)
Returns:
Tuple[str, bool]: (Status, Aktiv) - Status ist "online" oder "offline", Aktiv ist True/False
"""
if not ip_address:
return "offline", False
try:
# Windows-spezifischer Ping-Befehl mit Timeout
if os.name == 'nt': # Windows
cmd = ['ping', '-n', '1', '-w', str(timeout * 1000), ip_address]
else: # Unix/Linux/macOS
cmd = ['ping', '-c', '1', '-W', str(timeout), ip_address]
# Ping ausführen mit Timeout
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=timeout + 1 # Zusätzlicher Timeout für subprocess
)
# Erfolgreicher Ping (Return Code 0)
if result.returncode == 0:
return "online", True
else:
return "offline", False
except subprocess.TimeoutExpired:
printers_logger.warning(f"Ping-Timeout für Drucker {ip_address} nach {timeout} Sekunden")
return "offline", False
except Exception as e:
printers_logger.error(f"Fehler beim Ping für Drucker {ip_address}: {str(e)}")
return "offline", False
def check_multiple_printers_status(printers: List[Dict], timeout: int = 7) -> Dict[int, Tuple[str, bool]]:
"""
Überprüft den Status mehrerer Drucker parallel mit Timeout.
Args:
printers: Liste von Drucker-Dictionaries mit 'id' und 'ip_address'
timeout: Timeout in Sekunden pro Drucker (Standard: 7)
Returns:
Dict[int, Tuple[str, bool]]: Dictionary mit Drucker-ID als Key und (Status, Aktiv) als Value
"""
results = {}
# Parallel-Ausführung mit ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=min(len(printers), 10)) as executor:
# Futures für alle Drucker erstellen
future_to_printer = {
executor.submit(check_printer_status, printer.get('ip_address'), timeout): printer
for printer in printers
}
# Ergebnisse sammeln
for future in as_completed(future_to_printer, timeout=timeout + 2):
printer = future_to_printer[future]
try:
status, active = future.result()
results[printer['id']] = (status, active)
printers_logger.info(f"Drucker {printer['name']} ({printer.get('ip_address')}): {status}")
except Exception as e:
printers_logger.error(f"Fehler bei Status-Check für Drucker {printer['name']}: {str(e)}")
results[printer['id']] = ("offline", False)
return results
# UI-Routen
@app.route("/")
def index():
@@ -216,12 +298,18 @@ def admin_page():
# Benutzeranzahl
stats["total_users"] = db_session.query(User).count()
# Druckeranzahl
stats["total_printers"] = db_session.query(Printer).count()
# Druckeranzahl und Online-Status
all_printers = db_session.query(Printer).all()
stats["total_printers"] = len(all_printers)
stats["online_printers"] = len([p for p in all_printers if p.status == "online"])
# Aktive Jobs
# Aktive Jobs und Warteschlange
stats["active_jobs"] = db_session.query(Job).filter(
Job.status.in_(["scheduled", "running"])
Job.status.in_(["printing", "running"])
).count()
stats["queued_jobs"] = db_session.query(Job).filter(
Job.status == "scheduled"
).count()
# Erfolgsrate
@@ -259,11 +347,55 @@ def admin_page():
# System-Informationen laden
if active_tab == 'system':
import psutil
# CPU und Memory
cpu_percent = psutil.cpu_percent(interval=1)
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
# Uptime
boot_time = psutil.boot_time()
uptime_seconds = time.time() - boot_time
uptime_days = int(uptime_seconds // 86400)
uptime_hours = int((uptime_seconds % 86400) // 3600)
uptime_minutes = int((uptime_seconds % 3600) // 60)
# Datenbank-Status
db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'instance', 'database.db')
db_size = 0
if os.path.exists(db_path):
db_size = os.path.getsize(db_path) / (1024 * 1024) # MB
# Scheduler-Status
scheduler_running = False
scheduler_jobs = 0
try:
from utils.job_scheduler import scheduler
scheduler_running = scheduler.running
if hasattr(scheduler, 'get_jobs'):
scheduler_jobs = len(scheduler.get_jobs())
except:
pass
# Nächster Job
next_job = db_session.query(Job).filter(
Job.status == "scheduled"
).order_by(Job.created_at.asc()).first()
next_job_time = "Keine geplanten Jobs"
if next_job:
next_job_time = next_job.created_at.strftime("%d.%m.%Y %H:%M")
system_info = {
"cpu": psutil.cpu_percent(),
"memory": psutil.virtual_memory().percent,
"disk": psutil.disk_usage('/').percent,
"uptime": get_system_uptime_days()
"cpu_usage": round(cpu_percent, 1),
"memory_usage": round(memory.percent, 1),
"disk_usage": round((disk.used / disk.total) * 100, 1),
"uptime": f"{uptime_days}d {uptime_hours}h {uptime_minutes}m",
"db_size": f"{db_size:.1f} MB",
"db_connections": "Aktiv",
"scheduler_running": scheduler_running,
"scheduler_jobs": scheduler_jobs,
"next_job": next_job_time
}
# Logs laden
@@ -586,36 +718,58 @@ def finish_job(job_id):
@app.route("/api/printers", methods=["GET"])
@login_required
def get_printers():
"""Gibt alle Drucker mit aktuellem Status zurück (mit Ping-Check und 7-Sekunden-Timeout)."""
db_session = get_db_session()
try:
printers = db_session.query(Printer).all()
# Optimierte Drucker-Liste mit schneller Status-Bestimmung
# Drucker-Daten für Status-Check vorbereiten
printer_data = []
for printer in printers:
printer_data.append({
'id': printer.id,
'name': printer.name,
'ip_address': printer.ip_address,
'location': printer.location
})
# Status aller Drucker parallel überprüfen mit 7-Sekunden-Timeout
printers_logger.info(f"Starte Drucker-Status-Check für {len(printer_data)} Drucker")
status_results = check_multiple_printers_status(printer_data, timeout=7)
# Drucker-Liste mit aktuellem Status erstellen
printer_list = []
for printer in printers:
# Bestimme Status basierend auf hardkodierten Druckern
printer_config = PRINTERS.get(printer.name)
if printer_config:
status = "available" # Drucker verfügbar
active = True
if printer.id in status_results:
status, active = status_results[printer.id]
# Mapping für Frontend-Kompatibilität
if status == "online":
frontend_status = "available"
else:
frontend_status = "offline"
else:
status = "offline"
# Fallback falls kein Ergebnis vorliegt
frontend_status = "offline"
active = False
# Aktualisiere Status in der Datenbank
printer.status = status
# Status in der Datenbank aktualisieren
printer.status = frontend_status
printer.active = active
printer_data = printer.to_dict()
printer_data["status"] = status
printer_data["status"] = frontend_status
printer_data["active"] = active
printer_data["last_checked"] = datetime.now().isoformat()
printer_list.append(printer_data)
# Speichere Updates
db_session.commit()
db_session.close()
online_count = len([p for p in printer_list if p["status"] == "available"])
printers_logger.info(f"Drucker-Status-Check abgeschlossen: {online_count} von {len(printer_list)} Drucker verfügbar")
return jsonify({
"printers": printer_list
})
@@ -739,26 +893,35 @@ def get_activity():
@app.route("/api/printers/status", methods=["GET"])
@login_required
def get_printers_status():
"""Gibt den Status aller Drucker zurück - optimiert für schnelle Antwort."""
"""Gibt den Status aller Drucker zurück mit echtem Ping-Check und 7-Sekunden-Timeout."""
db_session = get_db_session()
try:
printers = db_session.query(Printer).all()
# Schnelle Status-Bestimmung basierend auf hardkodierten Druckern
# Drucker-Daten für Status-Check vorbereiten
printer_data = []
for printer in printers:
printer_data.append({
'id': printer.id,
'name': printer.name,
'ip_address': printer.ip_address,
'location': printer.location
})
# Status aller Drucker parallel überprüfen mit 7-Sekunden-Timeout
printers_logger.info(f"Starte Status-Check für {len(printer_data)} Drucker mit 7-Sekunden-Timeout")
status_results = check_multiple_printers_status(printer_data, timeout=7)
# Ergebnisse zusammenstellen und Datenbank aktualisieren
status_data = []
for printer in printers:
# Bestimme Status basierend auf IP-Adresse aus der Konfiguration
printer_config = PRINTERS.get(printer.name)
if printer_config:
# Drucker ist in der Konfiguration -> als online betrachten
status = "online"
active = True
if printer.id in status_results:
status, active = status_results[printer.id]
else:
# Drucker nicht in Konfiguration -> offline
status = "offline"
active = False
# Fallback falls kein Ergebnis vorliegt
status, active = "offline", False
# Aktualisiere den Status in der Datenbank für Konsistenz
# Status in der Datenbank aktualisieren
printer.status = status
printer.active = active
@@ -768,13 +931,16 @@ def get_printers_status():
"status": status,
"active": active,
"ip_address": printer.ip_address,
"location": printer.location
"location": printer.location,
"last_checked": datetime.now().isoformat()
})
# Speichere die aktualisierten Status
db_session.commit()
db_session.close()
printers_logger.info(f"Status-Check abgeschlossen: {len([s for s in status_data if s['status'] == 'online'])} von {len(status_data)} Drucker online")
return jsonify(status_data)
except Exception as e:
db_session.rollback()