anicli/core/loader.py - master - dig/anicli_ru

anicli/core/loader.py
128 строк4 KB
An0nX
refactor: modernize type hints, improve aniskip merging and history fallback
09 янв 2026, 00:43
09 янв 2026, 00:43925b7f8
import ast
import importlib.util
import sys
from pathlib import Path

from anicli.config import config
from anicli.core.base import BaseSource
from anicli.core.network import NetworkClient


class ModuleLoader:
    """
    Dynamic module loader for the providers directory.
    Uses AST to verify inheritance before import to ensure safety and strict typing.
    Handles per-source configuration and proxy injection.
    """

    def __init__(self, network: NetworkClient) -> None:
        """
        Initialize the loader.

        Args:
            network: The default global network client.
        """
        self.network = network
        self.providers: dict[str, BaseSource] = {}
        # Path relative to this file: anicli/core/loader.py -> anicli/providers
        self.providers_dir = Path(__file__).parent.parent / "providers"

    def load_all(self) -> None:
        """
        Scans the providers directory and loads all valid provider modules.
        """
        if not self.providers_dir.exists():
            return

        for file_path in self.providers_dir.glob("*.py"):
            if file_path.name == "__init__.py":
                continue

            self._load_module(file_path)

    def _load_module(self, path: Path) -> None:
        """
        Loads a single module given its path.
        Instantiates the provider with specific network settings if configured.

        Args:
            path: Path to the python file.
        """
        try:
            # 1. Read source code
            with path.open(encoding="utf-8") as f:
                source_code = f.read()

            # 2. AST Analysis: look for classes inheriting BaseSource
            tree = ast.parse(source_code)
            target_class_name = None

            for node in ast.walk(tree):
                if isinstance(node, ast.ClassDef):
                    for base in node.bases:
                        if isinstance(base, ast.Name) and base.id == "BaseSource":
                            target_class_name = node.name
                            break

            if not target_class_name:
                return  # No provider in file

            # 3. Import module
            module_name = f"anicli.providers.{path.stem}"
            spec = importlib.util.spec_from_file_location(module_name, path)
            if not spec or not spec.loader:
                return

            module = importlib.util.module_from_spec(spec)
            sys.modules[module_name] = module
            spec.loader.exec_module(module)

            # 4. Instantiation with Proxy Logic
            cls: type[BaseSource] = getattr(module, target_class_name)

            # Get config for specific source from TOML
            source_cfg = config.raw_sources.get(cls.source_id, {})

            # Check for specific proxy in source config
            # Priority: Source Config > Global Config
            source_proxy = source_cfg.get("proxy")

            client: NetworkClient
            if source_proxy:
                # Create a specialized client for this source
                # Clone global settings but override proxy
                new_settings = config.settings.network.model_copy(
                    update={"proxy": source_proxy}
                )
                client = NetworkClient(new_settings)
            else:
                # Use the global shared client (reuses connections/cookies)
                client = self.network

            instance = cls(client, source_cfg)
            self.providers[cls.source_id] = instance

        except Exception as e:
            # Log error to stderr to avoid breaking UI flow
            print(f"[ERROR] Failed to load module {path.name}: {e}", file=sys.stderr)

    def get(self, source_id: str) -> BaseSource | None:
        """
        Retrieve a loaded provider by its ID.

        Args:
            source_id: The unique identifier of the provider.

        Returns:
            Optional[BaseSource]: The provider instance or None if not found.
        """
        return self.providers.get(source_id)

    def get_all(self) -> list[BaseSource]:
        """
        Retrieve all loaded providers.

        Returns:
            List[BaseSource]: A list of all provider instances.
        """
        return list(self.providers.values())