from collections import defaultdict from datetime import date, timedelta from flask import current_app as app from sqlalchemy import Date, Integer, and_, cast, distinct, func, literal, union from sqlalchemy.orm import Query from typing import Any, Dict, Iterable, List, Optional, Tuple from apollo_main_db.apollo.models import ApolloMarket, BuzzCategory, BuzzUser from apollo_main_db.spotify.models import PlaylistIgnoredPlaylist, SpotifyAnalyticsPlaylistStreamInfo, SpotifyArtist, \ SpotifyPersonalizedPlaylist, SpotifyPlaylist, SpotifyPlaylistFollowers, SpotifyPlaylistFollowersHistory, \ SpotifyPlaylistLegacyUri, SpotifyPlaylistStreamSummary, SpotifyPlaylistTrackList2, \ SpotifyPlaylistTrackListHistory2, SpotifyPlaylistTrackListHistoryDates2, SpotifyTrack2, SpotifyTrackAlbum, \ SpotifyTrackArtist, SpotifyTrackPlaylistLastWeekHistory from src.constants.core import CURRENT_PLACEMENT_TYPE, MARKET_GLOBAL, NO_COUNTRY_CODE_PLACEHOLDER, \ PAST_PLACEMENT_TYPE, SPOTIFY_MARKET_GLOBAL, UNIX_EPOCH_START_DATE, VENDOR_SPOTIFY, ZERO_TREND, \ ElasticPlaylistsVendorsData, MusicServices, ReleaseTypes from src.db.base import session as db_session from src.legacy.core.clients import clients from src.legacy.core.util import DateRange, check_country_code_value, combine_artists_by_track, date_to_timestamp, \ generate_coordinates 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 from src.legacy.spotify.constants import SPOTIFY_PLAYLIST_IMAGE_URL_MASK, SPOTIFY_PLAYLIST_URI_PREFIX from src.legacy.spotify.vendor.base import SpotifyBase class SpotifyPlaylists(VendorPlaylists, SpotifyBase): @cache_requests( keys.SPOTIFY_TRACK_PLAYLISTS_FULL, keys.SPOTIFY_TRACK_PLAYLISTS_FULL_TTL, ids_index=None, ids_kwargs_name="isrc_list", cache_key_kwargs_names=( "markets_list", "search", "playlists_ids", "include_track_count", "include_username", "include_streams_7_days", "include_followers", "streams_markets_list", "include_owner", "category_id", ), cache_key_parts=None, item_key_name="isrc", ) def get_track_playlists_data( self, *, isrc_list: List[str] or str, markets_list: List[str] or None, search: str or None = None, playlists_ids: List[str] or None = None, include_track_count: bool = False, include_username: bool = False, include_streams_7_days: bool = False, include_followers: bool = False, streams_markets_list: List[str] or None = None, include_owner: bool or None = False, category_id: int or None = None, ) -> List[Dict[str, Any]]: """Get track playlists data. Arguments: isrc_list: Track ISRC value. markets_list: Owner markets filtering value. search: Search sting value. playlists_ids: List of playlists ids. include_track_count: Include track_count. include_username: Include username. include_streams_7_days: Include playlist streams for 7 days. include_followers: Include followers. streams_markets_list: Filter streams by market list. include_owner: Include playlist owner. category_id: BuzzCategory id to filter by. Returns: Track playlists data. """ filters = [SpotifyPlaylist.removed == 0] if search: filters.append(SpotifyPlaylist.name.ilike(f"%{search}%")) if markets_list: filters.append( func.coalesce(SpotifyPlaylist.country_code, BuzzUser.country_code, NO_COUNTRY_CODE_PLACEHOLDER).in_( markets_list ) ) if playlists_ids: filters.append(SpotifyPlaylistTrackList2.playlist_id.in_(playlists_ids)) track_subquery = ( db_session.query(SpotifyTrack2) .with_entities( SpotifyTrack2.id.label("id"), SpotifyTrack2.isrc.label("isrc"), ) .filter(SpotifyTrack2.isrc.in_(isrc_list)) .subquery() ) query_fields = [ SpotifyPlaylistTrackList2.playlist_id.label("id"), func.IF( cast(func.min(SpotifyPlaylistTrackList2.added_earliest_datetime), Date) == UNIX_EPOCH_START_DATE, None, cast(func.min(SpotifyPlaylistTrackList2.added_earliest_datetime), Date), ).label("added_date"), SpotifyPlaylist.name, func.coalesce(SpotifyPlaylist.country_code, BuzzUser.country_code, NO_COUNTRY_CODE_PLACEHOLDER).label( "country_code" ), SpotifyPlaylist.image.label("image_url"), (func.min(SpotifyPlaylistTrackList2.playlist_index) + 1).label("current_position"), track_subquery.c.id.label("track_id"), track_subquery.c.isrc.label("isrc"), SpotifyPersonalizedPlaylist.playlist_id.isnot(None).label("personalized"), cast(SpotifyPlaylist.update_timestamp, Date).label("updated_date"), ] query = ( db_session.query(SpotifyPlaylistTrackList2) .join(SpotifyPlaylist, SpotifyPlaylist.id == SpotifyPlaylistTrackList2.playlist_id) .join(track_subquery, track_subquery.c.id == SpotifyPlaylistTrackList2.track_id) .outerjoin( SpotifyPersonalizedPlaylist, SpotifyPersonalizedPlaylist.playlist_id == SpotifyPlaylistTrackList2.playlist_id, ) .outerjoin(BuzzUser, BuzzUser.user_name == SpotifyPlaylist.user_name) ) if include_owner or category_id: query = query.outerjoin(BuzzCategory, BuzzCategory.id == SpotifyPlaylist.buzz_category_id) query_fields.extend( [ SpotifyPlaylist.buzz_category_id.label("category_id"), BuzzCategory.name.label("category_name"), ] ) if category_id: filters.append(BuzzCategory.id == category_id) if include_track_count: query_fields.append(SpotifyPlaylist.track_count) if include_username: query_fields.append(SpotifyPlaylist.user_name.label("username")) if include_streams_7_days: subquery = db_session.query(func.sum(SpotifyPlaylistStreamSummary.streams_7_days)).filter( SpotifyPlaylistStreamSummary.playlist_id == SpotifyPlaylist.id ) if streams_markets_list: subquery = subquery.filter(SpotifyPlaylistStreamSummary.market.in_(streams_markets_list)) query_fields.append(func.IFNULL(subquery.as_scalar(), 0).cast(Integer).label("playlist_streams")) if include_followers: query = query.outerjoin( SpotifyPlaylistFollowers, SpotifyPlaylistFollowers.playlist_id == SpotifyPlaylist.id ) query_fields.append(func.IFNULL(SpotifyPlaylistFollowers.followers, 0).label("followers")) query = ( query.with_entities(*query_fields) .filter(*filters) .group_by(SpotifyPlaylistTrackList2.playlist_id, track_subquery.c.isrc) ) playlists_grouped_by_isrc = {} for item in query: playlist_data = item._asdict() playlist_data["image_url"] = SPOTIFY_PLAYLIST_IMAGE_URL_MASK.format(playlist_id=item.id) playlist_data["country_code"] = check_country_code_value(playlist_data["country_code"]) if not playlist_data["name"]: playlist_data["name"] = None del playlist_data["isrc"] 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) -> List[Any]: """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"]: 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.SPOTIFY_TRACK_PREVIOUS_PLAYLISTS_FULL, keys.SPOTIFY_TRACK_PREVIOUS_PLAYLISTS_FULL_TTL, ids_index=1, ids_kwargs_name=None, cache_key_kwargs_names=( "markets_list", "search", "exclude_current_playlists", "include_owner", "include_followers", "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, include_followers: bool = False, 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: Owner markets filtering value. search: Search string value. exclude_current_playlists: Exclude current playlists. include_owner: Include playlist owner. include_followers: Include playlist followers. category_id: BuzzCategory id to filter by. Returns: Track previous playlists data. """ filters = [SpotifyPlaylist.removed == 0, PlaylistIgnoredPlaylist.playlist_id.is_(None)] if search: filters.append(SpotifyPlaylist.name.ilike(f"%{search}%")) if markets_list: filters.append( func.coalesce(SpotifyPlaylist.country_code, BuzzUser.country_code, NO_COUNTRY_CODE_PLACEHOLDER).in_( markets_list ) ) track_ids_subquery = ( db_session.query(SpotifyTrack2.id.label("track_id")).filter(SpotifyTrack2.isrc.in_(isrc_list)).subquery() ) subquery = ( db_session.query( SpotifyPlaylistTrackListHistory2.playlist_id.label("playlist_id"), func.max(SpotifyPlaylistTrackListHistory2.date).label("removed_date"), SpotifyTrack2.isrc.label("isrc"), SpotifyPlaylistTrackListHistory2.track_id.label("track_id"), ) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackListHistory2.track_id) .filter(SpotifyTrack2.isrc.in_(isrc_list)) .group_by(SpotifyPlaylistTrackListHistory2.playlist_id, SpotifyTrack2.isrc) .subquery() ) query_fields = [ SpotifyPlaylist.id, SpotifyPlaylist.name, func.coalesce(SpotifyPlaylist.country_code, BuzzUser.country_code, NO_COUNTRY_CODE_PLACEHOLDER).label( "country_code" ), subquery.c.track_id, subquery.c.isrc, SpotifyPlaylist.image.label("image_url"), cast(SpotifyPlaylist.update_timestamp, Date).label("updated_date"), cast(SpotifyPlaylistTrackListHistoryDates2.timestamp, Date).label("removed_date"), SpotifyPersonalizedPlaylist.playlist_id.isnot(None).label("personalized"), func.IF( cast(SpotifyPlaylistTrackList2.added_earliest_datetime, Date) == UNIX_EPOCH_START_DATE, None, cast(SpotifyPlaylistTrackList2.added_earliest_datetime, Date), ).label("added_date"), ] query = ( db_session.query() .select_from(subquery) .join(SpotifyPlaylist, SpotifyPlaylist.id == subquery.c.playlist_id) .join( SpotifyPlaylistTrackListHistoryDates2, and_( SpotifyPlaylistTrackListHistoryDates2.playlist_id == subquery.c.playlist_id, SpotifyPlaylistTrackListHistoryDates2.date == subquery.c.removed_date, ), ) .outerjoin( PlaylistIgnoredPlaylist, and_( PlaylistIgnoredPlaylist.playlist_id == SpotifyPlaylist.id, PlaylistIgnoredPlaylist.music_service_id == MusicServices.SPOTIFY, ), ) .outerjoin(SpotifyPersonalizedPlaylist, SpotifyPersonalizedPlaylist.playlist_id == subquery.c.playlist_id) .outerjoin(BuzzUser, BuzzUser.user_name == SpotifyPlaylist.user_name) .outerjoin( SpotifyPlaylistTrackList2, and_( SpotifyPlaylist.id == SpotifyPlaylistTrackList2.playlist_id, SpotifyPlaylistTrackList2.track_id.in_(track_ids_subquery), ), ) ) if include_followers: query = query.outerjoin( SpotifyPlaylistFollowers, SpotifyPlaylistFollowers.playlist_id == SpotifyPlaylist.id ) query_fields.append(func.IFNULL(SpotifyPlaylistFollowers.followers, 0).label("followers")) if include_owner or category_id: query = query.outerjoin(BuzzCategory, BuzzCategory.id == SpotifyPlaylist.buzz_category_id) query_fields.extend( [ SpotifyPlaylist.buzz_category_id.label("category_id"), BuzzCategory.name.label("category_name"), ] ) if category_id: filters.append(BuzzCategory.id == category_id) if exclude_current_playlists: query = query.having(func.group_concat(SpotifyPlaylistTrackList2.playlist_id.distinct()).is_(None)) query = query.with_entities(*query_fields).filter(*filters).group_by(SpotifyPlaylist.id, subquery.c.isrc) playlists_grouped_by_isrc = {} for item in query: playlist_data = item._asdict() playlist_data["image_url"] = SPOTIFY_PLAYLIST_IMAGE_URL_MASK.format(playlist_id=item.id) playlist_data["country_code"] = check_country_code_value(playlist_data["country_code"]) if not playlist_data["name"]: playlist_data["name"] = None del playlist_data["isrc"] 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_playlists_trends_data( self, isrc: str or List[str], playlists_data: Dict[str, dict] ) -> Dict[str, dict]: """Get historical data for track playlists to calculate trends. Args: isrc: Track ISRC value. playlists_data (Dict[str, dict]): Playlists DB data. Returns: Dict[str, dict]: Map with trend and added date values for each track` playlist. """ if isinstance(isrc, str): isrc = [isrc] previous_date = date.today() - timedelta(days=14) playlists_ids = list(playlists_data.keys()) non_personalized_ids = list(item["id"] for item in playlists_data.values() if item["personalized"] is False) previous_positions_data = defaultdict(lambda: list()) result = defaultdict(lambda: dict(prev_position=None, changed_date=None, earliest_added_date=None)) trends_query = ( db_session.query( SpotifyPlaylistTrackListHistory2.playlist_id, (SpotifyPlaylistTrackListHistory2.playlist_index + 1).label("prev_position"), SpotifyPlaylistTrackListHistory2.date, ) .join(SpotifyTrack2, SpotifyPlaylistTrackListHistory2.track_id == SpotifyTrack2.id) .filter( SpotifyTrack2.isrc.in_(isrc), SpotifyPlaylistTrackListHistory2.playlist_id.in_(non_personalized_ids), SpotifyPlaylistTrackListHistory2.date > previous_date, ) .order_by(SpotifyPlaylistTrackListHistory2.date) ) earliest_added_dates_query = ( db_session.query( SpotifyPlaylistTrackListHistory2.playlist_id, func.min(SpotifyPlaylistTrackListHistory2.date).label("earliest_added_date"), ) .join(SpotifyTrack2, SpotifyPlaylistTrackListHistory2.track_id == SpotifyTrack2.id) .filter(SpotifyTrack2.isrc.in_(isrc), SpotifyPlaylistTrackListHistory2.playlist_id.in_(playlists_ids)) .group_by(SpotifyPlaylistTrackListHistory2.playlist_id) ) for item in trends_query: playlist_id = item.playlist_id if playlist_id and playlists_data[playlist_id]["current_position"] == item.prev_position: continue previous_positions_data[playlist_id].append(item._asdict()) for item in earliest_added_dates_query: result[item.playlist_id]["earliest_added_date"] = item.earliest_added_date for pl_id, data in previous_positions_data.items(): last_date = max(item["date"] for item in data if item["date"]) if not last_date: continue min_prev_position = min(item["prev_position"] for item in data if item["date"] == last_date) result[pl_id].update({"prev_position": min_prev_position, "changed_date": last_date}) return result def extend_playlists_data_with_trends( self, playlists_data: Dict[str, dict], trends_data: Dict[str, dict] ) -> List[dict]: """Add trends data to track playlists list. Args: playlists_data (Dict[str, dict]): Track playlists DB data. trends_data (Dict[str, dict]): Track playlists trends and added_date data. Returns: List[dict]: List of track playlists. """ today = date.today() for pl_id, _data in playlists_data.items(): historical_data = trends_data.get(pl_id, {}) prev_position = historical_data.get("prev_position") personalized = _data["personalized"] added_date = historical_data.get("earliest_added_date") or _data["added_date"] is_new = not personalized and not prev_position and added_date == today if personalized: trend = None elif not prev_position: trend = None if is_new else ZERO_TREND else: trend = _data["current_position"] - prev_position playlists_data[pl_id]["is_new"] = is_new playlists_data[pl_id]["trend"] = trend playlists_data[pl_id]["position_change_date"] = ( None if personalized else historical_data.get("changed_date") ) playlists_data[pl_id]["added_date"] = added_date return list(playlists_data.values()) def _get_track_in_playlist_db_data(self, playlist_id: str, isrc: str) -> Dict: """Retrieve DB stored data for track in specific playlist. Args: playlist_id (str): Playlist identifier. isrc (str): Track ISRC value. Returns: Dict: Mapped track in playlist data. """ filters = [ SpotifyTrack2.isrc == isrc, SpotifyPlaylist.removed == 0, SpotifyPlaylistTrackList2.playlist_id == playlist_id, ] earliest_added_date_subquery = ( db_session.query( func.IF( func.min(SpotifyPlaylistTrackListHistory2.date) == UNIX_EPOCH_START_DATE, None, func.min(SpotifyPlaylistTrackListHistory2.date), ).label("earliest_added_date"), SpotifyPlaylistTrackListHistory2.playlist_id.label("playlist_id"), ) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackListHistory2.track_id) .filter(SpotifyTrack2.isrc == isrc, SpotifyPlaylistTrackListHistory2.playlist_id == playlist_id) .group_by(SpotifyPlaylistTrackListHistory2.playlist_id) .subquery() ) track_in_playlist = ( db_session.query( SpotifyPlaylistTrackList2.playlist_id.label("playlist_id"), earliest_added_date_subquery.c.earliest_added_date, func.IF( cast(func.min(SpotifyPlaylistTrackList2.added_earliest_datetime), Date) == UNIX_EPOCH_START_DATE, None, cast(func.min(SpotifyPlaylistTrackList2.added_earliest_datetime), Date), ).label("added_date"), SpotifyPlaylist.name.label("playlist_name"), SpotifyPlaylist.country_code, SpotifyPlaylist.image.label("playlist_image"), (func.min(SpotifyPlaylistTrackList2.playlist_index) + 1).label("current_position"), SpotifyTrack2.id.label("track_id"), SpotifyTrack2.name.label("track_name"), SpotifyTrack2.isrc, SpotifyPersonalizedPlaylist.playlist_id.isnot(None).label("personalized"), cast(SpotifyPlaylist.update_timestamp, Date).label("changed_date"), ) .join(SpotifyPlaylist, SpotifyPlaylist.id == SpotifyPlaylistTrackList2.playlist_id) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackList2.track_id) .outerjoin( earliest_added_date_subquery, earliest_added_date_subquery.c.playlist_id == SpotifyPlaylistTrackList2.playlist_id, ) .outerjoin( SpotifyPersonalizedPlaylist, SpotifyPersonalizedPlaylist.playlist_id == SpotifyPlaylistTrackList2.playlist_id, ) .filter(*filters) .group_by(SpotifyPlaylistTrackList2.playlist_id) .first() ) return track_in_playlist._asdict() if track_in_playlist else {} def _get_track_in_past_playlist_db_data(self, playlist_id: str, isrc: str) -> Dict: """Retrieve DB stored data for track in specific past playlist. Args: playlist_id (str): Playlist identifier. isrc (str): Track ISRC value. Returns: Dict: Mapped track in playlist data. """ track_ids_subquery = ( db_session.query(SpotifyTrack2.id.label("track_id")).filter(SpotifyTrack2.isrc == isrc).subquery() ) earliest_added_date_subquery = ( db_session.query( func.IF( func.min(SpotifyPlaylistTrackListHistory2.date) == UNIX_EPOCH_START_DATE, None, func.min(SpotifyPlaylistTrackListHistory2.date), ).label("earliest_added_date"), SpotifyPlaylistTrackListHistory2.playlist_id.label("playlist_id"), ) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackListHistory2.track_id) .filter(SpotifyTrack2.isrc == isrc, SpotifyPlaylistTrackListHistory2.playlist_id == playlist_id) .group_by(SpotifyPlaylistTrackListHistory2.playlist_id) .subquery() ) filters = [ SpotifyTrack2.isrc == isrc, SpotifyPlaylist.removed == 0, SpotifyPlaylistTrackListHistory2.playlist_id == playlist_id, ] track_in_playlist = ( db_session.query(SpotifyPlaylistTrackListHistory2) .with_entities( SpotifyPlaylistTrackListHistory2.playlist_id.label("playlist_id"), earliest_added_date_subquery.c.earliest_added_date, SpotifyPlaylist.name.label("playlist_name"), SpotifyPlaylist.country_code, SpotifyPlaylist.image.label("playlist_image"), SpotifyTrack2.id.label("track_id"), SpotifyTrack2.name.label("track_name"), SpotifyTrack2.isrc, SpotifyPersonalizedPlaylist.playlist_id.isnot(None).label("personalized"), cast(SpotifyPlaylist.update_timestamp, Date).label("changed_date"), func.max(SpotifyPlaylistTrackListHistory2.date).label("removed_date"), ) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackListHistory2.track_id) .outerjoin( earliest_added_date_subquery, earliest_added_date_subquery.c.playlist_id == SpotifyPlaylistTrackListHistory2.playlist_id, ) .join(SpotifyPlaylist, SpotifyPlaylist.id == SpotifyPlaylistTrackListHistory2.playlist_id) .outerjoin( SpotifyPersonalizedPlaylist, SpotifyPersonalizedPlaylist.playlist_id == SpotifyPlaylistTrackListHistory2.playlist_id, ) .outerjoin( SpotifyPlaylistTrackList2, and_( SpotifyPlaylistTrackList2.playlist_id == playlist_id, SpotifyPlaylistTrackList2.track_id.in_(track_ids_subquery), ), ) .filter(*filters) .group_by(SpotifyPlaylistTrackListHistory2.playlist_id) .having(func.group_concat(SpotifyPlaylistTrackList2.playlist_id.distinct()).is_(None)) .first() ) return track_in_playlist._asdict() if track_in_playlist else {} def get_track_data_in_playlist(self, isrc: str, playlist_id: str, playlist_type: str) -> Dict: """Retrieve single track data in Spotify playlist from database. Args: playlist_id (str): Playlist identifier. isrc (str): Track ISRC value. playlist_type (str): Playlist type (current or past). Returns: Dict: Track in playlist data. """ if playlist_type == CURRENT_PLACEMENT_TYPE: return self._get_track_in_playlist_db_data(playlist_id, isrc) or {} elif playlist_type == PAST_PLACEMENT_TYPE: return self._get_track_in_past_playlist_db_data(playlist_id, isrc) or {} return {} def get_track_in_playlists_position_changes(self, isrc: str, playlist_id: str) -> Dict: """Retrieve from DB previous tracks' positions in playlist Args: playlist_id (str): Playlist identifier. isrc (str): Track ISRC value. Returns: Dict: Track in playlist data. """ return self._get_tracks_positions_changes_in_playlist(playlist_id, isrc) or {} def get_positions_changes_in_playlists(self, playlist_ids: Iterable[str], isrc: str) -> Dict: """Retrieve track previous and current position data for a list of playlists. Args: playlist_ids (Iterable[str]): Container with playlist identifiers. isrc (str): Track ISRC value. Returns: Dict: Track positions data for list of playlists. """ result = { pl_id: {"current_position": None, "previous_position": None, "changed_date": None} for pl_id in playlist_ids if pl_id } current_positions_query = ( db_session.query( SpotifyPlaylistTrackList2.playlist_id.label("playlist_id"), SpotifyPlaylistTrackList2.track_id.label("track_id"), (func.min(SpotifyPlaylistTrackList2.playlist_index) + 1).label("current_position"), cast(SpotifyPlaylist.update_timestamp, Date).label("updated_date"), ) .join(SpotifyTrack2, SpotifyPlaylistTrackList2.track_id == SpotifyTrack2.id) .join(SpotifyPlaylist, SpotifyPlaylist.id == SpotifyPlaylistTrackList2.playlist_id) .filter(SpotifyPlaylistTrackList2.playlist_id.in_(playlist_ids), SpotifyTrack2.isrc == isrc) .group_by(SpotifyPlaylistTrackList2.playlist_id) ) previous_date = date.today() - timedelta(days=14) prev_positions_query = ( db_session.query( SpotifyPlaylistTrackListHistory2.playlist_id, (SpotifyPlaylistTrackListHistory2.playlist_index + 1).label("prev_position"), SpotifyPlaylistTrackListHistory2.date, ) .join(SpotifyTrack2, SpotifyPlaylistTrackListHistory2.track_id == SpotifyTrack2.id) .filter( SpotifyTrack2.isrc == isrc, SpotifyPlaylistTrackListHistory2.playlist_id.in_(playlist_ids), SpotifyPlaylistTrackListHistory2.date > previous_date, ) .order_by(SpotifyPlaylistTrackListHistory2.date) ) for q in current_positions_query: result[q.playlist_id]["current_position"] = q.current_position result[q.playlist_id]["changed_date"] = q.updated_date previous_positions_data = defaultdict(lambda: list()) for item in prev_positions_query: playlist_id = item.playlist_id if playlist_id and result[playlist_id]["current_position"] == item.prev_position: continue previous_positions_data[playlist_id].append(item._asdict()) for pl_id, data in previous_positions_data.items(): last_date = max(item["date"] for item in data if item["date"]) if not last_date: continue min_prev_position = min(item["prev_position"] for item in data if item["date"] == last_date) result[pl_id].update({"previous_position": min_prev_position}) return result def get_personalized_playlists(self, playlist_ids: Iterable[str] = None) -> Iterable[str]: """Get what playlists are personalized and what not. Args: playlist_ids (Iterable[str]): Set of playlist ID. Returns: Iterable[str]: Personalized playlist ID. """ result = self.get_all_personalized_playlists() if playlist_ids is not None: result = list(set(playlist_ids) & set(self.get_all_personalized_playlists())) return result def get_personalized_playlist_tracklist( self, playlist_id: str, market: str or None = None, image_size: int or None = None, hydrate: bool = False ) -> List[dict]: """Retrieve personalized playlist data from vendor and combine it with DB data. Args: playlist_id (str): Playlist Identifier. market (str or None): Market value. image_size (int or None): Track cover image size. hydrate (bool): Populate objects with domain data Returns: List[str]: List with playlist placements tracks. """ tracklist, tracks_keys = self._get_personalized_playlist_tracklist_data_from_db(playlist_id) if not hydrate: return tracklist isrc_dates = self.get_playlist_tracks_dates(playlist_id) tracks_images = {} if market and image_size: tracks_images = { item["id"]: item["image_url"] for item in clients.vendor.get_tracks_images( tracks_keys["ids"], vendor=VENDOR_SPOTIFY, market=market, size=image_size ) } _tracklist = [] for track_data in tracklist: isrc = track_data["isrc"] added_date = isrc_dates.get(isrc, {}).get("earliest_added_date") or track_data["added_date"] track_data["added_date"] = added_date track_data["image_url"] = tracks_images.get(track_data["track_id"]) track_data["previous_position"] = None track_data["position_change_date"] = None track_data["trend"] = None track_data["is_new"] = False _tracklist.append(track_data) return _tracklist @staticmethod def get_playlist_tracks_dates(playlist_id: str) -> Dict: """Retrieve from DB playlist's tracks added and update dates values. Args: playlist_id (str): Playlist identifier. Returns: Dict: Map with added dates values for each track in playlist. """ historical_subquery = ( db_session.query( func.min(SpotifyPlaylistTrackListHistory2.date).label("earliest_added_date"), func.max(SpotifyPlaylistTrackListHistory2.date).label("latest_added_date"), cast(SpotifyPlaylist.update_timestamp, Date).label("update_date"), SpotifyPlaylistTrackListHistory2.track_id.label("track_id"), ) .join(SpotifyPlaylist, SpotifyPlaylistTrackListHistory2.playlist_id == SpotifyPlaylist.id) .filter(SpotifyPlaylistTrackListHistory2.playlist_id == playlist_id) .group_by(SpotifyPlaylistTrackListHistory2.track_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"), SpotifyTrack2.isrc.label("isrc"), ) .join(SpotifyTrack2, SpotifyTrack2.id == historical_subquery.c.track_id) .group_by(SpotifyTrack2.isrc) .having( func.max(historical_subquery.c.latest_added_date).label("latest_added_date") == historical_subquery.c.update_date ) ) result = {} for row in query: data = row._asdict() isrc = data.pop("isrc", None) if not isrc: continue result[isrc] = data return result @staticmethod def _get_personalized_playlist_tracklist_data_from_db(playlist_id: str) -> Tuple[List[Dict], Dict]: """Retrieve all required personalized playlist placements data from DB. Args: playlist_id (str): Playlist identifier. Returns: Tuple[List[Dict], Dict]: Tuple with personalized playlist data. """ query = ( ( db_session.query( SpotifyPlaylistTrackList2.track_id, (SpotifyPlaylistTrackList2.playlist_index + 1).label("current_position"), func.IF( cast(SpotifyPlaylistTrackList2.added_earliest_datetime, Date) == UNIX_EPOCH_START_DATE, None, cast(SpotifyPlaylistTrackList2.added_earliest_datetime, Date), ).label("added_date"), SpotifyTrack2.name, SpotifyTrack2.isrc, SpotifyTrack2.id.label("track_id"), SpotifyArtist.name.label("artist_name"), SpotifyTrackAlbum.album_id, ) .join(SpotifyPlaylist, SpotifyPlaylistTrackList2.playlist_id == SpotifyPlaylist.id) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackList2.track_id) .join(SpotifyTrackArtist, SpotifyTrackArtist.track_id == SpotifyTrack2.id) .join(SpotifyArtist, SpotifyArtist.id == SpotifyTrackArtist.artist_id) .outerjoin(SpotifyTrackAlbum, SpotifyTrackAlbum.track_id == SpotifyTrack2.id) .filter(SpotifyPlaylistTrackList2.playlist_id == playlist_id, SpotifyPlaylist.removed == 0) .order_by(SpotifyPlaylistTrackList2.playlist_index, SpotifyTrackArtist.order) ) if playlist_id else [] ) tracklist, tracks_keys = combine_artists_by_track(data=query.all()) return tracklist, tracks_keys def get_personalized_playlist_tracklist_data(self, playlist_id: str) -> Tuple[List[Dict], Dict]: """Retrieve all required personalized playlist placements data from DB. Args: playlist_id (str): Playlist identifier. Returns: Tuple[List[Dict], Dict]: Tuple with personalized playlist data. """ return self._get_personalized_playlist_tracklist_data_from_db(playlist_id) @staticmethod @cache_value(keys.TRACKS_POSITIONS_CHANGES_IN_PLAYLIST, keys.TRACKS_POSITIONS_CHANGES_IN_PLAYLIST_TTL) def _get_tracks_positions_changes_in_playlist(playlist_id: str, isrc: Optional[str]) -> Dict: """Retrieve from DB previous tracks' positions in playlist. Args: playlist_id (str): Playlist identifier. isrc_positions (Dict): Map with playlist`s tracks ISRC and positions values. Returns: Dict: Positions changes data for each track in playlist. """ today = date.today() previous_date = (today - timedelta(days=14)).strftime("%Y-%m-%d") positions = ( db_session.query( SpotifyTrack2.isrc.label("isrc"), (SpotifyPlaylistTrackListHistory2.playlist_index + 1).label("previous_position"), func.max(SpotifyPlaylistTrackListHistory2.date).label("changed_date"), ) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackListHistory2.track_id) .filter( SpotifyPlaylistTrackListHistory2.playlist_id == playlist_id, SpotifyPlaylistTrackListHistory2.date > previous_date, SpotifyPlaylistTrackListHistory2.date < today.strftime("%Y-%m-%d"), ) .group_by(SpotifyPlaylistTrackListHistory2.track_id, SpotifyPlaylistTrackListHistory2.playlist_index) .order_by(SpotifyPlaylistTrackListHistory2.date.desc()) ) if isrc: positions = positions.filter(SpotifyTrack2.isrc == isrc) result = {} for row in positions: data = row._asdict() isrc = data["isrc"] if isrc not in result: result[isrc] = {} prev_position, changed_date = int(data["previous_position"]), data["changed_date"] if prev_position in result[isrc] and changed_date < result[isrc][prev_position]: continue result[isrc][prev_position] = changed_date return result @staticmethod @cache_value(keys.PERSONALIZED_PLAYLISTS_IDS, keys.PERSONALIZED_PLAYLISTS_IDS_TTL) def get_all_personalized_playlists() -> List[str]: """Return list of all stored personalized playlists id.""" return [p[0] for p in db_session.query(SpotifyPersonalizedPlaylist.playlist_id)] def _get_tracks_playlists_current_query( self, isrc_list: Iterable[str], buzz_categories: Iterable[int], type_str: str ) -> Query: """Get current tracks playlists count. Args: isrc_list: Set of tracks ISRC. buzz_categories: Buzz category ID (like Spotify or Sony) list. type_str: Query type string. Returns: ISRC to buzz category ID to list of playlist ID and dates. """ return ( db_session.query( SpotifyTrack2.isrc.label("isrc"), SpotifyPlaylist.buzz_category_id.label("buzz_category_id"), func.count(distinct(SpotifyPlaylistTrackList2.playlist_id)).label("total"), literal(type_str).label("type"), ) .select_from(SpotifyPlaylistTrackList2) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackList2.track_id) .join(SpotifyPlaylist, SpotifyPlaylist.id == SpotifyPlaylistTrackList2.playlist_id) .outerjoin( PlaylistIgnoredPlaylist, and_( PlaylistIgnoredPlaylist.playlist_id == SpotifyPlaylist.uri, PlaylistIgnoredPlaylist.music_service_id == MusicServices.SPOTIFY, ), ) .filter(SpotifyTrack2.isrc.in_(isrc_list)) .filter(SpotifyPlaylist.removed == 0) .filter(PlaylistIgnoredPlaylist.playlist_id.is_(None)) .filter(SpotifyPlaylist.buzz_category_id.in_(buzz_categories)) .group_by(SpotifyTrack2.isrc, SpotifyPlaylist.buzz_category_id) ) def _get_tracks_playlists_added_query( self, isrc_list: Iterable[str], buzz_categories: Iterable[int], history_date: date, date_index: int, type_prefix: str, ) -> Query: """Get in what playlists selected tracks were added. Args: isrc_list: Set of tracks ISRC. buzz_categories: Buzz category ID (like Spotify or Sony) list. history_date: History date to compare with. date_index: Index in dates list to build type column. type_prefix: Prefix for type column. Returns: Query with ISRC, buzz category ID, playlists count and query type. """ return ( db_session.query( SpotifyTrack2.isrc.label("isrc"), SpotifyPlaylist.buzz_category_id.label("buzz_category_id"), func.count(distinct(SpotifyPlaylistTrackList2.playlist_id)).label("total"), literal(f"{type_prefix}{date_index}").label("type"), ) .select_from(SpotifyPlaylistTrackList2) .join(SpotifyTrack2, SpotifyTrack2.id == SpotifyPlaylistTrackList2.track_id) .join(SpotifyPlaylist, SpotifyPlaylist.id == SpotifyPlaylistTrackList2.playlist_id) .outerjoin( PlaylistIgnoredPlaylist, and_( PlaylistIgnoredPlaylist.playlist_id == SpotifyPlaylist.uri, PlaylistIgnoredPlaylist.music_service_id == MusicServices.SPOTIFY, ), ) .join( SpotifyPlaylistTrackListHistoryDates2, and_( SpotifyPlaylistTrackListHistoryDates2.playlist_id == SpotifyPlaylistTrackList2.playlist_id, SpotifyPlaylistTrackListHistoryDates2.date == history_date, ), ) .outerjoin( SpotifyTrackPlaylistLastWeekHistory, and_( SpotifyTrackPlaylistLastWeekHistory.isrc == SpotifyTrack2.isrc, SpotifyTrackPlaylistLastWeekHistory.playlist_id == SpotifyPlaylistTrackList2.playlist_id, SpotifyTrackPlaylistLastWeekHistory.date == history_date, ), ) .filter(SpotifyTrack2.isrc.in_(isrc_list)) .filter(SpotifyPlaylist.removed == 0) .filter(PlaylistIgnoredPlaylist.playlist_id.is_(None)) .filter(SpotifyPlaylist.buzz_category_id.in_(buzz_categories)) .filter(SpotifyTrackPlaylistLastWeekHistory.date.is_(None)) .group_by(SpotifyTrack2.isrc, SpotifyPlaylist.buzz_category_id) ) def _get_tracks_playlists_removed_query( self, isrc_list: Iterable[str], buzz_categories: Iterable[int], history_date: date, date_index: int, type_prefix: str, ) -> Query: """Get from what playlists selected tracks were removed. Args: isrc_list: Set of tracks ISRC. buzz_categories: Buzz category (like Spotify or Sony) list. history_date: History date to compare with. date_index: Index in dates list to build type column. type_prefix: Prefix for type column. Returns: Query with ISRC, buzz category ID, removed playlist count and query type. """ sub_query = ( db_session.query(SpotifyTrack2) .join(SpotifyPlaylistTrackList2, SpotifyPlaylistTrackList2.track_id == SpotifyTrack2.id) .filter(SpotifyTrack2.isrc == SpotifyTrackPlaylistLastWeekHistory.isrc) .filter(SpotifyPlaylistTrackList2.playlist_id == SpotifyTrackPlaylistLastWeekHistory.playlist_id) ) return ( db_session.query( SpotifyTrackPlaylistLastWeekHistory.isrc.label("isrc"), SpotifyPlaylist.buzz_category_id.label("buzz_category_id"), func.count(distinct(SpotifyTrackPlaylistLastWeekHistory.playlist_id)).label("total"), literal(f"{type_prefix}{date_index}").label("type"), ) .select_from(SpotifyTrackPlaylistLastWeekHistory) .join(SpotifyPlaylist, SpotifyPlaylist.id == SpotifyTrackPlaylistLastWeekHistory.playlist_id) .outerjoin( PlaylistIgnoredPlaylist, and_( PlaylistIgnoredPlaylist.playlist_id == SpotifyPlaylist.uri, PlaylistIgnoredPlaylist.music_service_id == MusicServices.SPOTIFY, ), ) .filter(SpotifyTrackPlaylistLastWeekHistory.isrc.in_(isrc_list)) .filter(SpotifyTrackPlaylistLastWeekHistory.date == history_date) .filter(SpotifyPlaylist.removed == 0) .filter(PlaylistIgnoredPlaylist.playlist_id.is_(None)) .filter(SpotifyPlaylist.buzz_category_id.in_(buzz_categories)) .filter(~sub_query.exists()) .group_by(SpotifyTrackPlaylistLastWeekHistory.isrc, SpotifyPlaylist.buzz_category_id) ) @cache_requests( keys.SPOTIFY_TRACK_PLAYLISTS_STATS, keys.SPOTIFY_TRACK_PLAYLISTS_STATS_TTL, cache_key_parts=(2, 3), item_key_name="isrc", ) def get_tracks_playlists_stats( self, isrc_list: List[str], buzz_categories: List[int], history_dates: List[date] ) -> List[Dict]: """Get tracks playlists statistics. Args: isrc_list: List of tracks ISRC. history_dates: Dates to compare with. buzz_categories: Playlist categories like Sony or Spotify. Returns: Tracks stats about playlists. """ type_current = "c" type_added = "a" type_removed = "r" queries = [self._get_tracks_playlists_current_query(isrc_list, buzz_categories, type_current)] for index, history_date in enumerate(history_dates): queries.append( self._get_tracks_playlists_added_query(isrc_list, buzz_categories, history_date, index, type_added) ) queries.append( self._get_tracks_playlists_removed_query(isrc_list, buzz_categories, history_date, index, type_removed) ) current_stats = defaultdict(lambda: defaultdict(lambda: 0)) history_stats = defaultdict(lambda: defaultdict(lambda: defaultdict(lambda: {"added": 0, "removed": 0}))) query = db_session.query(union(*queries).alias()) for row in query.all(): if row.type == type_current: current_stats[row.isrc][row.buzz_category_id] = row.total else: index = int(row.type.replace(type_added, "").replace(type_removed, "")) type_str = "added" if type_added in row.type else "removed" history_stats[row.isrc][row.buzz_category_id][history_dates[index]][type_str] = row.total result = [] for isrc in isrc_list: current_by_isrc = current_stats[isrc] history_by_isrc = history_stats[isrc] current_part = [] history_part = [] for category_id in buzz_categories: history_by_isrc_category = history_by_isrc[category_id] current_part.append({"category_id": category_id, "playlist_count": current_by_isrc[category_id]}) for history_date in history_dates: current_item = {"date": history_date, "category_id": category_id} current_item.update(history_by_isrc_category[history_date]) history_part.append(current_item) result.append({"isrc": isrc, "current": current_part, "history": history_part}) return result def extend_playlists_with_delphi_streams_data( self, playlists_data: List[dict], streams_data: dict, keep_empty_streams: bool = True ) -> List[dict]: """Extend playlists map with streams data from delphi. Args: playlists_data (List[dict]): Playlists list with track playlists db data. streams_data (dict): Track playlists streams data. keep_empty_streams (bool): Keep or skip records with empty current_streams values. Returns: List[dict]: List of playlists with streams data. """ streams_map = defaultdict(lambda: {"current_streams": 0, "prev_streams": 0}) for item in streams_data.get("items", []): pl_id = item["id"] current_streams, prev_streams = item["current_streams"], item["prev_streams"] streams_map[pl_id]["current_streams"] += current_streams or 0 streams_map[pl_id]["prev_streams"] += prev_streams or 0 playlists_items = [] for playlist_data in playlists_data: current_streams, prev_streams = None, None pl_id = playlist_data["id"] if pl_id in streams_map: current_streams, prev_streams = ( streams_map[pl_id]["current_streams"], streams_map[pl_id]["prev_streams"], ) item = { "current_streams": current_streams if current_streams else None, "prev_streams": prev_streams if prev_streams else None, **playlist_data, } if not current_streams and keep_empty_streams is False: continue playlists_items.append(item) return playlists_items def playlists_search_handler(self, query: str, size: int) -> List[dict]: """Search Spotify playlists in ElasticSearch and format retrieved data. Args: query (str): Search string. size (int): Request limit size. Returns: List[dict]: List with found playlists data. """ es_body = self.get_playlists_search_es_body( query, factor_field=ElasticPlaylistsVendorsData.FACTOR_FIELDS[VENDOR_SPOTIFY], field_name="name", size=size ) hits = app.elasticsearch.search( index=app.config["ELASTICSEARCH_PLAYLISTS_INDEX"], doc_type=ElasticPlaylistsVendorsData.DOC_TYPES[VENDOR_SPOTIFY], body=es_body, ).get("hits", {}) result, playlists_ids = {}, [] for item in hits.get("hits", []): item_source = item.get("_source") if not item_source: continue pl_id = item_source["playlistUri"].replace(SPOTIFY_PLAYLIST_URI_PREFIX, "") playlists_ids.append(pl_id) result[pl_id] = { "owner": item_source.get("userDisplayName") or item_source.get("user"), "image_url": SPOTIFY_PLAYLIST_IMAGE_URL_MASK.format(playlist_id=pl_id) if pl_id else None, "id": pl_id, "country_code": item_source.get("countryCode"), "personalized": False, "vendor": VENDOR_SPOTIFY, "followers": item_source.get("followers"), "name": item_source.get("name"), } personalized_playlists = self.get_personalized_playlists(playlists_ids) for pl_id in personalized_playlists: result[pl_id]["personalized"] = True return list(result.values()) def get_playlists_markets(self, playlist_ids: List[str]) -> List[dict]: """Returns playlists' owner market 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"] result[pl_id] = {"market": markets_data[0] if markets_data else None, "id": pl_id} return list(result.values()) @cache_requests(keys.SPOTIFY_PLAYLIST_MARKETS, keys.SPOTIFY_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(SpotifyPlaylist.country_code).label("code"), ApolloMarket.id, ApolloMarket.name, ApolloMarket.full_name, SpotifyPlaylist.id.label("playlist_id"), ) .join(ApolloMarket, ApolloMarket.code == SpotifyPlaylist.country_code) .filter(SpotifyPlaylist.id.in_(playlist_ids), ApolloMarket.is_primary.is_(True)) .group_by(SpotifyPlaylist.id, SpotifyPlaylist.country_code) ) 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], include_owner_data: bool or None = False) -> List[dict]: """Returns playlists' stored in DB data. Args: playlist_ids (List[str]): List of playlists' identifiers. include_owner_data bool: Add owner name and market code to response. Returns: List[dict]: List that contains DB stored playlists data. """ return self._get_playlists_data(playlist_ids, include_owner_data=include_owner_data) @cache_requests( keys.SPOTIFY_PLAYLIST_DATA, keys.SPOTIFY_PLAYLIST_DATA_TTL, cache_key_kwargs_names=("include_owner_data",), cache_key_parts=None, item_key_name="id", ) def _get_playlists_data(self, playlist_ids: List[str], include_owner_data: bool = False) -> List[dict]: """Inner method that query playlists' related data from DB and cached it. Args: playlist_ids (List[str]): List of playlists' identifiers. include_owner_data bool: Add owner name and market code to response. Returns: List[dict]: List of playlists' data from DB. """ query_fields = [ SpotifyPlaylist.id, SpotifyPlaylist.name, SpotifyPlaylist.followers, SpotifyPersonalizedPlaylist.playlist_id.isnot(None).label("personalized"), cast(SpotifyPlaylist.update_timestamp, Date).label("updated_date"), ] playlists_query = db_session.query(SpotifyPlaylist).outerjoin( SpotifyPersonalizedPlaylist, SpotifyPersonalizedPlaylist.playlist_id == SpotifyPlaylist.id ) if include_owner_data: query_fields.extend( [ BuzzUser.display_name.label("curator_name"), func.coalesce( SpotifyPlaylist.country_code, BuzzUser.country_code, NO_COUNTRY_CODE_PLACEHOLDER ).label("country_code"), ] ) playlists_query = playlists_query.outerjoin(BuzzUser, BuzzUser.user_name == SpotifyPlaylist.user_name) else: query_fields.extend( [ SpotifyPlaylist.user_name.label("curator_name"), func.lower(SpotifyPlaylist.country_code).label("country_code"), ] ) playlists_query = ( playlists_query.with_entities(*query_fields) .filter(SpotifyPlaylist.id.in_(playlist_ids)) .group_by(SpotifyPlaylist.id) ) result = [] for item in playlists_query: image_url = SPOTIFY_PLAYLIST_IMAGE_URL_MASK.format(playlist_id=item.id) item_data = {**item._asdict(), "image_url": image_url} result.append(item_data) return result @cache_requests( cache_key_template=keys.SPOTIFY_PLAYLIST_LISTENERS_GRAPH, cache_ttl=keys.SPOTIFY_PLAYLIST_LISTENERS_GRAPH_TTL, cache_key_parts=(2, 3), ) def get_playlists_listeners_graph_items( self, playlist_ids: List[str], markets: List[str], date_range: DateRange ) -> List: """ Retrieve and group by pair of playlist id and market listeners analytics info from DB. Args: playlist_ids (List[str]): List of playlists identifiers. markets (List[str]): List of markets. date_range (DateRange): Date range object. Returns: List: List with coordinates for each pair of playlist id and market. """ legacy_uri_query = db_session.query( SpotifyPlaylistLegacyUri.playlist_id, SpotifyPlaylistLegacyUri.playlist_uri ).filter(SpotifyPlaylistLegacyUri.playlist_id.in_(playlist_ids)) playlists_uri_map = {} for item in legacy_uri_query: playlists_uri_map[item.playlist_uri] = item.playlist_id all_playlists_pk = [*playlist_ids, *playlists_uri_map.keys()] analytics_query = ( db_session.query( SpotifyAnalyticsPlaylistStreamInfo.playlist_uri, SpotifyAnalyticsPlaylistStreamInfo.listeners, SpotifyAnalyticsPlaylistStreamInfo.date, SpotifyAnalyticsPlaylistStreamInfo.market, ) .filter( SpotifyAnalyticsPlaylistStreamInfo.account == ReleaseTypes.SONY, SpotifyAnalyticsPlaylistStreamInfo.date.between(date_range.start, date_range.end), SpotifyAnalyticsPlaylistStreamInfo.playlist_uri.in_(all_playlists_pk), SpotifyAnalyticsPlaylistStreamInfo.market.in_(markets), ) .group_by( SpotifyAnalyticsPlaylistStreamInfo.playlist_uri, SpotifyAnalyticsPlaylistStreamInfo.market, SpotifyAnalyticsPlaylistStreamInfo.date, ) ) result_dict = {} for item in analytics_query: _playlist_id = playlists_uri_map.get(item.playlist_uri) or item.playlist_uri if _playlist_id not in result_dict: result_dict[_playlist_id] = defaultdict(lambda: defaultdict(int)) result_dict[_playlist_id][item.market][date_to_timestamp(item.date)] += item.listeners if SPOTIFY_MARKET_GLOBAL not in markets: return [{"id": pl_id, "markets": _markets} for pl_id, _markets in result_dict.items()] global_analytics_query = ( db_session.query( SpotifyAnalyticsPlaylistStreamInfo.playlist_uri, func.sum(SpotifyAnalyticsPlaylistStreamInfo.listeners).label("listeners_sum"), SpotifyAnalyticsPlaylistStreamInfo.date, ) .filter( SpotifyAnalyticsPlaylistStreamInfo.account == ReleaseTypes.SONY, SpotifyAnalyticsPlaylistStreamInfo.date.between(date_range.start, date_range.end), SpotifyAnalyticsPlaylistStreamInfo.playlist_uri.in_(all_playlists_pk), ) .group_by(SpotifyAnalyticsPlaylistStreamInfo.playlist_uri, SpotifyAnalyticsPlaylistStreamInfo.date) ) for item in global_analytics_query: _playlist_id = playlists_uri_map.get(item.playlist_uri) or item.playlist_uri if _playlist_id not in result_dict: result_dict[_playlist_id] = defaultdict(lambda: defaultdict(int)) result_dict[_playlist_id][SPOTIFY_MARKET_GLOBAL][date_to_timestamp(item.date)] += int(item.listeners_sum) return [{"id": pl_id, "markets": _markets} for pl_id, _markets in result_dict.items()] @cache_value( keys.SPOTIFY_PLAYLIST_LISTENERS_RANGE, keys.SPOTIFY_PLAYLIST_LISTENERS_RANGE_TTL, key_args_indexes=(1, 2, 3, 4) ) def get_playlist_date_ranges_listeners( self, playlist_id: str, market: str, current_range: DateRange, prev_range: DateRange ) -> Dict[str, int]: """ Retrieve and count playlists listeners in current and previous date ranges. Args: playlist_id (str): Playlist identifier. market (str): Market value. current_range (DateRange): Current date range values. prev_range (DateRange): Previous date range values. Returns: Dict[str, int]: Map with listeners data for current and previous date ranges. """ legacy_uri_query = db_session.query( SpotifyPlaylistLegacyUri.playlist_id, SpotifyPlaylistLegacyUri.playlist_uri ).filter(SpotifyPlaylistLegacyUri.playlist_id == playlist_id) playlist_ids = [playlist_id] for item in legacy_uri_query: playlist_ids.append(item.playlist_uri) filters_list = [ SpotifyAnalyticsPlaylistStreamInfo.account == ReleaseTypes.SONY, SpotifyAnalyticsPlaylistStreamInfo.playlist_uri.in_(playlist_ids), ] if market != SPOTIFY_MARKET_GLOBAL: filters_list.append(SpotifyAnalyticsPlaylistStreamInfo.market == market) queries = [] for date_range in (current_range, prev_range): queries.append( ( db_session.query( SpotifyAnalyticsPlaylistStreamInfo.playlist_uri, SpotifyAnalyticsPlaylistStreamInfo.date, func.sum(SpotifyAnalyticsPlaylistStreamInfo.listeners).label("listeners_sum"), ) .filter( SpotifyAnalyticsPlaylistStreamInfo.date.between(date_range.start, date_range.end), *filters_list ) .group_by(SpotifyAnalyticsPlaylistStreamInfo.playlist_uri) ) ) current_range_query, prev_range_query = queries current_listeners, prev_listeners = None, None for item in current_range_query: current_listeners = (current_listeners or 0) + item.listeners_sum for item in prev_range_query: prev_listeners = (prev_listeners or 0) + item.listeners_sum return {"current_range": current_listeners, "prev_range": prev_listeners} @cache_value( keys.SPOTIFY_PLAYLIST_FOLLOWERS_GRAPH, keys.SPOTIFY_PLAYLIST_FOLLOWERS_GRAPH_TTL, key_args_indexes=(1, 2) ) def get_playlists_followers_graph_items( self, playlist_ids: List[str], date_range: DateRange ) -> List[Dict[str, dict]]: """ Retrieve and group by playlist id followers analytics info from DB. All calculations are for global. Args: playlist_ids (List[str]): List of playlists identifiers. date_range (DateRange): Date range object. Returns: List[Dict[str, dict]]: List of dictionaries with coordinates for each playlist id. """ items = defaultdict(lambda: defaultdict(int)) result = [] query = ( db_session.query( SpotifyPlaylistFollowersHistory.playlist_id, func.sum(SpotifyPlaylistFollowersHistory.followers).label("followers"), SpotifyPlaylistFollowersHistory.date, ) .filter( SpotifyPlaylistFollowersHistory.date.between(date_range.start, date_range.end), SpotifyPlaylistFollowersHistory.playlist_id.in_(playlist_ids), ) .group_by(SpotifyPlaylistFollowersHistory.playlist_id, SpotifyPlaylistFollowersHistory.date) ) for item in query: items[item.playlist_id][date_to_timestamp(item.date)] += int(item.followers) for playlist_id, followers_by_dates in items.items(): result.append( { "playlist_id": playlist_id, "country_code": MARKET_GLOBAL, "coordinates": generate_coordinates(followers_by_dates, date_range.start, date_range.end), } ) return result @cache_value( keys.SPOTIFY_PLAYLIST_FOLLOWERS_RANGE, keys.SPOTIFY_PLAYLIST_FOLLOWERS_RANGE_TTL, key_args_indexes=(1, 2, 3) ) def get_playlist_date_ranges_followers( self, playlist_id: str, current_range: DateRange, prev_range: DateRange ) -> Dict[str, int or None]: """ Retrieve and count playlists followers in current and previous date ranges. All calculations are for global. Args: playlist_id (str): Playlist identifier. current_range (DateRange): Current date range values. prev_range (DateRange): Previous date range values. Returns: Dict[str, int or None]: Map with followers data for current and previous date ranges. """ query_records = [] for date_range in (current_range, prev_range): for date_order in (SpotifyPlaylistFollowersHistory.date.desc(), SpotifyPlaylistFollowersHistory.date.asc()): query_records.append( ( db_session.query( SpotifyPlaylistFollowersHistory.playlist_id, SpotifyPlaylistFollowersHistory.date, SpotifyPlaylistFollowersHistory.followers, ) .filter( SpotifyPlaylistFollowersHistory.date.between(date_range.start, date_range.end), SpotifyPlaylistFollowersHistory.playlist_id == playlist_id, ) .order_by(date_order) .first() ) ) curr_range_last, curr_range_first, prev_range_last, prev_range_first = query_records current_followers = ( curr_range_last.followers - curr_range_first.followers if curr_range_first and curr_range_last else None ) prev_followers = ( prev_range_last.followers - prev_range_first.followers if prev_range_first and prev_range_last else None ) return {"current_range": current_followers, "prev_range": prev_followers}