146 lines
4.5 KiB
Python
146 lines
4.5 KiB
Python
"""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",
|
|
]
|