from apollo_utils.core.constants import MARKET_WORLDWIDE from datetime import date from flask import current_app as app from sqlalchemy import Column, case, func from sqlalchemy.orm import Query from typing import Dict, List, Optional, Tuple, Type, Union from apollo_main_db.apollo.models import ApolloKeyValueStorage, BuzzCategory, BuzzUser from apollo_main_db.spotify.models import SpotifyPersonalizedPlaylist, SpotifyPlaylist, \ SpotifyPlaylistDigestFollowersPlaylist, SpotifyPlaylistDigestStreamsPlaylist, SpotifyPlaylistFollowersWeekHistory, \ SpotifyPlaylistStreamsWeekHistory from src.constants.playlists.spotify.charts import PLAYLIST_CHART_COUNTRY_CODE_LIST, ChartMetric, ChartType, \ PlaylistCategory, PlaylistType from src.db.base import session as db_session from src.legacy.redis_db import keys as redis_keys from src.legacy.redis_db.decorators import cache_value def get_latest_dates() -> Dict[str, date]: """Get Spotify playlists charts latest dates. Returns: Types to dates mapping. """ key_mapping = { "playlist_digest_streams_date": "streams_max_date", "playlist_digest_followers_date": "followers_max_date", "playlist_digest_latest_date": "max_date", } latest_dates = ( db_session.query(ApolloKeyValueStorage.key, ApolloKeyValueStorage.value) .filter(ApolloKeyValueStorage.key.in_(key_mapping.keys())) .all() ) return {key_mapping[item.key]: (date.fromisoformat(item.value) if item.value else None) for item in latest_dates} def _apply_playlist_type_filter( query: Query, table: Type[Union[SpotifyPlaylistDigestFollowersPlaylist, SpotifyPlaylistDigestStreamsPlaylist]], playlists_type_list: Optional[List[PlaylistType]], ) -> Query: """Apply playlist type (personalized or not) filter. Args: query: Base query. table: Streams or followers playlists table. playlists_type_list: Playlist types. Returns: Filtered query. """ if playlists_type_list and len(playlists_type_list) == 1: if PlaylistType.PERSONALIZED in playlists_type_list: return query.join(SpotifyPersonalizedPlaylist, SpotifyPersonalizedPlaylist.playlist_id == table.playlist_id) else: return query.outerjoin( SpotifyPersonalizedPlaylist, SpotifyPersonalizedPlaylist.playlist_id == table.playlist_id ).filter(SpotifyPersonalizedPlaylist.playlist_id.is_(None)) return query def get_base_query( playlists_table: Type[Union[SpotifyPlaylistDigestStreamsPlaylist, SpotifyPlaylistDigestFollowersPlaylist]], metrics_table: Type[Union[SpotifyPlaylistStreamsWeekHistory, SpotifyPlaylistFollowersWeekHistory]], chart_date: date, owner_category_id_list: List[PlaylistCategory], playlists_type_list: Optional[List[PlaylistType]] = None, playlists_country_code_list: Optional[List[str]] = None, ) -> Query: """Get base query. Args: playlists_table: Playlists table. metrics_table: Metrics table. chart_date: Chart date. owner_category_id_list: Playlists owner category (Spotify, Sony, etc). playlists_type_list: Playlist types (personalized or not). playlists_country_code_list: Playlist country codes. Returns: Query. """ query = ( db_session.query(metrics_table) .join(playlists_table, playlists_table.playlist_id == metrics_table.playlist_id) .filter(metrics_table.date == chart_date) .filter(playlists_table.category_id.in_([i.value for i in owner_category_id_list])) ) query = _apply_playlist_type_filter(query, playlists_table, playlists_type_list) if playlists_country_code_list: query = query.filter(playlists_table.country_code.in_(playlists_country_code_list)) return query def _get_source( chart_metric: ChartMetric, ) -> Tuple[ Type[Union[SpotifyPlaylistDigestStreamsPlaylist, SpotifyPlaylistDigestFollowersPlaylist]], Type[Union[SpotifyPlaylistStreamsWeekHistory, SpotifyPlaylistFollowersWeekHistory]], ]: """Get source tables. Args: chart_metric: Source type. Returns: Playlists and metrics tables. """ if chart_metric == ChartMetric.STREAMS: return SpotifyPlaylistDigestStreamsPlaylist, SpotifyPlaylistStreamsWeekHistory else: return SpotifyPlaylistDigestFollowersPlaylist, SpotifyPlaylistFollowersWeekHistory def _get_columns( metrics_table: Type[Union[SpotifyPlaylistStreamsWeekHistory, SpotifyPlaylistFollowersWeekHistory]], chart_metric: ChartMetric, streams_country_code_list: Optional[List[str]], ) -> Tuple[Column, Column]: """Get value and change columns. Args: chart_metric: Metric type (streams or followers). streams_country_code_list: Playlist streams country codes. Returns: Value and change columns. """ if chart_metric == ChartMetric.STREAMS and streams_country_code_list: if len(streams_country_code_list) > (len(PLAYLIST_CHART_COUNTRY_CODE_LIST) / 2): value_column = metrics_table.value change_column = metrics_table.change for country_code in streams_country_code_list: value_column = value_column - getattr(metrics_table, f"value_{country_code}") change_column = change_column - getattr(metrics_table, f"change_{country_code}") else: country_code = streams_country_code_list[0] value_column = getattr(metrics_table, f"value_{country_code}") change_column = getattr(metrics_table, f"change_{country_code}") for country_code in streams_country_code_list[1:]: value_column = value_column + getattr(metrics_table, f"value_{country_code}") change_column = change_column + getattr(metrics_table, f"change_{country_code}") return value_column, change_column else: return metrics_table.value, metrics_table.change def _calc_sort_by(value_column: Column, change_column: Column, chart_type: ChartType) -> Column: """Calculate sort by column. Args: value_column: Value column. change_column: Change column. chart_type: Chart type (by value or by moves). Returns: Sort by column / expression. """ sort_by = value_column if chart_type == ChartType.VALUE else (change_column / (value_column - change_column)) if chart_type == ChartType.MOVE_ASC: return sort_by.asc() return sort_by.desc() def _set_columns_and_sorting( query: Query, playlists_table: Type[Union[SpotifyPlaylistDigestStreamsPlaylist, SpotifyPlaylistDigestFollowersPlaylist]], metrics_table: Type[Union[SpotifyPlaylistStreamsWeekHistory, SpotifyPlaylistFollowersWeekHistory]], value_column: Column, change_column: Column, chart_metric: ChartMetric, chart_type: ChartType, limit: int, ) -> Query: """Set columns, sorting order and limit. Args: query: Playlist digest query. playlists_table: Playlists table. metrics_table: Metrics table. value_column: Value column. change_column: Change column. chart_metric: Chart metric (streams or followers). chart_type: Chart type (by value or by moves). limit: Top playlists limit. """ sort_by = _calc_sort_by(value_column, change_column, chart_type) if chart_type != ChartType.VALUE: filter_value = app.config["PLAYLIST_DIGEST_MAJOR_MOVES_MIN_METRIC_DIFF"][chart_metric.value] query = query.filter( (change_column >= filter_value) if chart_type == ChartType.MOVE_DESC else (change_column <= -filter_value) ) return ( query.with_entities( metrics_table.playlist_id.label("playlist_id"), playlists_table.country_code.label("country_code"), value_column.label("value"), change_column.label("change"), ) .order_by(sort_by) .limit(limit) ) def _get_playlists_data(query: Query, chart_type: ChartType) -> Query: """Get charts playlists data. Args: query: Playlist digest query. chart_type: Chart type (by value or by moves). Returns: Playlist data query. """ sort_by = _calc_sort_by(query.c.value, query.c.change, chart_type) return ( db_session.query( query.c.playlist_id, SpotifyPlaylist.name, query.c.country_code, query.c.value, query.c.change, SpotifyPlaylist.uri.label("link"), func.IF(SpotifyPersonalizedPlaylist.playlist_id.is_(None), False, True).label("is_personalized"), func.IFNULL(BuzzUser.user_name, SpotifyPlaylist.user_name).label("owner_account_id"), func.IFNULL(BuzzUser.display_name, SpotifyPlaylist.user_name).label("owner_account_name"), BuzzUser.category_id.label("owner_category_id"), BuzzCategory.name.label("owner_category_name"), case( [(func.IFNULL(BuzzUser.country_code, "").in_(("_gl", "", "null")), MARKET_WORLDWIDE)], else_=BuzzUser.country_code, ).label("owner_country_code"), ) .join(SpotifyPlaylist, SpotifyPlaylist.id == query.c.playlist_id) .outerjoin(BuzzUser, BuzzUser.user_name == SpotifyPlaylist.user_name) .outerjoin(BuzzCategory, BuzzCategory.id == BuzzUser.category_id) .outerjoin(SpotifyPersonalizedPlaylist, SpotifyPersonalizedPlaylist.playlist_id == SpotifyPlaylist.id) .order_by(sort_by, SpotifyPlaylist.name) ) @cache_value(redis_keys.PLAYLIST_DIGEST_CHART, redis_keys.PLAYLIST_DIGEST_CHART_TTL) def get_chart( chart_metric: ChartMetric, chart_type: ChartType, chart_date: date, owner_category_id_list: List[PlaylistCategory], playlists_type_list: Optional[List[PlaylistType]] = None, playlists_country_code_list: Optional[List[str]] = None, streams_country_code_list: Optional[List[str]] = None, limit: int = 10, ) -> List[dict]: """Get playlist digest charts. Args: chart_metric: Metric type (streams or followers). chart_type: Chart type (by moves or by values). chart_date: Date. owner_category_id_list: Playlists owner category (Spotify, Sony, etc). playlists_type_list: Playlist types (personalized or not). playlists_country_code_list: Playlist country codes. streams_country_code_list: Playlist streams country codes. limit: Top playlists limit. Returns: Playlist digest charts data. """ playlists_table, metrics_table = _get_source(chart_metric) query = get_base_query( playlists_table, metrics_table, chart_date, owner_category_id_list, playlists_type_list, playlists_country_code_list, ) value_column, change_column = _get_columns(metrics_table, chart_metric, streams_country_code_list) query = _set_columns_and_sorting( query, playlists_table, metrics_table, value_column, change_column, chart_metric, chart_type, limit ) query = _get_playlists_data(query.subquery(), chart_type) return [i._asdict() for i in query.all()]