Trixy7/trixy_core/network/compression/zlib.py

# -*- coding: utf-8 -*-
"""
Zlib-Kompressor.

Implementiert Komprimierung mit dem zlib-Algorithmus,
der in Python standardmäßig verfügbar ist.
"""

from __future__ import annotations

import time
import zlib
from typing import Any

from trixy_core.network.compression.compressor import (
    Compressor,
    CompressionLevel,
    CompressionError,
    DecompressionError,
)


class ZlibCompressor(Compressor):
    """
    Kompressor basierend auf zlib.

    Verwendet den DEFLATE-Algorithmus für Komprimierung.
    Gut ausbalanciert zwischen Geschwindigkeit und Komprimierungsrate.

    Beispiel:
        compressor = ZlibCompressor(level=CompressionLevel.BEST)

        # Komprimieren
        compressed = compressor.compress(data)

        # Dekomprimieren
        original = compressor.decompress(compressed)

        # Statistiken
        print(compressor.stats.to_dict())
    """

    def __init__(
        self,
        level: CompressionLevel = CompressionLevel.DEFAULT,
        wbits: int = 15,
    ) -> None:
        """
        Initialisiert den Zlib-Kompressor.

        Args:
            level: Komprimierungsstufe (0-9)
            wbits: Fenstergröße für Komprimierung
                   15 = max Fenster (32KB)
                   -15 = raw deflate (ohne Header)
                   15+16 = gzip-kompatibel
        """
        super().__init__(level)
        self._wbits = wbits

    @property
    def name(self) -> str:
        return "zlib"

    @property
    def content_type(self) -> str:
        return "application/zlib"

    @property
    def min_size(self) -> int:
        """Zlib hat ~6 Byte Overhead."""
        return 50

    def compress(self, data: bytes) -> bytes:
        """
        Komprimiert Daten mit zlib.

        Args:
            data: Unkomprimierte Daten

        Returns:
            Komprimierte Daten

        Raises:
            CompressionError: Bei Komprimierungsfehler
        """
        if not data:
            return b""

        start = time.monotonic()

        try:
            compressed = zlib.compress(data, level=int(self.level))
        except zlib.error as e:
            self.stats.errors += 1
            raise CompressionError(f"Zlib-Komprimierung fehlgeschlagen: {e}") from e

        elapsed_ms = (time.monotonic() - start) * 1000
        self.stats.update_compression(len(data), len(compressed), elapsed_ms)

        return compressed

    def decompress(self, data: bytes) -> bytes:
        """
        Dekomprimiert zlib-Daten.

        Args:
            data: Komprimierte Daten

        Returns:
            Ursprüngliche Daten

        Raises:
            DecompressionError: Bei Dekomprimierungsfehler
        """
        if not data:
            return b""

        start = time.monotonic()

        try:
            decompressed = zlib.decompress(data)
        except zlib.error as e:
            self.stats.errors += 1
            raise DecompressionError(
                f"Zlib-Dekomprimierung fehlgeschlagen: {e}"
            ) from e

        elapsed_ms = (time.monotonic() - start) * 1000
        self.stats.update_decompression(len(data), len(decompressed), elapsed_ms)

        return decompressed

    def compress_with_header(
        self,
        data: bytes,
        include_size: bool = True,
    ) -> bytes:
        """
        Komprimiert mit zusätzlichem Header.

        Der Header enthält:
        - 4 Bytes: Originalgröße (wenn include_size=True)
        - Komprimierte Daten

        Args:
            data: Zu komprimierende Daten
            include_size: Originalgröße einschließen

        Returns:
            Komprimierte Daten mit Header
        """
        compressed = self.compress(data)

        if include_size:
            size_bytes = len(data).to_bytes(4, byteorder="big")
            return size_bytes + compressed

        return compressed

    def decompress_with_header(
        self,
        data: bytes,
        has_size: bool = True,
    ) -> bytes:
        """
        Dekomprimiert Daten mit Header.

        Args:
            data: Komprimierte Daten mit Header
            has_size: Ob Größen-Header vorhanden

        Returns:
            Dekomprimierte Daten
        """
        if has_size:
            # Größe lesen (4 Bytes)
            if len(data) < 4:
                raise DecompressionError("Daten zu kurz für Header")

            expected_size = int.from_bytes(data[:4], byteorder="big")
            compressed = data[4:]
        else:
            compressed = data

        decompressed = self.decompress(compressed)

        if has_size and len(decompressed) != expected_size:
            raise DecompressionError(
                f"Größe stimmt nicht: erwartet {expected_size}, "
                f"erhalten {len(decompressed)}"
            )

        return decompressed


class GzipCompressor(Compressor):
    """
    Kompressor für gzip-Format.

    Verwendet zlib mit gzip-kompatiblem Header.
    Kompatibel mit Standard-gzip-Tools.
    """

    def __init__(
        self,
        level: CompressionLevel = CompressionLevel.DEFAULT,
    ) -> None:
        super().__init__(level)

    @property
    def name(self) -> str:
        return "gzip"

    @property
    def content_type(self) -> str:
        return "application/gzip"

    @property
    def min_size(self) -> int:
        """Gzip hat ~18 Byte Overhead."""
        return 100

    def compress(self, data: bytes) -> bytes:
        """Komprimiert im gzip-Format."""
        if not data:
            return b""

        start = time.monotonic()

        try:
            import gzip
            compressed = gzip.compress(data, compresslevel=int(self.level))
        except Exception as e:
            self.stats.errors += 1
            raise CompressionError(f"Gzip-Komprimierung fehlgeschlagen: {e}") from e

        elapsed_ms = (time.monotonic() - start) * 1000
        self.stats.update_compression(len(data), len(compressed), elapsed_ms)

        return compressed

    def decompress(self, data: bytes) -> bytes:
        """Dekomprimiert gzip-Daten."""
        if not data:
            return b""

        start = time.monotonic()

        try:
            import gzip
            decompressed = gzip.decompress(data)
        except Exception as e:
            self.stats.errors += 1
            raise DecompressionError(
                f"Gzip-Dekomprimierung fehlgeschlagen: {e}"
            ) from e

        elapsed_ms = (time.monotonic() - start) * 1000
        self.stats.update_decompression(len(data), len(decompressed), elapsed_ms)

        return decompressed