Trixy7/trixy_core/utils/logging/handlers.py

# -*- 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()