ytsm/app/YtManagerApp/management/video_provider_manager.py

132 lines
4.8 KiB
Python
Raw Permalink Normal View History

2020-10-18 16:36:26 +00:00
import json
2020-04-10 21:30:24 +00:00
import logging
2020-10-18 16:36:26 +00:00
from typing import List, Dict, Union, Iterable, Optional
from django.db import transaction
2020-04-10 21:30:24 +00:00
from YtManagerApp.models import VideoProviderConfig, Video, Subscription
2020-10-18 16:36:26 +00:00
from YtManagerApp.providers.video_provider import VideoProvider, InvalidURLError, VideoProviderState
2020-04-10 21:30:24 +00:00
log = logging.getLogger("VideoProviderManager")
class VideoProviderManager(object):
def __init__(self, registered_providers: List[VideoProvider]):
self._registered_providers: Dict[str, VideoProvider] = {}
self._pending_configs: Dict[str, VideoProviderConfig] = {}
2020-10-18 16:36:26 +00:00
2020-04-10 21:30:24 +00:00
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
2020-04-22 21:47:27 +00:00
if provider.id in self._registered_providers:
log.error(f"Duplicate video provider {provider.id}")
2020-04-10 21:30:24 +00:00
return
# register
2020-04-22 21:47:27 +00:00
self._registered_providers[provider.id] = provider
log.info(f"Registered video provider {provider.id}")
2020-04-10 21:30:24 +00:00
# load configuration (if any)
2020-04-22 21:47:27 +00:00
if provider.id in self._pending_configs:
self._configure(provider, self._pending_configs[provider.id])
del self._pending_configs[provider.id]
2020-04-10 21:30:24 +00:00
2020-10-18 16:36:26 +00:00
def configure_provider(self, provider_id: str, config: Optional[Dict[str, any]]):
provider = self.get(provider_id)
if config is not None:
provider.configure(config)
with transaction.atomic():
cfg, _ = VideoProviderConfig.objects.get_or_create(provider_id=provider_id)
cfg.settings = json.dumps(config)
cfg.save()
provider.state = VideoProviderState.OK
else:
provider.unconfigure()
VideoProviderConfig.objects.filter(provider_id=provider_id).delete()
provider.state = VideoProviderState.NOT_CONFIGURED
def get_provider_config(self, provider_id: str):
cfg = VideoProviderConfig.objects.filter(provider_id=provider_id).first()
if cfg is not None:
return json.loads(cfg.settings)
return None
2020-04-10 21:30:24 +00:00
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)
2020-10-18 16:36:26 +00:00
provider.state = VideoProviderState.OK
2020-04-22 21:47:27 +00:00
log.info(f"Configured video provider {provider.id}")
2020-04-10 21:30:24 +00:00
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:
"""
2020-10-18 16:36:26 +00:00
for provider in self._registered_providers.values():
if provider.state == VideoProviderState.OK:
try:
provider.validate_subscription_url(url)
return
except InvalidURLError:
pass
2020-04-10 21:30:24 +00:00
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:
"""
2020-10-18 16:36:26 +00:00
for provider in self._registered_providers.values():
if provider.state == VideoProviderState.OK:
try:
provider.validate_subscription_url(url)
# Found the right provider
return provider.fetch_subscription(url)
except InvalidURLError:
pass
2020-04-10 21:30:24 +00:00
raise InvalidURLError("The given URL is not valid for any of the supported sites!")
2020-04-22 21:47:27 +00:00
2020-10-18 16:36:26 +00:00
def get_available_providers(self) -> Iterable[VideoProvider]:
return self._registered_providers.values()