from datetime import datetime import requests url = "https://qa-graphql-router.theorchard.io/graphql" headers = { "Content-Type": "application/json", "orchard-identity-id": "b273cf59-4035-4151-be6e-56acd02eae9c", "orchard-profile-id": "502648", "orchard-profile-type": "InsightsProfile", "apollographql-client-name": "test-insights", } def execute_new_query(variables): new_query = """ query NewGlobalSoundRecordingByIsrc( $isrc: String! $storeIds: [Int!] $streamCountries: [String!] $curatorCountries: [String!] $playlistTypes: [PlaylistType!] $playlistAppearances: String $limit: Int $offset: Int $orderBy: String $orderDir: String $storefrontEnabled: Boolean ) { globalSoundRecordingByIsrc(isrc: $isrc) { isrc analytics { playlistPlacements( storeIds: $storeIds, playlistTypes: $playlistTypes, streamCountries: $streamCountries, curatorCountries: $curatorCountries, playlistAppearances: $playlistAppearances, limit: $limit, offset: $offset, orderBy: $orderBy, orderDir: $orderDir, storefrontEnabled: $storefrontEnabled ) { totalCount placements { completionRate183Days completionRate1Day completionRate28Days completionRate365Days completionRate7Days completionRateAllTime curatorCountry curatorId curatorName currentPosition dateRemoved daysOnPlaylist firstPlay isrc lastAddedOnDate peakPosition playlistArtworkUrl playlistFollowerCount playlistGenres playlistId playlistName playlistTrackCount playlistType playlistUri positionChange previousPosition storeId storefront storefrontCount streams183Days streams1Day streams28Days streams365Days streams7Days streamsAllTime totalPlaylistStreams28Days totalPlaylistStreams7Days } } } } } """ response = requests.post( url, headers=headers, json={"query": new_query, "variables": variables}, ) assert response.status_code == 200 return response.json() def execute_old_query(variables): query = """ query OldGlobalSoundRecordingByIsrc( $isrc: String! $storeIds: [Int!] $streamCountries: [String!] $curatorCountries: [String!] $playlistTypes: [PlaylistType!] $playlistAppearances: String $limit: Int $offset: Int $orderBy: String $orderDir: String $storefrontEnabled: Boolean ) { globalSoundRecordingByIsrc(isrc: $isrc) { isrc playlistPlacements( storeIds: $storeIds, playlistTypes: $playlistTypes, streamCountries: $streamCountries, curatorCountries: $curatorCountries, playlistAppearances: $playlistAppearances, limit: $limit, offset: $offset, orderBy: $orderBy, orderDir: $orderDir, storefrontEnabled: $storefrontEnabled ) { totalCount placements { lastAddedOnDate dateRemoved firstPlay currentPosition peakPosition previousPosition positionChange daysOnPlaylist playlistPlacementStreams { growthPeriods { period value completionRate } } playlist { playlistId playlistName playlistUri playlistArtworkUrl playlistFollowerCount playlistTrackCount playlistType playlistGenres playlistCurator { curatorId curatorCountry curatorName } source { storeId } analytics { playlist { playlistId } streams7Days(streamCountries: $streamCountries) streams28Days(streamCountries: $streamCountries) } } storefrontCount(playlistAppearances: $playlistAppearances) } } } } """ response = requests.post( url, headers=headers, json={"query": query, "variables": variables}, ) assert response.status_code == 200 return response.json() def parse_growth_periods(old_placement): result = {} periods = old_placement["playlistPlacementStreams"]["growthPeriods"] for period in periods: if period["period"] == "_ALL_TIME": result["completionRateAllTime"] = period["completionRate"] result["streamsAllTime"] = period["value"] elif period["period"] == "_365_DAYS": result["completionRate365Days"] = period["completionRate"] result["streams365Days"] = period["value"] elif period["period"] == "_183_DAYS": result["completionRate183Days"] = period["completionRate"] result["streams183Days"] = period["value"] elif period["period"] == "_28_DAYS": result["completionRate28Days"] = period["completionRate"] result["streams28Days"] = period["value"] elif period["period"] == "_7_DAYS": result["completionRate7Days"] = period["completionRate"] result["streams7Days"] = period["value"] elif period["period"] == "_1_DAY": result["completionRate1Day"] = period["completionRate"] result["streams1Day"] = period["value"] return result def compare_placements(old_placement, new_placement, variables): old_placement_converted = { "curatorCountry": old_placement["playlist"]["playlistCurator"][ "curatorCountry" ], "curatorId": old_placement["playlist"]["playlistCurator"]["curatorId"], "curatorName": old_placement["playlist"]["playlistCurator"]["curatorName"], "currentPosition": old_placement["currentPosition"], "dateRemoved": old_placement["dateRemoved"], "daysOnPlaylist": old_placement["daysOnPlaylist"], "firstPlay": old_placement["firstPlay"], "isrc": variables["isrc"], "lastAddedOnDate": datetime.strptime( old_placement["lastAddedOnDate"], "%Y-%m-%d" ).date(), "peakPosition": old_placement["peakPosition"], "playlistArtworkUrl": old_placement["playlist"]["playlistArtworkUrl"], "playlistFollowerCount": old_placement["playlist"]["playlistFollowerCount"], "playlistGenres": old_placement["playlist"]["playlistGenres"], "playlistId": old_placement["playlist"]["playlistId"], "playlistName": old_placement["playlist"]["playlistName"], "playlistTrackCount": old_placement["playlist"]["playlistTrackCount"], "playlistType": old_placement["playlist"]["playlistType"], "playlistUri": old_placement["playlist"]["playlistUri"], "positionChange": old_placement["positionChange"], "previousPosition": old_placement["previousPosition"], "storeId": old_placement["playlist"]["source"]["storeId"], "storefrontCount": old_placement["storefrontCount"], "totalPlaylistStreams28Days": old_placement["playlist"]["analytics"][ "streams28Days" ], "totalPlaylistStreams7Days": old_placement["playlist"]["analytics"][ "streams7Days" ], } old_placement_converted.update(parse_growth_periods(old_placement)) if variables["storefrontEnabled"] is False: old_placement_converted["storefront"] = None lastAddedOnDate = new_placement["lastAddedOnDate"] dt = datetime.fromisoformat(lastAddedOnDate) new_placement["lastAddedOnDate"] = dt.date() # ignore firstPlay new_placement['firstPlay'] = None old_placement_converted['firstPlay'] = None # ignore playlistFollowerCount new_placement['playlistFollowerCount'] = None old_placement_converted['playlistFollowerCount'] = None assert new_placement == old_placement_converted def do_test(variables): old_payload = execute_old_query(variables) old_placements_wrapper = old_payload["data"]["globalSoundRecordingByIsrc"][ "playlistPlacements" ] old_placements = old_placements_wrapper["placements"] new_payload = execute_new_query(variables) new_placements_wrapper = new_payload["data"]["globalSoundRecordingByIsrc"][ "analytics" ]["playlistPlacements"] new_placements = new_placements_wrapper["placements"] assert old_placements_wrapper["totalCount"] == new_placements_wrapper["totalCount"] new_placements_iter = iter(new_placements) for old_placement in old_placements: new_placement = next(new_placements_iter) compare_placements(old_placement, new_placement, variables) def test_qraphql_1(): variables = { "isrc": "QMFMF2447070", "playlistTypes": [ "ALGORITHMIC", "CURATED", "EDITORIAL", "NEW_MUSIC_FRIDAY", "PERSONALIZED", ], "storeIds": [], "streamCountries": [], "curatorCountries": [], "playlistAppearances": "current", "orderBy": "streams_last_7_days", "orderDir": "desc", "limit": 200, "offset": 0, "storefrontEnabled": False, } do_test(variables) def test_qraphql_2(): variables = { "isrc": "USSM12501019", "playlistTypes": [ "ALGORITHMIC", "CURATED", "EDITORIAL", "NEW_MUSIC_FRIDAY", "PERSONALIZED", ], "storeIds": [], "streamCountries": [], "curatorCountries": [], "playlistAppearances": "current", "orderBy": "streams_last_7_days", "orderDir": "desc", "limit": 200, "offset": 0, "storefrontEnabled": False, } do_test(variables)