import logging import pandas import pytest from jsonbender import F as ApplyFunction from jsonbender import OptionalS from jsonbender import S as GetField from parameterized import parameterized from ui_automation_framework.utils import logger as cl import app.src.tests.decomission.api_init as api_init from app.src.tests.decomission.compare_helper import format_color, generate_provider, get_field_with_diff, merge_arrays, transform_json API = api_init.APIOverride() log = cl.Logger(logging.DEBUG) apollo_to_delphi_keys = { "trackUri": GetField("trackUri") >> ApplyFunction(lambda val: "" if val is None else val.split(":")[-1]), "name": GetField("trackName") >> ApplyFunction(lambda val: "" if val is None else val), "albumUpc": GetField("albumUpc") >> ApplyFunction(lambda val: "" if val is None else val), "artistName": GetField("artistName"), "artistsURLs": OptionalS("artists") >> ApplyFunction(lambda val: ", ".join(a["uri"].split(":")[-1] for a in val)), "isrc": GetField("isrc"), "albumReleaseDate": GetField("albumReleaseDate"), "position": GetField("position") >> ApplyFunction(lambda val: int(val) + 1), "globalStreams": GetField("globalStreams") >> ApplyFunction(lambda val: 0 if val is None else val), "marketStreams": GetField("marketStreams") >> ApplyFunction(lambda val: 0 if val is None else val), } market_c = "de" delphi_to_delphi_keys = { # modify to match "trackUri": GetField("spotify_track", "track_id") >> ApplyFunction(lambda val: "" if val is None else val), "name": GetField("spotify_track") >> ApplyFunction(lambda val: "" if val is None else val["name"]), "albumUpc": GetField("spotify_track", "album", "upc") >> ApplyFunction(lambda val: "" if val is None else val), "artistName": GetField("spotify_track", "artists") >> ApplyFunction(lambda val: ", ".join(v["name"] for v in val)), "artistsURLs": OptionalS("spotify_track", "artists") >> ApplyFunction(lambda val: ", ".join(a["url"].split(":")[-1] for a in val)), "isrc": GetField("isrc"), "albumReleaseDate": GetField("spotify_track", "album") >> ApplyFunction(lambda val: 0 if val in [None, ""] else val["release_date"]), "position": GetField("current") >> ApplyFunction(lambda val: val + 1), # CAN NOT GET FROM DELPHI "globalStreams": OptionalS("streaming_info") >> ApplyFunction(lambda val: 0 if val is None or val["worldwide"] is None or val["worldwide"]["7"] is None or val["worldwide"]["7"]["isrc_in_playlist"] is None else val["worldwide"]["7"]["isrc_in_playlist"]), "marketStreams": GetField("streaming_info") >> ApplyFunction(lambda val: 0 if val is None or val[market_c] is None or val[market_c]["7"] is None or val[market_c]["7"]["isrc_in_playlist"] is None else val[market_c]["7"]["isrc_in_playlist"]), } def extend_tracklist_with_streams(tracklist_resp_main, playlist_id, market, start_date, end_date): # get required isrcs isrcs_main = [str(t["isrc"]).upper() for t in tracklist_resp_main] # get response with streams resp_streams = API.get_apollo_response(streams_apollo_portal.format(id=playlist_id, start_date=start_date, end_date=end_date, isrcs="".join(f"isrc={isrc}&" for isrc in isrcs_main), market=market)) # get isrcs with streams isrcs_with_streams = [str(i["isrc"]).upper() for i in resp_streams] for isrc in isrcs_main: if isrc not in isrcs_with_streams: resp_streams.append({"isrc": isrc, "globalStreams": 0, "marketStreams": 0}) for item_main, streams in zip(sorted(tracklist_resp_main, key=lambda x: x["isrc"]), sorted(resp_streams, key=lambda x: x["isrc"])): item_main.update({"globalStreams": streams["globalStreams"]}) item_main.update({"marketStreams": streams["marketStreams"]}) return {"resp": tracklist_resp_main, "isrcs": isrcs_with_streams} tracklist_apollo_portal_local = "/spotifyplaylist/tracklisthistory/{date}?playlisturi=spotify%3Aplaylist%3A{id}®ion={market}" streams_apollo_portal = "/spotify-consumer-analytics-compat/tracks-in-playlist-summary-2?{isrcs}playlistId={id}&startDate={start_date}&endDate={end_date}&countryCode={market}" tracklist_delphi = "/v3/public/track-positions/playlists?start_date={start}&end_date={end}&dsp=spotify&playlist_id=spotify_{id}&streams_country_code=worldwide&streams_country_code={market}&include=track_info,streams_for_period&expand_isrc_by_track_positions=true" @parameterized.expand( [ # ('004Rf6OJOXidB5IK50Kt0b', 'us'), # ('37i9dQZF1DWX7rdRjOECPW', 'us'), # ('37i9dQZEVXbJ9SRaVj0yDF', 'us'), # # ('37i9dQZF1DXcBWIGoYBM5M', 'us'), # ('37i9dQZF1DXcBWIGoYBM5M', 'global'), # ('37i9dQZF1DX3rxVfibe1L0', 'us'), # ('37i9dQZF1DX3rxVfibe1L0', 'global'), # ("5H765bjiGWNovpZbfPeM3o", "us"), # ('5H765bjiGWNovpZbfPeM3o', 'global'), # ('37i9dQZF1E8zSJk2SWyOI4', 'us'), # ('37i9dQZF1DX3rxVfibe1L0', 'us'), # ("37i9dQZF1DXcBWIGoYBM5M", "us"), # ("37i9dQZF1DX5QPo288x03n", "us"), # ("0JFatPoPq82gNcPa4esOzj", "us"), # ('37i9dQZF1DXcBWIGoYBM5M', 'global'), # ('37i9dQZF1DX3rxVfibe1L0', 'us'), # ('37i9dQZF1DX3rxVfibe1L0', 'global'), # ("5H765bjiGWNovpZbfPeM3o", "us"), # ('5H765bjiGWNovpZbfPeM3o', 'global'), # ('6YtBXFCKiUj09rhpmP06Bp', 'global'), # ('37i9dQZF1DWT2SPAYawYcO', 'au'), # ("37i9dQZF1DWXti3N4Wp5xy", "us"), ("37i9dQZEVXbMDoHDwVN2tF", market_c) ] ) @pytest.mark.skip() def test_prev_tracklist(playlist_id, market): date_api = "2023-01-21" apollo_response_without_streams = API.get_apollo_response(tracklist_apollo_portal_local.format(id=playlist_id, market=market, date=date_api))["items"] extended_apollo_all = extend_tracklist_with_streams(apollo_response_without_streams, playlist_id, market, date_api, date_api) # isrcs = extended_apollo_all["isrcs"] apollo_response = extended_apollo_all["resp"] delphi_response = API.get_delphi_response( tracklist_delphi.format( id=playlist_id, market=market, start=date_api, end=date_api, ) )["items"] # isrcs=''.join(f'&isrc={isrc}' for isrc in isrcs) apollo_transformed_response = transform_json(apollo_to_delphi_keys, apollo_response) delphi_transformed_response = transform_json(delphi_to_delphi_keys, delphi_response) aggregated_matrix = merge_arrays(apollo_transformed_response, delphi_transformed_response, "isrc") index_for_all_fields = list(range(1, len(generate_provider(aggregated_matrix)) + 1)) subset_to_mark = range(3, len(aggregated_matrix) * 5, 5) pandas.DataFrame( { "provider": generate_provider(aggregated_matrix), "trackUri": get_field_with_diff("trackUri", aggregated_matrix), "name": get_field_with_diff("name", aggregated_matrix), "albumUpc": get_field_with_diff("albumUpc", aggregated_matrix), "artistName": get_field_with_diff("artistName", aggregated_matrix), "artistsURLs": get_field_with_diff("artistsURLs", aggregated_matrix), "isrc": get_field_with_diff("isrc", aggregated_matrix), "albumReleaseDate": get_field_with_diff("albumReleaseDate", aggregated_matrix), "position": get_field_with_diff("position", aggregated_matrix), "globalStreams": get_field_with_diff("globalStreams", aggregated_matrix), "marketStreams": get_field_with_diff("marketStreams", aggregated_matrix), }, index=index_for_all_fields, ).style.apply(format_color, subset=(subset_to_mark, slice(None))).to_excel(f"decomission/spotify_prev_tracklist/{date_api}_spotify_playlist_prev_tracklist_{market.upper()}_{playlist_id}_apollo_{len(apollo_transformed_response)}_delphi_{len(delphi_transformed_response)}.xlsx")