mirror of
https://github.com/chibicitiberiu/ytsm.git
synced 2024-02-24 05:43:31 +00:00
117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
import logging
|
|
from typing import List, Dict, Union, Iterable
|
|
|
|
from YtManagerApp.models import VideoProviderConfig, Video, Subscription
|
|
from YtManagerApp.providers.video_provider import VideoProvider, InvalidURLError
|
|
import json
|
|
from collections import namedtuple
|
|
|
|
log = logging.getLogger("VideoProviderManager")
|
|
|
|
VideoProviderInfo = namedtuple('VideoProviderInfo', ['id', 'name', 'is_configured', 'description'])
|
|
|
|
|
|
class VideoProviderManager(object):
|
|
def __init__(self, registered_providers: List[VideoProvider]):
|
|
self._registered_providers: Dict[str, VideoProvider] = {}
|
|
self._configured_providers: Dict[str, VideoProvider] = {}
|
|
self._pending_configs: Dict[str, VideoProviderConfig] = {}
|
|
for rp in registered_providers:
|
|
self.register_provider(rp)
|
|
self._load()
|
|
|
|
def register_provider(self, provider: VideoProvider) -> None:
|
|
"""
|
|
Registers a video provider
|
|
:param provider: Video provider
|
|
"""
|
|
# avoid duplicates
|
|
if provider.id in self._registered_providers:
|
|
log.error(f"Duplicate video provider {provider.id}")
|
|
return
|
|
|
|
# register
|
|
self._registered_providers[provider.id] = provider
|
|
log.info(f"Registered video provider {provider.id}")
|
|
|
|
# load configuration (if any)
|
|
if provider.id in self._pending_configs:
|
|
self._configure(provider, self._pending_configs[provider.id])
|
|
del self._pending_configs[provider.id]
|
|
|
|
def _load(self) -> None:
|
|
# Loads configuration from database
|
|
for config in VideoProviderConfig.objects.all():
|
|
provider = self._registered_providers.get(config.provider_id)
|
|
|
|
# provider not yet registered, keep it in the pending list
|
|
if provider is None:
|
|
self._pending_configs[config.provider_id] = config
|
|
log.warning(f"Provider {config.provider_id} not registered!")
|
|
continue
|
|
|
|
# configure
|
|
self._configure(provider, config)
|
|
|
|
def _configure(self, provider, config):
|
|
settings = json.loads(config.settings)
|
|
provider.configure(settings)
|
|
log.info(f"Configured video provider {provider.id}")
|
|
self._configured_providers[provider.id] = provider
|
|
|
|
def get(self, item: Union[str, Subscription, Video]):
|
|
"""
|
|
Gets provider for given item (subscription or video).
|
|
:param item: Provider ID, or subscription, or video
|
|
:return: Provider
|
|
"""
|
|
if isinstance(item, str):
|
|
return self._registered_providers[item]
|
|
elif isinstance(item, Video):
|
|
return self._registered_providers[item.subscription.provider_id]
|
|
elif isinstance(item, Subscription):
|
|
return self._registered_providers[item.provider_id]
|
|
return None
|
|
|
|
def validate_subscription_url(self, url: str):
|
|
"""
|
|
Validates given URL using all registered and configured provider.
|
|
:param url:
|
|
:return:
|
|
"""
|
|
for provider in self._configured_providers.values():
|
|
try:
|
|
provider.validate_subscription_url(url)
|
|
return
|
|
except InvalidURLError:
|
|
pass
|
|
|
|
raise InvalidURLError("The given URL is not valid for any of the supported sites!")
|
|
|
|
def fetch_subscription(self, url: str) -> Subscription:
|
|
"""
|
|
Validates given URL using all registered and configured provider.
|
|
:param url:
|
|
:return:
|
|
"""
|
|
for provider in self._configured_providers.values():
|
|
try:
|
|
provider.validate_subscription_url(url)
|
|
# Found the right provider
|
|
return provider.fetch_subscription(url)
|
|
except InvalidURLError:
|
|
pass
|
|
|
|
raise InvalidURLError("The given URL is not valid for any of the supported sites!")
|
|
|
|
def get_available_providers(self) -> Iterable[VideoProviderInfo]:
|
|
"""
|
|
Gets a list of available providers and some basic information about them.
|
|
:return: List of dictionary entries
|
|
"""
|
|
for key, provider in self._registered_providers.items():
|
|
yield VideoProviderInfo(id=key,
|
|
name=provider.name,
|
|
description=provider.description,
|
|
is_configured=(key in self._configured_providers))
|