Trixy7/trixy_core/config/versioning/diff.py

# -*- coding: utf-8 -*-
"""
Konfigurationsdifferenzen für Vergleich und Analyse.

Bietet Funktionen zum Vergleichen von Konfigurationsversionen
und Generieren von Änderungsberichten.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Sequence


class DiffType(Enum):
    """
    Typ einer Konfigurationsänderung.
    """

    ADDED = "added"
    """Neuer Schlüssel hinzugefügt."""

    REMOVED = "removed"
    """Schlüssel entfernt."""

    MODIFIED = "modified"
    """Wert geändert."""

    TYPE_CHANGED = "type_changed"
    """Typ des Wertes geändert."""


@dataclass
class DiffEntry:
    """
    Ein einzelner Unterschied zwischen zwei Konfigurationen.
    """

    path: str
    """Pfad zum geänderten Wert (z.B. 'network.host')."""

    diff_type: DiffType
    """Art der Änderung."""

    old_value: Any = None
    """Alter Wert (bei REMOVED, MODIFIED, TYPE_CHANGED)."""

    new_value: Any = None
    """Neuer Wert (bei ADDED, MODIFIED, TYPE_CHANGED)."""

    old_type: str = ""
    """Alter Typ (bei TYPE_CHANGED)."""

    new_type: str = ""
    """Neuer Typ (bei TYPE_CHANGED)."""

    def __str__(self) -> str:
        """String-Repräsentation."""
        if self.diff_type == DiffType.ADDED:
            return f"+ {self.path}: {self.new_value!r}"
        elif self.diff_type == DiffType.REMOVED:
            return f"- {self.path}: {self.old_value!r}"
        elif self.diff_type == DiffType.MODIFIED:
            return f"~ {self.path}: {self.old_value!r} -> {self.new_value!r}"
        elif self.diff_type == DiffType.TYPE_CHANGED:
            return (
                f"! {self.path}: {self.old_type}({self.old_value!r}) -> "
                f"{self.new_type}({self.new_value!r})"
            )
        return f"? {self.path}"


@dataclass
class ConfigDiff:
    """
    Differenz zwischen zwei Konfigurationen.

    Enthält alle Unterschiede und bietet Analyse-Methoden.
    """

    entries: list[DiffEntry] = field(default_factory=list)
    """Liste aller Unterschiede."""

    old_config_name: str = "old"
    """Name der alten Konfiguration."""

    new_config_name: str = "new"
    """Name der neuen Konfiguration."""

    @property
    def has_changes(self) -> bool:
        """Prüft, ob es Änderungen gibt."""
        return len(self.entries) > 0

    @property
    def additions(self) -> list[DiffEntry]:
        """Gibt alle Hinzufügungen zurück."""
        return [e for e in self.entries if e.diff_type == DiffType.ADDED]

    @property
    def removals(self) -> list[DiffEntry]:
        """Gibt alle Entfernungen zurück."""
        return [e for e in self.entries if e.diff_type == DiffType.REMOVED]

    @property
    def modifications(self) -> list[DiffEntry]:
        """Gibt alle Modifikationen zurück."""
        return [e for e in self.entries if e.diff_type == DiffType.MODIFIED]

    @property
    def type_changes(self) -> list[DiffEntry]:
        """Gibt alle Typänderungen zurück."""
        return [e for e in self.entries if e.diff_type == DiffType.TYPE_CHANGED]

    def filter_by_path(self, prefix: str) -> list[DiffEntry]:
        """
        Filtert Einträge nach Pfadpräfix.

        Args:
            prefix: Präfix zum Filtern (z.B. 'network.').

        Returns:
            Liste der passenden Einträge.
        """
        return [e for e in self.entries if e.path.startswith(prefix)]

    def summary(self) -> dict[str, int]:
        """
        Gibt eine Zusammenfassung der Änderungen zurück.

        Returns:
            Dictionary mit Anzahlen pro Typ.
        """
        return {
            "total": len(self.entries),
            "added": len(self.additions),
            "removed": len(self.removals),
            "modified": len(self.modifications),
            "type_changed": len(self.type_changes),
        }

    def to_dict(self) -> dict[str, Any]:
        """
        Konvertiert das Diff in ein Dictionary.

        Returns:
            Dictionary-Repräsentation.
        """
        return {
            "old": self.old_config_name,
            "new": self.new_config_name,
            "summary": self.summary(),
            "entries": [
                {
                    "path": e.path,
                    "type": e.diff_type.value,
                    "old_value": e.old_value,
                    "new_value": e.new_value,
                }
                for e in self.entries
            ],
        }

    def format_report(
        self,
        show_values: bool = True,
        colorize: bool = False,
    ) -> str:
        """
        Formatiert einen lesbaren Bericht.

        Args:
            show_values: Ob Werte angezeigt werden.
            colorize: Ob ANSI-Farben verwendet werden.

        Returns:
            Formatierter Bericht.
        """
        lines = [
            f"Konfigurationsdifferenz: {self.old_config_name} -> {self.new_config_name}",
            "=" * 60,
        ]

        summary = self.summary()
        lines.append(
            f"Änderungen: {summary['total']} "
            f"(+{summary['added']} -{summary['removed']} "
            f"~{summary['modified']} !{summary['type_changed']})"
        )
        lines.append("")

        # ANSI-Farben
        if colorize:
            GREEN = "\033[32m"
            RED = "\033[31m"
            YELLOW = "\033[33m"
            MAGENTA = "\033[35m"
            RESET = "\033[0m"
        else:
            GREEN = RED = YELLOW = MAGENTA = RESET = ""

        for entry in self.entries:
            if entry.diff_type == DiffType.ADDED:
                prefix = f"{GREEN}+{RESET}"
                if show_values:
                    lines.append(f"{prefix} {entry.path}: {entry.new_value!r}")
                else:
                    lines.append(f"{prefix} {entry.path}")

            elif entry.diff_type == DiffType.REMOVED:
                prefix = f"{RED}-{RESET}"
                if show_values:
                    lines.append(f"{prefix} {entry.path}: {entry.old_value!r}")
                else:
                    lines.append(f"{prefix} {entry.path}")

            elif entry.diff_type == DiffType.MODIFIED:
                prefix = f"{YELLOW}~{RESET}"
                if show_values:
                    lines.append(
                        f"{prefix} {entry.path}: "
                        f"{entry.old_value!r} -> {entry.new_value!r}"
                    )
                else:
                    lines.append(f"{prefix} {entry.path}")

            elif entry.diff_type == DiffType.TYPE_CHANGED:
                prefix = f"{MAGENTA}!{RESET}"
                if show_values:
                    lines.append(
                        f"{prefix} {entry.path}: "
                        f"[{entry.old_type}] -> [{entry.new_type}]"
                    )
                else:
                    lines.append(f"{prefix} {entry.path}")

        return "\n".join(lines)

    def __str__(self) -> str:
        """String-Repräsentation."""
        return self.format_report(show_values=False)

    def __len__(self) -> int:
        """Anzahl der Änderungen."""
        return len(self.entries)

    def __bool__(self) -> bool:
        """True wenn Änderungen vorhanden."""
        return self.has_changes


def _flatten_dict(
    data: dict[str, Any],
    prefix: str = "",
    separator: str = ".",
) -> dict[str, Any]:
    """Flacht ein verschachteltes Dictionary ab."""
    result: dict[str, Any] = {}

    for key, value in data.items():
        full_key = f"{prefix}{separator}{key}" if prefix else key

        if isinstance(value, dict):
            result.update(_flatten_dict(value, full_key, separator))
        else:
            result[full_key] = value

    return result


def diff_configs(
    old_config: dict[str, Any],
    new_config: dict[str, Any],
    old_name: str = "old",
    new_name: str = "new",
    ignore_paths: Sequence[str] | None = None,
    separator: str = ".",
) -> ConfigDiff:
    """
    Vergleicht zwei Konfigurationen und gibt die Unterschiede zurück.

    Args:
        old_config: Alte Konfiguration.
        new_config: Neue Konfiguration.
        old_name: Name der alten Konfiguration.
        new_name: Name der neuen Konfiguration.
        ignore_paths: Pfade, die ignoriert werden sollen.
        separator: Pfad-Trennzeichen.

    Returns:
        ConfigDiff mit allen Unterschieden.
    """
    ignore_set = set(ignore_paths or [])
    entries: list[DiffEntry] = []

    # Dicts abflachen
    old_flat = _flatten_dict(old_config, separator=separator)
    new_flat = _flatten_dict(new_config, separator=separator)

    all_keys = set(old_flat.keys()) | set(new_flat.keys())

    for key in sorted(all_keys):
        # Ignorierte Pfade überspringen
        if any(key.startswith(p) for p in ignore_set):
            continue

        old_value = old_flat.get(key)
        new_value = new_flat.get(key)

        if key not in old_flat:
            # Hinzugefügt
            entries.append(DiffEntry(
                path=key,
                diff_type=DiffType.ADDED,
                new_value=new_value,
            ))

        elif key not in new_flat:
            # Entfernt
            entries.append(DiffEntry(
                path=key,
                diff_type=DiffType.REMOVED,
                old_value=old_value,
            ))

        elif type(old_value) != type(new_value):
            # Typ geändert
            entries.append(DiffEntry(
                path=key,
                diff_type=DiffType.TYPE_CHANGED,
                old_value=old_value,
                new_value=new_value,
                old_type=type(old_value).__name__,
                new_type=type(new_value).__name__,
            ))

        elif old_value != new_value:
            # Wert geändert
            entries.append(DiffEntry(
                path=key,
                diff_type=DiffType.MODIFIED,
                old_value=old_value,
                new_value=new_value,
            ))

    return ConfigDiff(
        entries=entries,
        old_config_name=old_name,
        new_config_name=new_name,
    )


def apply_diff(
    config: dict[str, Any],
    diff: ConfigDiff,
    reverse: bool = False,
) -> dict[str, Any]:
    """
    Wendet ein Diff auf eine Konfiguration an.

    Args:
        config: Basis-Konfiguration.
        diff: Das anzuwendende Diff.
        reverse: Ob das Diff rückwärts angewendet werden soll.

    Returns:
        Aktualisierte Konfiguration.
    """
    import copy
    result = copy.deepcopy(config)

    def set_nested(data: dict, path: str, value: Any) -> None:
        keys = path.split(".")
        current = data
        for key in keys[:-1]:
            if key not in current:
                current[key] = {}
            current = current[key]
        current[keys[-1]] = value

    def delete_nested(data: dict, path: str) -> None:
        keys = path.split(".")
        current = data
        for key in keys[:-1]:
            if key not in current:
                return
            current = current[key]
        if keys[-1] in current:
            del current[keys[-1]]

    for entry in diff.entries:
        if reverse:
            # Rückwärts anwenden
            if entry.diff_type == DiffType.ADDED:
                delete_nested(result, entry.path)
            elif entry.diff_type == DiffType.REMOVED:
                set_nested(result, entry.path, entry.old_value)
            elif entry.diff_type in (DiffType.MODIFIED, DiffType.TYPE_CHANGED):
                set_nested(result, entry.path, entry.old_value)
        else:
            # Vorwärts anwenden
            if entry.diff_type == DiffType.ADDED:
                set_nested(result, entry.path, entry.new_value)
            elif entry.diff_type == DiffType.REMOVED:
                delete_nested(result, entry.path)
            elif entry.diff_type in (DiffType.MODIFIED, DiffType.TYPE_CHANGED):
                set_nested(result, entry.path, entry.new_value)

    return result