Trixy7/trixy_core/plugins/dependencies/graph.py

# -*- coding: utf-8 -*-
"""
Abhängigkeitsgraph für Plugins.

Verwaltet die Abhängigkeitsbeziehungen zwischen Plugins
und ermöglicht topologische Sortierung.
"""
from __future__ import annotations

from collections import defaultdict
from dataclasses import dataclass, field
from enum import IntEnum, auto
from typing import Any, Iterator


class CycleError(Exception):
    """Wird ausgelöst, wenn ein Zyklus erkannt wird."""

    def __init__(self, cycle: list[str]) -> None:
        self.cycle = cycle
        super().__init__(f"Zyklische Abhängigkeit erkannt: {' -> '.join(cycle)}")


class NodeState(IntEnum):
    """Zustand eines Knotens während der Traversierung."""
    UNVISITED = auto()
    VISITING = auto()
    VISITED = auto()


@dataclass
class DependencyNode:
    """
    Ein Knoten im Abhängigkeitsgraphen.

    Attributes:
        name: Plugin-Name
        version: Plugin-Version
        dependencies: Liste der Abhängigkeiten (Name -> Version-Requirement)
        optional_dependencies: Optionale Abhängigkeiten
        metadata: Zusätzliche Metadaten
    """
    name: str
    version: str = "1.0.0"
    dependencies: dict[str, str] = field(default_factory=dict)
    optional_dependencies: dict[str, str] = field(default_factory=dict)
    metadata: dict[str, Any] = field(default_factory=dict)

    def has_dependency(self, name: str) -> bool:
        """Prüft, ob eine Abhängigkeit existiert."""
        return name in self.dependencies or name in self.optional_dependencies

    def get_all_dependencies(self) -> dict[str, str]:
        """Gibt alle Abhängigkeiten zurück (required + optional)."""
        result = self.dependencies.copy()
        result.update(self.optional_dependencies)
        return result

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "name": self.name,
            "version": self.version,
            "dependencies": self.dependencies.copy(),
            "optional_dependencies": self.optional_dependencies.copy(),
            "metadata": self.metadata.copy(),
        }


class DependencyGraph:
    """
    Gerichteter Graph für Plugin-Abhängigkeiten.

    Ermöglicht:
    - Hinzufügen/Entfernen von Knoten
    - Zykluserkennung
    - Topologische Sortierung
    - Suche nach abhängigen Plugins

    Example:
        graph = DependencyGraph()

        # Knoten hinzufügen
        graph.add_node(DependencyNode(
            name="plugin_a",
            version="1.0.0",
            dependencies={"plugin_b": "^1.0.0"}
        ))
        graph.add_node(DependencyNode(
            name="plugin_b",
            version="1.2.0"
        ))

        # Ladereihenfolge berechnen
        order = graph.get_load_order()
        # ['plugin_b', 'plugin_a']

        # Abhängige finden
        dependents = graph.get_dependents("plugin_b")
        # ['plugin_a']

        # Als DOT exportieren
        dot = graph.to_dot()
    """

    def __init__(self) -> None:
        """Initialisiert den Graphen."""
        self._nodes: dict[str, DependencyNode] = {}
        self._edges: dict[str, set[str]] = defaultdict(set)  # from -> set(to)
        self._reverse_edges: dict[str, set[str]] = defaultdict(set)  # to -> set(from)

    @property
    def node_count(self) -> int:
        """Anzahl Knoten."""
        return len(self._nodes)

    @property
    def edge_count(self) -> int:
        """Anzahl Kanten."""
        return sum(len(targets) for targets in self._edges.values())

    def add_node(self, node: DependencyNode) -> None:
        """
        Fügt einen Knoten hinzu.

        Args:
            node: Der hinzuzufügende Knoten

        Raises:
            ValueError: Wenn der Knoten bereits existiert
        """
        if node.name in self._nodes:
            raise ValueError(f"Knoten existiert bereits: {node.name}")

        self._nodes[node.name] = node

        # Kanten für Abhängigkeiten hinzufügen
        for dep_name in node.dependencies.keys():
            self._edges[node.name].add(dep_name)
            self._reverse_edges[dep_name].add(node.name)

    def update_node(self, node: DependencyNode) -> None:
        """
        Aktualisiert einen existierenden Knoten.

        Args:
            node: Der zu aktualisierende Knoten
        """
        if node.name in self._nodes:
            self.remove_node(node.name)
        self.add_node(node)

    def remove_node(self, name: str) -> DependencyNode | None:
        """
        Entfernt einen Knoten.

        Args:
            name: Knotenname

        Returns:
            Der entfernte Knoten oder None
        """
        node = self._nodes.pop(name, None)
        if node is None:
            return None

        # Kanten entfernen
        if name in self._edges:
            for target in self._edges[name]:
                self._reverse_edges[target].discard(name)
            del self._edges[name]

        if name in self._reverse_edges:
            for source in self._reverse_edges[name]:
                self._edges[source].discard(name)
            del self._reverse_edges[name]

        return node

    def get_node(self, name: str) -> DependencyNode | None:
        """Gibt einen Knoten zurück."""
        return self._nodes.get(name)

    def has_node(self, name: str) -> bool:
        """Prüft, ob ein Knoten existiert."""
        return name in self._nodes

    def get_dependencies(self, name: str) -> set[str]:
        """
        Gibt die direkten Abhängigkeiten eines Knotens zurück.

        Args:
            name: Knotenname

        Returns:
            Set der Abhängigkeiten
        """
        return self._edges.get(name, set()).copy()

    def get_dependents(self, name: str) -> set[str]:
        """
        Gibt alle Knoten zurück, die von diesem abhängen.

        Args:
            name: Knotenname

        Returns:
            Set der abhängigen Knoten
        """
        return self._reverse_edges.get(name, set()).copy()

    def get_all_dependencies(self, name: str) -> set[str]:
        """
        Gibt alle transitiven Abhängigkeiten zurück.

        Args:
            name: Knotenname

        Returns:
            Set aller direkten und indirekten Abhängigkeiten
        """
        result: set[str] = set()
        stack = list(self._edges.get(name, set()))

        while stack:
            dep = stack.pop()
            if dep not in result:
                result.add(dep)
                stack.extend(self._edges.get(dep, set()))

        return result

    def get_all_dependents(self, name: str) -> set[str]:
        """
        Gibt alle Knoten zurück, die direkt oder indirekt abhängig sind.

        Args:
            name: Knotenname

        Returns:
            Set aller abhängigen Knoten
        """
        result: set[str] = set()
        stack = list(self._reverse_edges.get(name, set()))

        while stack:
            dep = stack.pop()
            if dep not in result:
                result.add(dep)
                stack.extend(self._reverse_edges.get(dep, set()))

        return result

    def has_cycle(self) -> bool:
        """
        Prüft, ob der Graph einen Zyklus enthält.

        Returns:
            True wenn ein Zyklus existiert
        """
        try:
            self._detect_cycle()
            return False
        except CycleError:
            return True

    def find_cycle(self) -> list[str] | None:
        """
        Findet einen Zyklus im Graphen.

        Returns:
            Liste der Knoten im Zyklus oder None
        """
        try:
            self._detect_cycle()
            return None
        except CycleError as e:
            return e.cycle

    def _detect_cycle(self) -> None:
        """Wirft CycleError wenn ein Zyklus gefunden wird."""
        state: dict[str, NodeState] = {
            name: NodeState.UNVISITED for name in self._nodes
        }
        path: list[str] = []

        def visit(name: str) -> None:
            if state.get(name, NodeState.UNVISITED) == NodeState.VISITED:
                return
            if state.get(name, NodeState.UNVISITED) == NodeState.VISITING:
                # Zyklus gefunden
                cycle_start = path.index(name)
                raise CycleError(path[cycle_start:] + [name])

            state[name] = NodeState.VISITING
            path.append(name)

            for dep in self._edges.get(name, set()):
                visit(dep)

            path.pop()
            state[name] = NodeState.VISITED

        for name in self._nodes:
            if state[name] == NodeState.UNVISITED:
                visit(name)

    def get_load_order(self) -> list[str]:
        """
        Berechnet die Ladereihenfolge (topologische Sortierung).

        Abhängigkeiten kommen vor den abhängigen Plugins.

        Returns:
            Liste der Knotennamen in Ladereihenfolge

        Raises:
            CycleError: Bei zyklischen Abhängigkeiten
        """
        self._detect_cycle()

        # Kahn's Algorithmus
        in_degree = {name: 0 for name in self._nodes}
        for sources in self._edges.values():
            for target in sources:
                if target in in_degree:
                    in_degree[target] += 1

        # Starte mit Knoten ohne eingehende Kanten
        queue = [name for name, degree in in_degree.items() if degree == 0]
        result: list[str] = []

        while queue:
            # Sortiert für deterministische Reihenfolge
            queue.sort()
            name = queue.pop(0)
            result.append(name)

            for dep in self._reverse_edges.get(name, set()):
                if dep in in_degree:
                    in_degree[dep] -= 1
                    if in_degree[dep] == 0:
                        queue.append(dep)

        # Umkehren: Abhängigkeiten zuerst
        result.reverse()
        return result

    def get_unload_order(self) -> list[str]:
        """
        Berechnet die Entlade-Reihenfolge.

        Umgekehrte Ladereihenfolge - abhängige Plugins zuerst.

        Returns:
            Liste der Knotennamen in Entlade-Reihenfolge
        """
        return list(reversed(self.get_load_order()))

    def get_missing_dependencies(self) -> dict[str, list[str]]:
        """
        Findet alle fehlenden Abhängigkeiten.

        Returns:
            Dictionary mit Knoten -> Liste fehlender Abhängigkeiten
        """
        missing: dict[str, list[str]] = {}

        for name, node in self._nodes.items():
            node_missing = []
            for dep_name in node.dependencies.keys():
                if dep_name not in self._nodes:
                    node_missing.append(dep_name)
            if node_missing:
                missing[name] = node_missing

        return missing

    def get_root_nodes(self) -> list[str]:
        """
        Gibt alle Knoten ohne Abhängigkeiten zurück.

        Returns:
            Liste der Root-Knoten
        """
        return [
            name for name, node in self._nodes.items()
            if not node.dependencies
        ]

    def get_leaf_nodes(self) -> list[str]:
        """
        Gibt alle Knoten ohne abhängige zurück.

        Returns:
            Liste der Blatt-Knoten
        """
        return [
            name for name in self._nodes
            if not self._reverse_edges.get(name)
        ]

    def to_dot(self, highlight: set[str] | None = None) -> str:
        """
        Exportiert den Graphen als DOT-Format.

        Args:
            highlight: Optional zu hervorhebende Knoten

        Returns:
            DOT-String für Graphviz
        """
        highlight = highlight or set()
        lines = ["digraph dependencies {"]
        lines.append('  rankdir=LR;')
        lines.append('  node [shape=box];')

        # Knoten
        for name, node in self._nodes.items():
            style = 'style=filled fillcolor=yellow' if name in highlight else ''
            label = f"{name}\\n{node.version}"
            lines.append(f'  "{name}" [label="{label}" {style}];')

        # Kanten
        for source, targets in self._edges.items():
            for target in sorted(targets):
                style = 'style=dashed' if target not in self._nodes else ''
                lines.append(f'  "{source}" -> "{target}" [{style}];')

        lines.append("}")
        return "\n".join(lines)

    def to_mermaid(self) -> str:
        """
        Exportiert den Graphen als Mermaid-Format.

        Returns:
            Mermaid-String
        """
        lines = ["graph LR"]

        for name, node in self._nodes.items():
            lines.append(f'  {name}["{name} v{node.version}"]')

        for source, targets in self._edges.items():
            for target in sorted(targets):
                lines.append(f'  {source} --> {target}')

        return "\n".join(lines)

    def __contains__(self, name: str) -> bool:
        return name in self._nodes

    def __iter__(self) -> Iterator[DependencyNode]:
        return iter(self._nodes.values())

    def __len__(self) -> int:
        return len(self._nodes)