import datetime import logging from collections import namedtuple from ui_automation_framework.utils import logger as cl from app.src.api.constants import APPLE, SPOTIFY from app.src.helpers.data_formaters import DataFormatter class ApiDataProvider: log = cl.Logger(logging.DEBUG) def __init__(self, _api): self.api_client = _api.common self.apollo_api_client = _api.apollo self.dsp_api_client = _api.dsp self.gate_api_client = _api.gate def get_common_top_track_for_apple_spotify(self, market="global"): top_data_spotify = self.apollo_api_client.get_charts_top("spotify", market, fields=["is_sony"])["body"]["tracks"] top_data_apple = self.apollo_api_client.get_charts_top("apple", market, fields="is_sony")["body"]["tracks"] for i_s in top_data_spotify: for i_a in top_data_apple: if i_s["isrc"] == i_a["isrc"]: return i_a["name"] def get_spotify_charts_artist_track_streams(self): test_data_spotify = self.apollo_api_client.get_charts_top( "spotify", "global", fields=["is_sony", "is_re_enter", "trends", "image_url", "chart_date"], order_by=["position"], ) spotify_track_name = [] spotify_artist_name = [] streams_volume = [] for item in test_data_spotify["body"]["tracks"]: spotify_track_name.append(item["name"]) spotify_artist_name.append(item["artist"]) streams_volume.append(item["streams"]) break self.log.info(spotify_artist_name[0] + " " + spotify_track_name[0] + " " + str(streams_volume[0])) return spotify_artist_name[0], spotify_track_name[0], streams_volume[0] def get_apple_charts_artist_track(self): test_data_spotify = self.apollo_api_client.get_charts_top( "apple", "global", fields=["is_sony", "is_re_enter", "trends", "image_url", "chart_date"], order_by="position", ) apple_track_name = [] apple_artist_name = [] for item in test_data_spotify["body"]["tracks"]: apple_track_name.append(item["name"]) apple_artist_name.append(item["artist"]) break self.log.info(apple_artist_name[0] + " " + apple_track_name[0]) return apple_artist_name[0], apple_track_name[0] def get_new_tracks_name(self, vendor, market, order, end): top_charts = self.apollo_api_client.get_top_charts(vendor, market, order, end) names = [] for item in top_charts["body"]["tracks"]: if item["is_new"]: names.append(item["name"]) self.log.info(names) return names def get_re_enter_tracks_name(self, vendor, market, order, end): top_charts = self.apollo_api_client.get_top_charts(vendor, market, order, end) names = [] for item in top_charts["body"]["tracks"]: if item["is_re_enter"]: names.append(item["name"]) return names def get_top_charts_date(self, vendor, market, order, end): top_charts = self.apollo_api_client.get_top_charts(vendor, market, order, end) chart_date = top_charts["body"]["tracks"][0]["chart_date"] self.log.info(chart_date) return chart_date def get_streams_track_data_in_playlist(self, isrc, vendor, playlist_id, market, start_date, end_date): track_data = self.dsp_api_client.get_analytics_v1_track_in_playlists_streams( "isrc={}".format(isrc), "vendor={}".format(vendor), "playlist_id={}".format(playlist_id), "markets={}".format(market), "start_date={}".format(start_date), "end_date={}".format(end_date), ) self.log.info(track_data) if not track_data["body"].get("items"): return None else: return ( track_data["body"]["items"][0]["markets"][0].get("current_range"), track_data["body"]["items"][0]["markets"][0].get("prev_range"), ) def get_spotify_track_data_in_playlist(self, isrc, playlist_id, market): track_data = self.apollo_api_client.get_v2_spotify_playlist_track(isrc, playlist_id, market) self.log.info(track_data["body"]) updated_date = datetime.datetime.strptime(track_data["body"]["changed_date"], "%Y-%m-%d").strftime("%m/%d/%y") if track_data["body"]["added_date"]: days_added = (datetime.datetime.today() - datetime.datetime.strptime(track_data["body"]["added_date"], "%Y-%m-%d")).days days_added = DataFormatter.convert_days_to_years(days_added) else: days_added = "N/A" trend_position = track_data["body"]["trend"] if track_data["body"]["country_code"] == "_gl": track_market = "Global" elif not track_data["body"]["country_code"]: track_market = "N/A" else: track_market = track_data["body"]["country_code"].upper() if not track_data["body"]["personalized"]: if track_data["body"]["previous_position"]: calculated_trend = track_data["body"]["previous_position"] - track_data["body"]["current_position"] else: calculated_trend = 0 else: calculated_trend = "N/A" response = namedtuple("Response", ["playlist_name", "track_market", "updated_date", "days_added", "added_date", "trend_position", "current_position", "personalized", "calculated_trend", "is_new"]) return response(playlist_name=track_data["body"]["playlist_name"], track_market=track_market, updated_date=updated_date, days_added=days_added, added_date=track_data["body"]["added_date"] if track_data["body"]["added_date"] else "N/A", trend_position=trend_position, current_position=track_data["body"]["current_position"], personalized=track_data["body"]["personalized"], calculated_trend=calculated_trend, is_new=track_data["body"]["is_new"]) def get_track_start_end_dates(self, isrc, delta=0, ignore_response_attaching=False): latest_date = self.dsp_api_client.get_latest_date_go(isrc, ignore_response_attaching=ignore_response_attaching) start_date = datetime.datetime.strftime(datetime.datetime.strptime(latest_date, "%Y-%m-%d") - datetime.timedelta(days=delta), "%Y-%m-%d") return ( datetime.datetime.strptime(start_date, "%Y-%m-%d").date(), datetime.datetime.strptime(latest_date, "%Y-%m-%d").date(), ) def get_personalized_playlist_data(self, isrc, market): playlists_data = self.gate_api_client.get_playlists_by_track(isrc=isrc, vendor=SPOTIFY, market=market, limit=50) for item in playlists_data.items: if item.personalized: return item.name, item.id def get_personalized_playlist_data_full_listeners(self, isrc, market, new_market, days_period): """ Finds Spotify playlist for which we have the listeners datain a given time range, market and return it's name and id :param isrc: isrc for a given track in the playlist :param market: global market :param new_market: specific market :param days_period: period of days in which we want to find the proper data for listeners :return: name and id of the playlist """ # ListenersByPlaylist = namedtuple("listeners_by_playlist", ["id", "name", "weight"]) # listeners_by_playlists = [] # playlists_data = self.gate_api_client.get_playlists_by_track(isrc, SPOTIFY, streams_market=market, limit=100) # # for item in playlists_data.items: # start_date, end_date = self.get_track_start_end_dates(isrc, days_period) # listeners_graph = self.apollo_api_client.get_v1_spotify_playlists_listeners_graph( # "spotify", start_date, end_date, f"{market},{new_market}", item.id) # with suppress(ValueError): # listeners_by_playlists.append(ListenersByPlaylist( # item.id, item.name, max(filter(lambda l: l.country_code == new_market, listeners_graph.items), # key=lambda x: x.weight).weight)) # playlist_a_big_graph = max(listeners_by_playlists, key=lambda x: x.weight) playlist_a_big_graph_name = "Today's Top Hits" playlist_a_big_graph_id = "37i9dQZF1DXcBWIGoYBM5M" return playlist_a_big_graph_name, playlist_a_big_graph_id # return playlist_a_big_graph.name, playlist_a_big_graph.id def get_non_personalized_playlist_data(self, isrc, market): playlists_data = self.gate_api_client.get_playlists_by_track(isrc=isrc, vendor=SPOTIFY, streams_market=market, limit=10, order_by=["-current_streams"]) for item in playlists_data.items: if not item.personalized: return item.name, item.id def get_apple_playlist_data(self, isrc, market): playlists_data = self.gate_api_client.get_playlists_by_track(isrc, APPLE, market=market, limit=50) for item in playlists_data.items: # Ignoring China as spotify search won't work with the market if not item.country_code == "cn": return item.name, item.id, item.country_code.upper() def get_apple_track_data_in_playlist(self, isrc, playlist_id, market): track_data = self.apollo_api_client.get_v2_apple_music_playlist_track(isrc, playlist_id, market) self.log.info(track_data["body"]) updated_date = datetime.datetime.strptime(track_data["body"]["changed_date"], "%Y-%m-%d").strftime("%m/%d/%y") days_added = (datetime.datetime.today() - datetime.datetime.strptime(track_data["body"]["added_date"], "%Y-%m-%d")).days response = namedtuple("Response", ["track_name", "days_added", "added_date", "updated_date", "current_position", "trend", "is_new"]) return response(track_name=track_data["body"].get("track_name", None), days_added=days_added, added_date=track_data["body"].get("added_date", None), updated_date=updated_date, current_position=abs(track_data["body"]["current_position"]), trend=track_data["body"]["trend"], is_new=track_data["body"]["is_new"]) def get_dates_range_from_api_formatted(self, isrc): latest_date = self.api_client.get_latest_date_portal(isrc)["body"]["date"] start_date = datetime.datetime.strftime(datetime.datetime.strptime(latest_date, "%Y-%m-%d") - datetime.timedelta(days=6), "%Y-%m-%d") return str(datetime.datetime.strptime(start_date, "%Y-%m-%d").strftime("%m/%d/%y") + "–" + datetime.datetime.strptime(latest_date, "%Y-%m-%d").strftime("%m/%d/%y")) def get_top_charts_spotify_tracks_data(self, market="global"): top_data = self.apollo_api_client.get_charts_top("spotify", market, fields="is_sony")["body"]["tracks"] isrcs = {track["isrc"]: track["name"] for track in top_data if track["is_sony"] and not track["is_new"]} for track_isrc, track_name in isrcs.items(): date = self.apollo_api_client.get_charts_summary(f"{track_isrc}", "spotify", 20, 0, market="global")["body"]["items"][0]["added_date"] if datetime.datetime.strptime(date, "%Y-%m-%d") <= (datetime.datetime.today() - datetime.timedelta(days=54)): yield track_name, track_isrc def get_top_charts_spotify_track_data(self, market="global"): return next(self.get_top_charts_spotify_tracks_data(market)) def get_top_charts_spotify_track_data_4_weeks(self, market="global"): top_data = self.apollo_api_client.get_charts_top("spotify", market, fields="is_sony")["body"]["tracks"] isrcs = {track["isrc"]: track["name"] for track in top_data if track["is_sony"] and not track["is_new"]} for k, v in isrcs.items(): date = self.apollo_api_client.get_charts_summary(f"{k}", "spotify", 20, 0, market="global")["body"]["items"][0]["added_date"] if datetime.datetime.strptime(date, "%Y-%m-%d") <= (datetime.datetime.today() - datetime.timedelta(days=28)): track_name = v track_isrc = k return track_name, track_isrc def get_top_charts_spotify_track_1_year_data(self, market="global"): top_data = self.apollo_api_client.get_charts_top("spotify", market, fields="is_sony")["body"]["tracks"] isrcs = {track["isrc"]: track["name"] for track in top_data if track["is_sony"] and not track["is_new"]} for k, v in isrcs.items(): date = self.apollo_api_client.get_charts_summary(f"{k}", "spotify", 20, 0, market="global")["body"]["items"][0]["added_date"] if datetime.datetime.strptime(date, "%Y-%m-%d") <= (datetime.datetime.today() - datetime.timedelta(days=365)): track_name = v track_isrc = k return track_name, track_isrc def get_top_charts_spotify_non_sony_track_data(self, market="global"): track_name = "" track_isrc = "" top_data = self.apollo_api_client.get_charts_top("spotify", market, fields="is_sony")["body"]["tracks"] for track in top_data: if not track["is_sony"] and not track["is_new"]: track_name = track["name"] track_isrc = track["isrc"] return track_name, track_isrc def get_top_charts_apple_non_sony_track_data(self, market="global"): track_name = "" track_isrc = "" top_data = self.apollo_api_client.get_charts_top("apple", market, fields="is_sony")["body"]["tracks"] for track in top_data: if not track["is_sony"] and not track["is_new"]: track_name = track["name"] track_isrc = track["isrc"] return track_name, track_isrc def get_top_track_data_for_spotify_and_apple_for_chart_widget(self, market="global"): track_name = "" track_isrc = "" charts_data_api_spotify = [] charts_data_api_apple = [] top_data_spotify = self.apollo_api_client.get_charts_top("spotify", market, fields="is_sony")["body"]["tracks"] top_data_apple = self.apollo_api_client.get_charts_top("apple", market, fields="is_sony")["body"]["tracks"] for s_track in top_data_spotify: for a_track in top_data_apple: if s_track["name"] == a_track["name"]: charts_data_api_spotify = self.apollo_api_client.get_charts_summary(s_track["isrc"], "spotify", 200, 0)["body"]["items"] charts_data_api_apple = self.apollo_api_client.get_charts_summary(s_track["isrc"], "apple", 200, 0)["body"]["items"] if len(charts_data_api_spotify) > 0 and len(charts_data_api_apple) > 0: track_isrc = s_track["isrc"] track_name = s_track["name"] return track_name, track_isrc, charts_data_api_spotify, charts_data_api_apple def get_top_charts_apple_track_data(self, market="us", index=0): top_data = self.apollo_api_client.get_charts_top("apple", market, fields="is_sony")["body"]["tracks"] self.log.info(top_data) track_name = top_data[index]["name"] track_isrc = top_data[index]["isrc"] track_artist = top_data[index]["artist"] return track_name, track_isrc, track_artist def get_sony_tracks_isrcs(self, vendor="spotify", market="global", ignore_response_attaching=False): top = self.apollo_api_client.get_charts_top(vendor, market, fields="is_sony", ignore_response_attaching=ignore_response_attaching)["body"]["tracks"] ids = [i["id"] for i in top if i["is_sony"]] sony_tracks = [i["isrc"] for i in top if i["is_sony"]] return sony_tracks, ids def get_non_sony_tracks_isrcs(self, vendor="spotify", market="global"): top = self.apollo_api_client.get_charts_top(vendor, market)["body"]["tracks"] ids = [i["id"] for i in top if not i["is_sony"]] non_sony_tracks = [i["isrc"] for i in top if not i["is_sony"]] return non_sony_tracks, ids def get_playlist_daily_avg_and_trend(self, vendor, start_date, end_date, playlist_id, market, days_in_period): trend = None daily_avg_raw = self.dsp_api_client.get_analytics_v1_playlist_date_range_streams( f"vendor={vendor}", f"start_date={start_date}", f"end_date={end_date}", f"playlist_id={playlist_id}", f"market={market}", )["body"] if daily_avg_raw["prev_range"]: trend = (daily_avg_raw.get("prev_range") - daily_avg_raw.get("current_range")) / daily_avg_raw.get("current_range") * 100 return daily_avg_raw["current_range"] / days_in_period, trend def get_playlist_streams_listeners_and_trend(self, start_date, end_date, playlist_id, market): trend = None streams_per_listeners_data_raw = self.gate_api_client.get_playlists_spotify_streams_per_listener(playlist_id, f"start_date={start_date}", f"end_date={end_date}", f"market={market}")["body"] if streams_per_listeners_data_raw["prev_streams_streams_per_listener"]: trend = (streams_per_listeners_data_raw.get("prev_streams_streams_per_listener") - streams_per_listeners_data_raw.get("current_streams_per_listener")) / streams_per_listeners_data_raw.get("current_streams_per_listener") * 100 return streams_per_listeners_data_raw["current_streams_per_listener"], trend def get_playlist_listeners_daily_avg_and_trend(self, playlist_id, start_date, end_date, market, days_in_period): trend = None listeners_data_raw = self.apollo_api_client.get_v1_spotify_playlists_listeners_range(playlist_id, start_date, end_date, market)["body"] if listeners_data_raw["prev_range"]: trend = (listeners_data_raw.get("prev_range") - listeners_data_raw.get("current_range")) / listeners_data_raw.get("current_range") * 100 return listeners_data_raw["current_range"] / days_in_period, trend def get_playlist_followers_daily_avg_trend_last_total(self, playlist_id, start_date, end_date, market, days_in_period): trend = None followers_data_raw = self.apollo_api_client.get_v1_spotify_playlists_followers_range(playlist_id, start_date, end_date, market)["body"] follower_coordinates = self.apollo_api_client.get_v1_spotify_playlists_followers_graph(f"id={playlist_id}", f"start_date={start_date}", f"end_date={end_date}", f"market={market}")["body"]["items"][0]["coordinates"] last_total = [i["y"] for i in follower_coordinates if i["y"] is not None][-1] if followers_data_raw["prev_range"]: trend = (followers_data_raw.get("prev_range") - followers_data_raw.get("current_range")) / followers_data_raw.get("current_range") * 100 return followers_data_raw["current_range"] / days_in_period, trend, last_total def star_track_spotify(self, isrc, track_id): self.apollo_api_client.post_v1_starred_tracks_spotify(isrc, track_id) def star_track_apple(self, isrc, track_id): self.apollo_api_client.post_v1_starred_tracks_apple(isrc, track_id) def unstar_track(self, isrc): self.apollo_api_client.delete_v1_starred_tracks(isrc) def get_top_market_with_entries(self, vendor): markets = {"Global": "global", "United States": "us", "Australia": "au", "United Kingdom": "gb", "France": "fr"} for k, v in markets.items(): tracks = self.apollo_api_client.get_charts_additions(vendor, v, fields=["name", "isrc", "is_new", "is_re_enter"])["body"]["tracks"] if tracks and len(tracks) > 2: return k, v def get_top_market_with_entries_for_market_selector(self, vendor): markets = {"United States": "us", "Australia": "au", "United Kingdom": "gb", "France": "fr", "Germany": "de"} markets_with_additions = [] for k, v in markets.items(): tracks = self.apollo_api_client.get_charts_additions(vendor, v, fields=["name", "isrc", "is_new", "is_re_enter"])["body"]["tracks"] if tracks and len(tracks) > 1: markets_with_additions.append(k) return markets_with_additions def get_top_market_with_exits(self, vendor): markets = {"Global": "global", "United States": "us", "Australia": "au", "United Kingdom": "gb", "France": "fr"} for k, v in markets.items(): tracks = self.apollo_api_client.get_charts_removals(vendor, v, fields=["name", "isrc", "is_new", "is_re_enter"])["body"]["tracks"] if tracks and len(tracks) > 2: return k, v def get_spotify_top_track_playlist_data(self, market="global"): """ Finds spotify playlist for sony top track with an actual data and return it's name and id :param market: market of the playlist :return: name of the appropriate playlist """ isrc = (self.get_top_charts_spotify_track_data(market))[1] playlists_data = self.gate_api_client.get_playlists_by_track(isrc, SPOTIFY, streams_market=market, limit=50).items for playlist in playlists_data: return playlist.name, playlist.id def get_apple_top_track_playlist_data(self, market="global"): """ Finds spotify playlist for apple track with an actual data and return it's name and id :param market: market of the playlist :return: name of the appropriate playlist """ isrc = (self.get_top_charts_apple_track_data(market))[1] playlists_data = self.gate_api_client.get_playlists_by_track(isrc, APPLE, market=market, limit=50).items for playlist in playlists_data: return playlist.name, playlist.id def get_tracklist_spotify_data_from_api(self, playlist_id, market): """ Return tracklist data for particular playlist and market :param playlist_id: :param market: :return: """ local = None sony = None frontline = None tracklist_data = self.gate_api_client.get_playlists_spotify_tracklist_graph(f"playlist_id={playlist_id}", f"market={market}")["body"] if tracklist_data["local_tracks"]: local = round(tracklist_data["local_tracks"] * 100 / tracklist_data["tracks_num"]) if local < 0: local = str("~0%") if tracklist_data["sony_tracks"]: sony = round(tracklist_data["sony_tracks"] * 100 / tracklist_data["tracks_num"]) if sony < 0: sony = str("~0%") if round(tracklist_data["frontline_tracks"]): frontline = round(tracklist_data["frontline_tracks"] * 100 / tracklist_data["tracks_num"]) if frontline < 0: local = str("~0%") self.log.info(local) self.log.info(sony) self.log.info(frontline) return str(local), str(sony), str(frontline) def get_tracklist_apple_data_from_api(self, playlist_id, market): """ Return tracklist data for particular playlist and market :param playlist_id: :param market: :return: """ local = None frontline = None tracklist_data = self.gate_api_client.get_playlists_apple_tracklist_graph(f"playlist_id={playlist_id}", f"market={market}")["body"] self.log.info(tracklist_data) if tracklist_data["local_tracks"]: local = round(tracklist_data["local_tracks"] * 100 / tracklist_data["tracks_num"]) if local < 0: local = str("~0%") if round(tracklist_data["frontline_tracks"]): frontline = round(tracklist_data["frontline_tracks"] * 100 / tracklist_data["tracks_num"]) if frontline < 0: local = str("~0%") elif tracklist_data["frontline_tracks"] == 0: frontline = 0 self.log.info(local) self.log.info(frontline) return str(local), str(frontline) def get_search_results_api(self, name): tracks = self.apollo_api_client.get_list_of_spotify_tracks( f"{name}", )["body"][ "tracks" ]["items"] list_tracks = {} list_tracks = {el["external_ids"]["isrc"]: el["name"] for el in tracks if el["external_ids"]["isrc"] not in list_tracks} return list_tracks def get_first_track_with_youtube_data_by_playlist(self, playlist_id): market = "us" tracklist = self.apollo_api_client.get_spotify_playlist_tracklist(playlist_id, market, "current_position", 100)["body"]["items"] for k, v in enumerate(tracklist): isrc = tracklist[k]["isrc"] youtube_data = self.gate_api_client.get_youtube_data( "group_by=date,video_id", "sort_by=-views", "views=1", "content_type=partner_uploaded,premium_ugc", "expand_to=related_isrcs", f"isrc={isrc}", ) if youtube_data["body"]["items"]: youtube_data = youtube_data["body"]["items"] return youtube_data else: raise Exception("There are no tracks with YouTube data")