#!/usr/bin/env python3 """ Robuste System-Control-Funktionen für wartungsfreien Produktionsbetrieb Bietet sichere Restart-, Shutdown- und Kiosk-Verwaltungsfunktionen """ import os import sys import subprocess import time import signal import psutil import threading from datetime import datetime, timedelta from typing import Dict, List, Optional, Tuple, Any from pathlib import Path import logging import json from contextlib import contextmanager from enum import Enum # Logging-Setup try: from utils.logging_config import get_logger system_logger = get_logger("system_control") except ImportError: logging.basicConfig(level=logging.INFO) system_logger = logging.getLogger("system_control") class SystemOperation(Enum): """Verfügbare System-Operationen""" RESTART = "restart" SHUTDOWN = "shutdown" KIOSK_RESTART = "kiosk_restart" KIOSK_ENABLE = "kiosk_enable" KIOSK_DISABLE = "kiosk_disable" SERVICE_RESTART = "service_restart" EMERGENCY_STOP = "emergency_stop" class SystemControlManager: """ Zentraler Manager für alle System-Control-Operationen. Bietet sichere und robuste Funktionen für wartungsfreien Betrieb. """ def __init__(self): self.is_windows = os.name == 'nt' self.pending_operations: Dict[str, Dict] = {} self.operation_history: List[Dict] = [] self.lock = threading.Lock() # Konfiguration self.config = { "restart_delay": 60, # Sekunden "shutdown_delay": 30, # Sekunden "kiosk_restart_delay": 10, # Sekunden "max_operation_history": 100, "safety_checks": True, "require_confirmation": True } # Service-Namen für verschiedene Plattformen self.services = { "https": "myp-https.service", "kiosk": "myp-kiosk.service", "watchdog": "kiosk-watchdog.service" } system_logger.info("🔧 System-Control-Manager initialisiert") def is_safe_to_operate(self) -> Tuple[bool, str]: """ Prüft ob System-Operationen sicher ausgeführt werden können. Returns: Tuple[bool, str]: (is_safe, reason) """ try: # Prüfe Systemlast load_avg = psutil.getloadavg()[0] if hasattr(psutil, 'getloadavg') else 0 if load_avg > 2.0: return False, f"Hohe Systemlast: {load_avg:.2f}" # Prüfe verfügbaren Speicher memory = psutil.virtual_memory() if memory.percent > 90: return False, f"Wenig verfügbarer Speicher: {memory.percent:.1f}% belegt" # Prüfe aktive Drucker-Jobs try: from models import get_db_session, Job db_session = get_db_session() active_jobs = db_session.query(Job).filter( Job.status.in_(["printing", "queued", "preparing"]) ).count() db_session.close() if active_jobs > 0: return False, f"Aktive Druckjobs: {active_jobs}" except Exception as e: system_logger.warning(f"Job-Prüfung fehlgeschlagen: {e}") # Prüfe kritische Prozesse critical_processes = ["chromium", "firefox", "python"] for proc in psutil.process_iter(['pid', 'name', 'cpu_percent']): try: if any(crit in proc.info['name'].lower() for crit in critical_processes): if proc.info['cpu_percent'] > 80: return False, f"Kritischer Prozess unter hoher Last: {proc.info['name']}" except (psutil.NoSuchProcess, psutil.AccessDenied): continue return True, "System ist sicher für Operationen" except Exception as e: system_logger.error(f"Fehler bei Sicherheitsprüfung: {e}") return False, f"Sicherheitsprüfung fehlgeschlagen: {e}" def schedule_operation(self, operation: SystemOperation, delay_seconds: int = None, user_id: str = None, reason: str = None, force: bool = False) -> Dict[str, Any]: """ Plant eine System-Operation mit Verzögerung. Args: operation: Art der Operation delay_seconds: Verzögerung in Sekunden (None = Standard) user_id: ID des anfragenden Benutzers reason: Grund für die Operation force: Sicherheitsprüfungen überspringen Returns: Dict mit Operation-Details """ with self.lock: # Sicherheitsprüfung (außer bei Force) if not force and self.config["safety_checks"]: is_safe, safety_reason = self.is_safe_to_operate() if not is_safe: return { "success": False, "error": f"Operation abgelehnt: {safety_reason}", "safety_check": False } # Standard-Verzögerung setzen if delay_seconds is None: delay_seconds = { SystemOperation.RESTART: self.config["restart_delay"], SystemOperation.SHUTDOWN: self.config["shutdown_delay"], SystemOperation.KIOSK_RESTART: self.config["kiosk_restart_delay"], SystemOperation.KIOSK_ENABLE: 5, SystemOperation.KIOSK_DISABLE: 5, SystemOperation.SERVICE_RESTART: 10, SystemOperation.EMERGENCY_STOP: 0 }.get(operation, 30) # Operations-ID generieren operation_id = f"{operation.value}_{int(time.time())}" scheduled_time = datetime.now() + timedelta(seconds=delay_seconds) # Operation speichern operation_data = { "id": operation_id, "operation": operation.value, "scheduled_time": scheduled_time, "delay_seconds": delay_seconds, "user_id": user_id, "reason": reason or "Keine Begründung angegeben", "force": force, "created_at": datetime.now(), "status": "scheduled" } self.pending_operations[operation_id] = operation_data # Operation in separatem Thread ausführen thread = threading.Thread( target=self._execute_delayed_operation, args=(operation_id,), daemon=True ) thread.start() system_logger.info(f"🕐 Operation geplant: {operation.value} in {delay_seconds}s") return { "success": True, "operation_id": operation_id, "scheduled_time": scheduled_time.isoformat(), "delay_seconds": delay_seconds, "message": f"Operation '{operation.value}' geplant für {scheduled_time.strftime('%H:%M:%S')}" } def _execute_delayed_operation(self, operation_id: str): """ Führt geplante Operation nach Verzögerung aus. Args: operation_id: ID der auszuführenden Operation """ try: operation_data = self.pending_operations.get(operation_id) if not operation_data: return # Warten bis zur geplanten Zeit scheduled_time = operation_data["scheduled_time"] wait_time = (scheduled_time - datetime.now()).total_seconds() if wait_time > 0: time.sleep(wait_time) # Status aktualisieren operation_data["status"] = "executing" operation_data["executed_at"] = datetime.now() # Operation ausführen operation = SystemOperation(operation_data["operation"]) result = self._execute_operation(operation, operation_data) # Ergebnis speichern operation_data["result"] = result operation_data["status"] = "completed" if result.get("success") else "failed" operation_data["completed_at"] = datetime.now() # In Historie verschieben self._move_to_history(operation_id) except Exception as e: system_logger.error(f"Fehler bei verzögerter Operation {operation_id}: {e}") if operation_id in self.pending_operations: self.pending_operations[operation_id]["status"] = "error" self.pending_operations[operation_id]["error"] = str(e) self._move_to_history(operation_id) def _execute_operation(self, operation: SystemOperation, operation_data: Dict) -> Dict[str, Any]: """ Führt die eigentliche System-Operation aus. Args: operation: Art der Operation operation_data: Operation-Daten Returns: Dict mit Ergebnis """ try: system_logger.info(f"▶️ Führe Operation aus: {operation.value}") if operation == SystemOperation.RESTART: return self._restart_system(operation_data) elif operation == SystemOperation.SHUTDOWN: return self._shutdown_system(operation_data) elif operation == SystemOperation.KIOSK_RESTART: return self._restart_kiosk(operation_data) elif operation == SystemOperation.KIOSK_ENABLE: return self._enable_kiosk(operation_data) elif operation == SystemOperation.KIOSK_DISABLE: return self._disable_kiosk(operation_data) elif operation == SystemOperation.SERVICE_RESTART: return self._restart_services(operation_data) elif operation == SystemOperation.EMERGENCY_STOP: return self._emergency_stop(operation_data) else: return {"success": False, "error": f"Unbekannte Operation: {operation.value}"} except Exception as e: system_logger.error(f"Fehler bei Operation {operation.value}: {e}") return {"success": False, "error": str(e)} def _restart_system(self, operation_data: Dict) -> Dict[str, Any]: """Startet das System neu.""" try: system_logger.warning("🔄 System-Neustart wird ausgeführt...") # Cleanup vor Neustart self._cleanup_before_restart() # System-Neustart je nach Plattform if self.is_windows: subprocess.run(["shutdown", "/r", "/t", "0"], check=True) else: subprocess.run(["sudo", "systemctl", "reboot"], check=True) return {"success": True, "message": "System-Neustart initiiert"} except subprocess.CalledProcessError as e: return {"success": False, "error": f"Neustart fehlgeschlagen: {e}"} except Exception as e: return {"success": False, "error": f"Unerwarteter Fehler: {e}"} def _shutdown_system(self, operation_data: Dict) -> Dict[str, Any]: """Fährt das System herunter.""" try: system_logger.warning("🛑 System-Shutdown wird ausgeführt...") # Cleanup vor Shutdown self._cleanup_before_restart() # System-Shutdown je nach Plattform if self.is_windows: subprocess.run(["shutdown", "/s", "/t", "0"], check=True) else: subprocess.run(["sudo", "systemctl", "poweroff"], check=True) return {"success": True, "message": "System-Shutdown initiiert"} except subprocess.CalledProcessError as e: return {"success": False, "error": f"Shutdown fehlgeschlagen: {e}"} except Exception as e: return {"success": False, "error": f"Unerwarteter Fehler: {e}"} def _restart_kiosk(self, operation_data: Dict) -> Dict[str, Any]: """Startet nur den Kiosk-Modus neu.""" try: system_logger.info("🖥️ Kiosk-Neustart wird ausgeführt...") success_count = 0 errors = [] # Kiosk-Service neustarten try: subprocess.run(["sudo", "systemctl", "restart", self.services["kiosk"]], check=True, timeout=30) success_count += 1 system_logger.info("✅ Kiosk-Service neugestartet") except Exception as e: errors.append(f"Kiosk-Service: {e}") # Watchdog-Service neustarten (falls vorhanden) try: subprocess.run(["sudo", "systemctl", "restart", self.services["watchdog"]], check=True, timeout=30) success_count += 1 system_logger.info("✅ Watchdog-Service neugestartet") except Exception as e: errors.append(f"Watchdog-Service: {e}") # X11-Session neustarten try: subprocess.run(["sudo", "systemctl", "restart", "getty@tty1.service"], check=True, timeout=30) success_count += 1 system_logger.info("✅ X11-Session neugestartet") except Exception as e: errors.append(f"X11-Session: {e}") if success_count > 0: return { "success": True, "message": f"Kiosk neugestartet ({success_count} Services)", "errors": errors if errors else None } else: return { "success": False, "error": "Alle Kiosk-Neustarts fehlgeschlagen", "details": errors } except Exception as e: return {"success": False, "error": f"Kiosk-Neustart fehlgeschlagen: {e}"} def _enable_kiosk(self, operation_data: Dict) -> Dict[str, Any]: """Aktiviert den Kiosk-Modus.""" try: system_logger.info("🖥️ Kiosk-Modus wird aktiviert...") # Kiosk-Service aktivieren und starten subprocess.run(["sudo", "systemctl", "enable", self.services["kiosk"]], check=True, timeout=30) subprocess.run(["sudo", "systemctl", "start", self.services["kiosk"]], check=True, timeout=30) # Watchdog aktivieren try: subprocess.run(["sudo", "systemctl", "enable", self.services["watchdog"]], check=True, timeout=30) subprocess.run(["sudo", "systemctl", "start", self.services["watchdog"]], check=True, timeout=30) except Exception as e: system_logger.warning(f"Watchdog-Aktivierung fehlgeschlagen: {e}") return {"success": True, "message": "Kiosk-Modus aktiviert"} except subprocess.CalledProcessError as e: return {"success": False, "error": f"Kiosk-Aktivierung fehlgeschlagen: {e}"} except Exception as e: return {"success": False, "error": f"Unerwarteter Fehler: {e}"} def _disable_kiosk(self, operation_data: Dict) -> Dict[str, Any]: """Deaktiviert den Kiosk-Modus.""" try: system_logger.info("🖥️ Kiosk-Modus wird deaktiviert...") # Kiosk-Service stoppen und deaktivieren subprocess.run(["sudo", "systemctl", "stop", self.services["kiosk"]], check=True, timeout=30) subprocess.run(["sudo", "systemctl", "disable", self.services["kiosk"]], check=True, timeout=30) # Watchdog stoppen try: subprocess.run(["sudo", "systemctl", "stop", self.services["watchdog"]], check=True, timeout=30) subprocess.run(["sudo", "systemctl", "disable", self.services["watchdog"]], check=True, timeout=30) except Exception as e: system_logger.warning(f"Watchdog-Deaktivierung fehlgeschlagen: {e}") return {"success": True, "message": "Kiosk-Modus deaktiviert"} except subprocess.CalledProcessError as e: return {"success": False, "error": f"Kiosk-Deaktivierung fehlgeschlagen: {e}"} except Exception as e: return {"success": False, "error": f"Unerwarteter Fehler: {e}"} def _restart_services(self, operation_data: Dict) -> Dict[str, Any]: """Startet wichtige Services neu.""" try: system_logger.info("🔄 Services werden neugestartet...") success_count = 0 errors = [] # HTTPS-Service neustarten try: subprocess.run(["sudo", "systemctl", "restart", self.services["https"]], check=True, timeout=60) success_count += 1 system_logger.info("✅ HTTPS-Service neugestartet") except Exception as e: errors.append(f"HTTPS-Service: {e}") # NetworkManager neustarten (falls nötig) try: subprocess.run(["sudo", "systemctl", "restart", "NetworkManager"], check=True, timeout=30) success_count += 1 system_logger.info("✅ NetworkManager neugestartet") except Exception as e: errors.append(f"NetworkManager: {e}") if success_count > 0: return { "success": True, "message": f"Services neugestartet ({success_count})", "errors": errors if errors else None } else: return { "success": False, "error": "Alle Service-Neustarts fehlgeschlagen", "details": errors } except Exception as e: return {"success": False, "error": f"Service-Neustart fehlgeschlagen: {e}"} def _emergency_stop(self, operation_data: Dict) -> Dict[str, Any]: """Notfall-Stopp aller Services.""" try: system_logger.warning("🚨 Notfall-Stopp wird ausgeführt...") # Flask-App stoppen try: os.kill(os.getpid(), signal.SIGTERM) except Exception as e: system_logger.error(f"Flask-Stopp fehlgeschlagen: {e}") return {"success": True, "message": "Notfall-Stopp initiiert"} except Exception as e: return {"success": False, "error": f"Notfall-Stopp fehlgeschlagen: {e}"} def _cleanup_before_restart(self): """Führt Cleanup-Operationen vor Neustart/Shutdown aus.""" try: system_logger.info("🧹 Cleanup vor Neustart/Shutdown...") # Shutdown-Manager verwenden falls verfügbar try: from utils.shutdown_manager import get_shutdown_manager shutdown_manager = get_shutdown_manager() shutdown_manager.shutdown(exit_code=0) except ImportError: system_logger.warning("Shutdown-Manager nicht verfügbar") # Datenbank-Cleanup try: from utils.database_cleanup import safe_database_cleanup safe_database_cleanup(force_mode_switch=False) except ImportError: system_logger.warning("Database-Cleanup nicht verfügbar") # Cache leeren self._clear_caches() except Exception as e: system_logger.error(f"Cleanup fehlgeschlagen: {e}") def _clear_caches(self): """Leert alle Caches.""" try: # User-Cache leeren from app import clear_user_cache, clear_printer_status_cache clear_user_cache() clear_printer_status_cache() # System-Cache leeren if not self.is_windows: subprocess.run(["sudo", "sync"], timeout=10) subprocess.run(["sudo", "echo", "3", ">", "/proc/sys/vm/drop_caches"], shell=True, timeout=10) except Exception as e: system_logger.warning(f"Cache-Clearing fehlgeschlagen: {e}") def _move_to_history(self, operation_id: str): """Verschiebt abgeschlossene Operation in Historie.""" with self.lock: if operation_id in self.pending_operations: operation_data = self.pending_operations.pop(operation_id) self.operation_history.append(operation_data) # Historie begrenzen if len(self.operation_history) > self.config["max_operation_history"]: self.operation_history = self.operation_history[-self.config["max_operation_history"]:] def cancel_operation(self, operation_id: str) -> Dict[str, Any]: """ Bricht geplante Operation ab. Args: operation_id: ID der abzubrechenden Operation Returns: Dict mit Ergebnis """ with self.lock: if operation_id not in self.pending_operations: return {"success": False, "error": "Operation nicht gefunden"} operation_data = self.pending_operations[operation_id] if operation_data["status"] == "executing": return {"success": False, "error": "Operation bereits in Ausführung"} operation_data["status"] = "cancelled" operation_data["cancelled_at"] = datetime.now() self._move_to_history(operation_id) system_logger.info(f"❌ Operation abgebrochen: {operation_id}") return {"success": True, "message": "Operation erfolgreich abgebrochen"} def get_pending_operations(self) -> List[Dict]: """Gibt alle geplanten Operationen zurück.""" with self.lock: return list(self.pending_operations.values()) def get_operation_history(self, limit: int = 20) -> List[Dict]: """Gibt Operation-Historie zurück.""" with self.lock: return self.operation_history[-limit:] if limit else self.operation_history def get_system_status(self) -> Dict[str, Any]: """Gibt aktuellen System-Status zurück.""" try: # Service-Status prüfen service_status = {} for name, service in self.services.items(): try: result = subprocess.run( ["sudo", "systemctl", "is-active", service], capture_output=True, text=True, timeout=10 ) service_status[name] = result.stdout.strip() except Exception as e: service_status[name] = f"error: {e}" # System-Metriken memory = psutil.virtual_memory() disk = psutil.disk_usage('/') # Aktive Operations pending_ops = len(self.pending_operations) return { "success": True, "timestamp": datetime.now().isoformat(), "services": service_status, "system_metrics": { "memory_percent": memory.percent, "memory_available_gb": memory.available / (1024**3), "disk_percent": disk.percent, "disk_free_gb": disk.free / (1024**3), "load_average": psutil.getloadavg()[0] if hasattr(psutil, 'getloadavg') else 0 }, "operations": { "pending": pending_ops, "history_count": len(self.operation_history) }, "is_safe": self.is_safe_to_operate()[0] } except Exception as e: return {"success": False, "error": str(e)} # Globaler System-Control-Manager _system_control_manager: Optional[SystemControlManager] = None _control_lock = threading.Lock() def get_system_control_manager() -> SystemControlManager: """ Singleton-Pattern für globalen System-Control-Manager. Returns: SystemControlManager: Globaler System-Control-Manager """ global _system_control_manager with _control_lock: if _system_control_manager is None: _system_control_manager = SystemControlManager() return _system_control_manager # Convenience-Funktionen def schedule_system_restart(delay_seconds: int = 60, user_id: str = None, reason: str = None, force: bool = False) -> Dict[str, Any]: """Plant System-Neustart.""" manager = get_system_control_manager() return manager.schedule_operation(SystemOperation.RESTART, delay_seconds, user_id, reason, force) def schedule_system_shutdown(delay_seconds: int = 30, user_id: str = None, reason: str = None, force: bool = False) -> Dict[str, Any]: """Plant System-Shutdown.""" manager = get_system_control_manager() return manager.schedule_operation(SystemOperation.SHUTDOWN, delay_seconds, user_id, reason, force) def restart_kiosk(delay_seconds: int = 10, user_id: str = None, reason: str = None) -> Dict[str, Any]: """Plant Kiosk-Neustart.""" manager = get_system_control_manager() return manager.schedule_operation(SystemOperation.KIOSK_RESTART, delay_seconds, user_id, reason) def get_system_status() -> Dict[str, Any]: """Gibt System-Status zurück.""" manager = get_system_control_manager() return manager.get_system_status()