# -*- coding: utf-8 -*- """ Windows-sichere Logging-Konfiguration für MYP Platform ====================================================== Robuste Logging-Konfiguration mit Windows-spezifischen Fixes für File-Locking-Probleme. """ import os import sys import time import logging import threading from datetime import datetime from functools import wraps from typing import Optional, Dict, Any from logging.handlers import RotatingFileHandler, TimedRotatingFileHandler # ===== WINDOWS-SICHERE LOGGING-KLASSE ===== class WindowsSafeRotatingFileHandler(RotatingFileHandler): """ Windows-sichere Implementierung von RotatingFileHandler. Behebt das WinError 32 Problem bei gleichzeitigen Log-Dateizugriffen. """ def __init__(self, filename, mode='a', maxBytes=0, backupCount=0, encoding=None, delay=False): # Verwende UTF-8 Encoding standardmäßig if encoding is None: encoding = 'utf-8' # Windows-spezifische Konfiguration self._windows_safe_mode = os.name == 'nt' self._rotation_lock = threading.Lock() super().__init__(filename, mode, maxBytes, backupCount, encoding, delay) def doRollover(self): """ Windows-sichere Log-Rotation mit verbessertem Error-Handling. """ if not self._windows_safe_mode: # Normale Rotation für Unix-Systeme return super().doRollover() # Windows-spezifische sichere Rotation with self._rotation_lock: try: if self.stream: self.stream.close() self.stream = None # Warte kurz bevor Rotation versucht wird time.sleep(0.1) # Versuche Rotation mehrmals mit exponentialem Backoff max_attempts = 5 for attempt in range(max_attempts): try: # Rotation durchführen super().doRollover() break except (PermissionError, OSError) as e: if attempt == max_attempts - 1: # Bei letztem Versuch: Erstelle neue Log-Datei ohne Rotation print(f"WARNUNG: Log-Rotation fehlgeschlagen - erstelle neue Datei: {e}") self._create_new_log_file() break else: # Warte exponentiell länger bei jedem Versuch wait_time = 0.5 * (2 ** attempt) time.sleep(wait_time) except Exception as e: print(f"KRITISCHER FEHLER bei Log-Rotation: {e}") # Notfall: Erstelle neue Log-Datei self._create_new_log_file() def _create_new_log_file(self): """ Erstellt eine neue Log-Datei als Fallback wenn Rotation fehlschlägt. """ try: # Füge Timestamp zum Dateinamen hinzu timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') base_name, ext = os.path.splitext(self.baseFilename) new_filename = f"{base_name}_{timestamp}{ext}" # Öffne neue Datei self.baseFilename = new_filename self.stream = self._open() except Exception as e: print(f"NOTFALL: Kann keine neue Log-Datei erstellen: {e}") # Letzter Ausweg: Console-Logging self.stream = sys.stderr # ===== GLOBALE LOGGING-KONFIGURATION ===== # Logger-Registry für Singleton-Pattern _logger_registry: Dict[str, logging.Logger] = {} _logging_initialized = False _init_lock = threading.Lock() def setup_logging(log_level: str = "INFO", base_log_dir: str = None) -> None: """ Initialisiert das zentrale Logging-System mit Windows-sicherer Konfiguration. Args: log_level: Logging-Level (DEBUG, INFO, WARNING, ERROR, CRITICAL) base_log_dir: Basis-Verzeichnis für Log-Dateien """ global _logging_initialized with _init_lock: if _logging_initialized: return try: # Bestimme Log-Verzeichnis if base_log_dir is None: current_dir = os.path.dirname(os.path.abspath(__file__)) base_log_dir = os.path.join(current_dir, '..', 'logs') # Erstelle Log-Verzeichnisse log_dirs = ['app', 'auth', 'jobs', 'printers', 'scheduler', 'errors'] for log_dir in log_dirs: full_path = os.path.join(base_log_dir, log_dir) os.makedirs(full_path, exist_ok=True) # Konfiguriere Root-Logger root_logger = logging.getLogger() root_logger.setLevel(getattr(logging, log_level.upper(), logging.INFO)) # Entferne existierende Handler um Duplikate zu vermeiden for handler in root_logger.handlers[:]: root_logger.removeHandler(handler) # Console-Handler für kritische Meldungen console_handler = logging.StreamHandler(sys.stdout) console_handler.setLevel(logging.WARNING) console_formatter = logging.Formatter( '%(asctime)s - %(name)s - [%(levelname)s] %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) console_handler.setFormatter(console_formatter) root_logger.addHandler(console_handler) _logging_initialized = True print(f"✅ Logging-System erfolgreich initialisiert (Level: {log_level})") except Exception as e: print(f"❌ KRITISCHER FEHLER bei Logging-Initialisierung: {e}") # Notfall-Konfiguration logging.basicConfig( level=getattr(logging, log_level.upper(), logging.INFO), format='%(asctime)s - %(name)s - [%(levelname)s] - %(message)s', handlers=[logging.StreamHandler(sys.stdout)] ) _logging_initialized = True def get_logger(name: str, log_level: str = None) -> logging.Logger: """ Erstellt oder gibt einen konfigurierten Logger zurück. Args: name: Name des Loggers (z.B. 'app', 'auth', 'jobs') log_level: Optionaler spezifischer Log-Level für diesen Logger Returns: Konfigurierter Logger """ global _logger_registry # Stelle sicher, dass Logging initialisiert ist if not _logging_initialized: setup_logging() # Prüfe Registry für existierenden Logger if name in _logger_registry: return _logger_registry[name] try: # Erstelle neuen Logger logger = logging.getLogger(name) # Setze spezifischen Level falls angegeben if log_level: logger.setLevel(getattr(logging, log_level.upper(), logging.INFO)) # Erstelle File-Handler mit Windows-sicherer Rotation log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', 'logs', name) os.makedirs(log_dir, exist_ok=True) log_file = os.path.join(log_dir, f'{name}.log') # Windows-sicherer RotatingFileHandler file_handler = WindowsSafeRotatingFileHandler( log_file, maxBytes=10*1024*1024, # 10MB backupCount=5, encoding='utf-8' ) # Detaillierter Formatter für File-Logs file_formatter = logging.Formatter( '%(asctime)s - [%(name)s] %(name)s - [%(levelname)s] %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) file_handler.setFormatter(file_formatter) # Handler hinzufügen logger.addHandler(file_handler) # Verhindere Propagation zu Root-Logger um Duplikate zu vermeiden logger.propagate = False # In Registry speichern _logger_registry[name] = logger return logger except Exception as e: print(f"❌ Fehler beim Erstellen des Loggers '{name}': {e}") # Fallback: Einfacher Logger ohne File-Handler fallback_logger = logging.getLogger(name) if name not in _logger_registry: _logger_registry[name] = fallback_logger return fallback_logger # ===== PERFORMANCE-MEASUREMENT DECORATOR ===== def measure_execution_time(logger: logging.Logger = None, task_name: str = "Task"): """ Decorator zur Messung und Protokollierung der Ausführungszeit von Funktionen. Args: logger: Logger-Instanz für die Ausgabe task_name: Bezeichnung der Aufgabe für die Logs Returns: Decorator-Funktion """ def decorator(func): @wraps(func) def wrapper(*args, **kwargs): start_time = time.time() # Verwende provided Logger oder erstelle Standard-Logger log = logger or get_logger("performance") try: # Führe Funktion aus result = func(*args, **kwargs) # Berechne Ausführungszeit execution_time = (time.time() - start_time) * 1000 # in Millisekunden # Protokolliere Erfolg log.info(f"✅ {task_name} '{func.__name__}' erfolgreich in {execution_time:.2f}ms") return result except Exception as e: # Berechne Ausführungszeit auch bei Fehlern execution_time = (time.time() - start_time) * 1000 # Protokolliere Fehler log.error(f"❌ {task_name} '{func.__name__}' fehlgeschlagen nach {execution_time:.2f}ms: {str(e)}") # Exception weiterleiten raise return wrapper return decorator # ===== STARTUP/DEBUG LOGGING ===== def log_startup_info(): """ Protokolliert System-Startup-Informationen. """ startup_logger = get_logger("startup") try: startup_logger.info("=" * 50) startup_logger.info("🚀 MYP Platform Backend wird gestartet...") startup_logger.info(f"🐍 Python Version: {sys.version}") startup_logger.info(f"💻 Betriebssystem: {os.name} ({sys.platform})") startup_logger.info(f"📁 Arbeitsverzeichnis: {os.getcwd()}") startup_logger.info(f"⏰ Startzeit: {datetime.now().isoformat()}") # Windows-spezifische Informationen if os.name == 'nt': startup_logger.info("🪟 Windows-Modus: Aktiviert") startup_logger.info("🔒 Windows-sichere Log-Rotation: Aktiviert") startup_logger.info("=" * 50) except Exception as e: print(f"❌ Fehler beim Startup-Logging: {e}") def debug_request(logger: logging.Logger, request) -> None: """ Detailliertes Request-Debugging. Args: logger: Logger für die Ausgabe request: Flask Request-Objekt """ try: logger.debug(f"📨 REQUEST: {request.method} {request.path}") logger.debug(f"🌐 Remote-Adresse: {request.remote_addr}") logger.debug(f"🔤 Content-Type: {request.content_type}") if request.args: logger.debug(f"❓ Query-Parameter: {dict(request.args)}") if request.form and logger.level <= logging.DEBUG: # Filtere sensible Daten aus Form-Daten safe_form = {k: '***' if 'password' in k.lower() else v for k, v in request.form.items()} logger.debug(f"📝 Form-Daten: {safe_form}") except Exception as e: logger.error(f"❌ Fehler beim Request-Debugging: {str(e)}") def debug_response(logger: logging.Logger, response, duration_ms: Optional[float] = None) -> None: """ Detailliertes Response-Debugging. Args: logger: Logger für die Ausgabe response: Flask Response-Objekt duration_ms: Optionale Ausführungszeit in Millisekunden """ try: status_emoji = "✅" if response.status_code < 400 else "❌" if response.status_code >= 500 else "⚠️" log_msg = f"📤 RESPONSE: {status_emoji} {response.status_code}" if duration_ms is not None: log_msg += f" ({duration_ms:.2f}ms)" logger.debug(log_msg) logger.debug(f"📏 Content-Length: {response.content_length or 'Unbekannt'}") except Exception as e: logger.error(f"❌ Fehler beim Response-Debugging: {str(e)}") # ===== NOTFALL-LOGGING ===== def emergency_log(message: str, level: str = "ERROR") -> None: """ Notfall-Logging das auch funktioniert wenn das Hauptsystem fehlschlägt. Args: message: Nachricht level: Log-Level """ try: # Versuche normales Logging logger = get_logger("emergency") getattr(logger, level.lower(), logger.error)(message) except: # Fallback zu Print timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S') print(f"[NOTFALL {timestamp}] [{level}] {message}") # Auto-Initialisierung beim Import if __name__ != "__main__": try: setup_logging() except Exception as e: print(f"❌ Auto-Initialisierung des Logging-Systems fehlgeschlagen: {e}")