# -*- coding: utf-8 -*- """ Erweiterte Log-Handler für verschiedene Output-Ziele. Bietet Rotating, Syslog und Async Handler für produktionsreifes Logging. """ from __future__ import annotations import asyncio import gzip import logging import os import queue import socket import threading from datetime import datetime from pathlib import Path from typing import Any, Callable class RotatingFileHandler(logging.Handler): """ Handler für rotierende Log-Dateien mit Größen- und Zeitlimits. Unterstützt automatische Kompression alter Logs und Aufräumung basierend auf Anzahl oder Alter der Backup-Dateien. """ def __init__( self, filename: str | Path, max_bytes: int = 10 * 1024 * 1024, # 10 MB backup_count: int = 5, rotate_on_time: bool = False, rotation_interval: str = "midnight", # midnight, hourly, daily compress: bool = True, encoding: str = "utf-8", ) -> None: """ Initialisiert den Rotating-Handler. Args: filename: Pfad zur Log-Datei. max_bytes: Maximale Dateigröße vor Rotation. backup_count: Anzahl der zu behaltenden Backup-Dateien. rotate_on_time: Ob zusätzlich zeitbasiert rotiert werden soll. rotation_interval: Rotationsintervall (midnight, hourly, daily). compress: Ob alte Dateien komprimiert werden sollen. encoding: Datei-Encoding. """ super().__init__() self.filename = Path(filename) self.max_bytes = max_bytes self.backup_count = backup_count self.rotate_on_time = rotate_on_time self.rotation_interval = rotation_interval self.compress = compress self.encoding = encoding self._stream: Any = None self._lock = threading.RLock() self._current_size = 0 self._last_rotation: datetime | None = None self._rotation_at: datetime | None = None # Verzeichnis erstellen falls nötig self.filename.parent.mkdir(parents=True, exist_ok=True) # Initiale Datei öffnen self._open_stream() # Nächste zeitbasierte Rotation berechnen if self.rotate_on_time: self._calculate_next_rotation() def _open_stream(self) -> None: """Öffnet den Datei-Stream.""" with self._lock: if self._stream: self._stream.close() self._stream = open( self.filename, "a", encoding=self.encoding, ) # Aktuelle Dateigröße ermitteln try: self._current_size = self.filename.stat().st_size except OSError: self._current_size = 0 def _calculate_next_rotation(self) -> None: """Berechnet den nächsten Rotationszeitpunkt.""" now = datetime.now() if self.rotation_interval == "midnight": # Nächster Mitternacht next_day = now.replace( hour=0, minute=0, second=0, microsecond=0 ) if next_day <= now: from datetime import timedelta next_day += timedelta(days=1) self._rotation_at = next_day elif self.rotation_interval == "hourly": # Nächste volle Stunde next_hour = now.replace(minute=0, second=0, microsecond=0) if next_hour <= now: from datetime import timedelta next_hour += timedelta(hours=1) self._rotation_at = next_hour elif self.rotation_interval == "daily": # Gleiche Zeit morgen from datetime import timedelta self._rotation_at = now + timedelta(days=1) def _should_rotate(self) -> bool: """Prüft, ob rotiert werden sollte.""" # Größenbasierte Rotation if self._current_size >= self.max_bytes: return True # Zeitbasierte Rotation if self.rotate_on_time and self._rotation_at: if datetime.now() >= self._rotation_at: return True return False def _do_rotation(self) -> None: """Führt die Dateirotation durch.""" with self._lock: if self._stream: self._stream.close() self._stream = None # Backup-Dateien rotieren for i in range(self.backup_count - 1, 0, -1): src = self._get_backup_name(i) dst = self._get_backup_name(i + 1) if src.exists(): if dst.exists(): dst.unlink() src.rename(dst) # Aktuelle Datei als erstes Backup if self.filename.exists(): backup = self._get_backup_name(1) if backup.exists(): backup.unlink() self.filename.rename(backup) # Optional komprimieren if self.compress: self._compress_file(backup) # Überzählige Backups löschen self._cleanup_old_backups() # Neue Datei öffnen self._open_stream() self._last_rotation = datetime.now() if self.rotate_on_time: self._calculate_next_rotation() def _get_backup_name(self, index: int) -> Path: """Generiert den Namen für eine Backup-Datei.""" suffix = f".{index}" if self.compress and index > 0: suffix += ".gz" return self.filename.with_suffix(self.filename.suffix + suffix) def _compress_file(self, filepath: Path) -> None: """Komprimiert eine Datei mit gzip.""" try: compressed = filepath.with_suffix(filepath.suffix + ".gz") with open(filepath, "rb") as f_in: with gzip.open(compressed, "wb") as f_out: f_out.write(f_in.read()) filepath.unlink() except Exception: # Bei Fehlern unkomprimierte Datei behalten pass def _cleanup_old_backups(self) -> None: """Entfernt überzählige Backup-Dateien.""" for i in range(self.backup_count + 1, self.backup_count + 100): backup = self._get_backup_name(i) if backup.exists(): backup.unlink() else: break def emit(self, record: logging.LogRecord) -> None: """ Schreibt einen Log-Record in die Datei. Args: record: Der zu schreibende Log-Record. """ try: with self._lock: if self._should_rotate(): self._do_rotation() msg = self.format(record) + "\n" self._stream.write(msg) self._stream.flush() self._current_size += len(msg.encode(self.encoding)) except Exception: self.handleError(record) def close(self) -> None: """Schließt den Handler und den Datei-Stream.""" with self._lock: if self._stream: self._stream.close() self._stream = None super().close() class SyslogHandler(logging.Handler): """ Handler für Syslog-Ausgabe (RFC 5424). Unterstützt UDP und TCP Verbindungen zu lokalen oder Remote-Syslog-Servern. """ # Syslog Facility Codes FACILITIES = { "kern": 0, "user": 1, "mail": 2, "daemon": 3, "auth": 4, "syslog": 5, "lpr": 6, "news": 7, "uucp": 8, "cron": 9, "authpriv": 10, "ftp": 11, "local0": 16, "local1": 17, "local2": 18, "local3": 19, "local4": 20, "local5": 21, "local6": 22, "local7": 23, } # Syslog Severity (entspricht ungefähr Python Log-Levels) SEVERITY = { logging.CRITICAL: 2, # CRITICAL logging.ERROR: 3, # ERROR logging.WARNING: 4, # WARNING logging.INFO: 6, # INFO logging.DEBUG: 7, # DEBUG } def __init__( self, address: tuple[str, int] = ("localhost", 514), facility: str = "user", use_tcp: bool = False, app_name: str = "trixy", ) -> None: """ Initialisiert den Syslog-Handler. Args: address: (Host, Port) Tupel für den Syslog-Server. facility: Syslog Facility (local0-local7, daemon, etc.). use_tcp: Ob TCP statt UDP verwendet werden soll. app_name: Anwendungsname für die Syslog-Nachricht. """ super().__init__() self.address = address self.facility = self.FACILITIES.get(facility, 1) self.use_tcp = use_tcp self.app_name = app_name self._socket: socket.socket | None = None self._lock = threading.Lock() self._connect() def _connect(self) -> None: """Stellt die Verbindung zum Syslog-Server her.""" with self._lock: if self._socket: self._socket.close() if self.use_tcp: self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self._socket.connect(self.address) else: self._socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) def _format_message(self, record: logging.LogRecord) -> bytes: """Formatiert eine Syslog-Nachricht nach RFC 5424.""" severity = self.SEVERITY.get(record.levelno, 6) priority = (self.facility << 3) | severity timestamp = datetime.fromtimestamp(record.created).isoformat() hostname = socket.gethostname() pid = os.getpid() msg = self.format(record) # RFC 5424 Format syslog_msg = ( f"<{priority}>1 {timestamp} {hostname} " f"{self.app_name} {pid} - - {msg}" ) return syslog_msg.encode("utf-8") def emit(self, record: logging.LogRecord) -> None: """ Sendet einen Log-Record an den Syslog-Server. Args: record: Der zu sendende Log-Record. """ try: msg = self._format_message(record) with self._lock: if self._socket is None: self._connect() if self.use_tcp: self._socket.send(msg + b"\n") else: self._socket.sendto(msg, self.address) except Exception: self.handleError(record) # Verbindung neu aufbauen bei Fehlern self._socket = None def close(self) -> None: """Schließt den Handler und die Socket-Verbindung.""" with self._lock: if self._socket: self._socket.close() self._socket = None super().close() class AsyncHandler(logging.Handler): """ Asynchroner Handler für nicht-blockierendes Logging. Logs werden in eine Queue geschrieben und von einem Hintergrund-Thread verarbeitet. """ def __init__( self, handler: logging.Handler, queue_size: int = 10000, flush_on_close: bool = True, overflow_policy: str = "drop", # drop, block, replace_oldest ) -> None: """ Initialisiert den Async-Handler. Args: handler: Der eigentliche Handler für die Verarbeitung. queue_size: Maximale Größe der Queue. flush_on_close: Ob die Queue beim Schließen geleert werden soll. overflow_policy: Verhalten bei voller Queue. """ super().__init__() self._handler = handler self._queue: queue.Queue[logging.LogRecord | None] = queue.Queue( maxsize=queue_size ) self._flush_on_close = flush_on_close self._overflow_policy = overflow_policy self._running = True self._dropped = 0 # Hintergrund-Thread starten self._thread = threading.Thread( target=self._process_queue, daemon=True, name="AsyncLogHandler", ) self._thread.start() def _process_queue(self) -> None: """Verarbeitet Log-Records aus der Queue.""" while self._running or not self._queue.empty(): try: record = self._queue.get(timeout=0.1) if record is None: break self._handler.handle(record) except queue.Empty: continue except Exception: # Fehler in der Verarbeitung ignorieren pass def emit(self, record: logging.LogRecord) -> None: """ Fügt einen Log-Record zur Queue hinzu. Args: record: Der zu logende Record. """ if not self._running: return try: if self._overflow_policy == "block": self._queue.put(record) elif self._overflow_policy == "replace_oldest": try: self._queue.put_nowait(record) except queue.Full: # Ältestes Element entfernen try: self._queue.get_nowait() self._dropped += 1 except queue.Empty: pass self._queue.put_nowait(record) else: # drop try: self._queue.put_nowait(record) except queue.Full: self._dropped += 1 except Exception: self.handleError(record) @property def dropped_count(self) -> int: """Gibt die Anzahl der verworfenen Records zurück.""" return self._dropped @property def queue_size(self) -> int: """Gibt die aktuelle Queue-Größe zurück.""" return self._queue.qsize() def flush(self) -> None: """Wartet bis die Queue leer ist.""" while not self._queue.empty(): pass self._handler.flush() def close(self) -> None: """Schließt den Handler und stoppt den Thread.""" self._running = False if self._flush_on_close: self.flush() # Sentinel-Wert zum Beenden des Threads try: self._queue.put_nowait(None) except queue.Full: pass self._thread.join(timeout=5.0) self._handler.close() super().close() class BufferedHandler(logging.Handler): """ Handler mit Pufferung für Batch-Verarbeitung. Sammelt Log-Records und schreibt sie in Batches. """ def __init__( self, handler: logging.Handler, capacity: int = 100, flush_interval: float = 5.0, flush_level: int | None = logging.ERROR, ) -> None: """ Initialisiert den Buffered-Handler. Args: handler: Der eigentliche Handler für die Ausgabe. capacity: Maximale Puffer-Größe vor automatischem Flush. flush_interval: Zeitintervall für automatisches Flushing (Sekunden). flush_level: Log-Level bei dem sofort geflusht wird. """ super().__init__() self._handler = handler self._capacity = capacity self._flush_interval = flush_interval self._flush_level = flush_level self._buffer: list[logging.LogRecord] = [] self._lock = threading.Lock() self._last_flush = datetime.now() # Hintergrund-Thread für zeitbasiertes Flushing self._running = True self._timer_thread = threading.Thread( target=self._timer_loop, daemon=True, name="BufferedLogTimer", ) self._timer_thread.start() def _timer_loop(self) -> None: """Prüft periodisch auf zeitbasiertes Flushing.""" while self._running: threading.Event().wait(1.0) # 1 Sekunde warten with self._lock: elapsed = (datetime.now() - self._last_flush).total_seconds() if elapsed >= self._flush_interval and self._buffer: self._do_flush() def _do_flush(self) -> None: """Schreibt alle gepufferten Records.""" for record in self._buffer: self._handler.handle(record) self._buffer.clear() self._last_flush = datetime.now() def emit(self, record: logging.LogRecord) -> None: """ Puffert einen Log-Record. Args: record: Der zu puffernde Record. """ with self._lock: self._buffer.append(record) # Sofortiges Flushing bei kritischen Levels if ( self._flush_level is not None and record.levelno >= self._flush_level ): self._do_flush() elif len(self._buffer) >= self._capacity: self._do_flush() def flush(self) -> None: """Flusht den Puffer.""" with self._lock: self._do_flush() self._handler.flush() def close(self) -> None: """Schließt den Handler.""" self._running = False self.flush() self._timer_thread.join(timeout=2.0) self._handler.close() super().close() class FilteringHandler(logging.Handler): """ Handler mit erweiterter Filterung. Unterstützt Filterung basierend auf Regex, Callables und Attributen. """ def __init__( self, handler: logging.Handler, include_patterns: list[str] | None = None, exclude_patterns: list[str] | None = None, level_filter: Callable[[logging.LogRecord], bool] | None = None, ) -> None: """ Initialisiert den Filtering-Handler. Args: handler: Der eigentliche Handler. include_patterns: Regex-Pattern für inkludierte Logger-Namen. exclude_patterns: Regex-Pattern für exkludierte Logger-Namen. level_filter: Optionale Funktion für Level-basierte Filterung. """ super().__init__() self._handler = handler self._level_filter = level_filter import re self._include_patterns = ( [re.compile(p) for p in include_patterns] if include_patterns else None ) self._exclude_patterns = ( [re.compile(p) for p in exclude_patterns] if exclude_patterns else None ) def _should_emit(self, record: logging.LogRecord) -> bool: """Prüft, ob der Record emittiert werden soll.""" # Include-Filter if self._include_patterns: if not any(p.search(record.name) for p in self._include_patterns): return False # Exclude-Filter if self._exclude_patterns: if any(p.search(record.name) for p in self._exclude_patterns): return False # Custom Level-Filter if self._level_filter: if not self._level_filter(record): return False return True def emit(self, record: logging.LogRecord) -> None: """ Emittiert einen Record falls er den Filter passiert. Args: record: Der zu filternde Record. """ if self._should_emit(record): self._handler.handle(record) def close(self) -> None: """Schließt den Handler.""" self._handler.close() super().close()