from __future__ import annotations from datetime import datetime from structlog import BoundLogger from typing import TYPE_CHECKING, Iterable, Mapping, MutableMapping, Type from src.api_client.atlas import VendorAPIClient from src.api_client.errors import AtlasError from src.api_utils.errors import BadRequest from src.data_info import PlaylistInfo, TrackInfo from src.enums import FieldCopyMode, ServiceType from src.playlist_sync.models import PlaylistSynchronizationLog from src.playlist_sync.services import PlaylistSyncService from src.service_account.services import ServiceAccountService from src.synchronizer.dsp import AppleMusicSynchronizer, BaseSynchronizer, DeezerSynchronizer, SpotifySynchronizer, \ YoutubeSynchronizer from src.synchronizer.errors import SynchronizerError, VendorSpecificSynchronizerError from src.synchronizer.utils import get_playlist_id if TYPE_CHECKING: from src.playlist_sync.models import PlaylistSynchronization from src.service_account.models import ServiceAccount __all__ = ["SyncManager"] class SyncManager: # TODO: For now it supports only spotify as a source of playlist synchronization # If we want to sync playlist from any other DSP we should implement a way to specify account that # should be used to get data service_type_dsp_map: Mapping[ServiceType, Type[BaseSynchronizer]] = { ServiceType.spotify: SpotifySynchronizer, ServiceType.youtube: YoutubeSynchronizer, ServiceType.deezer: DeezerSynchronizer, ServiceType.apple_music: AppleMusicSynchronizer, } def __init__(self, logger: BoundLogger, triggered_manually: bool = True): self._logger = logger self._vendor_api = VendorAPIClient.get_client(logger=self._logger) self._use_vendor_api_cache = not triggered_manually self.__synchronizers: MutableMapping[ServiceType, BaseSynchronizer] = {} def _get_playlist_info(self, playlist_id: str) -> PlaylistInfo: try: return self._vendor_api.get_spotify_playlist_info(playlist_id, use_cache=self._use_vendor_api_cache) except AtlasError as e: raise SynchronizerError(f"Can't get info for playlist '{playlist_id}'") from e def get_playlist_info(self, playlist_id: str) -> PlaylistInfo: return self._get_playlist_info(get_playlist_id(ServiceType.spotify, playlist_id)) def _get_source_playlist_tracks(self, playlist_id: str) -> Iterable[TrackInfo]: try: return self._vendor_api.get_spotify_playlist_tracks(playlist_id, use_cache=self._use_vendor_api_cache) except AtlasError: raise SynchronizerError(f"Can't get tracks from playlist '{playlist_id}'") def _get_dsp_synchronizer(self, service_account: "ServiceAccount") -> BaseSynchronizer: service_type: ServiceType | None = service_account.service_type if service_type is None: raise SynchronizerError(f"No service type for service account with id {service_account.id}") if service_type not in self.__synchronizers: synchronizer_cls = self.__class__.service_type_dsp_map.get(service_type) if synchronizer_cls is None: raise SynchronizerError(f"Unsupported DSP '{service_account.service_type}'") self.__synchronizers[service_type] = synchronizer_cls(service_account, logger=self._logger) return self.__synchronizers[service_type] def check_playlists(self, from_playlist_id: str, to_playlist_id: str, to_service_account_id: str) -> PlaylistInfo: from_playlist_id = get_playlist_id(ServiceType.spotify, from_playlist_id) playlist_info = self.get_playlist_info(from_playlist_id) if playlist_info is None: raise BadRequest(f"Can't get info for playlist '{from_playlist_id}'") if to_service_account_id and to_playlist_id: service_account = ServiceAccountService().select_by_pk(to_service_account_id) synchronizer = self._get_dsp_synchronizer(service_account) to_playlist_id = get_playlist_id(service_account.service_type, to_playlist_id) if not synchronizer.check_target_playlist(service_account, to_playlist_id): raise BadRequest(f"Can't access target playlist '{to_playlist_id}'") return playlist_info @staticmethod def _filter_duplicate_tracks(tracks: Iterable[TrackInfo]) -> Iterable[TrackInfo]: seen = set() for track in tracks: if track in seen: continue seen.add(track) yield track def synchronize( self, service: PlaylistSyncService, playlist_sync: "PlaylistSynchronization", triggered_manually: bool = False ) -> PlaylistSynchronizationLog: try: if not playlist_sync.service_account: raise SynchronizerError(f"No service account for playlist sync {playlist_sync.id}") if playlist_sync.service_account.service_type is None: raise SynchronizerError( f"No service type for service account with id {playlist_sync.service_account.id}" ) # supports only spotify for now from_playlist_id = get_playlist_id(ServiceType.spotify, playlist_sync.from_playlist_id) to_playlist_id = get_playlist_id(playlist_sync.service_account.service_type, playlist_sync.to_playlist_id) synchronizer = self._get_dsp_synchronizer(playlist_sync.service_account) source_playlist_tracks = self._get_source_playlist_tracks(from_playlist_id) source_playlist_tracks = list(self._filter_duplicate_tracks(source_playlist_tracks)) playlist_info: PlaylistInfo | None = None info: MutableMapping[str, str | None] = {} if playlist_sync.title_copy_mode == FieldCopyMode.copy_source: if playlist_info is None: playlist_info = self._get_playlist_info(from_playlist_id) info["title"] = playlist_info.title elif playlist_sync.title_copy_mode == FieldCopyMode.use_setting: info["title"] = playlist_sync.title if playlist_sync.description_copy_mode == FieldCopyMode.copy_source: if playlist_info is None: playlist_info = self._get_playlist_info(from_playlist_id) info["description"] = playlist_info.description elif playlist_sync.description_copy_mode == FieldCopyMode.use_setting: info["description"] = playlist_sync.description result = synchronizer.synchronize_tracks(to_playlist_id, source_playlist_tracks) to_playlist_info = synchronizer.get_playlist_info(to_playlist_id) service.update_by_pk( playlist_sync.id, to_playlist_title=to_playlist_info.title, to_playlist_image=to_playlist_info.image_url, updated_at=datetime.utcnow(), ) is_info_updated = synchronizer.synchronize_info(to_playlist_id, info, old_info=to_playlist_info) return PlaylistSynchronizationLog( playlist_synchronization_id=playlist_sync.id, made_change=result.has_changes or is_info_updated, added_tracks=result.inserted_track_count, deleted_tracks=result.deleted_track_count, source_tracks=result.source_track_count, target_track_count=result.target_track_count, triggered_manually=triggered_manually, ) except Exception as e: self._logger.exception(f"An error occurred during the synchronization {e}") if isinstance(e, VendorSpecificSynchronizerError): error_type = "VendorSpecific" error_text = str(e) elif isinstance(e, SynchronizerError): error_type = "Synchronization" error_text = str(e) else: error_type = "Internal" error_text = "Something went wrong during synchronization" return PlaylistSynchronizationLog( playlist_synchronization_id=playlist_sync.id, error_type=error_type, error_text=error_text, triggered_manually=triggered_manually, )