from datetime import datetime, timedelta from typing import Iterable, List, Optional, Set, Tuple from apollo_notifications.constants import APOLLO_GLOBAL_MARKET, APPLE_CATEGORY_ID, GLOBAL_MARKET, SONY_CATEGORY_ID, \ US_MARKET 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.apollo.models import BuzzUser from apollo_main_db.apple.models import AppleMusicPlaylistStatistics, AppleMusicSong, ApplePlaylist, \ ApplePlaylistTracklistHistoryReduced2, AppleWeeklyTopPlaylist from apollo_main_db.push_notifications import UserDeviceToken, VendorEnum class ApplePlaylistClient(NotificationsPlaylistClient): """Class to get Apple Music playlist data.""" vendor = VendorEnum.APPLE.name.lower() def __init__(self, config): super().__init__(config) self.top_playlist_threshold = config.TOP_PLAYLIST_THRESHOLD 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_starred_isrc_for_active_users(self, user_id_list: List[str]) -> List[str]: """Ger isrc list of tracks starred by active users from user_id_list.""" q = ( self.session.query(StarredContent.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.isrc) ) return [r[0] for r in q] def get_top_playlists_by_date_query( self, market: str, target_date: str, exclude: Optional[List[str]] = None ) -> Iterable: """Get top playlist by date Args: market (str): Market. target_date (str): Date "%Y-%m-%d" format. exclude (list): list of playlist ids to exclude. Returns: List of playlist_id """ filters = [ AppleWeeklyTopPlaylist.country_code == market, AppleWeeklyTopPlaylist.rank <= self.top_playlist_threshold, AppleWeeklyTopPlaylist.date.in_( (target_date, datetime.strptime(target_date, "%Y-%m-%d") - timedelta(days=1)) ), ] if exclude: filters.append(AppleWeeklyTopPlaylist.playlist_id.notin_(exclude)) top_playlists_query = ( self.session.query( AppleWeeklyTopPlaylist.playlist_id, ) .filter(*filters) .group_by(AppleWeeklyTopPlaylist.playlist_id) ) return [playlist_id for (playlist_id,) in top_playlists_query] 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 = [ AppleMusicPlaylistStatistics.market == statistics_market, AppleMusicPlaylistStatistics.tracklist_updated_at >= min_datetime, AppleMusicPlaylistStatistics.tracklist_updated_at < max_datetime, ] if exclude: filters.append(AppleMusicPlaylistStatistics.playlist_id.notin_(exclude)) if filter_by_playlists: filters.append(AppleMusicPlaylistStatistics.playlist_id.in_(filter_by_playlists)) playlists = self.session.query( AppleMusicPlaylistStatistics.playlist_id, ).filter(*filters) return [playlist_id for (playlist_id,) in 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], market: str, user_id_list: List[str], ) -> Query: """Get added tracks to top playlists between two dates. Args: last_date (str): Current date for track history "%Y-%m-%d" format. previous_date (str): Previous date for track history "%Y-%m-%d" format. 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. market (str): market to get data for. user_id_list: List of user_id to filter by. Returns: List of tuple(playlist_id, song_id,) """ starred_isrc_list = self.get_starred_isrc_for_active_users(user_id_list) end_date_tracks_query = ( self.session.query( ApplePlaylistTracklistHistoryReduced2.playlist_id.label("playlist_id"), ApplePlaylistTracklistHistoryReduced2.storefront.label("storefront"), AppleMusicSong.isrc.label("isrc"), ) .join( AppleMusicSong, and_( AppleMusicSong.id == ApplePlaylistTracklistHistoryReduced2.song_id, AppleMusicSong.storefront == market, AppleMusicSong.isrc.in_(starred_isrc_list), ), ) .filter( ApplePlaylistTracklistHistoryReduced2.date == previous_date, ApplePlaylistTracklistHistoryReduced2.storefront == market, ApplePlaylistTracklistHistoryReduced2.playlist_id.in_(previous_date_playlist_ids), ) .subquery() ) start_date_tracks_query = ( self.session.query( ApplePlaylistTracklistHistoryReduced2.playlist_id.label("playlist_id"), ApplePlaylistTracklistHistoryReduced2.storefront.label("storefront"), (ApplePlaylistTracklistHistoryReduced2.position + 1).label("position"), ApplePlaylistTracklistHistoryReduced2.song_id.label("track_id"), AppleMusicSong.name.label("track_name"), AppleMusicSong.isrc.label("isrc"), AppleMusicSong.artist_name.label("artist_name"), ApplePlaylist.name.label("playlist_name"), ApplePlaylist.artwork.label("playlist_image_url"), ApplePlaylist.curator_id.label("curator_id"), ) .join( AppleMusicSong, and_( AppleMusicSong.id == ApplePlaylistTracklistHistoryReduced2.song_id, AppleMusicSong.storefront == market, AppleMusicSong.isrc.in_(starred_isrc_list), ), ) .join(ApplePlaylist, ApplePlaylist.id == ApplePlaylistTracklistHistoryReduced2.playlist_id) .filter( ApplePlaylistTracklistHistoryReduced2.playlist_id.in_(last_date_playlist_ids), ApplePlaylistTracklistHistoryReduced2.storefront == market, ApplePlaylistTracklistHistoryReduced2.date == last_date, ) .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 get_allowed_categories_subquery(self): return ( self.session.query(BuzzUser.user_name) .filter( # filtering by category is required AG-6364 BuzzUser.category_id.in_([APPLE_CATEGORY_ID, SONY_CATEGORY_ID]), BuzzUser.user_name.isnot(None), ) .subquery() ) def filter_added_tracks_query(self, tracks_query: Query, user_id_list: List[str]) -> Query: """Filtering tracks query by starred tracks, active users and available playlists users categories. Args: tracks_query: Query with track_id, position, track_name, artist_name aliased columns. user_id_list: List of user_id to filter by. """ tracks_sub_query = tracks_query.subquery() categories_sub_query = self.get_allowed_categories_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(UserDeviceToken, UserDeviceToken.user_id == StarredContent.user_id) .join(tracks_sub_query, func.upper(StarredContent.isrc) == func.upper(tracks_sub_query.c.isrc)) .filter( UserDeviceToken.is_active.is_(True), UserDeviceToken.user_id.in_(user_id_list), tracks_sub_query.c.curator_id.in_(categories_sub_query), ) .group_by(StarredContent.user_id, tracks_sub_query.c.playlist_id, tracks_sub_query.c.isrc) ) def get_top_markets(self, target_date: str, filter_by_markets: Optional[List[str]] = None) -> Set[str]: """Get top playlists markets. Optionally filter by passed list. target_date: str (YYYY-mm-dd). filter_by: List of markets to filter by. """ markets = { m.country_code.lower() for m in self.session.query(AppleWeeklyTopPlaylist.country_code) .filter( AppleWeeklyTopPlaylist.date.in_( (target_date, datetime.strptime(target_date, "%Y-%m-%d") - timedelta(days=1)) ) ) .distinct(AppleWeeklyTopPlaylist.country_code) } # us is used instead global for real querying AG-4910, so we consider us as global if US_MARKET in markets: markets.add(GLOBAL_MARKET) else: markets -= {GLOBAL_MARKET} if filter_by_markets: markets &= set(filter_by_markets) return markets def get_updated_tracklist_markets( self, target_date: str, filter_by_playlists: Optional[List[str]] = None, filter_by_markets: Optional[Set[str]] = None, ) -> Set[str]: min_datetime, max_datetime = self.get_border_datetimes(target_date) filters = [ AppleMusicPlaylistStatistics.tracklist_updated_at >= min_datetime, AppleMusicPlaylistStatistics.tracklist_updated_at < max_datetime, AppleMusicPlaylistStatistics.market.isnot(None), ] if filter_by_playlists: filters.append(AppleMusicPlaylistStatistics.playlist_id.in_(filter_by_playlists)) if filter_by_markets: _filters = filter_by_markets.copy() if GLOBAL_MARKET in filter_by_markets: _filters.remove(GLOBAL_MARKET) _filters.add(APOLLO_GLOBAL_MARKET) filters.append(AppleMusicPlaylistStatistics.market.in_(_filters)) q = ( self.session.query(AppleMusicPlaylistStatistics.market) .filter(*filters) .distinct(AppleMusicPlaylistStatistics.market) ) return set([parse_gl_to_global(pl.market) for pl in q])