from abc import ABC, abstractmethod from collections import defaultdict from datetime import date, datetime, timedelta from typing import Dict, Iterable, List, Optional from natsort import natsort_keygen from sqlalchemy import and_, or_, text from sqlalchemy.orm import Query, aliased from apollo_main_db.apollo import ApolloMarket, StarredContent from charts.constants import THRESHOLD_MAJOR_MOVES from core.clients import clients from core.constants import DEFAULT_IMAGE_SIZE, DEFAULT_IMAGE_SIZE_MOBILE, MARKET_GLOBAL, SPOTIFY_MARKET_GLOBAL, \ IdType, VendorBitMask from core.util import sum_periods from main_db.base import session as db_session from redis_db.decorators import cache_value from redis_db.keys import CHART_AVAILABILITY, CHART_AVAILABILITY_TTL class VendorBase(ABC): """Base class for logic of specific vendor.""" vendor_name: str @abstractmethod def check_track_is_sony(self, ids_list: List[str], id_type: IdType, market: str = None) -> List[str]: """Method fetch sony/non-sony data for batch of tracks. Args: ids_list: list of tracks ids. market: Market code. id_type: type of identifiers in ids_list (id|isrc) Return: List of sony ids only """ class VendorCharts(VendorBase, ABC): # base query getters @abstractmethod def get_diff_tracks_by_dates_query( self, market: str, start_date: str, end_date: str, is_removed: bool = False ) -> Query: """Get track entities difference between two dates. Using this method we can get added or removed tracks in charts. Args: market: Str market. start_date: Str date in format %Y-%m-%d. end_date: Str date in format %Y-%m-%d. is_removed: should be True for searching removals query """ pass @abstractmethod def get_out_tracks_query( self, market: str, start_date: str, end_date: str, start: int, end: int, **kwargs ) -> Query: pass @abstractmethod def get_top_tracks_query( self, market: str, start_date: str, end_date: str, start: int, end: int, **kwargs ) -> Query: """Get track entities difference between two dates. Using this method we can get charts in some positions range. Arguments: market (str): Market country code. start_date: Str date in format %Y-%m-%d. end_date: Str date in format %Y-%m-%d. start (int): Position range start. end (int): Position range end. Returns: Query: Result query. """ @abstractmethod def get_major_moves_query( self, market: str, start_date: str, end_date: str, change: int = THRESHOLD_MAJOR_MOVES ) -> Query: """Get track entities difference between two dates. Using this method we can get major moves for tracks inside chart. Args: market: Str market. start_date: Str date in format %Y-%m-%d. end_date: Str date in format %Y-%m-%d. change: number of track changed positions to look for """ pass @abstractmethod def get_chart_track_summary_query(self, isrc: str) -> Query: pass # query handlers def order_query(self, tracks_query: Query, order_list: List[str]) -> Query: """Order query by its `moves` alias value. Arguments: tracks_query (Query): Query that will be ordered. order_list (List[str]): List of values for ordering. Returns: Query: Ordered query. """ subquery = tracks_query.subquery() order_str_list = [] for label in order_list: if label.startswith("-"): order_str_list.append(f"{label[1:]} DESC") else: order_str_list.append(label) order_str = ", ".join(order_str_list) return db_session.query(subquery).order_by(text(order_str)) @abstractmethod def get_tracks_is_new_and_re_entered( self, tracks_query: Query, market: str, prev_date: str, *args, **kwargs ) -> Query: """Append `is_re_enter` and `trends data to chart tracks query. Arguments: tracks_query (Query): chart tracks subquery. market (str): Market country code. prev_date (str): Previous chart query timestamp in format %Y-%m-%d. Returns: Query: Query with is_re_enter data. """ pass # result handlers def process_isrc_duplicates(self, data: Query or List, is_major_moves: bool = False) -> List[object]: """Check for removing duplicates, empty check as default.""" return data # external data handlers def get_track_extra_remote_data_map( self, track_ids: List[str], market: str, image_size: int = DEFAULT_IMAGE_SIZE_MOBILE ) -> Dict[str, str]: """Return external remote data for tracks_ids as track_id: {...some data map..} map.""" return {} @abstractmethod def get_track_extra_db_data_map(self, tracks_ids: List[str]) -> Dict[str, object]: """Get extra data fro charts from db. Arguments: tracks_ids : chart tracks ids list. Returns: Dict: track_id: {...some_extra_data..} """ pass # not cached data handlers @staticmethod def get_starred_tracks_query(user_id: str) -> Query: """Returns query of starred tracks query for current user. Arguments: user_id (str): User identifier. Returns: Query: Query with starred track data. """ return ( db_session.query(StarredContent) .filter(StarredContent.isrc.isnot(None), StarredContent.user_id == user_id) .group_by(StarredContent.isrc) ) @staticmethod @cache_value(CHART_AVAILABILITY, CHART_AVAILABILITY_TTL) def check_chart_is_available_in_market(vendor: str, market: str) -> bool: """Check if vendor provide chart data for chosen market. Arguments: vendor (str): Apple or Spotify. market (str): Market code value. Returns: bool: True - if chart data is available, False - if not. """ if market == MARKET_GLOBAL: market = SPOTIFY_MARKET_GLOBAL vendor_value = VendorBitMask.get_mask(vendor) return bool( db_session.query(ApolloMarket) .filter( ApolloMarket.charts_vendors.in_((vendor_value.value, VendorBitMask.BOTH_VENDORS.value)), ApolloMarket.code == market, ) .scalar() ) # utils def get_top_charts_market(self, charts_items: List[object]) -> Optional[str]: return None @abstractmethod def get_chart_last_dates(self, market: str, *args, convert_to_str: bool = False, **kwargs) -> List[datetime.date]: """Get two most recent dates of Spotify chart data. First is the date of the most recent chart data update in the database. It corresponds to the current most actual state of the chart. Second is the date of previous update. These two dates are usually needed to calculate changes in track's positions to see which tracks went up or down, entered or left the chart. Args: market: Market code. convert_to_str: True for returning string Returns: List of up to two datetime.date objects corresponding to dates of the most recent chart updates. """ pass @abstractmethod def get_chart_date(self, last_date: date or str) -> date or str: """Return date for specific chart.""" pass @abstractmethod def get_tracks_stats(self, isrc_list: List[str], market: str, chart_type: str, list_type: str) -> List[Dict]: """Get tracks statistics.""" pass class VendorTrackPlaylistHistory(VendorBase, ABC): track_table = None playlist_table = None weekly_top_playlist_table = None weekly_top_playlist_history_table = None def get_track_top_playlists_history( self, track_ids: List[int] or List[str], market: str = None, limit: Optional[int] = None, image_size: int = DEFAULT_IMAGE_SIZE, period: Optional[int] = None, ) -> Dict: """Get [Limit] of top playlists that contain [track_ids]. Determining the top is based on number of stream from ApplePlaylistStreamSummary DB table that updated by background jobs. Args: track_ids: Queried track IDs market: Market code, 2 chars limit: Number of playlist to receive. By default it is not set. image_size: Size of playlist image to get (used for apple only). period: Days count, upper interval border. Returns: List of top playlists with periods when track appears in it. """ market = market or MARKET_GLOBAL tracks_mapping = self._get_tracks_mapping(track_ids) original_tracks = {tracks_mapping[t_id].isrc: t_id for t_id in track_ids if t_id in tracks_mapping} # Get all first dates from consumerDB into dict {'isrc0' : 'start_date_0', ..} track_first_dates = clients.dsp.get_streams_first_date_any_vendor( list(original_tracks.keys()), self.vendor_name ) if not track_first_dates: return {"playlists": []} playlists_mapping = { p.id: {"name": p.name, "streams_latest": 0} for p in self._get_latest_date_ranked_playlists_query(market).all() } track_first_dates_parsed = {k: datetime.strptime(v, "%Y-%m-%d").date() for k, v in track_first_dates.items()} history_mapping = defaultdict(lambda: defaultdict(list)) for playlist_id, isrc, entry_date, exit_date in self._get_top_playlist_history_dates_query( playlists_mapping.keys(), track_first_dates_parsed, period, market ).all(): history_mapping[playlist_id][original_tracks[isrc]].append([entry_date, exit_date]) for item in self._get_playlist_streams_query(history_mapping.keys(), market).all(): playlists_mapping[item.playlist_id]["streams_latest"] = item.streams_latest playlist_id_to_image_url_map = self._get_playlists_images_urls(history_mapping.keys(), market, image_size) return self._format_data( history_mapping, playlists_mapping, tracks_mapping, playlist_id_to_image_url_map, track_first_dates, limit ) def _get_tracks_mapping(self, track_ids: List) -> Dict[str, object]: """Return dict of track_id: track_object, where track object contains id, name and isrc fields found for all ids from tracks_ids. """ aliased_track = aliased(self.track_table) return { t.id: t for t in ( db_session.query(self.track_table.id, self.track_table.name, self.track_table.isrc) .join(aliased_track, aliased_track.isrc == self.track_table.isrc) .filter(aliased_track.id.in_(track_ids)) .distinct() .all() ) } def _format_data( self, history_mapping, playlists_mapping, tracks_mapping, playlist_id_to_image_url_map, track_first_dates, limit ): result = [] for playlist_id, playlist_tracks in history_mapping.items(): playlist = { "id": playlist_id, "name": playlists_mapping[playlist_id]["name"], "streams": int(playlists_mapping[playlist_id]["streams_latest"]), "type": self.vendor_name, "image_url": playlist_id_to_image_url_map.get(playlist_id, ""), "tracks": [], } result.append(playlist) for track_id, history_dates in playlist_tracks.items(): track_data = tracks_mapping[track_id] track = { "id": track_id, "name": track_data.name, "isrc": track_data.isrc, "first_date": track_first_dates[track_data.isrc] if track_data.isrc in track_first_dates else False, "periods": sum_periods(history_dates), } playlist["tracks"].append(track) result = sorted(result, key=lambda x: (-x["streams"])) if limit: result = result[:limit] return {"playlists": result} @abstractmethod def _get_latest_date_ranked_playlists_query(self, market: str) -> Query: pass def _get_base_top_playlist_history_dates_query( self, playlists_ids: Iterable[str], tracks_isrc: Dict[str, date], period: Optional[int] = None ) -> Query: """Returns playlist history query with playlist_id, isrc, entry_date, exit_date. Filter by date range and lists of playlist ID and track isrc. """ track_filters = [] for track_isrc, first_stream_date in tracks_isrc.items(): current_filters = [ self.weekly_top_playlist_history_table.isrc == track_isrc, self.weekly_top_playlist_history_table.exit_date >= first_stream_date, ] if period: end_date = first_stream_date + timedelta(days=period) current_filters.append(self.weekly_top_playlist_history_table.entry_date <= end_date) track_filters.append(and_(*current_filters)) # For each track and playlist pair retrieve list of dates when the track was included in the playlist. history_query = ( db_session.query( self.weekly_top_playlist_history_table.playlist_id, self.weekly_top_playlist_history_table.isrc, self.weekly_top_playlist_history_table.entry_date, self.weekly_top_playlist_history_table.exit_date, ) .distinct() .filter(self.weekly_top_playlist_history_table.playlist_id.in_(playlists_ids)) .filter(or_(*track_filters)) ) return history_query def _get_top_playlist_history_dates_query( self, playlists_ids: Iterable[str], tracks_isrc: Dict[str, date], period: Optional[int] = None, market: Optional[str] = None, ) -> Query: return self._get_base_top_playlist_history_dates_query(playlists_ids, tracks_isrc, period) @abstractmethod def _get_playlist_streams_query(self, playlists_ids: Iterable[str], market: str) -> Query: pass @abstractmethod def _get_playlists_images_urls(self, playlists_ids: Iterable[str], *args, **kwargs) -> Dict[str, str]: pass class VendorPlaylists(VendorBase, ABC): @abstractmethod def get_playlist_data_map_for_track(self, isrc: str, *args, **kwargs) -> Dict[str, dict]: pass @abstractmethod def get_ordered_playlist_ids_by_streams(self, playlist_ids: Iterable[str], market: str) -> dict: pass def extend_playlists_with_streams_data(self, track_streams_data: dict, playlists_map: dict, key: str) -> list: """Extend list of playlists with streams value. Arguments: track_streams_data (dict): CA playlists streams data. playlists_map (dict): Playlists data. key (str): Streams key value. Returns: Extended with track streams data list of playlists. """ items, none_items = [], [] for k, v in playlists_map.items(): streams_value = track_streams_data.get(k) v[key] = streams_value if streams_value is None: none_items.append(v) continue items.append(v) items.sort(key=natsort_keygen(key=lambda pl: (-pl[key], pl["id"].lower()))) none_items.sort(key=natsort_keygen(key=lambda pl: pl["id"].lower())) return items + none_items