""" Amazon Music workflow tasks. Tasks to ingest and process Amazon Music data. """ import collections import datetime import gzip import os import shutil import zipfile import boto3 from botocore.exceptions import ClientError from garcon import task from garcon_contrib.aws.utils import garcon_s3 from garcon_contrib.dynamo_feed_status import garcon_feed_status from garcon_contrib.ftp import garcon_ftp from feed_ingestion.conf.config import merge_configs from feed_ingestion.flows import registered_executors from feed_ingestion.flows.amazon_music import config from feed_ingestion.flows.amazon_music.config import \ sme_file_name_regexp from feed_ingestion.flows.amazon_music.countries_config import \ count_entries, only_countries, org_countries_for_date from feed_ingestion.flows.amazon_music.generators import org_country_generator from feed_ingestion.flows.helpers import get_secret, get_sf_config from feed_ingestion.tasks import check_status from feed_ingestion.tasks import load_raw_table_tasks_sf from feed_ingestion.tasks.s3_tasks import _get_sme_s3_client, copy_s3_key from feed_ingestion.util import task_status from feed_ingestion.util.aws import s3 as s3utils from feed_ingestion.util.context_util import strtobool STOP_RESPONSE = {'stop': True} STAGING_PATH = 'file_stage_{report}_{licensor}' @task.decorate(timeout=1000) def bootstrap( activity, date, reload, soft_reload, report_name, countries=None, snowflake_error_limit=None, licensor=None, use_s3='False', snowflake_error_on_column_count_mismatch=None, extract_original_filename=False, populate_only='False', source=None): """Bootstrap workflow by injecting initial context from config. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). reload (str): If 'True' delete feed status in DynamoDB. soft_reload (str): If 'True' ignore some excessive steps. report_name (str): Name of the report to ingest. use_s3 (str): If flow should backfill data from S3 only. countries (str): Comma-separated list of country codes. Optional. If isn't provided then default from config is used snowflake_error_limit (int or None): Snowflake error limit. source (str): Source of staging data: 'sftp' (default, legacy partner drop files) or 'datapulse' (read from amazon_datapulse staging_raw tables in Snowflake). Returns: dict: Initial context for the workflow. """ if not licensor: licensor = 'theorchard' assert licensor in config.licensors, f'unsupported licensor "{licensor}"' source = source or config.SOURCE_SFTP assert source in config.sources, f'unsupported source "{source}"' if source == config.SOURCE_DATAPULSE: assert licensor in config.DATAPULSE_LICENSORS, ( f'source="datapulse" only supports licensors ' f'{config.DATAPULSE_LICENSORS}; got "{licensor}"' ) feed_name = '_'.join([config.feed_name, licensor, report_name]) date_obj = config.parse_date(date) if date else datetime.datetime.today() date = date_obj.strftime(config.DATE_FORMAT) use_s3 = use_s3 if use_s3 is not None else 'False' use_s3 = bool(strtobool(use_s3)) populate_only = populate_only if populate_only is not None else 'False' if reload == 'True': activity.logger.info( 'Delete status for feed: {} {} '.format(feed_name, date)) garcon_feed_status.delete_status(feed_name, date) else: overall_status = garcon_feed_status.get_overall_status(feed_name, date) if (overall_status == garcon_feed_status.STATUS_INGESTED or (populate_only == 'True' and overall_status == garcon_feed_status. STATUS_POPULATED_RAW_TABLE)): return STOP_RESPONSE if countries: country_list = _get_context_countries(countries) else: country_list = 'Empty countries' if not use_s3: country_list = only_countries( org_countries_for_date(licensor, report_name, date_obj.date())) assert country_list, 'Empty countries' if use_s3 and (licensor == 'awal' or licensor == 'altafonte'): s3_drop_bucket = config.awal_s3_drop_bucket if licensor == 'awal' \ else config.altafonte_s3_drop_bucket report_name_in_file = config.\ reports[report_name][f'{licensor}_report_name_in_filename'] expected_files = \ [file for file in s3utils.get_list_of_files_and_directories( s3_drop_bucket) if file[-1] != '/' and '_Activity_{date:%Y%m%d}_'.format(date=date_obj) in file and report_name_in_file in file] filenames = {file: file.split('/')[-1] for file in expected_files} duplicates = [item for item, count in collections.Counter( filenames.values()).items() if count > 1] versions = {file: {} for file in duplicates} for path, file in filenames.items(): if file in duplicates: versions[file].update( {path.split('/')[3].replace('version=', ''): path}) for file, versions_ in versions.items(): expected_version = max(versions.get(file).keys()) for version in versions_: if version != expected_version: expected_files.remove(versions_.get(version)) expected_files = {file.split('/')[-1]: file for file in expected_files} countries = [ file.split('_')[-1].split('.')[0] for file in expected_files] country_list = ','.join(countries) else: expected_files = _get_expected_files( date_obj, report_name, country_list, licensor) replace_archive_files = not task_status.is_completed_task( feed_name, date, 'set_overall_status_DOWNLOADED') drop_folder = config.reports[report_name].get( 'cucumber_drop_feed_name', config.reports[report_name]['cucumber_feed_name']) if licensor == 'sme' and report_name == 'unlimited': # for sme unlimited we have a different report folder # it's 'musicunlimited' instead of 'unlimited' report_location = 'musicunlimited' else: report_location = report_name s3_drop_location = config.s3['drop_path'][licensor].format( cucumber_feed_name=drop_folder, licensor=licensor, report_location=report_location, date=date_obj) archive_bucket = config.s3['archive_path'].format( cucumber_feed_name=config.reports[report_name]['cucumber_feed_name'], date=date_obj, licensor=licensor) if isinstance(snowflake_error_limit, int): snowflake_error_limit = snowflake_error_limit else: snowflake_error_limit = config.snowflake_error_limit if not use_s3 and licensor == 'altafonte': if 'ROE_EU' in country_list: country_list[country_list.index('ROE_EU')] = 'EU' if 'ROW_NA' in country_list: country_list[country_list.index('ROW_NA')] = 'NA' country_list = ','.join(country_list) common_kwargs = dict(licensor=licensor, use_s3=use_s3, country_list=country_list, source=source) snowflake_eoccm = snowflake_error_on_column_count_mismatch or 'true' clean_path = ''.join([archive_bucket, config.s3['clean_dir']]) if source == config.SOURCE_DATAPULSE: dimension_tables = config.dimension_tables_datapulse elif licensor == 'awal': dimension_tables = config.dimension_tables_awal elif licensor == 'altafonte': dimension_tables = config.dimension_tables_altafonte else: dimension_tables = config.dimension_tables if isinstance(soft_reload, bool): soft_reload_bool = soft_reload elif isinstance(soft_reload, str): soft_reload_bool = soft_reload.lower() == 'true' else: soft_reload_bool = False return dict( feed_name=feed_name, date=date, licensor=licensor, report_name=report_name, source=source, date_as_in_uuid=date.replace('-', ''), countries=country_list, use_s3=use_s3, populate_only=bool(strtobool(populate_only)), expected_files=expected_files, replace_archive_files=replace_archive_files, s3_drop_location=s3_drop_location, archive_bucket=archive_bucket, clean_path=clean_path, staging_raw_table=( config.DATAPULSE_STAGING_RAW_TABLE if source == config.SOURCE_DATAPULSE else config.reports[report_name]['staging_raw_table']), snowflake_error_limit=snowflake_error_limit, snowflake_error_on_column_count_mismatch=snowflake_eoccm, extract_original_filename=bool(extract_original_filename), common_kwargs=common_kwargs, dimension_tables=dimension_tables, jenkins_config=config.jenkins_config, soft_reload=soft_reload_bool ) def _list_keys_in_sme_s3(s3_bucket_and_path): sme_s3_client = _get_sme_s3_client() bucket_name, bucket_prefix = \ garcon_s3.extract_bucket_path(s3_bucket_and_path) s3_objects = sme_s3_client.list_objects( Bucket=bucket_name, Prefix=bucket_prefix ) s3_keys = [entry['Key'] for entry in s3_objects.get('Contents', [])] return s3_keys @task.decorate(timeout=600) def map_expected_files_to_sme(activity, date, s3_drop_location, expected_files): """Map files in SME S3 bucket to expected_files. It is possible to have multiple files for the same org, country, report_type In that scenario it take latest file from list as final and use it for ingestion writing a warning to logger. For example: 17:09:12 32282 PCO3_A_AU_20190220_20190220_Activity_20190221.txt.zip 17:09:12 32282 PCO3_A_AU_20190220_20190220_Activity_20190222.txt.zip 17:09:11 1038896 PCO3_A_AU_20190220_20190220_Activity_20190322.txt.zip Returns: dict: key 'map': mapping expected_file to path to file in SME bucket if destination file doesn't exist it maps to None value """ logger = activity.logger date_obj = datetime.datetime.strptime(date, config.DATE_FORMAT) destination_files = {} for key in _list_keys_in_sme_s3(s3_drop_location): file_name = key.split('/')[-1] match = sme_file_name_regexp.match(file_name) if not match: continue destination_file_name = config.file_template.format( org=match.group('org'), country=match.group('country'), date=date_obj, report_type=match.group('report_type'), ) if destination_file_name in destination_files: logger.warning(f'Overwriting {destination_file_name} ' f'with {file_name}') destination_files[destination_file_name] = key logger.info(f'{file_name} maps to {destination_file_name}') result_map = {} has_files = False for file_name in expected_files: destination_file = destination_files.get(file_name) if destination_file: has_files = True result_map[file_name] = destination_file return dict( map=result_map, has_files=has_files, ) @task.decorate(timeout=7200) def sme_grab_and_clean(activity, report_name, date, source_bucket_name, source_key_name, destination_bucket_name, destination_key_name, feed_name): """Download files from SME s3 bucket, clean and upload.""" logger = activity.logger destination_s3_bucket = boto3.resource('s3').Bucket( destination_bucket_name ) staging_dir_for_date = os.path.join( STAGING_PATH.format(report=report_name, licensor='sme'), date) os.makedirs(staging_dir_for_date, exist_ok=True) downloaded_file_path = os.path.join( staging_dir_for_date, os.path.basename(source_key_name) ) # check if it was already downloaded destination_s3_key_path = garcon_s3.get_destination_s3key_path( bucket=destination_bucket_name, destination_s3_key=destination_key_name ) current_status = garcon_feed_status.get_status( feed_name=feed_name, date=date, file_name=destination_s3_key_path) if current_status == garcon_feed_status.STATUS_DOWNLOADED: activity.logger.info( 'File {file} already DOWNLOADED, and ' '"reload" flag was not passed, skipping...'.format( file=destination_s3_key_path)) return {'skip': True} sme_s3_client = _get_sme_s3_client() logger.info(f'downloaded_file_path={downloaded_file_path}') sme_s3_client.download_file( source_bucket_name, source_key_name, downloaded_file_path ) logger.info('download completed') logger.info(f'uploading archive to {destination_key_name}') destination_s3_bucket.upload_file( downloaded_file_path, destination_key_name ) logger.info('upload completed') unzipped_file_path = _unzip_file( downloaded_file_path, staging_dir_for_date ) logger.info(f'unzipped_file_path={unzipped_file_path}') prepared_file_path = _prepare_file( staging_dir_for_date, unzipped_file_path, logger) logger.info(f'prepared_file_path={prepared_file_path}') clean_key_name = os.path.join( os.path.dirname(destination_key_name), 'clean', os.path.basename(destination_key_name).replace('zip', 'gz') ) logger.info(f'Uploading clean file to {clean_key_name}') destination_s3_bucket.upload_file(prepared_file_path, clean_key_name) logger.info('Upload completed') for file in ( downloaded_file_path, unzipped_file_path, prepared_file_path): os.remove(file) return {'success': True} @task.decorate(timeout=2000) def download_and_clean( activity, date, file_key, archive_bucket, report_name): """Download and clean files. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). file_key (str): the aws key for the file to clean. archive_bucket (str): name of the archive bucket. report_name (str): The name of the report (e.g Unlimited, Prime). Returns: None. """ staging_dir_for_date = os.path.join( STAGING_PATH.format(report=report_name, licensor='theorchard'), date) os.makedirs(staging_dir_for_date, exist_ok=True) s3_bucket = boto3.resource('s3').Bucket(archive_bucket) downloaded_file_path = _download_file_from_s3( staging_dir_for_date, file_key, s3_bucket) unzipped_file_path = _unzip_file( downloaded_file_path, staging_dir_for_date ) prepared_file_path = _prepare_file( staging_dir_for_date, unzipped_file_path, activity.logger) new_file_key = _upload_file_to_s3(prepared_file_path, s3_bucket, file_key) activity.logger.info( '{file_key} was downloaded and cleaned as {new_file_key}'.format( file_key=file_key, new_file_key=new_file_key)) for file in ( downloaded_file_path, unzipped_file_path, prepared_file_path): os.remove(file) @task.decorate(timeout=14400) def awal_grab_drop_files( activity, date, report_name, destination_key_name, s3_archive_path): """Download and clean files. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). report_name (str): The name of the report (e.g Unlimited, Prime). destination_key_name (str): Destination s3 path. s3_archive_path (str): Archive s3 path. Returns: dict: - file: filename - status: true/false successful - file_size: file size in bytes - exception: exception (on exception) """ destination_path, filename = os.path.split(destination_key_name) match = config.licensors_config['awal']['file_name_regexp'].match(filename) ftp_creds = dict(config.sftp) ftp_creds['username'] = get_secret( config.secrets_path, config.sftp['username']) ftp_creds['password'] = get_secret( config.secrets_path, config.sftp['password']) ftp_report_folder = config.reports[report_name]['awal_ftp_folder_name'] ftp_path = config.sftp['path'].format( country=match.group('country'), awal_ftp_folder_name=ftp_report_folder) response = garcon_ftp.copy_file_from_ftp_to_s3( activity, ftp_creds, ftp_path, filename, s3_archive_path, filename) if 'exception' in response: response['exception'] = repr(response['exception']) return response @task.decorate(timeout=14400) def altafonte_grab_drop_files( activity, date, report_name, destination_key_name, s3_archive_path): """Download and clean files. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). report_name (str): The name of the report (e.g Unlimited, Prime). destination_key_name (str): Destination s3 path. s3_archive_path (str): Archive s3 path. Returns: dict: - file: filename - status: true/false successful - file_size: file size in bytes - exception: exception (on exception) """ destination_path, filename = os.path.split(destination_key_name) match = config.licensors_config['altafonte']['file_name_regexp'].\ match(filename) ftp_creds = dict(config.sftp_altafonte) ftp_creds['username'] = get_secret( config.secrets_path, config.sftp_altafonte['username']) ftp_creds['password'] = get_secret( config.secrets_path, config.sftp_altafonte['password']) ftp_report_folder = config.\ reports[report_name]['altafonte_ftp_folder_name'] ftp_path = config.sftp_altafonte['path'].format( country=match.group('country'), altafonte_ftp_folder_name=ftp_report_folder) response = garcon_ftp.copy_file_from_ftp_to_s3( activity, ftp_creds, ftp_path, filename, s3_archive_path, filename) if 'exception' in response: response['exception'] = repr(response['exception']) return response @task.decorate(timeout=14400) def awal_grab_drop_files_s3( activity, date, source_key_name, destination_key_name): """Download and clean files. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). source_key_name (str): Source s3 path. destination_key_name (str): Archive s3 path. Returns: dict: - file: filename - status: true/false successful - file_size: file size in bytes - exception: exception (on exception) """ def concat_path(bucket, file): return 's3://' + bucket + '/' + file from_path_ = concat_path(config.awal_drop_bucket, source_key_name) to_path_ = concat_path(config.archive_bucket, destination_key_name) try: copy_s3_key(from_path_, to_path_) activity.logger.info( 'File was copied from {from_path} to {to_path}'.format( from_path=from_path_, to_path=to_path_)) except ClientError as err: activity.logger.error( ('Cannot copy files from {drop_location} ' 'to {archive_location}. {exception_body}').format( drop_location=from_path_, archive_location=to_path_, exception_body=err)) return {from_path_.split('/')[-1]: garcon_feed_status. STATUS_NOT_AVAILABLE} garcon_feed_status.set_status( config.feed_name, date, from_path_.split('/')[-1], status=garcon_feed_status.STATUS_DOWNLOADED) return {from_path_.split('/')[-1]: garcon_feed_status.STATUS_DOWNLOADED} @task.decorate(timeout=14400) def altafonte_grab_drop_files_s3( activity, date, source_key_name, destination_key_name): """Download and clean files. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). source_key_name (str): Source s3 path. destination_key_name (str): Archive s3 path. Returns: dict: - file: filename - status: true/false successful - file_size: file size in bytes - exception: exception (on exception) """ def concat_path(bucket, file): return 's3://' + bucket + '/' + file from_path_ = concat_path(config.altafonte_drop_bucket, source_key_name) to_path_ = concat_path(config.archive_bucket, destination_key_name) try: copy_s3_key(from_path_, to_path_) activity.logger.info( 'File was copied from {from_path} to {to_path}'.format( from_path=from_path_, to_path=to_path_)) except ClientError as err: activity.logger.error( ('Cannot copy files from {drop_location} ' 'to {archive_location}. {exception_body}').format( drop_location=from_path_, archive_location=to_path_, exception_body=err)) return {from_path_.split('/')[-1]: garcon_feed_status. STATUS_NOT_AVAILABLE} garcon_feed_status.set_status( config.feed_name, date, from_path_.split('/')[-1], status=garcon_feed_status.STATUS_DOWNLOADED) return {from_path_.split('/')[-1]: garcon_feed_status.STATUS_DOWNLOADED} @task.decorate(timeout=3600 * 2) def load_staging_raw_from_datapulse( activity, date, report_name, licensor, feed_name, sfdb_params, secrets_path): """Load amazon_music staging_raw from amazon_datapulse Snowflake tables. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). report_name (str): One of 'unlimited' / 'prime' / 'adsupported'. licensor (str): One of config.DATAPULSE_LICENSORS. feed_name (str): Contextified feed name. sfdb_params (dict): Snowflake connection params. secrets_path (str): Path to secrets in AWS Secrets Manager. """ if task_status.is_completed_task( feed_name, date, load_raw_table_tasks_sf.TASK_ID): activity.logger.info( f'load_staging_raw_from_datapulse for {date} already complete, ' 'skipping...' ) return sf_config = get_sf_config(secrets_path) sf_config_custom = merge_configs(sf_config, sfdb_params) ExecutorSR = registered_executors.get(feed_name) with ExecutorSR(sf_config_custom) as sf_executor: sf_executor.load_staging_raw_from_datapulse( date=date, report_name=report_name, licensor=licensor, ) @task.decorate(timeout=1000) def remove_file_stage(activity, date, report_name, licensor): """Remove the file_stage directory. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). report_name (str): The name of the report (e.g Unlimited, Prime). licensor (str): The licensor one of config.licensors Returns: None. """ staging_dir_for_date = os.path.join( STAGING_PATH.format(report=report_name, licensor=licensor), date) if os.path.exists(staging_dir_for_date): shutil.rmtree(staging_dir_for_date) @task.decorate(timeout=600) def remove_stale_stage_files(activity, report_name, licensor): """Remove stale files in stage directory. Args: activity (ActivityWorker): The Garcon activity worker. report_name (str): The name of the report (e.g Unlimited, Prime). licensor (str): The licensor one of config.licensors Returns: None. """ staging_path = STAGING_PATH.format(report=report_name, licensor=licensor) now_epoch = datetime.datetime.now().timestamp() cleanup_period = now_epoch - config.MAX_EXECUTION_TIMEOUT for root, dirs, files in os.walk(staging_path): for filename in files: file = os.path.join(root, filename) if os.stat(file).st_mtime < cleanup_period: os.remove(file) @task.decorate(timeout=36000) @check_status() def load_aggregated_table( activity, date, feed_name, licensor, report_name, sfdb_params, secrets_path=None, source=None): """Load aggregated staging raw table. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). feed_name (str): Feed name of workflow execution for status updates. licensor (str): licensor one of config.licensors report_name (str): The name of the report (e.g Unlimited, Prime). sfdb_params (dict): Dict with params to optionally override default ones (Snowflake db and schema name). secrets_path (str): Secrets manager path of the flow. source (str): Source of the staging data (config.sources). For datapulse the unified staging_raw table is read/scoped by feedid + licensor rather than by store-code orgs. """ if source == config.SOURCE_DATAPULSE: orgs = None else: date_obj = datetime.datetime.strptime(date, config.DATE_FORMAT).date() org_countries = org_countries_for_date(licensor, report_name, date_obj) orgs = list(org_countries.keys()) sf_config = get_sf_config(secrets_path) sf_config_custom = merge_configs(sf_config, sfdb_params) ExecutorAM = registered_executors.get(feed_name) with ExecutorAM(sf_config_custom) as executor: executor.clean_aggregated_staging_raw_table(date, orgs, source=source) activity.logger.info('{table} was cleaned for {date} and orgs {orgs}' .format(table=config.aggregated_staging_raw_table, orgs=orgs, date=date)) executor.load_aggregated_staging_raw_table(date, source=source) activity.logger.info('{table} was loaded for {date}'.format( table=config.aggregated_staging_raw_table, date=date)) @task.decorate(timeout=1000) def monitor_drop_location(activity, date, s3_drop_location, report_name, licensor, skip_monitor=None): """Check appearance of any new unexpected files or new stable countries. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). s3_drop_location (str): Drop location on S3. report_name (str): The name of the report (e.g Unlimited, Prime). licensor (str): one of config.licensors skip_monitor (str): If 'True' skip monitor. Returns: dict: With a subject and a message. """ assert licensor in config.licensors # todo implement monitoring awal licensor in next PR if skip_monitor == 'True' or licensor in ['awal', 'altafonte']: return {'skip_monitor': skip_monitor} unexpected_files, unstable_countries_files = _filter_files( date, s3_drop_location, licensor, report_name) result = {} if unexpected_files: unexpected_files_report = _unexpected_files_report( unexpected_files, licensor, report_name) result.update(unexpected_files_report) if unstable_countries_files: unstable_countries_report = _unstable_countries_report( unstable_countries_files, licensor, report_name) result.update(unstable_countries_report) return result def _unexpected_files_report(unexpected_files, licensor, report_name): """Compose a report with subject and message with unexpected files. Args: unexpected_files (list[str]): List of unexpected files. licensor (str): one of config.licensors report_name (str): The name of the report (e.g Unlimited, Prime). Returns: dict: Subject and message with unexpected files. """ return { 'unexpected_files_subject': ( config.UNEXPECTED_FILES_NOTIFICATION_SUBJECT.format( report=report_name.capitalize(), licensor=licensor)), 'unexpected_files_message': ( config.UNEXPECTED_FILES_NOTIFICATION_MESSAGE.format( '\n'.join(unexpected_files))) } def _unstable_countries_report(unstable_countries_files, licensor, report_name): """Compose a report with subject and message with unstable countries. Args: unstable_countries_files (list[str]): List of unexpected files. licensor (str): on of config.licensors report_name (str): The name of the report (e.g Unlimited, Prime). Returns: dict: Subject and message with unstable countries. """ countries_stability = _evaluate_stability( unstable_countries_files, licensor, report_name) if any(countries_stability.values()): new_stable_countries_message = _new_stable_countries_message( countries_stability) return { 'new_stable_countries_subject': ( config.NEW_STABLE_COUNTRIES_NOTIFICATION_SUBJECT.format( report=report_name.capitalize(), licensor=licensor)), 'new_stable_countries_message': new_stable_countries_message} return {} def _new_stable_countries_message(countries_stability): """Compose a message with unstable countries. Args: countries_stability (dict): Country code and stability value (True or False). Returns: str: Message with unstable countries. """ country_statuses = [] for country, stable in countries_stability.items(): stability = 'stable' if stable else 'unstable' country_statuses.append('{} {}'.format(country, stability)) return config.NEW_STABLE_COUNTRIES_NOTIFICATION_MESSAGE.format( '\n'.join(country_statuses)) def _evaluate_stability(unstable_countries_files, licensor, report_name): """Evaluate stability for each unstable country. Args: unstable_countries_files (list[str]): List of unstable countries files in drop location. licensor (str): on of config.licensors report_name (str): The name of the report (e.g Unlimited, Prime). Returns: dict: Country code and stability value (True or False). """ country_files = _count_files_for_countries(unstable_countries_files) result = {} for country in config.reports[report_name]['unstable_countries']: result[country] = _is_country_stable(country_files.get(country), licensor, report_name) return result def _count_files_for_countries(unstable_countries_files): """Count files for each country in drop location. Args: unstable_countries_files (list[str]): List of unstable countries files in drop location. Returns: dict: Country code, date and number of files on drop location. """ country_files = collections.defaultdict(collections.Counter) for file in unstable_countries_files: match = config.file_name_regexp.match(file) country_files[match.group('country')][match.group('file_date')] += 1 return country_files def _is_country_stable(country_files, licensor, report_name): """Evaluate whether a country stable or not. Args: country_files (dict): Dates and corresponding number of files. licensor (str): on of config.licensors report_name (str): one of config.reports Returns: bool: Is country stable or not. """ if not country_files: return False number_of_stable_days = 0 today = datetime.datetime.today() org_countries = org_countries_for_date(licensor, report_name, today.date()) # Required number of files to satisfy ingestion condition required_number_of_files = \ count_entries(org_countries) \ * len(config.licensors_config[licensor]['report_types']) for date_shift in range(config.MONITORING_PERIOD_OF_UNSTABLE_COUNTRIES): required_date = today - datetime.timedelta( days=config.FEED_DELAY + date_shift) file_date = required_date.strftime('%Y%m%d') if country_files.get(file_date, 0) == required_number_of_files: number_of_stable_days += 1 return ( number_of_stable_days / config.MONITORING_PERIOD_OF_UNSTABLE_COUNTRIES > config.REQUIRED_PERCENT_OF_STABLE_DAYS) def _filter_files(date, s3_path, licensor, report_name): """Filter files for monitoring. Args: date (date): reporting date. s3_path (str): S3 path of drop location. licensor (str): on of config.licensors report_name (str): The name of the report (e.g Unlimited, Prime). Returns: tuple(list, list): List of unexpected and unstable countries files. """ if licensor == 'sme': all_files = _list_keys_in_sme_s3(s3_path) else: all_files = s3utils.get_list_of_files_and_directories(s3_path) date_obj = config.parse_date(date) org_countries = org_countries_for_date(licensor, report_name, date_obj.date()) countries = only_countries(org_countries) orgs = list(org_countries.keys()) unstable_countries = config.reports[report_name]['unstable_countries'] if licensor == 'sme': regexp = config.sme_file_name_regexp else: regexp = config.file_name_regexp unexpected_files, unstable_countries_files = [], [] for file_path in all_files: if not _key_is_directory(file_path): filename = os.path.basename(file_path) match = regexp.match(filename) if _unexpected_file(match, countries, orgs, unstable_countries): unexpected_files.append(filename) elif _unstable_country_file(match, countries, unstable_countries): unstable_countries_files.append(filename) return unexpected_files, unstable_countries_files def _key_is_directory(s3_key): return s3_key[-1] == '/' def _unexpected_file(re_match, countries, orgs, unstable_countries): """Check if the file is unexpected or not. File is unexpected if it was not matched with basic reg expression from config or filename contains unexpected organisation value or unexpected report type or unexpected country. Args: re_match (SRE_Match): Match object with config reg expression. countries (list): The list of expected countries. orgs (list): The list of expected orgs. unstable_countries (list): The list of unstable countries. Returns: bool: True if file is unexpected, otherwise False. """ return ( not re_match or not re_match.group('report_type') in config.report_types or not re_match.group('org').upper() in orgs or ( not re_match.group('country').upper() in countries and not re_match.group('country').upper() in unstable_countries ) ) def _unstable_country_file(re_match, countries, unstable_countries): """Check if the file is one of the unstable countries. Args: re_match (SRE_Match): Match object with config reg expression. countries (list): The list of expected countries. unstable_countries (list): The list of unstable countries. Returns: bool: True if file is one of the unstable countries, otherwise False. """ return ( not re_match.group('country').upper() in countries and re_match.group('country').upper() in unstable_countries ) def _get_expected_files(date, report_name, only_countries, licensor): """Generate list of expected files.""" expected_files = [] date_obj = date.date() generator = org_country_generator(report_name, only_countries, licensor, date_obj) report_name_in_file_name = \ 'altafonte_report_name_in_filename' if licensor == 'altafonte' \ else 'awal_report_name_in_filename' for org, country in generator: for report_type in config.licensors_config[licensor]['report_types']: expected_files.append( config.licensors_config[licensor]['file_template'].format( date=date, org=org, country=country, report_type=report_type, report=config.reports[report_name][ report_name_in_file_name]) ) return sorted(expected_files) def _download_file_from_s3(staging_path, file_key, s3_bucket): destination = os.path.join(staging_path, os.path.basename(file_key)) s3_bucket.download_file(file_key, destination) return destination def _upload_file_to_s3(file_path, s3_bucket, file_key): new_file_key = os.path.join( os.path.dirname(file_key), 'clean', os.path.basename(file_path)) s3_bucket.upload_file(file_path, new_file_key) return new_file_key def _unzip_file(zipped_file_path, unzipped_file_dir): with zipfile.ZipFile(zipped_file_path) as the_file: the_file.extractall(unzipped_file_dir) return zipped_file_path.replace('.zip', '') def _prepare_file(staging_path, file_path, logger): """Create new file with common separators.""" gz_file_name = os.path.basename(file_path) + '.gz' prepared_file_path = os.path.join(staging_path, gz_file_name) report_type = _get_report_type(file_path) malformed_rows = 0 with open(file_path) as infile, \ gzip.open(prepared_file_path, 'wb') as outfile: for line in infile: try: prepared_line = _prepare_report_line(report_type, line) except Exception: logger.error(f'malformed_row: {line} , file {file_path}') malformed_rows = malformed_rows + 1 if prepared_line: outfile.write(prepared_line.encode('utf-8')) if malformed_rows >= config.malformed_rows_limit: raise ValueError(f'File {file_path} is corrupted ') return prepared_file_path def _get_report_type(file_name): """Get report_type by file name.""" for report_type in config.report_types: if report_type in file_name: return report_type.lower() def _prepare_report_line(report_type, line): """Change separators and filter lines with additional information. Amazon includes two rows at the beginning and at the end of a file, which do not match the columns expected for the file. Length of these rows usually == SKIP_LINE_COLUMNS. Just to avoid using ON_ERROR=CONTINUE while loading data into snowflake this rows will be skipped. Args: report_type (str): The report type: activity, playlist or user. line (str): The line from source file Returns: str or None: String with new tab separators or None if count of columns in the line <= config.SKIP_LINE_COLUMNS. """ line_values = line.replace('\t', ' ').split('#*#') if len(line_values) <= config.SKIP_LINE_COLUMNS: return if _report_row_handlers.get(report_type): line_values = _report_row_handlers[report_type](line_values) return '\t'.join(line_values) def _playlist_report_row_handler(row): """Process playlist report row. Since 2019-01-01 Amazon has sent playlist reports with two additional columns. In order to support new and original versions we extend old rows with two empty values. """ if len(row) == 5: row[-1] = row[-1].strip() row.extend(['', '\n']) return row def _try_datetime_format(str_value): """Convert datetime value to common format. For some SME activity files timestamp columns can have different timestamp format. Args: str_value (str): datetime as str Returns: str: datetime in format %Y%m%dT%H:%M:%S or str_value unchanged if the format was not detected """ try_format = None if 'Z' in str_value: if '.' in str_value: try_format = '%Y-%m-%dT%H:%M:%S.%fZ' else: try_format = '%Y-%m-%dT%H:%M:%SZ' else: if ' ' in str_value: if '.' in str_value: try_format = '%Y-%m-%d %H:%M:%S.%f' else: try_format = '%Y-%m-%d %H:%M:%S' if try_format: dt = datetime.datetime.strptime(str_value, try_format) return dt.strftime('%Y%m%dT%H:%M:%S') return str_value def _activity_report_row_handler(row): """Process activity report row. Some of the Amazon Music activity reports contain 28 or 29 columns depending on country and report (JP country and AdSupported report files have 29 columns others 28). Since 2019-07-22 all files have 29 columns. """ if len(row) == 28: row[-1] = row[-1].strip() row.append('\n') # column 'timestamp' row[20] = _try_datetime_format(row[20]) # column 'serviceLoggingTimestamp' row[25] = _try_datetime_format(row[25]) return row _report_row_handlers = { 'playlist': _playlist_report_row_handler, 'user': None, 'activity': _activity_report_row_handler } def _get_context_countries(countries): country_list = [country.strip() for country in countries.split(',')] # noqa; 6 is provided for ROE_EU and ROW_NA if not all(len(country) in (2, 6) and country.isupper() for country in country_list): raise ValueError( 'Countries value has incorrect format! Please provide ' 'comma-separated list with two-character ' 'upper case country codes.\n' '{} was provided.'.format(countries) ) return country_list