import time from typing import List import allure from app.src.api.api_client import ApiClient from app.src.api.models.chart_moves_model import ChartMovesResponse from app.src.api.models.listeners_graph import ListenersGraph from app.src.api.models.playlist_by_track import PlaylistByTrack from app.src.api.models.playlist_item_model import PlaylistSearchResult from app.src.api.models.top_charts_track import TopChartsTrack from app.src.helpers.allure_helper import attach_response_info class ApolloApiClient(ApiClient): @allure.step("Getting total number of records for the spotify in /gate-api/v3/playlists/spotify/by-track/") def get_spotify_playlists_positions_mobile(self, market, isrc, limit, offset, ignore_response_attaching=False): resp = self.session.get( self.host_endpoint + "/gate-api/v3/playlists/spotify/by-track/", params=dict(market=market, isrc=isrc, limit=limit, is_sony="false", offset=offset), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting total number of records for the apple in /gate-api/v3/playlists/apple/by_track/") def get_apple_playlists_positions_mobile(self, market, isrc, limit, offset, is_sony, ignore_response_attaching=False): resp = self.session.get( self.host_endpoint + "/gate-api/v3/playlists/apple/by-track/", params=dict(market=market, isrc=isrc, limit=limit, offset=offset, is_sony=is_sony), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting total number of records for the spotify in apollo-api/spotify/v1/filtr-track-playlists/") def get_spotify_playlists_positions_portal(self, isrc, market, limit, offset, ignore_response_attaching=False): resp = self.session.get( self.host_endpoint + "/apollo-api/spotify/v1/filtr-track-playlists/", params=dict(isrc=isrc, limit=limit, offset=offset, market=market), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting trends for the spotify track playlists in /gate-api/v2/playlists/spotify/by-track/trends/") def get_spotify_playlists_trends_go(self, isrc, ids, ignore_response_attaching=False): resp = self.session.get( self.host_endpoint + "/gate-api/v2/playlists/spotify/by-track/trends/", params=dict(isrc=isrc, playlist_ids=ids), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting charts summary info for the track in /gate-api/charts/summary/") def get_charts_summary(self, isrc, vendor, limit, offset, market="global", sort_by=None, ignore_response_attaching=False): resp = self.session.get( self.host_endpoint + "/gate-api/charts/summary/", params=dict( vendor=vendor, isrc=isrc, limit=limit, offset=offset, market=market, fields="chart_date", sort_by=sort_by, ), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting top charts info for the track accordingly to the market in /gate-api/charts/top/") def get_top_charts(self, vendor, market, order_by, end, ignore_response_attaching=False): resp = self.session.get( self.host_endpoint + "/gate-api/charts/top/", params=dict(vendor=vendor, market=market, order_by=order_by, end=end, compact="false", fields="chart_date"), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting response from /gate-api/charts/additions/") def get_charts_additions(self, vendor, market, order_by: List = None, fields: List = None, distributors: List = None, limit=None, ignore_response_attaching=False): api_string = "/gate-api/charts/additions/" resp = self.session.get( self.host_endpoint + api_string, params=dict(vendor=vendor, market=market, order_by=order_by, fields=fields, distributors=distributors, limit=limit), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting response from /gate-api/charts/removals/") def get_charts_removals(self, vendor, market, order_by: List = None, fields: List = None, distributors: List = None, limit=None, ignore_response_attaching=False): api_string = "/gate-api/charts/removals/" resp = self.session.get( self.host_endpoint + api_string, params=dict(vendor=vendor, market=market, order_by=order_by, fields=fields, distributors=distributors, limit=limit), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting response from /gate-api/charts/moves/") def get_charts_moves(self, vendor, market, order_by: List = None, fields: List = None, distributors: List = None, ignore_response_attaching=False): api_string = "/gate-api/charts/moves/" resp = self.session.get( self.host_endpoint + api_string, params=dict(vendor=vendor, market=market, order_by=order_by, fields=fields, distributors=distributors), timeout=self.timeout, ) attach_response_info(resp, ignore_response_attaching) return ChartMovesResponse(**resp.json()) @allure.step("Getting response from /gate-api/charts/top/") def get_charts_top(self, vendor, market, fields: List = None, order_by: List = None, limit=None, ignore_response_attaching=False): api_string = "/gate-api/charts/top/" resp = self.session.get( self.host_endpoint + api_string, params=dict(vendor=vendor, market=market, fields=fields, order_by=order_by, limit=limit, distributors=["sme", "theorchard", "awal"]), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting max_date from /apollo-api/charts/date-range/") def get_max_date(self, *parameters, token, ignore_response_attaching=False): # TODO: investigate usage headers = {"authorization": "Bearer {}".format(token)} api_string = "/apollo-api/charts/date-range/?type=daily&" params_list = "&".join(parameters) resp = self.session.get(self.host_endpoint + api_string + params_list, headers=headers, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response["body"]["max_date"] @allure.step("Getting starred tracks from /apollo-api/v1/starred-tracks/detailed/") def get_starred_tracks_detailed(self, vendors, market=None, offset=None, limit=None, related=None, ignore_response_attaching=False): # TODO: is vendors a list? api_string = "/apollo-api/v1/starred-tracks/detailed/" resp = self.session.get( self.host_endpoint + api_string, params=dict(vendors=vendors, market=market, offset=offset, limit=limit, related=related), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting starred tracks from /apollo-api/v1/starred-tracks/") def get_starred_tracks(self, ignore_response_attaching=False): api_string = "/apollo-api/v1/starred-tracks/" resp = self.session.get(self.host_endpoint + api_string, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting is sony flag from /apollo-api/is-sony/") def get_is_sony_go(self, spotify_ids, market, ignore_response_attaching=False): # Outdated endpoint api_string = "/apollo-api/is-sony/" resp = self.session.get(self.host_endpoint + api_string, params=dict(spotify_ids=spotify_ids, market=market), timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting portal playlists data from /spotify/v1/filtr-track-playlists/") def get_spotify_playlists_data_portal(self, isrc, market, offset, limit, ignore_response_attaching=False): api_string = "/gate-api/v1/playlists/spotify/by-track/" resp = self.session.get( self.host_endpoint + api_string, params=dict(isrc=isrc, market=market, offset=offset, limit=limit), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting Go playlists data from /gate-api/v3/playlists/spotify/by-track/") def get_spotify_playlists_data_go(self, isrc, streams_market, streams_type, keep_zero_streams, offset, limit, order_by=None, start_date=None, end_date=None, ignore_response_attaching=False): api_string = "/gate-api/v3/playlists/spotify/by-track/" resp = self.session.get( self.host_endpoint + api_string, params=dict(isrc=isrc, streams_market=streams_market, start_date=start_date, end_date=end_date, streams_type=streams_type, keep_zero_streams=keep_zero_streams, offset=offset, limit=limit, order_by=order_by), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting Go Apple playlists data from /gate-api/v3/playlists/apple/by-track/") def get_apple_playlists_data_go(self, isrc, market, streams_type, keep_zero_streams, offset, limit, order_by=None, start_date=None, end_date=None, ignore_response_attaching=False): api_string = "/gate-api/v3/playlists/apple/by-track/" resp = self.session.get( self.host_endpoint + api_string, params=dict(isrc=isrc, market=market, start_date=start_date, end_date=end_date, limit=limit, streams_type=streams_type, keep_zero_streams=keep_zero_streams, offset=offset, order_by=order_by), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting Go Amazon playlists data from /gate-api/playlists/amazon/by-track/") def get_amazon_playlists_data_go(self, isrc, market, streams_type, keep_zero_streams, offset, limit, order_by=None, start_date=None, end_date=None, ignore_response_attaching=False): api_string = "/gate-api/playlists/amazon/by-track/" resp = self.session.get( self.host_endpoint + api_string, params=dict(isrc=isrc, market=market, start_date=start_date, end_date=end_date, limit=limit, streams_type=streams_type, keep_zero_streams=keep_zero_streams, offset=offset, order_by=order_by), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting chart placement for Go from /apollo-api/v1/charts/summary/") def get_chart_placement_go(self, *parameters, token, ignore_response_attaching=False): # TODO: investigate usage headers = {"authorization": "Bearer {}".format(token)} api_string = "/apollo-api/v1/charts/summary/?" params_list = "&".join(parameters) resp = self.session.get(self.host_endpoint + api_string + params_list, headers=headers, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting track's playlist data from /gate-api/v2/playlists/spotify/track/") def get_v2_spotify_playlist_track(self, isrc, playlist_id, market, ignore_response_attaching=False): api_string = "/gate-api/v2/playlists/spotify/track/" resp = self.session.get( self.host_endpoint + api_string, params=dict(isrc=isrc, playlist_id=playlist_id, market=market), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting track's playlist data from /gate-api/v2/playlists/apple/track/") def get_v2_apple_music_playlist_track(self, isrc, playlist_id, market, ignore_response_attaching=False): api_string = "/gate-api/v2/playlists/apple/track/" resp = self.session.get( self.host_endpoint + api_string, params=dict(isrc=isrc, playlist_id=playlist_id, market=market.lower()), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting track's playlist data from /gate-api/v1/playlists/apple/tracklist/") def get_v1_apple_music_playlist_tracklist(self, playlist_id, market, order_by, limit, offset=None, ignore_response_attaching=False): api_string = "/gate-api/v1/playlists/apple/tracklist/" resp = self.session.get( self.host_endpoint + api_string, params=dict(playlist_id=playlist_id, market=market, order_by=order_by, limit=limit, offset=offset), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response def get_amazon_music_playlist_tracklist(self, playlist_id, market, order_by, limit, offset=None, ignore_response_attaching=False): api_string = "/gate-api/playlists/amazon/tracklist/" resp = self.session.get( self.host_endpoint + api_string, params=dict(playlist_id=playlist_id, market=market, order_by=order_by, limit=limit, offset=offset), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting track's playlist data from /gate-api/v1/playlists/spotify/tracklist/") def get_spotify_playlist_tracklist(self, playlist_id, market=None, order_by=None, limit=20, offset=None, ignore_response_attaching=False): api_string = "/gate-api/v1/playlists/spotify/tracklist/" resp = self.session.get( self.host_endpoint + api_string, params=dict(playlist_id=playlist_id, market=market, order_by=order_by, limit=limit, offset=offset), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response def post_v1_starred_tracks(self, isrc, uri, ignore_response_attaching=False): api_string = "/apollo-api/v1/starred-tracks/" payload = {"isrc": isrc, "uri": uri} resp = self.session.post(self.host_endpoint + api_string, json=payload) attach_response_info(resp, ignore_response_attaching) return {"status_code": resp.status_code, "body": resp.json()} @allure.step("POST /apollo-api/v1/starred-tracks/") def post_v1_starred_tracks_spotify(self, isrc, track_id, ignore_response_attaching=False): return self.post_v1_starred_tracks(isrc, "spotify:track:{}".format(track_id), ignore_response_attaching) @allure.step("POST /apollo-api/v1/starred-tracks/") def post_v1_starred_tracks_apple(self, isrc, track_id, ignore_response_attaching=False): return self.post_v1_starred_tracks(isrc, track_id, ignore_response_attaching) @allure.step("DELETE /apollo-api/v1/starred-tracks/") def delete_v1_starred_tracks(self, isrc, ignore_response_attaching=False): api_string = f"/apollo-api/v1/starred-tracks/?isrc={isrc}" resp = self.session.delete(self.host_endpoint + api_string) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /user-data-api/v1/favorites/playlists/") def get_v1_star_playlist(self, ignore_response_attaching=False): api_string = "/user-data-api/v1/favorites/playlists/" resp = self.session.get(self.host_endpoint + api_string, headers={"X-App-Slug": "apollo"}) attach_response_info(resp, ignore_response_attaching) return {"status_code": resp.status_code, "body": resp.json()} @allure.step("POST /user-data-api/v1/favorites/playlists/") def post_v1_star_playlist(self, playlist_id, vendor, country_code=None, ignore_response_attaching=False): api_string = "/user-data-api/v1/favorites/playlists/" headers = {"X-App-Slug": "apollo"} if country_code: payload = {"id": playlist_id, "vendor": vendor, "country_code": country_code} else: payload = {"id": playlist_id, "vendor": vendor} resp = self.session.post(self.host_endpoint + api_string, headers=headers, json=payload) attach_response_info(resp, ignore_response_attaching) return {"status_code": resp.status_code, "body": resp.json()} @allure.step("DELETE /user-data-api/v1/favorites/playlists/") def delete_v1_star_playlist(self, playlist_id, vendor, country_code=None): api_string = "/user-data-api/v1/favorites/playlists/" headers = {"X-App-Slug": "apollo"} if country_code: payload = {"id": playlist_id, "vendor": vendor, "country_code": country_code} else: payload = {"id": playlist_id, "vendor": vendor} self.session.delete(self.host_endpoint + api_string, headers=headers, json=payload) @allure.step("GET /apollo-api/push-notifications/") def get_push_notifications(self, limit=20, offset=0, search=None, type=None, ignore_response_attaching=False): # api_string = "/apollo-api/push-notifications/?is_new=false&offset=0&limit=20" api_string = "/user-data-api/users/messages/feed/?" headers = {"X-App-Slug": "apollo"} resp = self.session.get( self.host_endpoint + api_string, params=dict(limit=limit, offset=offset, search=search, type=type), headers=headers, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /apollo-api/markets/") def get_markets(self, ignore_response_attaching=False): api_string = "/apollo-api/markets/" resp = self.session.get(self.host_endpoint + api_string, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /gate-api/v3/playlists/apple/by-track/") def get_v1_apple_music_track_top_playlists(self, isrc, start_date, end_date, market: List, limit, order_by, keep_zero_streams, ignore_response_attaching=False): api_string = "/gate-api/v3/playlists/apple/by-track/" resp = self.session.get( self.host_endpoint + api_string, params=dict( isrc=isrc, start_date=start_date, end_date=end_date, market=market, limit=limit, order_by=order_by, keep_zero_streams=keep_zero_streams, ), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /gate-api/v3/playlists/spotify/by-track/") def get_v1_spotify_track_top_playlists(self, isrc, start_date, end_date, market, limit, streams_market, order_by, ignore_response_attaching=False): api_string = "/gate-api/v3/playlists/spotify/by-track/" resp = self.session.get( self.host_endpoint + api_string, params=dict( isrc=isrc, start_date=start_date, end_date=end_date, market=market, limit=limit, streams_market=streams_market, order_by=order_by, ), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /gate-api/v3/playlists/spotify/by-track/") def get_v3_spotify_track_playlists(self, isrc, market, order_by, recent_adds_only, limit, streams_market=None, ignore_response_attaching=False): api_string = "/gate-api/v3/playlists/spotify/by-track/" resp = self.session.get( self.host_endpoint + api_string, params=dict( isrc=isrc, market=market, order_by=order_by, recent_adds_only=recent_adds_only, limit=limit, streams_market=streams_market, ), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /gate-api/v3/playlists/apple/by-track/") def get_v1_apple_music_track_playlists(self, isrc, market: List, order_by, recent_adds_only, limit, ignore_response_attaching=False): api_string = "/gate-api/v3/playlists/apple/by-track/" resp = self.session.get( self.host_endpoint + api_string, params=dict(isrc=isrc, market=market, order_by=order_by, recent_adds_only=recent_adds_only, limit=limit), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /apollo-api/search/history/") def get_search_history(self, _type, limit, vendor: List = None, ignore_response_attaching=False): api_string = "/apollo-api/search/recent/" resp = self.session.get(self.host_endpoint + api_string, params=dict(type=_type, limit=limit, vendor=vendor), timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("POST /apollo-api/search/history/") def post_search_history(self, payload): api_string = "/apollo-api/search/recent/" payload = payload self.session.post(self.host_endpoint + api_string, json=payload) @allure.step("GET /apollo-api/v1/spotify/playlists/listeners/graph/") def get_v1_spotify_playlists_listeners_graph(self, vendor, start_date, end_date, market, _id, ignore_response_attaching=False): api_string = "/apollo-api/v1/spotify/playlists/listeners/graph/" resp = self.session.get( self.host_endpoint + api_string, params=dict(vendor=vendor, start_date=start_date, end_date=end_date, market=market, id=_id), timeout=self.timeout, ) attach_response_info(resp, ignore_response_attaching) return ListenersGraph(**resp.json()) @allure.step("GET /apollo-api/v1/spotify/playlists/listeners/range/") def get_v1_spotify_playlists_listeners_range(self, playlist_id, start_date, end_date, market, ignore_response_attaching=False): api_string = f"/apollo-api/v1/spotify/playlists/listeners/range/{playlist_id}/" resp = self.session.get( self.host_endpoint + api_string, params=dict(start_date=start_date, end_date=end_date, market=market), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /apollo-api/v1/spotify/playlists/followers/range/") def get_v1_spotify_playlists_followers_range(self, playlist_id, start_date, end_date, market, ignore_response_attaching=False): api_string = f"/apollo-api/v1/spotify/playlists/followers/range/{playlist_id}/" resp = self.session.get( self.host_endpoint + api_string, params=dict(start_date=start_date, end_date=end_date, market=market), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /apollo-api/v1/spotify/playlists/followers/graph/") def get_v1_spotify_playlists_followers_graph(self, *parameters, ignore_response_attaching=False): api_string = "/apollo-api/v1/spotify/playlists/followers/graph/?" params_list = "&".join(parameters) resp = self.session.get(self.host_endpoint + api_string + params_list, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /gate-api/v3/playlists/spotify/by-track/history/") def get_v3_spotify_track_previous_playlists(self, *parameters, ignore_response_attaching=False): api_string = "/gate-api/v3/playlists/spotify/by-track/history/?" params_list = "&".join(parameters) resp = self.session.get(self.host_endpoint + api_string + params_list, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET Portal /gate-api/v1/playlists/spotify/by-track/history/") def get_portal_v1_spotify_track_previous_playlists(self, *parameters, ignore_response_attaching=False): api_string = "/gate-api/v1/playlists/spotify/by-track/history/?" params_list = "&".join(parameters) resp = self.session.get(self.host_endpoint + api_string + params_list, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET Portal /gate-api/v1/playlists/apple/by-track/history/") def get_portal_v1_apple_track_previous_playlists(self, *parameters, ignore_response_attaching=False): api_string = "/gate-api/v1/playlists/apple/by-track/history/?" params_list = "&".join(parameters) resp = self.session.get(self.host_endpoint + api_string + params_list, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("GET /gate-api/v3/playlists/apple/by-track/history/") def get_v3_apple_music_track_previous_playlists(self, *parameters, ignore_response_attaching=False): api_string = "/gate-api/v3/playlists/apple/by-track/history/?" params_list = "&".join(parameters) resp = self.session.get(self.host_endpoint + api_string + params_list, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response @allure.step("Getting list of tracks /vendor-api/spotify/v1/search/") def get_list_of_spotify_tracks(self, name, ignore_response_attaching=False): resp = self.session.get( self.host_endpoint + "/vendor-api/spotify/v1/search?limit=20&offset=0&vendor=spotify&query={}".format(name), timeout=self.timeout, ) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response def get_playlist_search_results(self, query, vendor: List = None): query = query[:50] # Length must be between 2 and 50 resp = self.session.get(self.host_endpoint + "/gate-api/v1/playlists/search/", params=dict(offset=0, limit=30, query=query, vendor=vendor)) return [PlaylistSearchResult(**item, position=i) for i, item in enumerate(resp.json()["items"], start=1)] def get_full_charts_data(self, vendor, market, fields=None, order_by="position", timestamp=None): timestamp = timestamp or round(time.time() * 1000) fields = fields or [ "artist", "artists", "change", "id", "image_url", "is_starred", "isrc", "name", "position", "is_sony", "is_new", "is_re_enter", "is_starred_track", ] tracks = self.session.get( self.host_endpoint + "/gate-api/charts/top/", params=dict(vendor=vendor, market=market, fields=fields, order_by=order_by, timestamp=timestamp), ).json()["tracks"] return [TopChartsTrack(**kv) for kv in tracks] def get_spotify_track_playlists(self, **kwargs): api_string = '/gate-api/v3/playlists/spotify/by-track/' payload = kwargs # timestamp = timestamp or round(time.time() * 1000) # order_by = order_by or ['-added_date', 'current_position'] response = self.session.get(self.host_endpoint + api_string, params=payload, timeout=self.timeout) return [PlaylistByTrack(**kv, position_ui=i) for i, kv in enumerate(response.json()["items"], start=1)] @allure.step("GET /gate-api/v1/tracks/playlists/hot-hits/") def get_hot_hits_playlists(self, ignore_response_attaching=False, **kwargs): api_string = "/gate-api/v1/tracks/playlists/hot-hits/?" payload = kwargs resp = self.session.get(self.host_endpoint + api_string, params=payload, timeout=self.timeout) response = {"status_code": resp.status_code, "body": resp.json()} attach_response_info(resp, ignore_response_attaching) return response["body"]