from collections import OrderedDict, defaultdict from datetime import date, timedelta from flask import current_app as app from sqlalchemy import BIGINT, Date, and_, cast, func, or_ from sqlalchemy.orm import Query, aliased from typing import Any, Dict, Iterable, List, Set, Tuple from apollo_main_db.apollo.models import ApolloMarket, BuzzCategory, BuzzUser from apollo_main_db.apple.models import AppleMusicAlbum, AppleMusicContainerStreamSummary, AppleMusicCurator, \ AppleMusicSong, ApplePlaylist, ApplePlaylistStorefrontData, ApplePlaylistTrackList, ApplePlaylistTracklistHistory from src.constants.core import DEFAULT_IMAGE_SIZE, NO_COUNTRY_CODE_PLACEHOLDER, TOP_MARKETS, UNIX_EPOCH_START_DATE, \ VENDOR_APPLE, ZERO_TREND, ElasticPlaylistsVendorsData, MusicServices from src.db.base import session as db_session from src.legacy.apple_music.constants import APPLE_GLOBAL_MARKET, PlaylistMarketsIncludeFields from src.legacy.apple_music.vendor.base import AppleMusicBase from src.legacy.core.util import check_country_code_value, multikeysort from src.legacy.core.vendor import VendorPlaylists from src.legacy.redis_db import keys from src.legacy.redis_db.decorators import cache_requests, cache_value class AppleMusicPlaylists(VendorPlaylists, AppleMusicBase): def _get_global_market_track_playlists_query(self, isrc: str) -> Query: """ Build query for retrieving track playlists information in global market. Args: isrc (str): Track ISRC value. Returns: Query: Track playlist information query. """ filters = [ApplePlaylist.removed == 0] songs = db_session.query(AppleMusicSong.id).filter(AppleMusicSong.isrc == isrc).group_by(AppleMusicSong.id) songs_ids = [q.id for q in songs] playlists_tracks_subquery = ( db_session.query( ApplePlaylistTrackList.song_id.label("song_id"), ApplePlaylistTrackList.playlist_id.label("playlist_id"), ApplePlaylistTrackList.storefront.label("storefront"), ) .filter(ApplePlaylistTrackList.song_id.in_(songs_ids)) .group_by(ApplePlaylistTrackList.playlist_id, ApplePlaylistTrackList.song_id) .subquery() ) query = ( db_session.query( ApplePlaylist.id, cast(ApplePlaylist.latest_update, Date).label("changed_date"), func.coalesce(playlists_tracks_subquery.c.storefront, BuzzUser.country_code).label("country_code"), ApplePlaylist.name.label("name"), ApplePlaylist.artwork.label("image_url"), playlists_tracks_subquery.c.song_id, ) .join(playlists_tracks_subquery, playlists_tracks_subquery.c.playlist_id == ApplePlaylist.id) .outerjoin( BuzzUser, and_( BuzzUser.user_name == ApplePlaylist.curator_id, BuzzUser.music_service == MusicServices.APPLEMUSIC ), ) .filter(*filters) .group_by(ApplePlaylist.id) ) return query def _get_market_track_playlists_query(self, isrc: str, market: str) -> Query: """ Build query for retrieving track playlists information in selected (non-global) market. Args: isrc (str): Track ISRC value. market (str): Country code. Returns: Query: Track playlist information query. """ filters = [AppleMusicSong.isrc == isrc, ApplePlaylist.removed == 0] previous_date = date.today() - timedelta(days=14) query = ( db_session.query( ApplePlaylist.id, func.coalesce(ApplePlaylistTrackList.storefront, BuzzUser.country_code).label("country_code"), AppleMusicSong.id.label("song_id"), AppleMusicSong.isrc, cast(ApplePlaylist.latest_update, Date).label("changed_date"), func.IF( cast(ApplePlaylistTrackList.added_datetime, Date) == UNIX_EPOCH_START_DATE, None, cast(ApplePlaylistTrackList.added_datetime, Date), ).label("added_date"), func.coalesce(ApplePlaylistStorefrontData.name, ApplePlaylist.name).label("name"), func.coalesce(ApplePlaylistStorefrontData.artwork, ApplePlaylist.artwork).label("image_url"), (func.min(ApplePlaylistTrackList.position) + 1).label("position"), func.IF( and_( ApplePlaylistTrackList.previous_position.isnot(None), cast(ApplePlaylistTrackList.latest_position_change, Date) > previous_date, ), (func.min(ApplePlaylistTrackList.position) - ApplePlaylistTrackList.previous_position), ZERO_TREND, ).label("trend"), ) .join( ApplePlaylistTrackList, and_( ApplePlaylistTrackList.playlist_id == ApplePlaylist.id, ApplePlaylistTrackList.storefront == market ), ) .join( AppleMusicSong, and_( AppleMusicSong.id == ApplePlaylistTrackList.song_id, AppleMusicSong.storefront == ApplePlaylistTrackList.storefront, ), ) .outerjoin( ApplePlaylistStorefrontData, and_( ApplePlaylistStorefrontData.playlist_id == ApplePlaylist.id, ApplePlaylistStorefrontData.storefront == market, ), ) .outerjoin( BuzzUser, and_( BuzzUser.user_name == ApplePlaylist.curator_id, BuzzUser.music_service == MusicServices.APPLEMUSIC ), ) .filter(*filters) .group_by(ApplePlaylist.id, ApplePlaylistStorefrontData.storefront) ) return query @cache_requests( keys.APPLE_TRACK_PLAYLISTS_FULL, keys.APPLE_TRACK_PLAYLISTS_FULL_TTL, ids_index=None, ids_kwargs_name="isrc_list", cache_key_kwargs_names=( "markets_list", "image_size", "search", "playlists_ids", "include_username", "include_playlist_count", "include_streams_7_days", "include_isrc", "include_owner", "category_id", ), cache_key_parts=None, item_key_name="isrc", ) def get_track_playlists_data( self, *, isrc_list: List[str], markets_list: List[str] or None, image_size: int, search: str or None = None, playlists_ids: List[str] or None = None, include_username: bool = False, include_playlist_count: bool = False, include_streams_7_days: bool = False, include_isrc: bool = True, include_owner: bool = False, category_id: int or None = None, ) -> List[Dict[str, Any]]: """Get track playlists data. Args: isrc_list: List of track ISRC values. markets_list (List[str] or None): List of storefront markets. image_size (int): Playlists artwork image size value. search (str or None): Search string. playlists_ids (List[str] or None): List of playlists ids. include_owner: Include playlist owner. category_id: BuzzCategory id to filter by. Returns: List[Dict[str, Any]]: Track playlists data. """ adjusted_markets_list = [*markets_list, None] today = date.today() previous_date = today - timedelta(days=14) song_ids = ( db_session.query(AppleMusicSong) .with_entities(AppleMusicSong.id.label("song_id")) .distinct() .filter( AppleMusicSong.isrc.in_(isrc_list), AppleMusicSong.storefront.in_(adjusted_markets_list), ) ) playlist_tracklist_filters = [ ApplePlaylistTrackList.storefront.in_(adjusted_markets_list), ApplePlaylistTrackList.song_id.in_(song_ids), ] if playlists_ids: playlist_tracklist_filters.append(ApplePlaylistTrackList.playlist_id.in_(playlists_ids)) playlist_tracklist_subquery = ( db_session.query(ApplePlaylistTrackList) .with_entities( ApplePlaylistTrackList.playlist_id.label("id"), ApplePlaylistTrackList.song_id.label("song_id"), ApplePlaylistTrackList.storefront.label("storefront"), ApplePlaylistTrackList.playlist_id.label("playlist_id"), func.IF( cast(ApplePlaylistTrackList.added_datetime, Date) == UNIX_EPOCH_START_DATE, None, cast(ApplePlaylistTrackList.added_datetime, Date), ).label("added_date"), ApplePlaylistTrackList.position.label("current_position"), func.IF( and_( ApplePlaylistTrackList.previous_position.isnot(None), cast(ApplePlaylistTrackList.latest_position_change, Date) > previous_date, ), ApplePlaylistTrackList.previous_position + 1, None, ).label("prev_position"), func.IF( cast(ApplePlaylistTrackList.latest_position_change, Date) > previous_date, cast(ApplePlaylistTrackList.latest_position_change, Date), None, ).label("position_change_date"), ) .filter(*playlist_tracklist_filters) .subquery() ) music_song_subquery = ( db_session.query(AppleMusicSong) .join( playlist_tracklist_subquery, and_( playlist_tracklist_subquery.c.song_id == AppleMusicSong.id, playlist_tracklist_subquery.c.storefront == AppleMusicSong.storefront, ), ) .filter( AppleMusicSong.storefront.in_(adjusted_markets_list), AppleMusicSong.isrc.in_(isrc_list), ) .with_entities( AppleMusicSong.id.label("track_id"), AppleMusicSong.isrc.label("isrc"), playlist_tracklist_subquery.c.id.label("id"), playlist_tracklist_subquery.c.song_id.label("song_id"), playlist_tracklist_subquery.c.storefront.label("storefront"), playlist_tracklist_subquery.c.playlist_id.label("playlist_id"), playlist_tracklist_subquery.c.added_date.label("added_date"), playlist_tracklist_subquery.c.current_position.label("current_position"), playlist_tracklist_subquery.c.prev_position.label("prev_position"), playlist_tracklist_subquery.c.position_change_date.label("position_change_date"), ) .subquery() ) tracklist_history_subquery = ( db_session.query(ApplePlaylistTracklistHistory) .with_entities( func.min(ApplePlaylistTracklistHistory.date).label("earliest_added_date"), ApplePlaylistTracklistHistory.song_id.label("song_id"), ApplePlaylistTracklistHistory.storefront.label("storefront"), ApplePlaylistTracklistHistory.playlist_id.label("playlist_id"), ) .filter( ApplePlaylistTracklistHistory.storefront.in_(markets_list), ApplePlaylistTracklistHistory.song_id.in_(song_ids), ) .group_by( ApplePlaylistTracklistHistory.song_id, ApplePlaylistTracklistHistory.storefront, ApplePlaylistTracklistHistory.playlist_id, ) .order_by(None) .subquery() ) query = ( db_session.query(music_song_subquery) .join(ApplePlaylist, ApplePlaylist.id == music_song_subquery.c.playlist_id) .outerjoin( ApplePlaylistStorefrontData, and_( ApplePlaylistStorefrontData.playlist_id == music_song_subquery.c.playlist_id, ApplePlaylistStorefrontData.storefront == music_song_subquery.c.storefront, ), ) .outerjoin( tracklist_history_subquery, and_( tracklist_history_subquery.c.playlist_id == music_song_subquery.c.playlist_id, tracklist_history_subquery.c.storefront == music_song_subquery.c.storefront, tracklist_history_subquery.c.song_id == music_song_subquery.c.song_id, ), ) ) query_fields = [ cast(ApplePlaylist.latest_update, Date).label("updated_date"), func.coalesce(ApplePlaylist.name, ApplePlaylistStorefrontData.name).label("name"), func.coalesce(ApplePlaylist.artwork, ApplePlaylistStorefrontData.artwork).label("image_url"), tracklist_history_subquery.c.earliest_added_date.label("earliest_added_date"), music_song_subquery.c.track_id.label("track_id"), music_song_subquery.c.isrc.label("isrc"), music_song_subquery.c.id.label("id"), music_song_subquery.c.storefront.label("country_code"), music_song_subquery.c.added_date.label("added_date"), (func.min(music_song_subquery.c.current_position) + 1).label("current_position"), music_song_subquery.c.prev_position.label("prev_position"), music_song_subquery.c.position_change_date.label("position_change_date"), ] filters = [ ApplePlaylist.removed == 0, ] if include_username: query = query.outerjoin(AppleMusicCurator, AppleMusicCurator.id == ApplePlaylist.curator_id) query_fields.append(AppleMusicCurator.name.label("username")) if include_streams_7_days: query = query.outerjoin( AppleMusicContainerStreamSummary, and_( AppleMusicContainerStreamSummary.container_id == ApplePlaylist.id, AppleMusicContainerStreamSummary.country_code == music_song_subquery.c.storefront, ), ) query_fields.append( func.IFNULL(AppleMusicContainerStreamSummary.streams_7_days, 0).label("playlist_streams") ) if include_playlist_count: aliased_tracklist = aliased(ApplePlaylistTrackList) subquery = ( db_session.query(func.count(aliased_tracklist.song_id)) .filter(aliased_tracklist.playlist_id == ApplePlaylistTrackList.playlist_id) .filter(aliased_tracklist.storefront == ApplePlaylistTrackList.storefront) ).as_scalar() query_fields.append(func.IFNULL(subquery, 0).label("track_count")) if include_owner or category_id: query_fields.append(ApplePlaylist.curator_id.label("curator_id")) query = ( query.with_entities(*query_fields) .filter(*filters) .group_by(ApplePlaylist.id, music_song_subquery.c.storefront, music_song_subquery.c.isrc) ) if include_owner or category_id: subquery = query.subquery() buzz_user_subquery = ( db_session.query(BuzzUser) .with_entities( BuzzUser.category_id.label("category_id"), cast(BuzzUser.user_name, BIGINT).label("user_name") ) .filter(BuzzUser.music_service == MusicServices.APPLEMUSIC, BuzzUser.category_id.isnot(None)) .subquery() ) query = ( db_session.query(subquery) .outerjoin( buzz_user_subquery, buzz_user_subquery.c.user_name == subquery.c.curator_id, ) .outerjoin(BuzzCategory, BuzzCategory.id == buzz_user_subquery.c.category_id) .with_entities( subquery.c.id.label("id"), subquery.c.country_code.label("country_code"), subquery.c.track_id.label("track_id"), subquery.c.isrc.label("isrc"), subquery.c.updated_date.label("updated_date"), subquery.c.added_date.label("added_date"), subquery.c.earliest_added_date.label("earliest_added_date"), subquery.c.name.label("name"), subquery.c.image_url.label("image_url"), subquery.c.current_position.label("current_position"), subquery.c.prev_position.label("prev_position"), subquery.c.position_change_date.label("position_change_date"), BuzzCategory.id.label("category_id"), BuzzCategory.name.label("category_name"), *[subquery.c.playlist_streams.label("playlist_streams")] if include_streams_7_days else [], *[subquery.c.username.label("username")] if include_username else [], *[subquery.c.track_count.label("track_count")] if include_playlist_count else [], ) ) if category_id: query = query.filter(BuzzCategory.id == category_id) playlists_grouped_by_isrc = {} for item in query: if search and (not item.name or search not in item.name.lower()): continue playlist_data = item._asdict() try: image_url = item.image_url.format(w=image_size, h=image_size) except (KeyError, AttributeError): image_url = None country_code = check_country_code_value(item.country_code) added_date = playlist_data.pop("earliest_added_date", None) or playlist_data["added_date"] prev_position = playlist_data.pop("prev_position", None) is_new = not prev_position and added_date == today if not prev_position: trend = None if is_new else ZERO_TREND else: trend = item.current_position - prev_position if not playlist_data["name"]: playlist_data["name"] = None if not include_isrc: del playlist_data["isrc"] playlist_data["image_url"] = image_url playlist_data["country_code"] = country_code playlist_data["added_date"] = added_date playlist_data["is_new"] = is_new playlist_data["trend"] = trend playlists_grouped_by_isrc.setdefault(item.isrc, []) playlists_grouped_by_isrc[item.isrc].append(playlist_data) result = [{"isrc": isrc, "data": playlists_grouped_by_isrc.get(isrc, [])} for isrc in isrc_list] return result def flatten_playlists_data(self, track_playlists_data: List[Dict], map_result: bool = False) -> Tuple[Any, Set]: """Filter list of objects by removed_date field, showing only items after specific date. Args: track_playlists_data (List[Dict]): Playlists data. map_result(bool): True when mapped result is needed Returns: Playlists and Playlists Ids """ items = {} if map_result else [] playlist_ids = set() for isrc_item in track_playlists_data: for playlist in isrc_item["data"]: playlist_id = playlist["id"] + "_" + playlist["country_code"] if playlist_id not in playlist_ids: if map_result: items[playlist_id] = playlist else: items.append(playlist) playlist_ids.add(playlist_id) return items, playlist_ids @cache_requests( keys.APPLE_TRACK_PREVIOUS_PLAYLISTS_FULL, keys.APPLE_TRACK_PREVIOUS_PLAYLISTS_FULL_TTL, ids_index=1, ids_kwargs_name=None, cache_key_kwargs_names=( "markets_list", "search", "exclude_current_playlists", "image_size", "include_owner", "category_id", ), cache_key_parts=None, item_key_name="isrc", ) def get_track_previous_playlists_data( self, isrc_list: Iterable[str], *, markets_list: List[str] or None = None, search: str or None = None, exclude_current_playlists: bool = True, image_size: int = DEFAULT_IMAGE_SIZE, include_owner: bool = False, category_id: int or None = None, ) -> List[Dict[str, Any]]: """Get track previous playlists data. Arguments: isrc_list: Track ISRC value. markets_list: List of storefront markets. search: Search string value. exclude_current_playlists: Exclude current playlists. image_size (int): Playlists artwork image size value. include_owner: Include playlist owner. category_id: BuzzCategory id to filter by. Returns: Track previous playlists data. """ song_ids = ( db_session.query(AppleMusicSong) .with_entities(AppleMusicSong.id.label("song_id")) .distinct() .filter( AppleMusicSong.isrc.in_(isrc_list), AppleMusicSong.storefront.in_(markets_list), ) ) subquery = ( db_session.query(ApplePlaylistTracklistHistory) .with_entities( cast(func.max(ApplePlaylistTracklistHistory.date), Date).label("removed_date"), func.min(ApplePlaylistTracklistHistory.date).label("earliest_added_date"), ApplePlaylistTracklistHistory.song_id.label("song_id"), ApplePlaylistTracklistHistory.storefront.label("storefront"), ApplePlaylistTracklistHistory.playlist_id.label("playlist_id"), ) .filter( ApplePlaylistTracklistHistory.storefront.in_(markets_list), ApplePlaylistTracklistHistory.song_id.in_(song_ids), ) .group_by( ApplePlaylistTracklistHistory.song_id, ApplePlaylistTracklistHistory.storefront, ApplePlaylistTracklistHistory.playlist_id, ) .subquery() ) music_song_fields = [ subquery.c.playlist_id.label("playlist_id"), subquery.c.earliest_added_date.label("earliest_added_date"), cast(func.max(subquery.c.removed_date), Date).label("removed_date"), AppleMusicSong.storefront.label("country_code"), AppleMusicSong.id.label("track_id"), AppleMusicSong.isrc.label("isrc"), func.IF( cast(ApplePlaylistTrackList.added_datetime, Date) == UNIX_EPOCH_START_DATE, None, cast(ApplePlaylistTrackList.added_datetime, Date), ).label("added_date"), ] subquery2 = ( db_session.query(AppleMusicSong) .join( subquery, and_( subquery.c.song_id == AppleMusicSong.id, subquery.c.storefront == AppleMusicSong.storefront, ), ) .outerjoin( ApplePlaylistTrackList, and_( subquery.c.playlist_id == ApplePlaylistTrackList.playlist_id, AppleMusicSong.storefront == ApplePlaylistTrackList.storefront, AppleMusicSong.id == ApplePlaylistTrackList.song_id, ), ) ) if exclude_current_playlists: music_song_fields.append(func.count(ApplePlaylistTrackList.playlist_id).label("current_count")) subquery2 = ( subquery2.with_entities(*music_song_fields) .filter( AppleMusicSong.storefront.in_(markets_list), AppleMusicSong.isrc.in_(isrc_list), ) .group_by( subquery.c.playlist_id, subquery.c.storefront, AppleMusicSong.isrc, ) ) if exclude_current_playlists: subquery2 = subquery2.having(func.group_concat(ApplePlaylistTrackList.playlist_id.distinct()).is_(None)) subquery2 = subquery2.subquery() query = db_session.query(subquery2).join(ApplePlaylist, ApplePlaylist.id == subquery2.c.playlist_id) additional_fields = [] additional_filters = [] if exclude_current_playlists: additional_filters.append(subquery2.c.current_count == 0) if include_owner or category_id: buzz_user_subquery = ( db_session.query(BuzzUser) .with_entities( BuzzUser.category_id.label("category_id"), cast(BuzzUser.user_name, BIGINT).label("user_name") ) .filter(BuzzUser.music_service == MusicServices.APPLEMUSIC, BuzzUser.category_id.isnot(None)) .subquery() ) query = query.outerjoin( buzz_user_subquery, buzz_user_subquery.c.user_name == ApplePlaylist.curator_id, ).outerjoin(BuzzCategory, BuzzCategory.id == buzz_user_subquery.c.category_id) if include_owner: additional_fields.extend( [ BuzzCategory.id.label("category_id"), BuzzCategory.name.label("category_name"), ] ) if category_id: additional_filters.append(BuzzCategory.id == category_id) query = query.with_entities( ApplePlaylist.id.label("id"), ApplePlaylist.name.label("name"), subquery2.c.removed_date.label("removed_date"), subquery2.c.country_code.label("country_code"), subquery2.c.track_id.label("track_id"), subquery2.c.isrc.label("isrc"), subquery2.c.added_date.label("added_date"), subquery2.c.earliest_added_date.label("earliest_added_date"), cast(ApplePlaylist.latest_update, Date).label("updated_date"), ApplePlaylist.artwork.label("image_url"), *additional_fields, ).filter(*additional_filters) playlists_grouped_by_isrc = {} for item in query: if search and (not item.name or search not in item.name.lower()): continue playlist_data = item._asdict() try: image_url = item.image_url.format(w=image_size, h=image_size) except (KeyError, AttributeError): image_url = None country_code = check_country_code_value(item.country_code) if not playlist_data["name"]: playlist_data["name"] = None del playlist_data["isrc"] playlist_data["image_url"] = image_url playlist_data["country_code"] = country_code playlist_data["personalized"] = False added_date = playlist_data.pop("earliest_added_date", None) or playlist_data["added_date"] playlist_data["added_date"] = added_date playlists_grouped_by_isrc.setdefault(item.isrc, []) playlists_grouped_by_isrc[item.isrc].append(playlist_data) result = [{"isrc": isrc, "data": playlists_grouped_by_isrc.get(isrc, [])} for isrc in isrc_list] return result def get_track_in_playlist_data( self, isrc: str, playlist_id: str, market: str, image_size: int ) -> Tuple[object, dict]: """Get track in playlist query and context data. Arguments: isrc (str): ISRC. playlist_id (str): Playlist identifier. market (str): Market code value. image_size (int): Image height value. Returns: Tuple[object, dict]: Tuple with query object and additional data for schema. """ filters = [AppleMusicSong.isrc == isrc, ApplePlaylist.removed == 0, ApplePlaylist.id == playlist_id] today = date.today() previous_date = today - timedelta(days=14) track_in_playlist = ( db_session.query( ApplePlaylist.id.label("playlist_id"), func.coalesce(ApplePlaylistTrackList.storefront, BuzzUser.country_code).label("country_code"), AppleMusicSong.id.label("track_id"), AppleMusicSong.name.label("track_name"), AppleMusicSong.isrc, func.coalesce(ApplePlaylistStorefrontData.name, ApplePlaylist.name).label("playlist_name"), func.coalesce(ApplePlaylistStorefrontData.artwork, ApplePlaylist.artwork).label("playlist_image"), (func.min(ApplePlaylistTrackList.position) + 1).label("current_position"), cast(ApplePlaylist.latest_update, Date).label("changed_date"), AppleMusicSong.artwork_url.label("track_image"), func.IF( and_( ApplePlaylistTrackList.previous_position.isnot(None), cast(ApplePlaylistTrackList.latest_position_change, Date) > previous_date, ), ApplePlaylistTrackList.previous_position + 1, None, ).label("previous_position"), ) .join( ApplePlaylistTrackList, and_( ApplePlaylistTrackList.playlist_id == ApplePlaylist.id, ApplePlaylistTrackList.storefront == market ), ) .join( AppleMusicSong, and_( AppleMusicSong.id == ApplePlaylistTrackList.song_id, AppleMusicSong.storefront == ApplePlaylistTrackList.storefront, ), ) .outerjoin( ApplePlaylistStorefrontData, and_( ApplePlaylistStorefrontData.playlist_id == ApplePlaylist.id, ApplePlaylistStorefrontData.storefront == market, ), ) .outerjoin( BuzzUser, and_( BuzzUser.user_name == ApplePlaylist.curator_id, BuzzUser.music_service == MusicServices.APPLEMUSIC ), ) .filter(*filters) .group_by(ApplePlaylist.id, ApplePlaylistStorefrontData.storefront) .first() ) context = {} if not track_in_playlist: return track_in_playlist, context added_date_data = self._get_track_added_dates_values_in_playlists( [playlist_id], track_in_playlist.track_id, market ) trend = ( track_in_playlist.current_position - track_in_playlist.previous_position if track_in_playlist.previous_position else ZERO_TREND ) added_date = next((pl["added_date"] for pl in added_date_data if pl["id"] == playlist_id), None) no_trend = not trend and added_date and added_date != today and added_date == track_in_playlist.changed_date image_map = {"track_image": track_in_playlist.track_image, "playlist_image": track_in_playlist.playlist_image} context = {} for field, url in image_map.items(): try: context[field] = url.format(w=image_size, h=image_size) except (KeyError, AttributeError): context[field] = None context.update({"no_trend": no_trend, "trend": trend, "today": today, "added_date": added_date}) return track_in_playlist, context def get_track_in_past_playlist_data( self, isrc: str, playlist_id: str, market: str, image_size: int ) -> Tuple[object, dict]: """Get track in playlist query and context data. Arguments: isrc (str): ISRC. playlist_id (str): Playlist identifier. market (str): Market code value. image_size (int): Image height value. Returns: Tuple[object, dict]: Tuple with query object and additional data for schema. """ filters = [AppleMusicSong.isrc == isrc, ApplePlaylist.removed == 0, ApplePlaylist.id == playlist_id] today = date.today() track_in_playlist = ( db_session.query( ApplePlaylist.id.label("playlist_id"), func.coalesce(ApplePlaylistTracklistHistory.storefront, BuzzUser.country_code).label("country_code"), AppleMusicSong.id.label("track_id"), AppleMusicSong.name.label("track_name"), AppleMusicSong.isrc, func.coalesce(ApplePlaylistStorefrontData.name, ApplePlaylist.name).label("playlist_name"), func.coalesce(ApplePlaylistStorefrontData.artwork, ApplePlaylist.artwork).label("playlist_image"), cast(ApplePlaylist.latest_update, Date).label("changed_date"), cast(func.max(ApplePlaylistTracklistHistory.date), Date).label("removed_date"), AppleMusicSong.artwork_url.label("track_image"), ) .join( ApplePlaylistTracklistHistory, and_( ApplePlaylistTracklistHistory.playlist_id == ApplePlaylist.id, ApplePlaylistTracklistHistory.storefront == market, ), ) .join( AppleMusicSong, and_( AppleMusicSong.id == ApplePlaylistTracklistHistory.song_id, AppleMusicSong.storefront == ApplePlaylistTracklistHistory.storefront, ), ) .outerjoin( ApplePlaylistStorefrontData, and_( ApplePlaylistStorefrontData.playlist_id == ApplePlaylist.id, ApplePlaylistStorefrontData.storefront == market, ), ) .outerjoin( BuzzUser, and_( BuzzUser.user_name == ApplePlaylist.curator_id, BuzzUser.music_service == MusicServices.APPLEMUSIC ), ) .outerjoin( ApplePlaylistTrackList, and_( ApplePlaylistTrackList.playlist_id == ApplePlaylist.id, ApplePlaylistTrackList.song_id == AppleMusicSong.id, ApplePlaylistTrackList.storefront == market, ), ) .filter(*filters) .group_by(ApplePlaylist.id, ApplePlaylistStorefrontData.storefront) .having(func.group_concat(ApplePlaylistTrackList.playlist_id.distinct()).is_(None)) .first() ) context = {} if not track_in_playlist: return track_in_playlist, context added_date_data = self._get_track_added_dates_values_in_playlists( [playlist_id], track_in_playlist.track_id, market ) added_date = next((pl["added_date"] for pl in added_date_data if pl["id"] == playlist_id), None) image_map = {"track_image": track_in_playlist.track_image, "playlist_image": track_in_playlist.playlist_image} context = {} for field, url in image_map.items(): try: context[field] = url.format(w=image_size, h=image_size) except (KeyError, AttributeError): context[field] = None context.update({"today": today, "added_date": added_date}) return track_in_playlist, context @cache_value(keys.RAW_APPLE_TRACK_PLAYLISTS, keys.RAW_APPLE_TRACK_PLAYLISTS_TTL, key_args_indexes=(1, 2)) def _get_playlist_data_from_db(self, isrc: str, market: str) -> Dict[str, dict]: """ Execute query and store it in a map. Args: isrc (str): Track ISRC value. market (str): Market value. Returns: Dict[str, dict]: Track playlist data map. """ if market == APPLE_GLOBAL_MARKET: query = self._get_global_market_track_playlists_query(isrc) else: query = self._get_market_track_playlists_query(isrc, market) return {q.id: q._asdict() for q in query} def _get_track_added_dates_values_in_playlists( self, playlist_ids: Iterable[str], track_id: str, market: str ) -> list: """Get track playlists added date. Arguments: track_id (str): Apple Music track identifier. market (str): Market code. playlist_ids (Iterable[str]): List of playlists ids. Returns: list: Playlists added dates values. """ result = {pl_id: {"id": pl_id, "added_date": None} for pl_id in playlist_ids if pl_id} if market == APPLE_GLOBAL_MARKET: return list(result.values()) isrc_query = db_session.query(AppleMusicSong.isrc).filter( AppleMusicSong.storefront == market, AppleMusicSong.id == track_id ) tracks_subquery = ( db_session.query(AppleMusicSong.id.label("song_id"), AppleMusicSong.isrc.label("isrc")) .distinct(AppleMusicSong.id) .filter(AppleMusicSong.isrc.in_(isrc_query), AppleMusicSong.storefront == market) .subquery() ) earliest_date_subquery = ( db_session.query( ApplePlaylistTracklistHistory.playlist_id.label("playlist_id"), ApplePlaylistTracklistHistory.storefront.label("storefront"), func.min(ApplePlaylistTracklistHistory.date).label("earliest_date"), tracks_subquery, ) .join(tracks_subquery, tracks_subquery.c.song_id == ApplePlaylistTracklistHistory.song_id) .filter( ApplePlaylistTracklistHistory.storefront == market, ApplePlaylistTracklistHistory.playlist_id.in_(playlist_ids), ) .group_by(ApplePlaylistTracklistHistory.playlist_id, tracks_subquery.c.song_id, tracks_subquery.c.isrc) .subquery() ) earliest_added_date_query = ( db_session.query( ApplePlaylistTracklistHistory.playlist_id.label("id"), func.IF( func.min(earliest_date_subquery.c.earliest_date) == UNIX_EPOCH_START_DATE, None, func.min(earliest_date_subquery.c.earliest_date), ).label("earliest_added_date"), ApplePlaylistTracklistHistory.position.label("earliest_position"), AppleMusicSong.isrc, ) .join( earliest_date_subquery, and_( earliest_date_subquery.c.playlist_id == ApplePlaylistTracklistHistory.playlist_id, earliest_date_subquery.c.storefront == ApplePlaylistTracklistHistory.storefront, earliest_date_subquery.c.earliest_date == ApplePlaylistTracklistHistory.date, earliest_date_subquery.c.song_id == ApplePlaylistTracklistHistory.song_id, ), ) .join( AppleMusicSong, and_( AppleMusicSong.id == ApplePlaylistTracklistHistory.song_id, AppleMusicSong.storefront == ApplePlaylistTracklistHistory.storefront, ), ) .filter( ApplePlaylistTracklistHistory.storefront == market, ApplePlaylistTracklistHistory.playlist_id.in_(playlist_ids), AppleMusicSong.isrc.in_(isrc_query), ) .group_by(ApplePlaylistTracklistHistory.playlist_id, AppleMusicSong.isrc) ) for item in earliest_added_date_query: result[item.id]["added_date"] = item.earliest_added_date no_added_dates_ids = [pl_id for pl_id, data in result.items() if data["added_date"] is None] if not no_added_dates_ids: return list(result.values()) added_date_query = ( db_session.query( ApplePlaylistTrackList.playlist_id.label("id"), func.IF( cast(ApplePlaylistTrackList.added_datetime, Date) == UNIX_EPOCH_START_DATE, None, cast(ApplePlaylistTrackList.added_datetime, Date), ).label("added_date"), ) .filter( ApplePlaylistTrackList.storefront == market, ApplePlaylistTrackList.playlist_id.in_(no_added_dates_ids), ApplePlaylistTrackList.song_id == track_id, ) .group_by(ApplePlaylistTrackList.playlist_id) ) for item in added_date_query: result[item.id]["added_date"] = item.added_date return list(result.values()) @cache_requests(keys.APPLE_PLAYLISTS_ADDED_DATES, keys.APPLE_PLAYLISTS_ADDED_DATES_TTL, cache_key_parts=(2, 3)) def get_track_playlists_added_date(self, playlist_ids: Iterable[str], track_id: str, market: str) -> list: """Get track playlists added date wrapper with cache decorator. Arguments: track_id (str): Apple Music track identifier. market (str): Market code. playlist_ids (Iterable[str]): List of playlists ids. Returns: list: Playlists added dates values. """ return self._get_track_added_dates_values_in_playlists(playlist_ids, track_id, market) @staticmethod def get_playlist_tracklist_allocation_graph_query(playlist_id: str, market: str) -> Query: """Return query with basic data for allocated playlist graphs for each track from tracklist. Args: playlist_id (str): id of playlist to get data for. market (str): market code to get tracks for. Return: Query with basic data for allocated playlist graphs for each track from tracklist. """ query = ( db_session.query( (ApplePlaylistTrackList.position + 1).label("current_position"), ApplePlaylistTrackList.song_id.label("track_id"), func.IF( cast(AppleMusicSong.release_date, Date) == UNIX_EPOCH_START_DATE, None, cast(AppleMusicSong.release_date, Date), ).label("release_date"), func.IF( cast(AppleMusicAlbum.release_date, Date) == UNIX_EPOCH_START_DATE, None, cast(AppleMusicAlbum.release_date, Date), ).label("album_release_date"), AppleMusicSong.isrc.label("isrc"), ) .outerjoin( AppleMusicSong, and_(AppleMusicSong.storefront == market, AppleMusicSong.id == ApplePlaylistTrackList.song_id), ) .outerjoin( AppleMusicAlbum, and_( AppleMusicAlbum.storefront == AppleMusicSong.storefront, AppleMusicAlbum.id == AppleMusicSong.album_id, ), ) .filter(ApplePlaylistTrackList.storefront == market, ApplePlaylistTrackList.playlist_id == playlist_id) .group_by(ApplePlaylistTrackList.position) ) return query def get_playlist_tracklist(self, playlist_id: str, market: str, image_size: int) -> List[Dict]: """Retrieve Apple playlist tracklist data from DB. Args: playlist_id (str): Playlist identifier. market (str): Market code. image_size (int): Image size. Returns: List[Dict]: List of playlist`s tracks. """ tracklist = self._get_playlist_tracklist_data_from_db(playlist_id, market) isrc_added_dates = self._get_playlist_tracks_added_dates(playlist_id, market) today = date.today() result = [] for track in tracklist: _isrc = track["isrc"] artist = track.pop("artist") added_date = isrc_added_dates.get(_isrc, {}).get("earliest_date") or track.get("added_date") prev_position, update_date = track["previous_position"], track["update_date"] trend = track["current_position"] - prev_position if prev_position else ZERO_TREND # If there is no previous position and track was added today is_new = not trend and added_date == today # if there is no previous position and track added date was not today e.g. no trend to show and not bluedot no_trend = not trend and added_date and added_date != today and added_date == update_date try: image_url = track["image_url"].format(w=image_size, h=image_size) except (KeyError, AttributeError): image_url = None result.append( { **track, "added_date": added_date, "trend": None if is_new or no_trend else trend, "artists": [artist], "image_url": image_url, "is_new": is_new, } ) return result def _get_playlist_tracklist_data_from_db(self, playlist_id: str, market: str) -> List[Dict]: """Retrieve playlist's list of tracks from DB. Args: playlist_id (str): Playlist identifier. market (str): Market code. Returns: Tuple[List[Dict], List[str]]: Playlist's list of tracks and list of ISRC values. """ previous_date = date.today() - timedelta(days=14) query = ( db_session.query( (ApplePlaylistTrackList.position + 1).label("current_position"), ApplePlaylistTrackList.song_id.label("track_id"), func.IF( cast(ApplePlaylistTrackList.added_datetime, Date) == UNIX_EPOCH_START_DATE, None, cast(ApplePlaylistTrackList.added_datetime, Date), ).label("added_date"), func.IF( and_( ApplePlaylistTrackList.previous_position.isnot(None), cast(ApplePlaylistTrackList.latest_position_change, Date) > previous_date, ), cast(ApplePlaylistTrackList.latest_position_change, Date), None, ).label("position_change_date"), func.IF( and_( ApplePlaylistTrackList.previous_position.isnot(None), cast(ApplePlaylistTrackList.latest_position_change, Date) > previous_date, ), (ApplePlaylistTrackList.previous_position + 1), None, ).label("previous_position"), AppleMusicSong.isrc.label("isrc"), AppleMusicSong.name.label("name"), AppleMusicSong.artist_name.label("artist"), AppleMusicSong.artwork_url.label("image_url"), cast(ApplePlaylist.latest_update, Date).label("update_date"), ) .outerjoin( AppleMusicSong, and_(AppleMusicSong.storefront == market, AppleMusicSong.id == ApplePlaylistTrackList.song_id), ) .join(ApplePlaylist, ApplePlaylist.id == ApplePlaylistTrackList.playlist_id) .filter(ApplePlaylistTrackList.storefront == market, ApplePlaylistTrackList.playlist_id == playlist_id) .group_by(ApplePlaylistTrackList.position) ) result = [] for item in query: track = item._asdict() result.append(track) return result @staticmethod def _get_playlist_tracks_added_dates(playlist_id: str, market: str) -> Dict: """Retrieve from DB playlist's tracks added dates values. Args: playlist_id (str): Playlist identifier. market (str): Market code. Returns: Dict: Playlist added dates values for each track in playlist. """ historical_subquery = ( db_session.query( func.min(ApplePlaylistTracklistHistory.date).label("earliest_added_date"), func.max(ApplePlaylistTracklistHistory.date).label("latest_added_date"), cast(ApplePlaylist.latest_update, Date).label("update_date"), ApplePlaylistTracklistHistory.song_id.label("track_id"), ) .join(ApplePlaylist, ApplePlaylistTracklistHistory.playlist_id == ApplePlaylist.id) .filter( ApplePlaylistTracklistHistory.playlist_id == playlist_id, ApplePlaylistTracklistHistory.storefront == market, ) .group_by(ApplePlaylistTracklistHistory.song_id) .subquery(name="historical_subquery") ) query = ( db_session.query( func.min(historical_subquery.c.earliest_added_date).label("earliest_added_date"), func.max(historical_subquery.c.latest_added_date).label("latest_added_date"), historical_subquery.c.update_date.label("update_date"), AppleMusicSong.isrc.label("isrc"), ) .join( AppleMusicSong, and_(AppleMusicSong.id == historical_subquery.c.track_id, AppleMusicSong.storefront == market), ) .group_by(AppleMusicSong.isrc) .having( func.max(historical_subquery.c.latest_added_date).label("latest_added_date") == historical_subquery.c.update_date ) ) result = {} for item in query: track = item._asdict() isrc, earliest_date = track.pop("isrc"), track.pop("earliest_added_date") track["earliest_date"] = earliest_date if earliest_date != UNIX_EPOCH_START_DATE else None result[isrc] = track return result def playlists_search_handler(self, query: str, size: int, image_size: int) -> List[dict]: """Search Apple Music playlists in ElasticSearch and format retrieved data. Args: query (str): Search string. size (int): Request limit size. image_size (int): Playlist image size. Returns: List[dict]: List with found playlists data. """ es_body = self.get_playlists_search_es_body( query, factor_field=ElasticPlaylistsVendorsData.FACTOR_FIELDS[VENDOR_APPLE], field_name="name", size=size ) hits = app.elasticsearch.search( index=app.config["ELASTICSEARCH_PLAYLISTS_INDEX"], doc_type=ElasticPlaylistsVendorsData.DOC_TYPES[VENDOR_APPLE], body=es_body, ).get("hits", {}) result = [] for item in hits.get("hits", []): item_source = item.get("_source") if not item_source: continue image = item_source.get("artworkUrl") result.append( { "owner": item_source.get("curator"), "image_url": image.format(w=image_size, h=image_size) if image else None, "id": item_source["playlistId"], "country_code": item_source.get("countryCode"), "vendor": VENDOR_APPLE, "name": item_source.get("name"), } ) return result def get_playlists_markets(self, playlist_ids: List[str], include: List[str]) -> List[dict]: """Returns playlists' storefronts markets data. Args: playlist_ids (List[str]): List of playlists' identifiers. include (List[str): List with extra fields to show in response. Returns: List[dict]: List that contains markets data for each requested playlist. """ playlists_markets_list = self._get_playlists_markets_data(playlist_ids) result = {} for item in playlists_markets_list: pl_id, markets_data = item["id"], item["markets"] top_markets_dict = OrderedDict.fromkeys(TOP_MARKETS, None) other_markets_dict = {} for market_data in markets_data: code = market_data["code"] if code in top_markets_dict.keys(): top_markets_dict[code] = market_data else: other_markets_dict[code] = market_data top_markets = list([v for v in top_markets_dict.values() if v]) other_markets = list(other_markets_dict.values()) country_code = top_markets[0] if top_markets else (other_markets[0] if other_markets else None) result[pl_id] = {"market": country_code, "id": pl_id} if PlaylistMarketsIncludeFields.TOP_MARKETS in include: result[pl_id]["top_markets"] = top_markets if PlaylistMarketsIncludeFields.OTHER_MARKETS in include: result[pl_id]["other_markets"] = other_markets if PlaylistMarketsIncludeFields.OTHER_MARKETS in include: # List of other markets has to be sorted by name result[pl_id]["other_markets"] = multikeysort(result[pl_id]["other_markets"], ["name"]) return list(result.values()) @cache_requests(keys.APPLE_PLAYLIST_MARKETS, keys.APPLE_PLAYLIST_MARKETS_TTL, cache_key_parts=None) def _get_playlists_markets_data(self, playlist_ids: List[str]) -> List[dict]: """Inner method that query playlists' markets data from DB and cached it. Args: playlist_ids (List[str]): List of playlists' identifiers. Returns: List[dict]: List of raw playlists' markets data. """ markets_query = ( db_session.query( func.lower(ApplePlaylistStorefrontData.storefront).label("code"), ApolloMarket.id, ApolloMarket.name, ApolloMarket.full_name, ApplePlaylistStorefrontData.playlist_id.label("playlist_id"), ) .join(ApolloMarket, ApolloMarket.code == ApplePlaylistStorefrontData.storefront) .filter(ApplePlaylistStorefrontData.playlist_id.in_(playlist_ids), ApolloMarket.is_primary.is_(True)) .group_by(ApplePlaylistStorefrontData.playlist_id, ApplePlaylistStorefrontData.storefront) ) result = defaultdict(lambda: {"markets": []}) for item in markets_query: item_data = item._asdict() pl_id = item_data.pop("playlist_id") result[pl_id]["markets"].append(item_data) result[pl_id]["id"] = pl_id return list(result.values()) def get_playlists_data(self, playlist_ids: List[str], market: str) -> List[dict]: """Returns playlists' stored in DB storefront related data. Args: playlist_ids (List[str]): List of playlists' identifiers. market (str): Country code value. Returns: List[dict]: List that contains storefront related data for each requested playlist in selected market. """ return self._get_playlists_data_in_market(playlist_ids, market) def _get_playlists_data(self, filter_condition): playlists_query = ( db_session.query( ApplePlaylistStorefrontData.playlist_id.label("id"), func.coalesce(ApplePlaylistStorefrontData.name, ApplePlaylist.name).label("name"), func.coalesce(ApplePlaylistStorefrontData.artwork, ApplePlaylist.artwork).label("image_url"), func.lower(ApplePlaylistStorefrontData.storefront).label("country_code"), BuzzUser.display_name.label("curator_name"), BuzzUser.user_name.label("curator_id"), cast(ApplePlaylist.latest_update, Date).label("updated_date"), ) .join(ApplePlaylist, ApplePlaylist.id == ApplePlaylistStorefrontData.playlist_id) .outerjoin( BuzzUser, and_( BuzzUser.user_name == ApplePlaylist.curator_id, BuzzUser.music_service == MusicServices.APPLEMUSIC ), ) .filter(filter_condition) .group_by(ApplePlaylistStorefrontData.playlist_id, ApplePlaylistStorefrontData.storefront) ) return [item._asdict() for item in playlists_query] @cache_requests(keys.APPLE_PLAYLIST_DATA, keys.APPLE_PLAYLIST_DATA_TTL, cache_key_parts=(2,)) def _get_playlists_data_in_market(self, playlist_ids: List[str], market: str) -> List[dict]: """Inner method that query playlists' storefront related data from DB and cached it. Args: playlist_ids (List[str]): List of playlists' identifiers. market (str): Country code value. Returns: List[dict]: List with playlists storefront data. """ return self._get_playlists_data( and_( ApplePlaylistStorefrontData.playlist_id.in_(playlist_ids), ApplePlaylistStorefrontData.storefront == market, ) ) def get_playlists_markets_data(self, playlist_data: List[dict]) -> List[dict]: """Get playlists data by list of pairs: playlist ID and market (country_code). Args: playlist_data: List of playlist ID and market pairs. Returns: Playlists info. """ # tuple_().in_ is much slower for mysql 5.6 return self._get_playlists_data( or_( *[ and_( ApplePlaylistStorefrontData.playlist_id == i["playlist_id"], ApplePlaylistStorefrontData.storefront == i["country_code"], ) for i in playlist_data ] ) ) @cache_requests( keys.APPLE_TRACK_PLAYLISTS_OWNER_FULL, keys.APPLE_TRACK_PLAYLISTS_OWNER_FULL_TTL, ids_index=None, ids_kwargs_name="playlists_list", cache_key_kwargs_names=( "playlists_list", "markets_list", ), cache_key_parts=None, item_key_name="playlist", ) def get_apple_owner_market_data( self, playlists_list: List[str], markets_list: List[str] or None, ) -> List[Dict[str, Any]]: """Get owners data for playlists with missing country codes. Arguments: playlists_list: Playlists IDs. markets_list: List of storefront markets. Returns: Owners markets data. """ filters = [] if markets_list: filters.append(BuzzUser.country_code.in_([*markets_list, None])) subquery = ( db_session.query(ApplePlaylist) .with_entities( ApplePlaylist.id.label("id"), ApplePlaylist.curator_id.label("curator_id"), ) .filter(ApplePlaylist.id.in_(playlists_list)) ).subquery() query = ( db_session.query(subquery) .outerjoin( BuzzUser, and_( BuzzUser.music_service == MusicServices.APPLEMUSIC, BuzzUser.user_name == subquery.c.curator_id, ), ) .with_entities( subquery.c.id.label("id"), func.coalesce(BuzzUser.country_code, NO_COUNTRY_CODE_PLACEHOLDER).label("country_code"), ) .filter(*filters) ) result = dict(query.all()) return result def fill_missing_countries( self, items: List[Dict], owners_data: List[Dict], markets_list: List[str] or None, ): """Fill country codes and filter. Arguments: items: Playlists data. owners_data: Owners markets data. markets_list: List of storefront markets. Returns: Full playlists data. """ for isrc_item in items: for playlist in isrc_item["data"]: if not playlist["country_code"]: playlist["country_code"] = owners_data.get(playlist["id"], NO_COUNTRY_CODE_PLACEHOLDER) isrc_item["data"] = [i for i in isrc_item["data"] if i["country_code"] in markets_list] return items