import re class RenamingStrategy: def __init__(self, logger, regex): self._logger = logger self._regex = re.compile(regex) self._licensor_mapping = { 'sony': 'sme', 'smej': 'smejp', 'smejintl': 'smejpintl', 'theorchard': 'theorchard' } def can_handle(self, key: str) -> bool: """ Checks if current strategy can handle a key. A preliminary check that is called to determine if a strategy can process a given key. By default, this method returns true if key matches regexp but it can be overridden in child classes as needed. Args: key: S3 key Returns: True if key is supported by strategy """ return True if self._regex.fullmatch(key) else False def _do_rename(self, key: str) -> str: raise NotImplementedError() def rename(self, key: str) -> str: """ Transforms S3 key for spotify data according to renaming strategy. :param key: Original S3 key. :return: Transformed S3 key. """ if not key: raise ValueError('Key must not be empty') return self._do_rename(key) @staticmethod def select_renaming_strategy(strategies, key: str): renaming_strategy = None # Select appropriate renaming strategy from the list for strategy in strategies: if strategy.can_handle(key): renaming_strategy = strategy break return renaming_strategy class SpotifyRenamingStrategy(RenamingStrategy): def __init__(self, logger): pattern = '^spotify/\\S*/(\\w+)_(\\d{4}-\\d{2}-\\d{2})_(\\w+)_(v[0-9]+(?=\\.))[.0-9]*_?((?<=_)[a-z]+)?_(\\w+)\\.ndjson$' super().__init__(logger, pattern) def _do_rename(self, key: str) -> str: if not key: raise ValueError('Key must not be empty') matching_result = self._regex.findall(key) if not matching_result: raise ValueError('Unable match spotify key: `{0}`'.format(key)) (dsp, date, licensor, version, country, report) = matching_result[0] transformed = '{dsp}/{rep}/{ver}/report_date={dt}/report_licensor={lic}/{filename}.gz'.format( dsp=dsp, rep=report, ver=version, dt=date, lic=self._licensor_mapping[licensor], filename=report + '_' + date.replace('-', '') + (('_' + country.upper()) if country else '') ) self._logger.debug('Spotify key transformed: %s -> %s', key, transformed) return transformed class AppleMusicRenamingStrategy(RenamingStrategy): def _transform_key(self, values_tuple: tuple) -> str: raise NotImplementedError def _do_rename(self, key: str) -> str: matching_result = self._regex.findall(key) if not matching_result: raise ValueError('Unable match apple key: `{0}`'.format(key)) transformed = self._transform_key(matching_result[0]) self._logger.debug('AppleMusic key transformed: %s -> %s', key, transformed) return transformed @staticmethod def build_filename(date_str: str, report, vendor_id, version): return '_'.join(['AppleMusic', report, vendor_id, date_str.replace('-', ''), version.upper()]) class ApolloAppleMusicRenamingStrategy(AppleMusicRenamingStrategy): def __init__(self, logger): pattern = '^apple/\\S*/(\\w+)_(\\d{4}-\\d{2}-\\d{2})_([a-z]+)_([0-9]+)_(v[0-9_]+)_([a-zA-Z]+)\\.tsv$' super().__init__(logger, pattern) def _transform_key(self, values_tuple: tuple) -> str: (dsp, date, licensor, vendor_id, version, report) = values_tuple return '{dsp}/{rep}/{ver}/report_date={dt}/report_licensor={lic}/{filename}.txt.gz'.format( dsp=dsp, rep=str(report)[2:].lower(), ver=version, dt=date, lic=self._licensor_mapping[licensor], filename=AppleMusicRenamingStrategy.build_filename(date, str(report)[2:], vendor_id, version) ) class TheOrchardAppleMusicRenamingStrategy(AppleMusicRenamingStrategy): __DSP = 'apple' __REPORT_VERSION = 'v1_0' __LICENSOR = 'theorchard' def __init__(self, logger): pattern = '^apple/(\\d{4}-\\d{2}-\\d{2})/AppleMusic_([a-zA-Z]+)_([0-9]+)_\\d{4}\\d{2}\\d{2}\\.txt\\.gz$' super().__init__(logger, pattern) def _transform_key(self, values_tuple: tuple) -> str: (date, report, vendor_id) = values_tuple return '{dsp}/{rep}/{ver}/report_date={dt}/report_licensor={lic}/{filename}.txt.gz'.format( dsp=self.__DSP, rep=str(report).lower(), ver=self.__REPORT_VERSION, dt=date, lic=self.__LICENSOR, filename=AppleMusicRenamingStrategy.build_filename(date, report, vendor_id, self.__REPORT_VERSION) )