from datetime import datetime, timedelta from typing import Iterable, List, Optional, Set, Tuple from apollo_notifications.constants import SONY_CATEGORY_ID, SPOTIFY_CATEGORY_ID from apollo_notifications.playlists.client import NotificationsPlaylistClient from apollo_notifications.utils import parse_gl_to_global, parse_global_to_gl from sqlalchemy import and_, func from sqlalchemy.orm import Query from apollo_main_db.apollo import StarredContent from apollo_main_db.push_notifications import UserDeviceToken, VendorEnum from apollo_main_db.spotify import SpotifyArtist, SpotifyPlaylist, SpotifyPlaylistStatistics, \ SpotifyPlaylistTrackListHistory2Reduced2, SpotifyTrack2, SpotifyTrackArtist, SpotifyWeeklyTopPlaylist class SpotifyPlaylistClient(NotificationsPlaylistClient): """Class to get Spotify playlist data.""" vendor = VendorEnum.SPOTIFY.name.lower() def __init__(self, config): super().__init__(config) self.top_playlist_threshold = config.TOP_PLAYLIST_THRESHOLD def encode_playlist_id_to_ascii(self, playlists: Iterable) -> List[str]: # FIXME: For some reason SpotifyPlaylistTrackListHistory2Reduced2.playlist_id has # ascii charset, so we should encode our string to ascii. encoded_top_playlists = [] for (playlist,) in playlists: try: encoded_top_playlists.append(playlist.encode("ascii")) except UnicodeEncodeError: continue return encoded_top_playlists def get_border_datetimes(self, target_date: str) -> Tuple[datetime, datetime]: target_min = datetime.strptime(target_date, "%Y-%m-%d") return target_min, target_min + timedelta(days=1) def get_top_playlists_by_date_query( self, market: str, target_date: str, exclude: Optional[List[str]] = None ) -> List[str]: """Get top playlist by date Args: market (str): Market. target_date (str): Date. exclude (list): list of playlist ids to exclude. Returns: List of playlist_id """ max_date_top_query = self.session.query(func.max(SpotifyWeeklyTopPlaylist.date)).filter( SpotifyWeeklyTopPlaylist.date <= target_date ) filters = [ func.lower(SpotifyWeeklyTopPlaylist.country) == market.lower(), SpotifyWeeklyTopPlaylist.rank <= self.top_playlist_threshold, SpotifyWeeklyTopPlaylist.date == max_date_top_query, SpotifyWeeklyTopPlaylist.playlist_id.isnot(None), ] if exclude: filters.append(SpotifyWeeklyTopPlaylist.playlist_id.notin_(exclude)) top_playlists = ( self.session.query( SpotifyWeeklyTopPlaylist.playlist_id, ) .filter(*filters) .group_by(SpotifyWeeklyTopPlaylist.playlist_id) ) return self.encode_playlist_id_to_ascii(top_playlists) def get_updated_playlists_by_date_query( self, market: str, target_date: str, exclude: Optional[List[str]] = None, filter_by_playlists: Optional[List[str]] = None, ) -> List[str]: """Get top playlist by date Args: market (str): Market. target_date (str): Date. exclude (list): list of playlist ids to exclude. filter_by_playlists (list): list of playlist ids to filter by. Returns: List of playlist_id """ min_datetime, max_datetime = self.get_border_datetimes(target_date) statistics_market = parse_global_to_gl(market) filters = [ SpotifyPlaylistStatistics.market == statistics_market, SpotifyPlaylistStatistics.tracklist_updated_at >= min_datetime, SpotifyPlaylistStatistics.tracklist_updated_at < max_datetime, ] if exclude: filters.append(SpotifyPlaylistStatistics.playlist_id.notin_(exclude)) if filter_by_playlists: filters.append(SpotifyPlaylistStatistics.playlist_id.in_(filter_by_playlists)) playlists = self.session.query( SpotifyPlaylistStatistics.playlist_id, ).filter(*filters) return self.encode_playlist_id_to_ascii(playlists) def get_added_tracks_query( self, last_date: str, previous_date: str, last_date_playlist_ids: List[str], previous_date_playlist_ids: List[str], ) -> Query: """Get added tracks to top playlists between two dates Args: last_date (str): Current date for track history. previous_date (str): Previous date for track history. last_date_playlist_ids (list of str): List of target playlist ids for the last date. previous_date_playlist_ids (list of str): List of target playlist ids for the previous date. Returns: List of tuple(playlist_uri, track_id,) """ end_date_tracks_query = ( self.session.query( SpotifyPlaylistTrackListHistory2Reduced2.playlist_id.label("playlist_id"), SpotifyTrack2.isrc.label("isrc"), ) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackListHistory2Reduced2.track_id) .filter( SpotifyPlaylistTrackListHistory2Reduced2.playlist_id.in_(previous_date_playlist_ids), SpotifyPlaylistTrackListHistory2Reduced2.date == previous_date, ) .subquery() ) start_date_tracks_query = ( self.session.query( SpotifyPlaylistTrackListHistory2Reduced2.playlist_id.label("playlist_id"), (SpotifyPlaylistTrackListHistory2Reduced2.playlist_index + 1).label("position"), SpotifyPlaylistTrackListHistory2Reduced2.track_id.label("track_id"), SpotifyTrack2.name.label("track_name"), SpotifyTrack2.isrc.label("isrc"), SpotifyArtist.name.label("artist_name"), SpotifyPlaylist.name.label("playlist_name"), ) .filter( SpotifyPlaylistTrackListHistory2Reduced2.playlist_id.in_(last_date_playlist_ids), SpotifyPlaylistTrackListHistory2Reduced2.date == last_date, SpotifyTrackArtist.order == 0, SpotifyPlaylist.buzz_category_id.in_((SONY_CATEGORY_ID, SPOTIFY_CATEGORY_ID)), ) .join(SpotifyTrackArtist, SpotifyTrackArtist.track_id == SpotifyPlaylistTrackListHistory2Reduced2.track_id) .join(SpotifyArtist, SpotifyArtist.id == SpotifyTrackArtist.artist_id) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackListHistory2Reduced2.track_id) .join(SpotifyPlaylist, SpotifyPlaylist.id == SpotifyPlaylistTrackListHistory2Reduced2.playlist_id) .subquery() ) return ( self.session.query(start_date_tracks_query) .outerjoin( end_date_tracks_query, and_( start_date_tracks_query.c.playlist_id == end_date_tracks_query.c.playlist_id, start_date_tracks_query.c.isrc == end_date_tracks_query.c.isrc, ), ) .filter( end_date_tracks_query.c.isrc.is_(None), ) ) def filter_by_starred_and_active(self, tracks_query: Query, user_id_list: List[str]) -> Query: """Filtering tracks query by starred tracks, and users, join with User devices, exclude already sent pushes. Args: tracks_query: Query with isrc, position, track_name, artist_name aliased columns. user_id_list: List of user_id to filter by. """ tracks_sub_query = tracks_query.subquery() return ( self.session.query( StarredContent.user_id, func.group_concat(UserDeviceToken.token.op("ORDER BY")(UserDeviceToken.created_at.desc())).label( "device_tokens" ), tracks_sub_query, ) .join(tracks_sub_query, func.upper(StarredContent.isrc) == func.upper(tracks_sub_query.c.isrc)) .join(UserDeviceToken, UserDeviceToken.user_id == StarredContent.user_id) .filter(UserDeviceToken.is_active.is_(True), UserDeviceToken.user_id.in_(user_id_list)) .group_by(StarredContent.user_id, tracks_sub_query.c.playlist_id, tracks_sub_query.c.isrc) ) def get_top_markets(self, filter_by_markets: Optional[Set[str]] = None) -> Set[str]: """Get playlists markets. Optionally filter by passed list. filter_by: List of markets to filter by. """ q = self.session.query(SpotifyWeeklyTopPlaylist.country).distinct(SpotifyWeeklyTopPlaylist.country) markets = set([pl.country.lower() if pl.country else None for pl in q]) if filter_by_markets: markets = markets & filter_by_markets return markets def get_updated_tracklist_markets( self, target_date: str, filter_by_playlists: Optional[List[str]] = None ) -> Set[str]: min_datetime, max_datetime = self.get_border_datetimes(target_date) filters = [ SpotifyPlaylistStatistics.tracklist_updated_at >= min_datetime, SpotifyPlaylistStatistics.tracklist_updated_at < max_datetime, ] if filter_by_playlists: filters.append(SpotifyPlaylistStatistics.playlist_id.in_(filter_by_playlists)) q = ( self.session.query(SpotifyPlaylistStatistics.market) .filter(*filters) .distinct(SpotifyPlaylistStatistics.market) ) return set([parse_gl_to_global(pl.market) for pl in q])