"""Music Analytics Report Flow tasks.""" from datetime import datetime, timedelta import os from garcon import task from garcon_contrib.dynamo_feed_status import garcon_feed_status from feed_ingestion.conf.config import merge_configs from feed_ingestion.flows import MusicAnalyticsReportsSE from feed_ingestion.flows.helpers import get_sf_config from feed_ingestion.flows.music_analytics_reports import config from feed_ingestion.flows.music_analytics_reports.config import \ music_analytics_utils_config as mau_c from feed_ingestion.flows.music_analytics_reports.music_analytics_report \ import MusicAnalyticsReport from feed_ingestion.tasks import STOP_RESPONSE from feed_ingestion.tasks.apple_podcasts_reporter_tasks import _upload_to_s3 from feed_ingestion.util.apple_music_analytics import \ EmptyReportException, MusicAnalyticsAPI, MusicAnalyticsUtils @task.decorate(timeout=1000) def bootstrap(activity, date, reload, report_type): """Bootstrap workflow by getting the correct configurations. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). reload (str): If "True" delete feed status in dynamodb. report_type (str): Report type. Returns: dict: Context. """ if not date: date = (datetime.today() - timedelta(days=1)).strftime('%Y-%m-%d') if report_type not in config.cucumber_feed_name: return {'stop': f'Provide the correct report_type. ' f'Must be in ${", ".join(config.cucumber_feed_name)}'} if report_type == 'excluded_streams': date = datetime.strptime( date, '%Y-%m-%d').replace(day=1).strftime('%Y-%m-%d') activity.logger.info('Bootstrap flow for {}'.format(date)) feed = config.feed_name if reload == 'True': activity.logger.info( 'Delete status for feed: {} {} '.format(feed, date)) garcon_feed_status.delete_status(feed, date) else: overall_status = garcon_feed_status.get_overall_status(feed, date) if overall_status == garcon_feed_status.STATUS_INGESTED: return STOP_RESPONSE temp_staging_raw_table = config.snowflake_tables['temp_staging']. \ format(date=date, report_type=report_type).replace('-', '') temp_errors_table = config.snowflake_tables['temp_errors']. \ format(date=date, report_type=report_type).replace('-', '') staging_raw_table = config.snowflake_tables['staging_raw']. \ format(report_type=report_type) errors_table = config.snowflake_tables['errors']. \ format(report_type=report_type) return dict( feed_name=feed, temp_staging_raw_table=temp_staging_raw_table, temp_errors_table=temp_errors_table, staging_raw_table=staging_raw_table, errors_table=errors_table, date=date, secrets_path=config.secrets_path, reload=reload, report_type=report_type, ) @task.decorate(timeout=1000) def get_report_file(activity, date, report_type): """Extract a report from Apple's Reporter Tool and upload it to S3. Args: activity (ActivityWorker): The activity worker. date (str): The date of report. report_type (str): MA report type. """ utils = MusicAnalyticsUtils( mau_c.get('jar'), mau_c.get('key_id'), mau_c.get('team_id'), mau_c.get('private_key')) api = MusicAnalyticsAPI(utils) _report = ''.join([w.capitalize() for w in 'in_review'.split('_')]) activity.logger.info('Getting Apple Music Analytics ' '{report} report for {date}.'. format(report=_report, date=date)) try: raw = api.get_report_raw_data(report_type, date) report = MusicAnalyticsReport.get_report_by_type(report_type, raw) file = report.write_report_data_to_file() activity.logger.info( f'Data from Apple Music Analytics {_report} ' f"report has been loaded to {file.get('report')}.") return {'file': file} except EmptyReportException as ex: activity.logger.error(ex.message) return {'stop': True} @task.decorate(timeout=1000) def load_report_to_s3(activity, file, date, feed_name, report_type): """Extract a report from MA Utils and upload it to S3. Args: activity (ActivityWorker): The activity worker. file (dict): The dict with filenames. date (str): The date of report. feed_name (str): Feed name. report_type (str): Report type. Returns: Dict of full_s3_path report and error files. """ filename = file.get('report') _report = config.cucumber_feed_name[report_type] copy_on_s3_result = _upload_to_s3( filename, config.s3['report'].format( date=date, feed_name=_report), activity) os.remove(filename) if file.get('errors'): errors = file.get('errors') activity.logger.info( f'During validating Apple Music Analytics {_report} ' f'report the errors were collected at {errors} file.') _upload_to_s3(errors, config.s3['errors'].format( date=date, feed_name=_report), activity) os.remove(errors) if feed_name is not None: if not copy_on_s3_result: garcon_feed_status.set_status( feed_name, date, filename, status=garcon_feed_status.STATUS_NOT_AVAILABLE) return {'failed': True} garcon_feed_status.set_status( feed_name, date, filename, status=garcon_feed_status.STATUS_DOWNLOADED) return { 'full_s3_path': { k: config.s3[k].format( date=date, feed_name=_report) + v for k, v in file.items()}, 'failed': False } @task.decorate(timeout=12600) def load_temp_errors_table( activity, date, aws, key_dir: dict, temp_errors_table: str, sfdb_params, secrets_path=None): """Copy the temp_staging_raw data to the feed's staging_raw table. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). temp_errors_table (str): A tenp. table name. 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. """ sf_config = get_sf_config(secrets_path) sf_config_custom = merge_configs(sf_config, sfdb_params) with MusicAnalyticsReportsSE(sf_config_custom) as sf_executor: activity.logger.info( f'{sf_executor.errors_table} was cleaned from ' 'the rows with the date (or date range) ' 'of the current workflow run') sf_executor.load_temp_errors_table( temp_errors_table, aws, key_dir)