""" This module will provide you with methods to easily use any endpoint from the [Luminate API](https://app.swaggerhub.com/apis/LUMINATE-DATA/MusicConnect-API/). You'll also find some methods for tracks and artists endpoints to fetch data directly in a clean pandas DataFrame. ???+ warning "Pre-requisites" To use this module, the following variables must be defined in your .env file: `luminateLogin`, `luminatePassword`, `luminateApiKey` See the [.env file configuration][configure-your-env-file] section if needed To use this module in your script, import the module like so. ``` from djagitit.api import luminate ``` """ from dotenv import load_dotenv import requests import os import pandas as pd base_url = 'https://api.musicconnect.mrc-data.com/api/' load_dotenv() class Lumos : """Class for interacting with Luminate API.""" def __init__(self): """Initialize Lumos class with API credentials.""" # Set the authentication URL for the API self.url = 'https://api.musicconnect.mrc-data.com/auth' # Get the API key, username, and password from the environment variables self.api_key = os.getenv("luminateApiKey") self.username = os.getenv("luminateLogin") self.password = os.getenv("luminatePassword") # Authenticate and get the headers for API requests self.headers = self.__auth() def __auth(self): """Authenticate and obtain headers for both POST and GET for API requests.""" # Define headers for the initial authentication request headers = { 'accept': 'application/json', 'x-api-key': self.api_key, 'Content-Type': 'application/x-www-form-urlencoded', } # Define the data payload for the authentication request data = { 'username': self.username, 'password': self.password, } # Send the authentication request response = requests.post(self.url, headers=headers, data=data) # Extract the access token from the response access_token = response.json().get('access_token') # If the authentication was successful, prepare the headers for subsequent API requests if response.status_code == 200 : # Headers for POST requests headers_post = { 'accept': 'application/vnd.mrc-data.dashboard.v1+json', 'Authorization': access_token, 'x-api-key': self.api_key, 'Content-Type': 'application/vnd.mrc-data.dashboard.v1+json', } # Headers for GET requests headers_get = { 'accept': 'application/vnd.mrc-data.search.v1+json', 'Authorization': access_token, 'x-api-key': self.api_key, } # Return both sets of headers return headers_post, headers_get # If the authentication failed, raise an exception else: raise Exception(f'failed to obtain access_token {response.text}') def post(self, endpoint, params): """ Makes a POST request to the API. Args: endpoint (str): The API endpoint to make the request to. params (dict): The parameters to include in the POST request. Returns: r (http response): The response from the API. """ # Try to make a POST request to the API try : r = requests.post(f'{base_url}{endpoint}', headers=self.headers[0], json=params) # If the request is successful, raise a status r.raise_for_status() # If there is an HTTP error during the request, handle it except requests.exceptions.HTTPError as e: # Create a response message with the error response_message = f"Unexpected HTTP error: {e}" print(response_message) # Return None if there is an error return None # If the request is successful, return the response return r def get(self, endpoint, params): """ Makes a GET request to the API. Args: endpoint (str): The API endpoint to make the request to. params (dict): The parameters to include in the GET request. Returns: r (http response): The response from the API. """ # Try to make a GET request to the API try : r = requests.get(f'{base_url}{endpoint}', headers=self.headers[1], params=params) # If the request is successful, raise a status r.raise_for_status() # If there is an HTTP error during the request, handle it except requests.exceptions.HTTPError as e: response_message = f"Unexpected HTTP error: {e}" print(response_message) # Return None if there is an error return None # If the request is successful, return the response return r def __utilsMissingWeeks(self, missing_weeks, weeks): """Handle missing weeks from the results.""" # Check if there are any missing weeks if len(missing_weeks)!=0: # Check f the number of missing weeks is not equal to the total number of weeks if len(missing_weeks) != len(weeks) : # Print the missing weeks print('these weeks are missing: ' + ','.join(str(item) for item in missing_weeks)) else : # If all weeks are missing, print a message indicating no results print('there are no results for your selection') def __setTrackEndpoint(self, id, song_level): """Pick API endpoint for track data.""" # Check if the id is a string (ie should be an ISRC) but starts with a digit (ie should be a song_id) if isinstance(id, str) and id[:2].isdigit(): return print('song_id is an integer value') # Check if the id is a string (ie is an ISRC) if isinstance(id, str): # If song_level is not activated, the endpoint has to be the isrc one if not song_level: endpoint = f'isrc/data/{id}' # song_level is activated, the endpoint has to be the isrc/song one' else : endpoint = f'isrc/data/{id}/song' # If the id is not a string (ie is a song_id), the endpoint has to be the song one else: endpoint = f'song/{id}/data' # Return the endpoint return endpoint def __setApiCallParams(self, week=None, daily_details=False, country=None, metric=None): """Define parameters for track data api call.""" params = {} # Set the 'metric' parameter if it is provided, else set it by default to 'Summary' params.update({'metric': f'{metric}'}) if metric else params.update({'metric': 'Summary'}) # Set the 'country' parameter if it is provided, else set it by default to 'G1' params.update({'country': f'{country}'}) if country else params.update({'country': 'G1'}) # Set the 'week_id' parameter if it is provided params.update({'week_id': week}) if week else None # Set the 'day_id' parameter to 0 if daily_details is True params.update({'day_id': 0}) if daily_details else None # Return the parameters return params def __apiCallTrack(self, id, week=None, daily_details=False, song_level=False, country=None): """Make API call and return response data.""" # Set the endpoint endpoint = self.__setTrackEndpoint(id, song_level) # Set the parameters params = self.__setApiCallParams(week, daily_details, country) # Make a POST request to the API r = self.post(endpoint, params) # Parse the JSON response data = r.json() # If the 'metrics' field in the response is empty, print a message and return if len(data['metrics']) == 0: print(f'No data for track {id} on week {week}') return None # Return the parsed data return data def __apiCallArtist(self, artist_id, week=None, daily_details=False, country=None): """Get artist data, used in the artist_split function.""" # Set the endpoint endpoint = f'artist/{artist_id}/data' # Set the parameters params = self.__setApiCallParams(week=week, daily_details=daily_details, country=country) # Make a POST request to the API r = self.post(endpoint, params) # Parse the JSON response data = r.json() # If the 'metrics' field in the response is empty, print a message and return if len(data['metrics']) == 0: print(f'No data for artist {artist_id} on week {week}') return None # Return the parsed data return data def __processData(self, entity, data, id, daily_details=None, song_level=None, country='G1'): """Process response data and return final dataframe.""" # Check if the entity is valid valid_entities = {'track', 'artist'} if entity not in valid_entities: raise ValueError(f"'{entity}' not valid, entity must be one of 'track' or 'artist'") # Extract metrics data from the response data metrics_data = data['metrics'] # Create a dataframe from with the first element of metrics data df = pd.DataFrame(metrics_data[0]['value']) # If daily details are requested, rename value columns with the metric name and drop unused columns if daily_details: df.rename(columns={'value': metrics_data[0]['name']}, inplace=True) df.drop(columns=['name'], inplace=True) # Otherwise, pivot the dataframe to get all value in columns and prefix value columns with the metric name else: df = df.pivot_table(columns='name', values='value', aggfunc='first').reset_index(drop=True) df.columns = [metrics_data[0]['name']+' ' + col.upper() for col in df.columns.tolist()] # Add additional columns to the dataframe df['country'] = country if country else 'G1' df['id'] = id df['artist'] = data['artist_name'] # If the entity is a track, add the title to the dataframe if entity=='track': df['title'] = data['title_name'] df['week'] = data['week_id'] # Reorder the columns in the dataframe temp_cols = df.columns.tolist() if entity=='track': new_cols = temp_cols[-5:] + temp_cols[:-5] elif entity=='artist': new_cols = temp_cols[-4:] + temp_cols[:-4] df = df[new_cols] # Process all other metrics in the response data for metric in metrics_data[1:]: metric_df = pd.DataFrame(metric['value']) # If daily details are requested, rename value columns with the metric name and keep only the value column if daily_details: metric_df.rename(columns={'value': metric['name']}, inplace=True) metric_df = metric_df[[metric['name']]] # Otherwise, pivot the dataframe to get all value in columns and prefix value columns with the metric name else: metric_df = metric_df.pivot_table(columns='name', values='value', aggfunc='first').reset_index(drop=True) metric_df.columns = [metric['name']+' ' + col.upper() for col in metric_df.columns.tolist()] # Merge the metric dataframe with the main dataframe df = pd.merge(df, metric_df, left_index=True, right_index=True) # Rename the id column based on the entity and id type # Rename the id column based on the entity, id type (isrc or song_id) and song_level if entity=='track': id_column = 'song_id' if not isinstance(id, str) else 'song_isrc' if song_level else 'isrc' elif entity=='artist': id_column = 'artist_id' df.rename(columns={'id': id_column}, inplace=True) # Clean the column names df.columns = list(map(lambda x: x.replace('Streaming On-Demand', 'Streaming').replace('Digital Songs','Downloads'), df.columns)) # Return the final dataframe return df def __track_volumes(self, id, weeks=None, song_level=False, daily_details=False, country=None): """ISRC or SONG level audio and video streaming split.""" # Check if weeks is not a list, if not convert it to a list if not isinstance(weeks, list): weeks = [weeks] # Initialize empty lists for storing dataframes and missing weeks df_list = [] missing_weeks = [] for week in weeks: # Call the API for each week and process the data data = self.__apiCallTrack(id, week, daily_details, song_level, country) # If no data is returned for a week, add it to the missing weeks list if data is None: missing_weeks.append(week) else: week_df = self.__processData('track', data, id, daily_details, song_level, country) # Add the dataframe for each week to the list df_list.append(week_df) # Handle any missing weeks self.__utilsMissingWeeks(missing_weeks, weeks) # Concatenate all the weekly dataframes into one df = pd.concat(df_list) if len(df_list) > 0 else pd.DataFrame() # Return the final dataframe return df def isrc_volumes_daily(self, isrc, weeks=None, country=None): """ Get daily volumes for a given ISRC. Args: isrc (str): The ISRC to get volumes for. weeks (list of int or int, optional): List of weeks to get data for. Defaults to None and will get data for the latest week available. country (str, optional): The country to get data for. Defaults to None and will get data at the G1 level. Returns: df (pd.DataFrame): A dataframe with the daily volumes for the given ISRC. """ return self.__track_volumes(isrc, weeks, daily_details=True, country=country) def isrc_volumes_weekly(self, isrc, weeks=None, country=None): """ Get weekly volumes for a given ISRC. Args: isrc (str): The ISRC to get volumes for. weeks (list of int or int, optional): List of weeks to get data for. Defaults to None and will get data for the latest week available. country (str, optional): The country to get data for. Defaults to None and will get data at the G1 level. Returns: df (pd.DataFrame): A dataframe with the weekly volumes for the given ISRC. """ return self.__track_volumes(isrc, weeks, daily_details=False, country=country) def song_volumes_daily(self, id, weeks=None, country=None): """ Get daily volumes for a given song. Args: id (int or str): The song id to get volumes for or an isrc to get volumes for at the song level. weeks (list of int or int, optional): List of weeks to get data for. Defaults to None and will get data for the latest week available. country (str, optional): The country to get data for. Defaults to None and will get data at the G1 level. Returns: df (pd.DataFrame): A dataframe with the daily volumes for the given song. """ song_level = True if isinstance(id, str) else False return self.__track_volumes(id, weeks, song_level=song_level, daily_details=True, country=country) def song_volumes_weekly(self, id, weeks=None, country=None): """ Get weekly volumes for a given song. Args: id (int or str): The song id to get volumes for or an isrc to get volumes for at the song level. weeks (list of int or int, optional): List of weeks to get data for. Defaults to None and will get data for the latest week available. country (str, optional): The country to get data for. Defaults to None and will get data at the G1 level. Returns: df (pd.DataFrame): A dataframe with the weekly volumes for the given song. """ song_level = True if isinstance(id, str) else False return self.__track_volumes(id, weeks, song_level=song_level, daily_details=False, country=country) def __artist_volumes(self, id, weeks=None, daily_details=False, country=None): """Artist audio and video streaming split.""" # Check if weeks is not a list, if not convert it to a list if not isinstance(weeks, list): weeks = [weeks] # Initialize empty lists for storing dataframes and missing weeks df_list = [] missing_weeks = [] for week in weeks: # Call the API for each week and process the data data = self.__apiCallArtist(id, week, daily_details, country) if data is None: # If no data is returned for a week, add it to the missing weeks list missing_weeks.append(week) else: week_df = self.__processData('artist', data, id, daily_details=daily_details, country=country) # Add the dataframe for each week to the list df_list.append(week_df) # Handle any missing weeks self.__utilsMissingWeeks(missing_weeks, weeks) # Concatenate all the weekly dataframes into one df = pd.concat(df_list) if len(df_list) > 0 else pd.DataFrame() # Return the final dataframe return df def artist_volumes_daily(self, artist_id, weeks=None, country=None): """ Get daily volumes for a given artist. Args: artist_id (int): The artist id to get volumes for. weeks (list of int or int, optional): List of weeks to get data for. Defaults to None and will get data for the latest week available. country (str, optional): The country to get data for. Defaults to None and will get data at the G1 level. Returns: df (pd.DataFrame): A dataframe with the daily volumes for the given artist. """ return self.__artist_volumes(artist_id, weeks, daily_details=True, country=country) def artist_volumes_weekly(self, artist_id, weeks=None, country=None): """ Get weekly volumes for a given artist. Args: artist_id (int): The artist id to get volumes for. weeks (list of int or int, optional): List of weeks to get data for. Defaults to None and will get data for the latest week available. country (str, optional): The country to get data for. Defaults to None and will get data at the G1 level. Returns: df (pd.DataFrame): A dataframe with the weekly volumes for the given artist. """ return self.__artist_volumes(artist_id, weeks, daily_details=False, country=country) def song_isrcs(self,song_id): """from a track's song_id , get all global isrcs linked to it from the API.""" # Define the endpoint endpoint = f'relationship/song-isrc/{song_id}' # Set the parameters params = {"country": "G1"} # Make a GET request to the API r = self.get(endpoint, params) # Parse the JSON response data = r.json() # Convert the data to a pandas DataFrame df = pd.DataFrame(data) # Return the DataFrame return df