import datetime import logging import pandas import pytest from jsonbender import F as ApplyFunction from jsonbender import If, K 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, get_field_with_diff_custom_date, merge_arrays, transform_json API = api_init.APIOverride() log = cl.Logger(logging.DEBUG) apollo_to_delphi_keys = { "id": GetField("id") >> ApplyFunction(lambda val: "" if val is None else str(val)), "storefront": GetField("storefront"), "name": GetField("name"), "artistName": GetField("artistName"), "position": GetField("position"), "positionChange": GetField("positionChange") >> ApplyFunction(lambda val: "" if val is None else val), "positionChangeDate": GetField("positionChangeDate") >> ApplyFunction(lambda val: "" if val is None else val[:10]), "isrc": GetField("isrc"), "trackReleaseDate": If(GetField("trackReleaseDate") == K(None), GetField("albumReleaseDate"), GetField("trackReleaseDate")), "earliestAdded": GetField("earliestAdded") >> ApplyFunction(lambda val: "" if val is None else val[:10]), "earliestPosition": GetField("earliestPosition"), "marketStreams": GetField("marketStreams") >> ApplyFunction(lambda val: 0 if val is None else val), } delphi_to_delphi_keys = { "id": GetField("apple_track") >> ApplyFunction(lambda val: "" if val is None or val["track_id"] is None else val["track_id"]), "storefront": GetField("country_code"), "name": GetField("apple_track") >> ApplyFunction(lambda val: "" if val is None or val["name"] is None else val["name"]), # 'artistName': GetField('apple_track') >> ApplyFunction(lambda val: "" if val is None or val['artists'] or val['artists'] == [] else ', '.join(v['artists']['name'] for v in val)), "artistName": GetField("apple_track", "artists") >> ApplyFunction(lambda val: "" if len(val) == 0 else " & ".join(v["name"] for v in val)), "position": GetField("current") >> ApplyFunction(lambda val: int(val) + 1), "positionChange": GetField("trend_change_14_days") >> ApplyFunction(lambda val: "" if val is None else val), "positionChangeDate": GetField("last_date_change_14_days"), "isrc": GetField("isrc"), "trackReleaseDate": If(GetField("apple_track", "release_date") == K(None), GetField("apple_track", "album", "release_date"), GetField("apple_track", "release_date")), "earliestAdded": GetField("earliest_position_date_time") >> ApplyFunction(lambda val: "" if val is None else val[:10]), "earliestPosition": GetField("earliest_position"), "marketStreams": GetField("streaming_info", "us") >> ApplyFunction(lambda val: 0 if val is None or val["7"] is None or val["7"]["isrc_in_playlist"] is None else val["7"]["isrc_in_playlist"]), } def extend_tracklist_with_streams(tracklist_resp_main, playlist_id, market): # get required isrcs isrcs_main = [t["isrc"] for t in tracklist_resp_main] end_date = API.get_latest_date()["apple"] end_date = datetime.datetime.strptime(end_date, "%Y-%m-%d").date() start_date = end_date - datetime.timedelta(days=6) # get response with streams resp_streams_all = [] for isrc in isrcs_main: if isrc is None: resp_streams_all.append({"isrc": isrc, "globalStreams": 0, "marketStreams": 0}) else: resp_streams_one_isrc = API.get_apollo_response(streams_apollo_portal.format(id=playlist_id, start_date=start_date, end_date=end_date, isrc=isrc, market=market))["items"] if len(resp_streams_one_isrc) == 0: resp_streams_all.append({"isrc": isrc, "globalStreams": 0, "marketStreams": 0}) else: resp_streams_all.append({"isrc": isrc, "globalStreams": sum(item["globalStreams"] for item in resp_streams_one_isrc), "marketStreams": sum(item["localStreams"] for item in resp_streams_one_isrc)}) for item_main, streams in zip(sorted(tracklist_resp_main, key=lambda x: (x["isrc"] is not None, x["isrc"])), sorted(resp_streams_all, key=lambda x: (x["isrc"] is not None, x["isrc"]))): item_main.update({"globalStreams": streams["globalStreams"]}) item_main.update({"marketStreams": streams["marketStreams"]}) return tracklist_resp_main # tracklist_apollo_portal_local = '/applemusic/playlists/{id}?country={market}' # streams_apollo_portal = '/apple-consumer-analytics/track-in-container?containerId={id}&startDate={start_date}&endDate={end_date}&isrc={isrc}&countryCode={market}' # tracklist_delphi = '/v3/public/track-positions/playlists/current-tracklist?playlist_id=apple_{id}&dsp=apple&country_code={market}&streams_country_code={}market&limit=2000&offset=0' tracklist_apollo_portal_local = "/applemusic/playlists/{id}?country={market}" streams_apollo_portal = "/apple-consumer-analytics/track-in-container?containerId={id}&startDate={start_date}&endDate={end_date}&isrc={isrc}&countryCode={market}" tracklist_delphi = "/v3/public/track-positions/playlists/current-tracklist?playlist_id=apple_{id}&dsp=apple&country_code={market}&streams_country_code={market}&include=track_info,streams_for_period" @parameterized.expand( [ # ('pl.005c5a0f52744716b0021df43fe61fc6', 'us'), # ('pl.0e61531414db4ee8a9b34fada69845c8', 'us'), # ('pl.9a964a33c1484aec8fdb0cac3e7771ed', 'us'), # 500 error from delphi # Today’s Hits ("pl.606afcbb70264d2eb2b51d8dbcfa6a12", "us"), # ("pl.4fe671d3bd994738ace5ae974daebc10", 'au'), # ("pl.559ea7b879434003bb7621b1d60d3403", 'gb'), # ("pl.567c541f63414e798be5cf214e155557", 'us'), # ("pl.56e0d896d28b4291ba7e02af84605fc7", 'us'), # ('pl.2b0e6e332fdf4b7a91164da3162127b5', 'au'), # ('pl.6ea68024aeae49ca9e81eaac6ef929cf', 'us'), # ('pl.0061bc57171e4f848b377de3db21cf71', 'us'), # ('pl.5ee8333dbe944d9f9151e97d92d1ead9', 'fr') ] ) @pytest.mark.skip() def test_curr_tracklist(playlist_id, market): apollo_response_without_streams = API.get_apollo_response(tracklist_apollo_portal_local.format(id=playlist_id, market=market))["tracklist"] delphi_response = API.get_delphi_response(tracklist_delphi.format(id=playlist_id, market=market))["items"] apollo_response = extend_tracklist_with_streams(apollo_response_without_streams, playlist_id, market) 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), "id": get_field_with_diff("id", aggregated_matrix), "storefront": get_field_with_diff("storefront", aggregated_matrix), "name": get_field_with_diff("name", aggregated_matrix), "artistName": get_field_with_diff("artistName", aggregated_matrix), "position": get_field_with_diff("position", aggregated_matrix), "positionChange": get_field_with_diff("positionChange", aggregated_matrix), "positionChangeDate": get_field_with_diff_custom_date("positionChangeDate", aggregated_matrix), "isrc": get_field_with_diff("isrc", aggregated_matrix), "trackReleaseDate": get_field_with_diff("trackReleaseDate", aggregated_matrix), "earliestAdded": get_field_with_diff_custom_date("earliestAdded", aggregated_matrix), "earliestPosition": get_field_with_diff("earliestPosition", 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/apple_playlist_curr_tracklist/{playlist_id}_{market.upper()}_apple_playlist_curr_tracklist_total_apollo_{len(apollo_transformed_response)}_delphi_{len(delphi_transformed_response)}.xlsx")