Trixy7/trixy_core/plugins/dependencies/resolver.py

# -*- coding: utf-8 -*-
"""
Abhängigkeitsauflösung für Plugins.

Löst Abhängigkeiten auf und berechnet die optimale
Plugin-Ladereihenfolge.
"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

from trixy_core.plugins.dependencies.graph import (
    DependencyGraph,
    DependencyNode,
    CycleError,
)
from trixy_core.plugins.versioning.semver import parse_version
from trixy_core.plugins.versioning.compatibility import (
    VersionChecker,
    CompatibilityResult,
)


class ResolutionError(Exception):
    """Fehler bei der Abhängigkeitsauflösung."""

    def __init__(
        self,
        message: str,
        missing: list[str] | None = None,
        incompatible: dict[str, str] | None = None,
        cycle: list[str] | None = None,
    ) -> None:
        super().__init__(message)
        self.missing = missing or []
        self.incompatible = incompatible or {}
        self.cycle = cycle


@dataclass
class Dependency:
    """
    Eine Plugin-Abhängigkeit.

    Attributes:
        name: Name des Plugins
        version_requirement: Versionsanforderung
        optional: Ist optional
        reason: Grund für die Abhängigkeit
    """
    name: str
    version_requirement: str = "*"
    optional: bool = False
    reason: str = ""

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "name": self.name,
            "version_requirement": self.version_requirement,
            "optional": self.optional,
            "reason": self.reason,
        }


@dataclass
class ResolutionResult:
    """
    Ergebnis einer Abhängigkeitsauflösung.

    Attributes:
        success: Auflösung erfolgreich
        load_order: Ladereihenfolge
        resolved: Aufgelöste Abhängigkeiten
        missing: Fehlende Abhängigkeiten
        incompatible: Inkompatible Versionen
        warnings: Warnungen
    """
    success: bool
    load_order: list[str] = field(default_factory=list)
    resolved: dict[str, str] = field(default_factory=dict)  # name -> version
    missing: list[str] = field(default_factory=list)
    incompatible: dict[str, CompatibilityResult] = field(default_factory=dict)
    warnings: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        """Konvertiert zu Dictionary."""
        return {
            "success": self.success,
            "load_order": self.load_order,
            "resolved": self.resolved,
            "missing": self.missing,
            "incompatible": {
                k: v.to_dict() for k, v in self.incompatible.items()
            },
            "warnings": self.warnings,
        }


class DependencyResolver:
    """
    Löst Plugin-Abhängigkeiten auf.

    Erstellt einen Abhängigkeitsgraphen, prüft Versionskompatibilität
    und berechnet die optimale Ladereihenfolge.

    Example:
        resolver = DependencyResolver()

        # Plugin-Informationen registrieren
        resolver.register_plugin(
            "my_plugin",
            version="1.0.0",
            dependencies={
                "core_plugin": "^2.0.0",
                "util_plugin": "~1.5.0",
            }
        )

        resolver.register_plugin("core_plugin", "2.1.0")
        resolver.register_plugin("util_plugin", "1.5.3")

        # Auflösen
        result = resolver.resolve()

        if result.success:
            for plugin in result.load_order:
                load_plugin(plugin)
        else:
            print(f"Fehler: {result.missing}, {result.incompatible}")

        # Visualisierung
        print(resolver.graph.to_dot())
    """

    def __init__(self, strict: bool = False) -> None:
        """
        Initialisiert den Resolver.

        Args:
            strict: Strikte Versionsprüfung
        """
        self._graph = DependencyGraph()
        self._checker = VersionChecker(strict=strict)
        self._available: dict[str, str] = {}  # name -> version

    @property
    def graph(self) -> DependencyGraph:
        """Der Abhängigkeitsgraph."""
        return self._graph

    def register_plugin(
        self,
        name: str,
        version: str = "1.0.0",
        dependencies: dict[str, str] | None = None,
        optional_dependencies: dict[str, str] | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> DependencyNode:
        """
        Registriert ein Plugin.

        Args:
            name: Plugin-Name
            version: Plugin-Version
            dependencies: Erforderliche Abhängigkeiten
            optional_dependencies: Optionale Abhängigkeiten
            metadata: Zusätzliche Metadaten

        Returns:
            Der erstellte Knoten
        """
        node = DependencyNode(
            name=name,
            version=version,
            dependencies=dependencies or {},
            optional_dependencies=optional_dependencies or {},
            metadata=metadata or {},
        )

        if self._graph.has_node(name):
            self._graph.update_node(node)
        else:
            self._graph.add_node(node)

        self._available[name] = version
        return node

    def unregister_plugin(self, name: str) -> bool:
        """
        Entfernt ein Plugin.

        Args:
            name: Plugin-Name

        Returns:
            True wenn entfernt
        """
        if self._graph.remove_node(name):
            self._available.pop(name, None)
            return True
        return False

    def resolve(
        self,
        targets: list[str] | None = None,
        include_optional: bool = False,
    ) -> ResolutionResult:
        """
        Löst Abhängigkeiten auf.

        Args:
            targets: Spezifische Plugins zum Auflösen (None = alle)
            include_optional: Optionale Abhängigkeiten einbeziehen

        Returns:
            ResolutionResult
        """
        warnings: list[str] = []
        missing: list[str] = []
        incompatible: dict[str, CompatibilityResult] = {}

        # Fehlende Abhängigkeiten finden
        missing_deps = self._graph.get_missing_dependencies()
        for plugin, deps in missing_deps.items():
            for dep in deps:
                node = self._graph.get_node(plugin)
                if node and dep in node.optional_dependencies:
                    if include_optional:
                        warnings.append(
                            f"Optionale Abhängigkeit '{dep}' für '{plugin}' nicht gefunden"
                        )
                else:
                    missing.append(f"{dep} (benötigt von {plugin})")

        # Versionskompatibilität prüfen
        for node in self._graph:
            for dep_name, requirement in node.dependencies.items():
                if dep_name not in self._available:
                    continue

                result = self._checker.check(requirement, self._available[dep_name])
                if not result.compatible:
                    incompatible[f"{node.name}->{dep_name}"] = result
                elif result.warnings:
                    for warning in result.warnings:
                        warnings.append(f"{node.name}->{dep_name}: {warning}")

        # Zyklus prüfen
        try:
            load_order = self._graph.get_load_order()

            # Nur angeforderte Plugins filtern
            if targets:
                needed = set(targets)
                for target in targets:
                    needed.update(self._graph.get_all_dependencies(target))
                load_order = [p for p in load_order if p in needed]

        except CycleError as e:
            return ResolutionResult(
                success=False,
                missing=missing,
                incompatible=incompatible,
                warnings=[f"Zyklische Abhängigkeit: {' -> '.join(e.cycle)}"],
            )

        # Erfolg bestimmen
        success = not missing and not incompatible

        return ResolutionResult(
            success=success,
            load_order=load_order,
            resolved={name: self._available[name] for name in load_order if name in self._available},
            missing=missing,
            incompatible=incompatible,
            warnings=warnings,
        )

    def check_can_install(
        self,
        name: str,
        version: str,
        dependencies: dict[str, str] | None = None,
    ) -> ResolutionResult:
        """
        Prüft, ob ein Plugin installiert werden kann.

        Args:
            name: Plugin-Name
            version: Plugin-Version
            dependencies: Plugin-Abhängigkeiten

        Returns:
            ResolutionResult
        """
        dependencies = dependencies or {}
        missing = []
        incompatible: dict[str, CompatibilityResult] = {}

        for dep_name, requirement in dependencies.items():
            if dep_name not in self._available:
                missing.append(dep_name)
            else:
                result = self._checker.check(requirement, self._available[dep_name])
                if not result.compatible:
                    incompatible[dep_name] = result

        success = not missing and not incompatible

        return ResolutionResult(
            success=success,
            missing=missing,
            incompatible=incompatible,
        )

    def check_can_uninstall(self, name: str) -> tuple[bool, list[str]]:
        """
        Prüft, ob ein Plugin deinstalliert werden kann.

        Args:
            name: Plugin-Name

        Returns:
            (Kann deinstalliert werden, Liste abhängiger Plugins)
        """
        dependents = list(self._graph.get_dependents(name))
        return (not dependents, dependents)

    def get_install_order(self, plugins: list[str]) -> list[str]:
        """
        Berechnet die Installationsreihenfolge für mehrere Plugins.

        Args:
            plugins: Liste der zu installierenden Plugins

        Returns:
            Sortierte Installationsreihenfolge
        """
        needed: set[str] = set(plugins)

        for plugin in plugins:
            needed.update(self._graph.get_all_dependencies(plugin))

        order = self._graph.get_load_order()
        return [p for p in order if p in needed]

    def get_uninstall_order(self, plugins: list[str]) -> list[str]:
        """
        Berechnet die Deinstallationsreihenfolge.

        Args:
            plugins: Liste der zu entfernenden Plugins

        Returns:
            Sortierte Deinstallationsreihenfolge
        """
        to_remove: set[str] = set(plugins)

        # Alle abhängigen auch entfernen
        for plugin in plugins:
            to_remove.update(self._graph.get_all_dependents(plugin))

        order = self._graph.get_unload_order()
        return [p for p in order if p in to_remove]

    def get_affected_by_update(
        self,
        name: str,
        new_version: str,
    ) -> list[str]:
        """
        Findet alle Plugins, die von einem Update betroffen sind.

        Args:
            name: Plugin-Name
            new_version: Neue Version

        Returns:
            Liste betroffener Plugins
        """
        affected = []
        dependents = self._graph.get_all_dependents(name)

        for dependent in dependents:
            node = self._graph.get_node(dependent)
            if node and name in node.dependencies:
                requirement = node.dependencies[name]
                result = self._checker.check(requirement, new_version)
                if not result.compatible:
                    affected.append(dependent)

        return affected

    def clear(self) -> None:
        """Löscht alle registrierten Plugins."""
        self._graph = DependencyGraph()
        self._available.clear()

    def get_statistics(self) -> dict[str, Any]:
        """Gibt Statistiken zurück."""
        return {
            "total_plugins": self._graph.node_count,
            "total_dependencies": self._graph.edge_count,
            "root_plugins": len(self._graph.get_root_nodes()),
            "leaf_plugins": len(self._graph.get_leaf_nodes()),
            "has_cycles": self._graph.has_cycle(),
            "missing_dependencies": sum(
                len(deps) for deps in self._graph.get_missing_dependencies().values()
            ),
        }