""" Apple Music Analytics Utils. Util module for working with Music Analytics Java application to request In Review report from Music Analytics API. Requires following environment variables: JAR_FILE - A Reporter jar file name (needs to be in the swf direcotry) PRIVATE_KEY - Private key value for auth key (provided from Secrets Manager) General Guide https://help.apple.com/itc/musicanalyticsapi/#/itc120a2ce63 """ from abc import ABC, abstractmethod import datetime import os import re import shlex import command as cmd import requests def set_token(f): """Decorate with setting a fresh token.""" def set_token_wrapper(self, *args, **kwargs): self.generate_key_and_set_token() return f(self, *args, **kwargs) return set_token_wrapper class Validator(ABC): """Validator abstract class.""" def __setattr__(self, key, value): """Set attribute.""" self.validate(key, value) ABC.__setattr__(self, key, value) @abstractmethod def validate(self, key, value): """Validate.""" pass class TeamID(Validator): """Validator for team ID.""" def __init__(self, team_id): """Initialise team ID validator.""" self.team_id = team_id def validate(self, key, value): """Validate team ID.""" if len(value) != 36: raise ValueError(f'Expected {value} to be a 36-character string.') class EmptyReportException(Exception): """Empty report exception.""" def __init__(self, rptg_date, not_today=False): """Initialise date and message.""" self.date = rptg_date self.message = f'Music Analytics API has ' \ f'returned 200 status code, however ' \ f'the report does not exist for the date {rptg_date}.' \ if not not_today else f'{rptg_date} is the future date.' class PrivateKeyFilePath(Validator): """Validator for private key path.""" def __init__(self, key_id): """Private key file path validator.""" __private_key_file = 'AuthKey_' + key_id + '.p8' self.file = __private_key_file def validate(self, key, value): """Validate AuthKey path/filename.""" if not re.findall(r'(/(\w)+)*/*AuthKey_([A-Z0-9]{10,}).p8', value): raise ValueError( f'Expected file {value} using ' f'AuthKey_.p8 name-pattern.\n' f'And the Key ID should be 10-character uppercase string.') class MusicAnalyticsUtils(object): """Class encapsulating convenience methods to handle iTunes reports.""" def __init__( self, music_analytics_jar, key_id, team_id, private_key): """Initialize an ITunesReporter object. Object is vendor & date specific. Args: music_analytics_jar (str): Path to the Analytics app jar file. key_id (str): Key ID supplied by Apple. team_id (str): Team ID for private key file. """ self.music_analytics_jar = music_analytics_jar self.key_id, self.team_id = key_id, TeamID(team_id).team_id self.__private_key = private_key self.__private_key_file_path = PrivateKeyFilePath( self.key_id).file def get_private_key_file(self): """Create or open the private key file.""" try: _private_key_file = open(self.__private_key_file_path, 'r') # if a file is corrupted, just delete it, # the method will create a new one. except FileNotFoundError: _private_key = str( '-----BEGIN PRIVATE KEY-----\n' + # noqa: W504 self.__private_key.replace('\\n', '\n') + # noqa: W504 '\n-----END PRIVATE KEY-----') _private_key_file = open(self.__private_key_file_path, 'w+') _private_key_file.write(_private_key), _private_key_file.close() return _private_key_file def get_private_key_path_from_os(self): """Get private key path from os.""" return self.__private_key_file_path def generate_fresh_token( self, private_key_path=None): """Generate a fresh token for the application. For Arg meaning, see: https://help.apple.com/itc/musicanalyticsapi/#/itc098ad2577 Args: private_key_path (str): A private key file path. Returns: str: Output from Utils Application. """ if not private_key_path: private_key_path = self.get_private_key_path_from_os() cmd_str = (f'java -jar {self.music_analytics_jar} ' f'-c {self.team_id} ' # noqa f'-f {private_key_path}') command = shlex.split(cmd_str) try: _response = cmd.run(command) output = _response.output.decode('utf-8') if output[-1] == '\n': output = output[:-1] except cmd.core.CommandException as exception: if 'Was expecting pattern AuthKey_.p8' \ in exception.message: print(f'The file {private_key_path} has a wrong name.\n' f'Should be named as AuthKey_.p8.') if f'Problem parsing private key from file' \ f' {private_key_path}' in exception.message: print(f'The file {private_key_path} seems to be corrupted.\n' f'Delete it, and it will be regenerated.') if 'Unable to access jarfile' in exception.message: print(f'The .jar file {self.music_analytics_jar} seems to ' f'have a wrong name or be corrupted. ' f'Specify the correct path to it.') raise exception os.remove(private_key_path) return output class MusicAnalyticsAPI(object): """Music Analytics API class.""" def __init__(self, _utils: MusicAnalyticsUtils): """Music Analytics API initialisation.""" self.utils = _utils self.base_url = 'https://musicanalytics.apple.com/reports' self.__token = None self.__header = None self.__empty_report_len = 247 def generate_key_and_set_token(self): """Generate kay and token for headers.""" self.utils.get_private_key_file() self.__token = self.utils.generate_fresh_token() self.__header = {'Authorization': f'Bearer {self.__token}'} def get_in_review_report(self, rptg_date): """Get InReview report endpoint response. Args: rptg_date (str): YYYY-MM-DD str for report requested (will provide sample if not set when methods are executed). """ if datetime.datetime.strptime(rptg_date, '%Y-%m-%d') > \ datetime.datetime.today(): raise EmptyReportException(rptg_date, not_today=True) params = {'rptg_date': rptg_date} _response = requests.get( f'{self.base_url}/in-review/v1', params=params, headers=self.__header) if len(_response.text) > self.__empty_report_len: return _response.text if _response.status_code == 404: raise Exception('The request returned 404 status code.') else: raise EmptyReportException(rptg_date) def get_excluded_streams_report(self, start_date): """Get InReview report endpoint response. Args: start_date (str): YYYY-MM-DD str for report requested (will provide sample if not set when methods are executed). """ if datetime.datetime.strptime(start_date, '%Y-%m-%d') > \ datetime.datetime.today(): raise EmptyReportException(start_date, not_today=True) params = {'start_date': start_date} _response = requests.get( f'{self.base_url}/excluded-streams/v1', params=params, headers=self.__header) if len(_response.text) > self.__empty_report_len: return _response.text if _response.status_code == 404: raise Exception('The request returned 404 status code.') else: raise EmptyReportException(start_date) @set_token def get_report_raw_data(self, report_type, date): """Get report raw data by report_type.""" if report_type == 'in_review': return self.get_in_review_report(rptg_date=date) if report_type == 'excluded_streams': return self.get_excluded_streams_report(start_date=date)