""" Hot Hits utilities. """ from collections import defaultdict from datetime import date, datetime from sqlalchemy import and_, func from sqlalchemy.engine.result import RowProxy from sqlalchemy.orm import Query from typing import List, Optional, Tuple, Union from apollo_main_db.apollo.models import ApolloKeyValueStorage, ApolloMarket, ApolloMarketRegion, ApolloRegion, \ ApolloRegionDetail, MarketRegionTypeEnum from apollo_main_db.spotify.models import MarketRankTypeEnum, SpotifyHotHitsPlaylistTrackStats, SpotifyMarketRank, \ SpotifyPlaylist, SpotifyViewPlaylist, ViewPlaylistTypeEnum from src.db.base import session from src.legacy.apollo_api.constants import REGION_ALL_CODE from src.legacy.core.calculations.trends import get_track_in_playlist_trend, is_track_in_playlist_new_entry from src.legacy.spotify.constants import SPOTIFY_PLAYLIST_IMAGE_URL_MASK from src.legacy.spotify.util import get_avg_and_top_positions, get_playlists_regions # Core functions def get_latest_hot_hits_date(with_time: bool = False) -> Optional[Union[date, datetime]]: """Get the latest available Hot Hits date. Args: with_time: Return recent datetime instead date. Returns: The latest Hot Hits date. """ hothits_latest_data_datetime = ( session.query(ApolloKeyValueStorage.value).filter(ApolloKeyValueStorage.key == "hot_hits_latest_ts").first() ) if not hothits_latest_data_datetime or not hothits_latest_data_datetime[0]: return hothits_latest_data_datetime = datetime.fromisoformat(hothits_latest_data_datetime[0]) if with_time: hothits_job_latest_run_datetime = ( session.query(ApolloKeyValueStorage.value).filter(ApolloKeyValueStorage.key == "hh_job_last_run_ts").first() ) if hothits_job_latest_run_datetime and hothits_job_latest_run_datetime[0]: hothits_job_latest_run_datetime = hothits_job_latest_run_datetime[0].split(".")[0] hothits_job_latest_run_datetime = datetime.fromisoformat(hothits_job_latest_run_datetime) if hothits_latest_data_datetime.date() >= hothits_job_latest_run_datetime.date(): return hothits_job_latest_run_datetime return hothits_latest_data_datetime else: return hothits_latest_data_datetime.date() def get_track_in_playlist_entry_date( entry_date: Optional[date], current_entry_date: Optional[date], in_history: int = 1 ) -> Optional[date]: """Calculate final entry_date value based on values from historical and current tracklist tables. Implemented to provide consistency with track/tracklist APIs, that logic of calculating added_date works by rule: searching for track isrc in playlists history, if there is nothing, select the date from the current table. Args: entry_date: added_date from stats hothits table. current_entry_date: added_date from current tracklits in_history: int flag of was 'entry_date' taken from historical table or Spotify (timestamp from current tracklist table). Added to distinguish the case when no any data was found in historical table: in this situation we should show entry_date from current tracklist table. Return: entry_date """ return entry_date if bool(in_history) else current_entry_date # Track (single) in playlists view def _get_hot_hits_single_track_in_playlists_data( isrc: str, hot_hits_date: date, updated_only: bool = True, with_name: bool = True ) -> list: """Get single track in Hot Hits playlists data. Args: isrc: ISRC. hot_hits_date: hot hits latest date. updated_only: Updated playlists (markets) only. with_name: Include playlist name in result (uses obsolete table). Returns: Hot Hits track playlists data. """ query_field_list = [ SpotifyViewPlaylist.playlist_id, SpotifyViewPlaylist.market_code.label("code"), SpotifyMarketRank.rank.label("top_market"), SpotifyHotHitsPlaylistTrackStats.isrc, SpotifyHotHitsPlaylistTrackStats.entry_date.label("entry_date"), SpotifyHotHitsPlaylistTrackStats.added_date.label("current_entry_date"), SpotifyHotHitsPlaylistTrackStats.in_history, ApolloRegion.name.label("region_name"), ApolloMarketRegion.region_code, (SpotifyHotHitsPlaylistTrackStats.latest_position + 1).label("position"), (SpotifyHotHitsPlaylistTrackStats.previous_position + 1).label("previous_position"), SpotifyHotHitsPlaylistTrackStats.latest_date, (SpotifyHotHitsPlaylistTrackStats.peak_position + 1).label("peak_position"), func.IFNULL(ApolloMarketRegion.market_name_override, ApolloMarket.name).label("name"), ApolloRegionDetail.rank.label("rank"), ApolloRegionDetail.name_override.label("name_override"), ] query = ( session.query(SpotifyViewPlaylist) .outerjoin( SpotifyHotHitsPlaylistTrackStats, and_( SpotifyHotHitsPlaylistTrackStats.playlist_id == SpotifyViewPlaylist.playlist_id, SpotifyHotHitsPlaylistTrackStats.isrc == isrc, SpotifyHotHitsPlaylistTrackStats.latest_date == hot_hits_date, ), ) .join( SpotifyMarketRank, and_( SpotifyMarketRank.market_code == SpotifyViewPlaylist.market_code, SpotifyMarketRank.type == MarketRankTypeEnum.HH, ), ) .join(ApolloMarket, ApolloMarket.code == SpotifyViewPlaylist.market_code) .join( ApolloMarketRegion, and_( ApolloMarketRegion.market_code == SpotifyViewPlaylist.market_code, ApolloMarketRegion.type == MarketRegionTypeEnum.HH, ), ) .join(ApolloRegion, ApolloRegion.code == ApolloMarketRegion.region_code) .outerjoin( ApolloRegionDetail, and_( ApolloRegion.code == ApolloRegionDetail.code, ApolloRegionDetail.type == MarketRegionTypeEnum.HH, ), ) ) if with_name: query_field_list.append(SpotifyPlaylist.name.label("playlist_name")) query = query.outerjoin(SpotifyPlaylist, SpotifyPlaylist.id == SpotifyViewPlaylist.playlist_id) query = query.with_entities(*query_field_list).filter(SpotifyViewPlaylist.type == ViewPlaylistTypeEnum.HH) if updated_only: query = query.filter(SpotifyViewPlaylist.last_date == hot_hits_date) return query.all() def get_hot_hits_single_track_in_playlists( isrc: str, region_code: str, include_inactive_markets: bool = False, with_name: bool = True ) -> Tuple[List[dict], List[dict], dict, int, int]: """Get Hot Hits track playlists. Args: isrc: ISRC. region_code: Chosen region code. include_inactive_markets: Include markets where track was not added. with_name: Include playlist name in result (uses obsolete table). Returns: Hot Hits track playlists data, regions list, chosen region data, avg and top 10 total. """ hh_latest_date = get_latest_hot_hits_date() track_in_playlists = _get_hot_hits_single_track_in_playlists_data(isrc, hh_latest_date, with_name=with_name) playlist_list = [] average_position, top_10_total = get_avg_and_top_positions(track_in_playlists) region_list, region_data = get_playlists_regions(track_in_playlists, region_code, include_inactive_markets) for item in track_in_playlists: if not include_inactive_markets and not item.position: continue if region_code and region_code != REGION_ALL_CODE and region_code != item.region_code: continue playlist_data = item._asdict() entry_date = get_track_in_playlist_entry_date( playlist_data.pop("entry_date", None), playlist_data.pop("current_entry_date", None), in_history=playlist_data.pop("in_history", 1), ) trend = get_track_in_playlist_trend(item.position, item.previous_position, none_as=0) playlist_data["entry_date"] = entry_date playlist_data["image_url"] = SPOTIFY_PLAYLIST_IMAGE_URL_MASK.format(playlist_id=item.playlist_id) playlist_data["position_change"] = trend and -trend # 'and' to avoid None; frontend expects -trend here playlist_data["added_days"] = (hh_latest_date - entry_date).days if entry_date else 0 playlist_data["is_new"] = is_track_in_playlist_new_entry(entry_date, current_date=hh_latest_date) playlist_data["region"] = playlist_data.pop("region_name") playlist_data["market"] = playlist_data["code"] del playlist_data["region_code"], playlist_data["rank"], playlist_data["name_override"] playlist_list.append(playlist_data) return playlist_list, region_list, region_data, average_position, top_10_total # Tracks (bulk) in playlists view def get_hot_hits_multiple_tracks_in_playlists_query(isrc_list: List[str], hot_hits_date: date) -> Query: """Get hot hits tracks-in-playlists query. Args: isrc_list: Tracks ISRC list. hot_hits_date: Hot hits date. Returns: Query. """ return ( session.query(SpotifyHotHitsPlaylistTrackStats) .join(SpotifyViewPlaylist, SpotifyViewPlaylist.playlist_id == SpotifyHotHitsPlaylistTrackStats.playlist_id) .filter(SpotifyHotHitsPlaylistTrackStats.isrc.in_(isrc_list)) .filter(SpotifyViewPlaylist.last_date == hot_hits_date) .filter(SpotifyViewPlaylist.type == ViewPlaylistTypeEnum.HH) ) def _get_playlist_data(item: RowProxy, extra_data: dict = None, with_name: bool = True) -> dict: """Get base playlist data. Args: item: Playlist DB data. with_name: Include playlist name in result. Returns: Base response data. """ return { "playlist_id": item.playlist_id, "playlist_image": SPOTIFY_PLAYLIST_IMAGE_URL_MASK.format(playlist_id=item.playlist_id), "market": item.market, "rank": item.rank, "region": item.region, **(extra_data if extra_data else {}), **({"playlist_name": item.playlist_name} if with_name else {}), } def format_hot_hits_multiple_tracks_in_playlists( isrc_list: List[str], hot_hits_date: date, playlist_list: list, stats_list: List[SpotifyHotHitsPlaylistTrackStats], with_name: bool = True, ) -> List[dict]: """Get tracks hot hits response items. Args: isrc_list: Tracks ISRC list. hot_hits_date: Hot hits state for selected tracks on this date. playlist_list: HH playlists. stats_list: HH stats records. with_name: Include playlist name in result. Returns: Response items. """ updated_playlists = {item.playlist_id: item for item in playlist_list if item.latest_date == hot_hits_date} current_playlists = defaultdict(list) previous_playlists = defaultdict(list) featured_playlists = defaultdict(list) for row in stats_list: if row.playlist_id not in updated_playlists: continue featured_playlists[row.isrc].append(row.playlist_id) playlist = updated_playlists[row.playlist_id] entry_date = get_track_in_playlist_entry_date(row.entry_date, row.added_date, in_history=row.in_history) data = _get_playlist_data( playlist, { "entry_date": entry_date, "peak_position": row.peak_position + 1, "peak_date": row.peak_date, }, with_name=with_name, ) if row.latest_date == hot_hits_date: data.update( { "position": row.latest_position + 1, "trend": row.latest_trend, "is_entry": is_track_in_playlist_new_entry(entry_date, current_date=row.latest_date), } ) current_playlists[row.isrc].append(data) else: data["exit_date"] = row.latest_date previous_playlists[row.isrc].append(data) result_items = [] for isrc in isrc_list: track_current = current_playlists[isrc] track_featured_playlists = featured_playlists[isrc] avg_position = (sum(i["position"] for i in track_current) / len(track_current)) if track_current else None track_non_featured = [ _get_playlist_data(item, with_name=with_name) for item in updated_playlists.values() if item.playlist_id not in track_featured_playlists ] result_items.append( { "isrc": isrc, "non_featured": track_non_featured, "previous": previous_playlists[isrc], "current": track_current, "avg_position": None if avg_position is None else round(avg_position), "current_count": len(track_current), } ) return result_items def get_hot_hits_multiple_tracks_in_playlists(isrc_list: List[str], with_name: bool = True) -> dict: """Get hot hits track-in-playlist data mor multiple tracks. Args: isrc_list: Tracks ISRC list. with_name: Include playlist name in result (uses obsolete table). Returns: Tracks hot hots data. """ query_field_list = [ SpotifyViewPlaylist.playlist_id, SpotifyViewPlaylist.market_code.label("market"), ApolloRegion.name.label("region"), SpotifyMarketRank.rank, SpotifyViewPlaylist.last_date.label("latest_date"), ] query = ( session.query(SpotifyViewPlaylist) .join( ApolloMarketRegion, and_( ApolloMarketRegion.market_code == SpotifyViewPlaylist.market_code, ApolloMarketRegion.type == MarketRegionTypeEnum.HH, ), ) .join(ApolloRegion, ApolloRegion.code == ApolloMarketRegion.region_code) .join( SpotifyMarketRank, and_( SpotifyMarketRank.market_code == SpotifyViewPlaylist.market_code, SpotifyMarketRank.type == MarketRankTypeEnum.HH, ), ) ) if with_name: query_field_list.append(SpotifyPlaylist.name.label("playlist_name")) query = query.outerjoin(SpotifyPlaylist, SpotifyPlaylist.id == SpotifyViewPlaylist.playlist_id) playlist_list = ( query.with_entities(*query_field_list).filter(SpotifyViewPlaylist.type == ViewPlaylistTypeEnum.HH).all() ) hot_hits_date = get_latest_hot_hits_date() result_items = [] if hot_hits_date: stats_list = get_hot_hits_multiple_tracks_in_playlists_query(isrc_list, hot_hits_date).all() result_items = format_hot_hits_multiple_tracks_in_playlists( isrc_list, hot_hits_date, playlist_list, stats_list, with_name ) playlist_list = [ { "playlist_image": SPOTIFY_PLAYLIST_IMAGE_URL_MASK.format(playlist_id=i.playlist_id), **i._asdict(), } for i in playlist_list ] return { "items": result_items, "hot_hits_date": hot_hits_date, "regions": [i["region"] for i in playlist_list], "outdated": [i for i in playlist_list if i["latest_date"] != hot_hits_date], "playlists": playlist_list, "playlists_by_markets": {i["market"]: i["playlist_id"] for i in playlist_list}, }