ViewIT/addon/core/plugin_manager.py

#!/usr/bin/env python3
"""Plugin-Erkennung und -Verwaltung fuer ViewIT.

Dieses Modul laedt dynamisch alle Plugins aus dem `plugins/` Verzeichnis,
instanziiert sie und cached die Instanzen im RAM.
"""

from __future__ import annotations

import importlib.util
import inspect
import sys
from pathlib import Path
from types import ModuleType

try:  # pragma: no cover - Kodi runtime
    import xbmc  # type: ignore[import-not-found]
except ImportError:  # pragma: no cover
    xbmc = None

from plugin_interface import BasisPlugin

PLUGIN_DIR = Path(__file__).resolve().parent.parent / "plugins"
_PLUGIN_CACHE: dict[str, BasisPlugin] | None = None


def _log(message: str, level: int = 1) -> None:
    if xbmc is not None:
        xbmc.log(f"[ViewIt] {message}", level)


def import_plugin_module(path: Path) -> ModuleType:
    """Importiert eine einzelne Plugin-Datei als Python-Modul."""
    spec = importlib.util.spec_from_file_location(path.stem, path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Modul-Spezifikation fuer {path.name} fehlt.")
    module = importlib.util.module_from_spec(spec)
    sys.modules[spec.name] = module
    try:
        spec.loader.exec_module(module)
    except Exception:
        sys.modules.pop(spec.name, None)
        raise
    return module


def discover_plugins() -> dict[str, BasisPlugin]:
    """Laedt alle Plugins aus `plugins/*.py` und cached Instanzen im RAM."""
    global _PLUGIN_CACHE
    if _PLUGIN_CACHE is not None:
        return _PLUGIN_CACHE

    plugins: dict[str, BasisPlugin] = {}
    if not PLUGIN_DIR.exists():
        _PLUGIN_CACHE = plugins
        return plugins

    for file_path in sorted(PLUGIN_DIR.glob("*.py")):
        if file_path.name.startswith("_"):
            continue
        try:
            module = import_plugin_module(file_path)
        except Exception as exc:
            _log(f"Plugin-Datei {file_path.name} konnte nicht geladen werden: {exc}", 2)
            continue

        preferred = getattr(module, "Plugin", None)
        if inspect.isclass(preferred) and issubclass(preferred, BasisPlugin) and preferred is not BasisPlugin:
            plugin_classes = [preferred]
        else:
            plugin_classes = [
                obj
                for obj in module.__dict__.values()
                if inspect.isclass(obj) and issubclass(obj, BasisPlugin) and obj is not BasisPlugin
            ]
            plugin_classes.sort(key=lambda cls: cls.__name__.casefold())

        for cls in plugin_classes:
            try:
                instance = cls()
            except Exception as exc:
                _log(f"Plugin {cls.__name__} konnte nicht geladen werden: {exc}", 2)
                continue
            if getattr(instance, "is_available", True) is False:
                reason = getattr(instance, "unavailable_reason", "Nicht verfuegbar.")
                _log(f"Plugin {cls.__name__} deaktiviert: {reason}", 2)
                continue
            plugin_name = str(getattr(instance, "name", "") or "").strip()
            if not plugin_name:
                _log(
                    f"Plugin {cls.__name__} wurde ohne Name registriert und wird uebersprungen.",
                    2,
                )
                continue
            if plugin_name in plugins:
                _log(
                    f"Plugin-Name doppelt ({plugin_name}), {cls.__name__} wird uebersprungen.",
                    2,
                )
                continue
            plugins[plugin_name] = instance

    plugins = dict(sorted(plugins.items(), key=lambda item: item[0].casefold()))
    _PLUGIN_CACHE = plugins
    return plugins


def plugin_has_capability(plugin: BasisPlugin, capability: str) -> bool:
    """Prueft ob ein Plugin eine bestimmte Faehigkeit hat."""
    getter = getattr(plugin, "capabilities", None)
    if callable(getter):
        try:
            capabilities = getter()
        except Exception:
            capabilities = set()
        try:
            return capability in set(capabilities or [])
        except Exception:
            return False
    # Backwards compatibility: Popular via POPULAR_GENRE_LABEL constant.
    if capability == "popular_series":
        return _popular_genre_label(plugin) is not None
    return False


def _popular_genre_label(plugin: BasisPlugin) -> str | None:
    label = getattr(plugin, "POPULAR_GENRE_LABEL", None)
    if isinstance(label, str) and label.strip():
        return label.strip()
    return None


def popular_genre_label(plugin: BasisPlugin) -> str | None:
    """Gibt das POPULAR_GENRE_LABEL des Plugins zurueck, falls vorhanden."""
    return _popular_genre_label(plugin)


def plugins_with_popular() -> list[tuple[str, BasisPlugin, str]]:
    """Liefert alle Plugins die 'popular_series' unterstuetzen."""
    results: list[tuple[str, BasisPlugin, str]] = []
    for plugin_name, plugin in discover_plugins().items():
        if not plugin_has_capability(plugin, "popular_series"):
            continue
        label = _popular_genre_label(plugin) or ""
        results.append((plugin_name, plugin, label))
    return results


def series_url_params(plugin: BasisPlugin, title: str) -> dict[str, str]:
    """Liefert series_url Parameter fuer Kodi-Navigation, falls vom Plugin bereitgestellt."""
    getter = getattr(plugin, "series_url_for_title", None)
    if not callable(getter):
        return {}
    try:
        series_url = str(getter(title) or "").strip()
    except Exception:
        return {}
    return {"series_url": series_url} if series_url else {}