"""Provider registry. Concrete provider implementations live in the `Provider/` package. This module is the single source of truth for provider discovery. """ from __future__ import annotations from typing import Any, Dict, Optional, Type import sys from SYS.logger import log from ProviderCore.base import Provider, SearchProvider, FileProvider, SearchResult from Provider.alldebrid import AllDebrid from Provider.bandcamp import Bandcamp from Provider.libgen import Libgen from Provider.matrix import Matrix from Provider.openlibrary import OpenLibrary from Provider.soulseek import Soulseek, download_soulseek_file from Provider.telegram import Telegram from Provider.youtube import YouTube from Provider.fileio import FileIO from Provider.zeroxzero import ZeroXZero _PROVIDERS: Dict[str, Type[Provider]] = { # Search-capable providers "alldebrid": AllDebrid, "libgen": Libgen, "openlibrary": OpenLibrary, "soulseek": Soulseek, "bandcamp": Bandcamp, "youtube": YouTube, "telegram": Telegram, # Upload-capable providers "0x0": ZeroXZero, "file.io": FileIO, "matrix": Matrix, } def _supports_search(provider: Provider) -> bool: return provider.__class__.search is not Provider.search def _supports_upload(provider: Provider) -> bool: return provider.__class__.upload is not Provider.upload def get_provider(name: str, config: Optional[Dict[str, Any]] = None) -> Optional[Provider]: """Get a provider by name (unified registry).""" provider_class = _PROVIDERS.get((name or "").lower()) if provider_class is None: log(f"[provider] Unknown provider: {name}", file=sys.stderr) return None try: provider = provider_class(config) if not provider.validate(): log(f"[provider] Provider '{name}' is not available", file=sys.stderr) return None return provider except Exception as exc: log(f"[provider] Error initializing '{name}': {exc}", file=sys.stderr) return None def list_providers(config: Optional[Dict[str, Any]] = None) -> Dict[str, bool]: """List all providers and their availability.""" availability: Dict[str, bool] = {} for name, provider_class in _PROVIDERS.items(): try: provider = provider_class(config) availability[name] = provider.validate() except Exception: availability[name] = False return availability def get_search_provider(name: str, config: Optional[Dict[str, Any]] = None) -> Optional[SearchProvider]: """Get a search-capable provider by name (compat API).""" provider = get_provider(name, config) if provider is None: return None if not _supports_search(provider): log(f"[provider] Provider '{name}' does not support search", file=sys.stderr) return None return provider # type: ignore[return-value] def list_search_providers(config: Optional[Dict[str, Any]] = None) -> Dict[str, bool]: """List all search providers and their availability.""" availability: Dict[str, bool] = {} for name, provider_class in _PROVIDERS.items(): try: provider = provider_class(config) availability[name] = bool(provider.validate() and _supports_search(provider)) except Exception: availability[name] = False return availability def get_file_provider(name: str, config: Optional[Dict[str, Any]] = None) -> Optional[FileProvider]: """Get an upload-capable provider by name (compat API).""" provider = get_provider(name, config) if provider is None: return None if not _supports_upload(provider): log(f"[provider] Provider '{name}' does not support upload", file=sys.stderr) return None return provider # type: ignore[return-value] def list_file_providers(config: Optional[Dict[str, Any]] = None) -> Dict[str, bool]: """List all file providers and their availability.""" availability: Dict[str, bool] = {} for name, provider_class in _PROVIDERS.items(): try: provider = provider_class(config) availability[name] = bool(provider.validate() and _supports_upload(provider)) except Exception: availability[name] = False return availability __all__ = [ "SearchResult", "Provider", "SearchProvider", "FileProvider", "get_provider", "list_providers", "get_search_provider", "list_search_providers", "get_file_provider", "list_file_providers", "download_soulseek_file", ]