from __future__ import annotations import re from functools import partial from typing import Any, Iterable, Mapping, Sequence, Tuple from src.api_client.dsp import YoutubeApiClient from src.data_info import PlaylistInfo, TrackInfo, YoutubeVideoInfo from src.enums import ServiceType from src.playlist_sync.services import FlaggedChannelService from ..errors import SynchronizerError, YoutubeSynchronizerError from ..utils import compare_sequences from .base import BaseSynchronizer __all__ = ["YoutubeSynchronizer"] class YoutubeSynchronizer(BaseSynchronizer[YoutubeApiClient]): service_type = ServiceType.youtube vendor_specific_sync_error_cls = YoutubeSynchronizerError _extra_patterns: Tuple[re.Pattern, ...] = ( re.compile(r"\((?P.*)\)"), # parenthesis re.compile(r"-(?P.*)$"), # dash ) _official_suffixes = ( "official", "officiel", "offiziell", "ufficiale", "offisiell", "oficial", "officiell", "channel", ) _blacklist_partial = ( "remaster", "album", "full length", ) _blacklist_exact = ( "single version", "radio edit", "single edit", "radio version", "radio mix", "single mix", "stereo version", "extended version", "original mix", "single remix", ) _blacklist_title = ("karaoke", "behind the scenes", "full album") def __init__(self, *args: Any, **kwargs: Any): super().__init__(*args, **kwargs) self._flagged_channel_service = FlaggedChannelService() self._target_item_ids: list[str] = [] @staticmethod def _is_video_title_contains_track_info(video: YoutubeVideoInfo, *, track_info: TrackInfo) -> bool: return track_info.title in video.title and all(artist in video.title for artist in track_info.artists) @classmethod def _is_video_title_not_contains_blacklisted_words(cls, video: YoutubeVideoInfo, *, track_info: TrackInfo) -> bool: for word in cls._blacklist_title: if ( word in video.title and word not in track_info.title and any(word not in artist for artist in track_info.artists) ): return False return True def _is_video_not_from_blacklisted_channel(self, video: YoutubeVideoInfo) -> bool: return ( video.channel_id not in self._flagged_channel_service.blacklisted_channels and video.channel_title not in self._flagged_channel_service.blacklisted_channels ) def _is_video_from_whitelisted_channel(self, video: YoutubeVideoInfo) -> bool: return ( video.channel_id in self._flagged_channel_service.whitelisted_channels or video.channel_title in self._flagged_channel_service.whitelisted_channels ) @classmethod def _is_video_from_official_channel(cls, video: YoutubeVideoInfo, *, track_info: TrackInfo) -> bool: return any(f"{track_info.artists[0]} {suffix}" in video.channel_title for suffix in cls._official_suffixes) def _match_video(self, videos: Iterable[YoutubeVideoInfo], track_info: TrackInfo) -> YoutubeVideoInfo | None: # We want the video name to really contain the track and artist name videos = filter(partial(self._is_video_title_contains_track_info, track_info=track_info), videos) # We exclude videos that have certain blacklisted words in the title (when the word is not in the track title) videos = filter(partial(self._is_video_title_not_contains_blacklisted_words, track_info=track_info), videos) # Videos are checked against a list of blacklisted channels, and not used if from one of those channels. videos = list(filter(self._is_video_not_from_blacklisted_channel, videos)) # We prefer videos by VEVO users. vevo_videos = list(filter(lambda v: "vevo" in v.channel_title, videos)) if vevo_videos: return vevo_videos[0] # If the video is from a whitelisted channel, we like that. whitelisted_channel_videos = list(filter(self._is_video_not_from_blacklisted_channel, videos)) if whitelisted_channel_videos: return whitelisted_channel_videos[0] # We prefer users with artist name official_channel_videos = list( filter(partial(self._is_video_from_official_channel, track_info=track_info), videos) ) if official_channel_videos: return official_channel_videos[0] # If all else fails, just return first video (after the filtering above) if videos: return videos[0] return None def _search_videos(self, search_queries: Sequence[str]) -> Iterable[YoutubeVideoInfo]: seen_video_ids = set() for query in search_queries: videos = self.api_client.search(query) for video in videos: if video.id in seen_video_ids: continue seen_video_ids.add(video.id) yield video def _get_track_from_dsp(self, track_info: TrackInfo) -> str: self._logger.debug( f"Looking for track '{track_info.title}' by '{', '.join(track_info.artists)}' " f"with ISRC {track_info.isrc}" ) search_queries = self._get_search_queries(track_info) videos = self._search_videos(search_queries) matched_video = self._match_video(videos, track_info) if matched_video is None: raise SynchronizerError( f"Can't find video for track '{track_info.title}' by '{', '.join(track_info.artists)}' " f"with ISRC {track_info.isrc}" ) return matched_video.id def _filter_extra_info(self, extra: str) -> str: for blacklisted in self.__class__._blacklist_partial: if blacklisted in extra: return "" for blacklisted in self.__class__._blacklist_exact: extra.replace(blacklisted, "") return extra.strip() def _parse_track_name(self, name: str) -> Tuple[str, str]: extra = "" for pattern in self.__class__._extra_patterns: match = pattern.match(name) if match: name = pattern.sub("", name).strip() extra = match.group("extra").strip() break if extra: extra = self._filter_extra_info(extra) return name, extra def _get_search_queries(self, track_info: TrackInfo) -> Sequence[str]: track_name, extra = self._parse_track_name(track_info.title) raw_search_query = f"{track_info.artists[0]} - {track_name}" quoted_search_query = f'"{raw_search_query}"' if extra: raw_search_query = f"{raw_search_query} {extra}" quoted_search_query = f"{quoted_search_query} {extra}" return quoted_search_query, raw_search_query def _insert_videos(self, playlist_id: str, video_ids: Iterable[str], position: int): for i, video_id in enumerate(video_ids): self.api_client.insert_video(playlist_id, video_id, position + i) def _delete_videos(self, playlist_item_ids: Iterable[str]): for playlist_item_id in playlist_item_ids: self.api_client.delete_playlist_item(playlist_item_id) def _get_tracks_from_dsp(self, tracks: Iterable[TrackInfo]) -> Mapping[TrackInfo, str]: mapping = {} for track in tracks: mapping[track] = self._get_track_from_dsp(track) return mapping def _get_target_track_ids(self, playlist_id: str) -> Iterable[str]: self._target_item_ids = [] for item in self.api_client.get_playlist(playlist_id): self._target_item_ids.append(item.playlist_item_id) yield item.id def _synchronize_tracks( self, playlist_id: str, source_track_ids: Sequence[str], target_track_ids: Sequence[str] ) -> tuple[int, int]: target_item_ids = self._target_item_ids inserted = 0 deleted = 0 for action, t1, t2, s1, s2 in compare_sequences(source_track_ids, target_track_ids): if action == "insert": self._insert_videos(playlist_id, source_track_ids[s1:s2], t1) inserted += s2 - s1 if action == "delete": self._delete_videos(target_item_ids[t1:t2]) deleted += t2 - t1 if action == "replace": self._insert_videos(playlist_id, source_track_ids[s1:s2], t1) self._delete_videos(target_item_ids[t1:t2]) inserted += s2 - s1 deleted += t2 - t1 return inserted, deleted def get_playlist_info(self, playlist_id: str) -> PlaylistInfo: return self.api_client.get_playlist_info(playlist_id) def _update_playlist_info(self, playlist_id: str, old_info: PlaylistInfo, new_info: Mapping[str, str | None]): snippet_info = {"title": old_info.title, "description": old_info.description} snippet_info.update(new_info) self.api_client.update_playlist_info(playlist_id, snippet_info)