""" Apple Music Streams Garcon tasks. Tasks to ingest Apple Music Streams data. """ from collections import defaultdict from datetime import date as date_module from datetime import datetime from datetime import timedelta from functools import partial import os from botocore.exceptions import ClientError from garcon import task from garcon_contrib.dynamo_feed_status import garcon_feed_status from garcon_contrib.dynamo_feed_status.garcon_feed_status import \ STATUS_DOWNLOADED from snowflake_connector.etl_connector import SnowflakeSQLExecutor from snowflake_connector.etl_connector import SQLLoader from feed_ingestion.common import apple_id_mapping from feed_ingestion.conf.config import merge_configs from feed_ingestion.flows.apple_music_streams import config, utils from feed_ingestion.flows.apple_music_streams.snowflake_executor \ import AppleMusicStreams from feed_ingestion.flows.apple_music_streams.utils import \ get_available_reports, get_missing_vendors from feed_ingestion.flows.apple_music_streams.utils import \ get_fact_analytics_report from feed_ingestion.flows.apple_music_streams.utils import \ get_skip_and_saves_reports from feed_ingestion.flows.apple_music_streams.vendor_accounts import \ contexts_config, get_vendors from feed_ingestion.flows.helpers import get_sf_config from feed_ingestion.tasks import check_status from feed_ingestion.tasks import overall_status_tasks from feed_ingestion.tasks.feed_status_tasks import file_is_optional from feed_ingestion.tasks.feed_status_tasks import \ get_contexts_config_for_report from feed_ingestion.tasks.overall_status_tasks import \ set_overall_status_enhanced from feed_ingestion.util import check_nonconcurrent_workflows from feed_ingestion.util import task_status from feed_ingestion.util.aws.s3 import copy_s3_key, \ get_list_of_files_and_directories from feed_ingestion.util.context_util import get_context_values, Reload from feed_ingestion.util.log_status import log_feed_ingestion_completed_status sql_loader = SQLLoader(__file__) STOP_RESPONSE = {'stop': True} STAGING_RAW_TASK_ID = 'staging_raw_table_tasks' @task.decorate(timeout=1000) def check_concurrent_status(activity, licensor, domain): """Check if there is any active executions. Args: activity (ActivityWorker): The Garcon activity worker. licensor (str): The licensor to ingest. domain (str): The flow domain (f.e. prod_feed_ingestion). Returns: dict: {} or STOP_RESPONSE from check_ingested_status decorator """ return check_nonconcurrent_workflows.check_concurrent_status( domain, licensor, config.max_concurrent_executions[licensor]) @task.decorate(timeout=1000) def bootstrap( activity, date, reload, reports, licensor, snowflake_error_limit=None, use_s3=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 or None): If 'True' delete all feed statuses in DynamoDB. reports (str or None): List of the reports to ingest (optional). licensor (str): The licensor to ingest. snowflake_error_limit (int or None): Snowflake error limit. use_s3 (str or None): If 'True' use s3 instead of Apple api. Returns: dict: Initial context for the workflow. """ # date is the date passed in or yesterday's date activity.logger.info(f'Starting Apple Music streams ' f'for {config.feed_name}-{licensor}') reload = Reload.from_value(reload) activity.logger.info(f'Reload is set to {reload}') if not date: date_obj = date_module.today() - timedelta(days=1) else: date_obj = datetime.strptime(date, '%Y-%m-%d') date_str = date_obj.strftime('%Y-%m-%d') available_reports_list = get_available_reports(date) reports_list = get_context_values(reports, available_reports_list) overall_feed_name = '_'.join([config.feed_name, licensor]) is_soft_reload = reload == Reload.SOFT is_full_reload = reload == Reload.TRUE if is_full_reload: activity.logger.info('Delete status for feed: {} {} '.format( overall_feed_name, date)) garcon_feed_status.delete_status(overall_feed_name, date) elif is_soft_reload: activity.logger.info('Soft reload is set. ' 'Will not delete status for feed: {} {} '.format( overall_feed_name, date)) # prepare feed status for soft reload task_status.soft_reload_update_status_clearing_tasks( context_date=date_str, feed_name=overall_feed_name, ) # prepare fact_analytics_feed_name (streams report) for soft reload fact_analytics_report = utils.get_fact_analytics_report( date=date_str, licensor=licensor, ) fact_analytics_feed_name = (f'{overall_feed_name}_' f'{fact_analytics_report}') task_status.soft_reload_update_status_clearing_tasks( context_date=date_str, feed_name=fact_analytics_feed_name, ) else: overall_status = garcon_feed_status.get_overall_status( overall_feed_name, date) completed = task_status.is_completed_overall_job( overall_feed_name, date) if overall_status == garcon_feed_status.STATUS_INGESTED and completed: activity.logger.info('Job is already completed.') return STOP_RESPONSE reports_status_names = defaultdict(dict) for report_name in set(reports_list + config.common_reports): # if report is not available for this licensor if licensor not in config.reports[report_name]['licensors']: continue report_feed_name = _get_feed_name(report_name, licensor) if is_full_reload and report_name in reports_list: if (config.reports[report_name].get('end_date') and licensor != 'awal'): raise ValueError( f'{report_name} is available. It can not be re-ingested.') garcon_feed_status.delete_status(report_feed_name, date) # if report is not ingested yet if (not is_soft_reload and not check_report(report_name, report_feed_name, date)): activity.logger.info( f'Report {report_name} was already ingested. Skipping') continue # get list of vendor accounts for the licensor and the report_name vendors = config.reports[report_name].get( 'vendors', get_vendors(date, licensor, report_name)) garcon_feed_status.set_missing_files(report_feed_name, date, []) task_status.delete_newcontexts(report_feed_name, date) reports_status_names[report_name] = { 'feed_name': report_feed_name, 'vendors': vendors } # init contexts, if they don't exist contexts_field_value = task_status.get_report_contexts( report_feed_name, date) if not contexts_field_value: activity.logger.info(f'Creating contexts for {report_name}') conf_optional, conf_contexts = ( get_contexts_config_for_report( report_name, contexts_config, licensor)) task_status.create_report_contexts( report_feed_name, date, vendors, conf_contexts, conf_optional) if isinstance(snowflake_error_limit, int): snowflake_error_limit = snowflake_error_limit else: snowflake_error_limit = config.snowflake_error_limit result = { 'date': date_str, 'processed_datetime': datetime.now().isoformat(), 'reports_status_names': reports_status_names, 's3_archive_bucket': config.s3['archive_bucket'].format( date=date_obj), 's3_drop_bucket': config.s3['drop_bucket'].format( date=date_obj), 'feed_name_for_fact_analytics': _get_feed_name( get_fact_analytics_report(date, licensor), licensor), 'snowflake_error_limit': snowflake_error_limit, 'use_s3': use_s3, 'soft_reload': is_soft_reload, 'fact_analytics_kwargs': dict(licensor=licensor), 'jenkins_config': config.jenkins_config, 'contexts': contexts_config } return result @task.decorate(timeout=600) def grab_drop_files( activity, feed_name, date, s3_archive_path, s3_download_path, filename, reporter_account, report_type=None, contexts_config=None, licensor=None): """Archive certain file from the drop location to archive location. Args: activity (ActivityWorker): The activity worker. feed_name (str): Name of feed being ingested. date (str): Reporting date (YYYY-MM-DD). s3_archive_path (str): Archive location on S3. s3_download_path (str): Drop location on S3. filename (str): The name of file to copy. reporter_account (str): Apple Reporter Account ('ORCHARD' or 'IODA'). report_type(str): Report name, e.g. amContainer. contexts_config(dict): config for required and optional contexts. licensor (str): Name of the licensor Returns: dict: Adds an entry to the context with a key of the file name and a value of whether the file was Downloaded or Not Available. if file Not Available for optional vendor do nothing """ if contexts_config is None: contexts_config = dict() from_path = '{s3_path}{filename}'.format( s3_path=s3_download_path, filename=filename) to_path = '{s3_path}{filename}'.format( s3_path=s3_archive_path, filename=filename) 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)) # will be handled in update_feed_file_status if file_is_optional(report_type, contexts_config, filename, licensor): return else: return {filename: garcon_feed_status.STATUS_NOT_AVAILABLE} garcon_feed_status.set_status( feed_name, date, filename, status=garcon_feed_status.STATUS_DOWNLOADED) vendor_id = reporter_account task_status.add_newcontext(feed_name, date, vendor_id) task_status.update_report_context_status( feed_name, date, vendor_id, task_status.CONTEXT_STATUS_IN_PROGRESS) return {filename: garcon_feed_status.STATUS_DOWNLOADED} @task.decorate(timeout=600) def grab_drop_files_awal( activity, feed_name, date, s3_archive_path, report_name, filename, reporter_account): """Archive certain file from the drop location to archive location. Args: activity (ActivityWorker): The activity worker. feed_name (str): Name of feed being ingested. date (str): Reporting date (YYYY-MM-DD). s3_archive_path (str): Archive location on S3. report_name (str): The report name. filename (str): The name of file to copy. reporter_account (str): Apple Reporter Account ('ORCHARD' or 'IODA'). Returns: dict: Adds an entry to the context with a key of the file name and a value of whether the file was Downloaded or Not Available. """ def get_version(rules, lookup_date): """Get version of the report by date.""" version = None for rule in rules: period_start_str = rule['since'] period_start = datetime.strptime( period_start_str, '%Y-%m-%d').date() if period_start > lookup_date: break version = rule['version'] return version parsed_date = datetime.strptime(date, '%Y-%m-%d').date() version = get_version( config.awal_s3_versions[report_name].get('versions', []), parsed_date) if not version: return {filename: garcon_feed_status.STATUS_NOT_AVAILABLE} from_path = config.awal_s3_path.format( drop_bucket=config.awal_drop_bucket, s3_folder=config.awal_s3_versions[report_name]['s3_folder'], version=version, date=parsed_date) files = get_list_of_files_and_directories(from_path) if not files: return {filename: garcon_feed_status.STATUS_NOT_AVAILABLE} # there is usually only one file file = os.path.basename(files[0]) to_path = f'{s3_archive_path}{file}' try: copy_s3_key(f'{from_path}{file}', 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 {filename: garcon_feed_status.STATUS_NOT_AVAILABLE} garcon_feed_status.set_status( feed_name, date, filename, status=garcon_feed_status.STATUS_DOWNLOADED) # add current vendor to set of new_contexts vendor_id = reporter_account task_status.add_newcontext(feed_name, date, vendor_id) task_status.update_report_context_status( feed_name, date, vendor_id, task_status.CONTEXT_STATUS_IN_PROGRESS) return {filename: garcon_feed_status.STATUS_DOWNLOADED} @task.decorate(timeout=600) def check_available_reports(activity, date, reports_status_names, s3_path, licensor, is_soft_reload=False): """Return new reports which are available. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). reports_status_names (dict): List of reports, which should be processed. licensor (str): The licensor to ingest. s3_path: is required in order to better parse missing_files field is_soft_reload: (bool) True for soft reload Returns: dict: of reports and their statuses, which files were uploaded on S3 statuses of possibility loading staging_raw and facts tables if report is missing for optional vendor than file will be ignored. """ available_reports = defaultdict(dict) load_fact_analytics = False update_library_reports = False has_new_vendors = False for report_name, description in reports_status_names.items(): feed_name = description['feed_name'] missing_vendors = get_missing_vendors(feed_name, date, s3_path) in_progress_contexts = task_status.get_report_in_progress_contexts( feed_name, date) if in_progress_contexts: has_new_vendors = True if (task_status.is_completed_task(feed_name, date, 'reporter_to_s3') and (report_name in config.common_reports or in_progress_contexts) or is_soft_reload): # exclude missing optional vendor out of the reports list available_reports[report_name] = \ {'feed_name': feed_name, 'vendors': [vendor for vendor in description['vendors'] if vendor not in missing_vendors] # 'new_vendors': list(new_vendors) } if not has_new_vendors and not is_soft_reload: activity.logger.warning( 'Skip processing, because there are no new vendors') return STOP_RESPONSE else: activity.logger.info(f'Available reports: {available_reports.keys()}') if get_fact_analytics_report(date, licensor) in available_reports.keys(): if licensor in config.active_licensors: load_fact_analytics = True if (not available_reports.keys() & set(config.common_reports) or len(available_reports) == 1): activity.logger.warning( f'Stop processing because it is not enough available_reports. ' f'available_reports={available_reports.keys()}') return STOP_RESPONSE if available_reports.keys() & set(config.reports_to_update): update_library_reports = True if load_fact_analytics: log_feed_ingestion_completed_status(activity, f'{config.feed_name}_{licensor}', date, STATUS_DOWNLOADED) return { 'available_reports': available_reports, 'load_fact_analytics': load_fact_analytics or is_soft_reload, 'update_library_reports': update_library_reports or is_soft_reload} @task.decorate(timeout=1000) @check_status(task_id=STAGING_RAW_TASK_ID) def load_staging_raw_table( activity, date, processed_datetime, vendors, sfdb_params, staging_raw_table, report_name, feed_name, licensor, secrets_path=None): """Load data to staging_raw_apple_music_streams. Runs for each vendor separately. Args: activity (ActivityWorker): The garcon activity worker. date (str): YYYY-MM-DD date of rows to delete from staging_raw table. processed_datetime (str): processing timestamp. vendors (dict): Vendor accounts and filenames for that licensor. sfdb_params (dict): Dictionary stores Snowflake params. staging_raw_table (str): The staging raw table name. report_name (str): Name of report being ingested. feed_name (str): Name of feed being ingested. licensor (str): Name of the licensor to ingest. secrets_path (str): Secrets manager path of the flow. """ activity.logger.info('Deleting {} from {}'.format( date, staging_raw_table)) sql_loader = SQLLoader(__file__, date=date) sf_config = merge_configs(get_sf_config(secrets_path), sfdb_params) executor = AppleMusicStreams(sf_config) consumer_db, consumer_schema = executor.staging_raw_location(report_name) executor.execute_query( sql_loader, 'delete_from_staging_raw', { 'db': sf_config['db'], 'schema': sf_config['schema'], 'consumer_db': consumer_db, 'consumer_schema': consumer_schema, 'staging_raw_table': staging_raw_table, 'date': date, 'vendor_ids': list(vendors.keys()) } ) for vendor, filename in vendors.items(): activity.logger.info('Loading {} into {} [{}]'.format( date, staging_raw_table, vendor)) params = { 'report_name': report_name, 'licensor': licensor, 'vendor': vendor, 'processed_datetime': processed_datetime, 'filename': filename, 'temp_table_amcontent': get_temp_table_name( date, 'amContent', vendor, licensor), 'temp_table_amsubreference': get_temp_table_name( date, 'amSubscriptionReference', vendor, licensor), 'apple_id_mapping_table': config.apple_id_mapping_tables[ licensor]} try: AppleMusicStreams(sf_config).load_staging_raw_table( date, staging_raw_table, temp_staging_raw_table=get_temp_table_name( date, report_name, vendor, licensor), **params) except Exception as e: # handle case, when e.g. vendor for ContentDemographics exists, # but for Content it doesn't exist. if 'does not exist' in str(e): # just log error, but don't fail flow activity.logger.error( f'Failed to load_staging_raw_table: {staging_raw_table}, ' f'{vendor}, {str(e)}') else: raise e @task.decorate(timeout=14400) def update_apple_id_mapping( activity, date, sfdb_params, licensor, skip_mapping=None, secrets_path=None): """Update apple_id_mapping table. Args: activity (ActivityWorker): The activity worker. date (str): YYYY-MM-DD date of rows to delete from staging_raw table. sfdb_params (dict): Dictionary stores Snowflake params. skip_mapping (str): if 'True' skip apple_id_mapping process. secrets_path (str): Secrets manager path of the flow. """ if skip_mapping == 'True': return {'skip_mapping': True} sf_config = merge_configs(get_sf_config(secrets_path), sfdb_params) sql_loader = SQLLoader(apple_id_mapping.query_path, folder='/queries') if licensor == 'theorchard': query_names = [ '00_ams_insert_new_entries', '01_update_vendor_offer_code', '02_update_vendor_identifier', '03_clean_blank_apple_release_id', '04_clean_blank_orchard_release_id', '05_clean_zero_apple_track_id', '06_clean_zero_orchard_track_id', ('07_update_apple_track_id', {'term': 1}), ('08_update_apple_release_id', {'term': 1}), '09_update_apple_release_id_one_track', '10_update_apple_release_id_from_upc', '10_ams_update_finetunes_apple_release_id', '10_ams_update_phonofile_apple_release_id', '11_update_apple_track_id_one_track', '12_update_orchard_track_id', '13_update_orchard_release_id', '14_update_season_pass_orchard_release_track_id', '15_update_with_vendor_identifier_mapping', '16_update_ioda_video_mapping_case', '17_update_ioda_tv_mapping_case', '18_ams_update_orchard_release_track_id_from_isrc', '19_update_orchard_release_id_from_manufacturer_upc_in_vid', '20_update_orchard_release_id_from_manufacturer_upc_in_upc', ] elif licensor == 'awal': query_names = ['24_populate_awal_apple_id_mapping'] else: query_names = [ '23_populate_sony_apple_id_mapping', '26_populate_sony_apple_id_mapping_derived', ] common_params = { 'db': sf_config['db'], 'schema': sf_config['schema'], # the ams_* mapping queries read the migrated summary streams table, # which always lives in the consumer reporting db/schema 'consumer_db': config.consumer_sf['db'], 'consumer_schema': config.consumer_sf['schema'], 'date': date, 'vendor_ids': get_vendors(date, licensor, get_fact_analytics_report( date, licensor)), 'staging_raw_table': config.reports[get_fact_analytics_report( date, licensor)]['staging_raw_table'] } with SnowflakeSQLExecutor(sf_config) as executor: for query in query_names: if isinstance(query, tuple): query_name, add_params = query else: query_name, add_params = query, {} activity.logger.info('Mapping: ' + query_name) params = common_params.copy() params.update(add_params) executor.execute_query(sql_loader, query_name, params) @task.decorate(timeout=7200) def drop_temp_table(activity, temp_table_name, sfdb_params, secrets_path=None): """Drop temporary table. Args: activity (ActivityWorker): The garcon activity worker. temp_table_name (str): name of table to drop. sfdb_params (dict): Dictionary stores Snowflake params. secrets_path (str): Secrets manager path of the flow. """ activity.logger.info('Dropping temp table {}'.format(temp_table_name)) sf_config = merge_configs(get_sf_config(secrets_path), sfdb_params) with SnowflakeSQLExecutor(sf_config) as executor: executor.execute_query( sql_loader, 'drop_temp_table', { 'db': sf_config['db'], 'schema': sf_config['schema'], 'table_name': temp_table_name, } ) def check_report(report_name, report_feed_name, date): """Check status if report is already ingested. Args: report_name (str): Name of the report to ingest. report_feed_name (str): Feed name for status updates. date (str): Reporting date (YYYY-MM-DD). Returns: (bool): True if report need to be ingested, otherwise False. """ completed = task_status.is_completed_report(report_feed_name, date) status = garcon_feed_status.get_overall_status(report_feed_name, date) return (report_name in config.common_reports or status != garcon_feed_status.STATUS_INGESTED or (status == garcon_feed_status.STATUS_INGESTED and not completed)) @task.decorate(timeout=1000) @check_status(task_id='update_staging_raw_library_reports') def update_staging_raw_library_reports( activity, date, feed_name, report_name, sfdb_params, licensor, secrets_path=None): """Update upc and isrc for amLibraryEvents, amTotalLibraryAdds. Runs for each vendor separately. Args: activity (ActivityWorker): The garcon activity worker. date (str): YYYY-MM-DD date of rows to delete from staging_raw table. feed_name (str): Name of feed being ingested. report_name (str): Name of report being ingested. sfdb_params (dict): Dictionary stores Snowflake params. licensor (str): Name of the licensor to ingest. secrets_path (str): Secrets manager path of the flow. """ table_name = config.reports[report_name]['staging_raw_table'] activity.logger.info('Updating upc/isrc for table {}'.format(table_name)) sf_config = merge_configs(get_sf_config(secrets_path), sfdb_params) SnowflakeSQLExecutor(sf_config).execute_query( sql_loader, 'update_staging_raw_library_reports', { 'db': sf_config['db'], 'schema': sf_config['schema'], 'staging_raw_table': table_name, 'date': date, 'vendor_ids': get_vendors(date, licensor) } ) @task.decorate(timeout=14000) def load_aggregated_skips_and_saves( activity, date, sfdb_params, licensor, secrets_path=None, available_reports=None, is_soft_reload=False): """Load from amLibraryEvents and amNonRoyaltyStreams tables. Args: activity (ActivityWorker): The activity worker. date (str): Date of the data being process (YYYY-MM-DD). sfdb_params (dict): Dict with params to optionally override default ones (Snowflake db and schema name). licensor (str): Name of the licensor to ingest. secrets_path (str): Secrets manager path of the flow. available_reports (dict): map of processed reports. is_soft_reload (bool): flag if it is soft-reload When is_soft_reload is True it forcingly runs skips_and_saves queries. """ # check if all the required tables for all the licensors are already loaded if not available_reports: available_reports = {} required_task_id = 'set_overall_status_POPULATED_RAW_TABLE' feed_names = [_get_feed_name(report_name, licensor) for report_name in get_skip_and_saves_reports( date, licensor)] if not all(map( partial( task_status.is_completed_task, datestamp=date, task_name=required_task_id), feed_names) ): activity.logger.info( 'load_aggregated_skips_and_saves task skipped, ' 'since not all the upstream tasks done') return # set/check status apple_music_theorchard or apple_music_sme overall_feed_name = '_'.join([config.feed_name, licensor]) task_id = 'load_aggregated_skips_and_saves' has_changes = False for val in available_reports.values(): if val['feed_name'] in feed_names: has_changes = True if (task_status.is_completed_task(overall_feed_name, date, task_id) and not has_changes and not is_soft_reload): activity.logger.info( 'Task {task_id} of {feed_name} for {date} already complete, and ' '"reload" flag was not passed, skipping...'.format( task_id=task_id, feed_name=overall_feed_name, date=date)) return sf_config = merge_configs(get_sf_config(secrets_path), sfdb_params) with AppleMusicStreams(sf_config) as sf_executor: sf_executor.delete_from_aggregated_skips_and_saves(date, licensor) activity.logger.info( 'Data was deleted from load_aggregated_skips_and_saves table') sf_executor.load_saves_into_aggregated_skips_and_saves(date, licensor) activity.logger.info( 'Saves were loaded into load_aggregated_skips_and_saves table') sf_executor.merge_skips_into_aggregated_skips_and_saves(date, licensor) activity.logger.info( 'Skips were loaded into aggregated_skips_and_saves table') sf_executor.update_streams_in_skips_and_saves_aggregated_streams( date, licensor) activity.logger.info( 'Streams were updated in aggregated_skips_and_saves table') task_status.mark_completed_task(overall_feed_name, date, task_id) activity.logger.info( 'Task {task_id} of {feed_name} for {date} completed'.format( task_id=task_id, feed_name=overall_feed_name, date=date)) @task.decorate(timeout=600) def set_status_ingested(activity, date, reports_status_names, licensor): """Set report status INGESTED if all tasks completed. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). reports_status_names (dict): List of reports, which should be processed. licensor (str): Name of the licensor to ingest. """ def check_only_staging_raw_report(report_name, overall_status): """Check if this report load data only in staging_raw table.""" status = garcon_feed_status.STATUS_POPULATED_RAW_TABLE return (overall_status == status and (report_name != get_fact_analytics_report(date, licensor) or licensor not in config.active_licensors)) for report_name, description in reports_status_names.items(): feed_name = description['feed_name'] overall_status = garcon_feed_status.get_overall_status( feed_name, date) if check_only_staging_raw_report(report_name, overall_status): set_overall_status_enhanced( feed_name, date, garcon_feed_status.STATUS_INGESTED, activity) processed_contexts = task_status.mark_report_processed_contexts( feed_name, date) activity.logger.info(f'Change context status {report_name} ' f'to processed for: {processed_contexts}') activity.logger.info( 'Setting status for feed: {feed_name} with date: {date} ' 'to: {status}, via a task.' .format( feed_name=feed_name, date=date, status=garcon_feed_status.STATUS_INGESTED )) elif report_name in config.common_reports: # handle commonReports processed_contexts = task_status.mark_report_processed_contexts( feed_name, date) activity.logger.info(f'Change context status {report_name} ' f'to processed for: {processed_contexts}') @task.decorate(timeout=1000) def set_status_ingested_to_fact_analytics_report( activity, date, feed_name, status): """Set the overall status of a feed for fact analytics report. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYYM-MM-DD). feed_name (str): Name of the feed. status (str): Status constant in util.garcon_feed_status. """ overall_status_tasks.set_overall_status(activity, date, feed_name, status) task_status.mark_report_processed_contexts(feed_name, date) @task.decorate(timeout=600) def set_overall_status_ingested(activity, date, licensor): """Set overall feed status INGESTED if all reports are completed. Args: activity (ActivityWorker): The Garcon activity worker. date (str): Reporting date (YYYY-MM-DD). licensor (str): Name of the licensor to ingest. """ date_obj = datetime.strptime(date, '%Y-%m-%d') is_completed = True not_completed_reports = [] for report_name in get_available_reports(date): if date_obj < datetime(2019, 11, 1) and report_name == 'amShazam': continue if date_obj < datetime(2020, 11, 9) and report_name == \ 'amSubscriptionReference': continue if (licensor, report_name) == ('theorchard', 'amSongs'): continue if licensor in config.reports[report_name]['licensors']: feed_name = _get_feed_name(report_name, licensor) if not task_status.is_completed_report(feed_name, date): is_completed = False not_completed_reports.append(report_name) overall_status = garcon_feed_status.get_overall_status( feed_name, date) if (overall_status != garcon_feed_status.STATUS_INGESTED and report_name not in config.common_reports): return STOP_RESPONSE overall_feed_name = '_'.join([config.feed_name, licensor]) set_overall_status_enhanced( overall_feed_name, date, garcon_feed_status.STATUS_INGESTED, activity) task_status.mark_completed_overall_job( overall_feed_name, date, is_completed) task_status.set_values(overall_feed_name, date, 'not_completed_reports', not_completed_reports) activity.logger.info( 'Setting status for feed: {feed_name} with date: {date} ' 'to: {status}, via a task, ' 'is is_completed processing: {is_completed}'.format( feed_name=config.feed_name, date=date, status=garcon_feed_status.STATUS_INGESTED, is_completed=is_completed)) # it is needed to add staging_raw_table_tasks to completed tasks # for amContent to prevent loading it into snowflake in case of soft reload for report_name in config.common_reports: task_status.mark_completed_task( _get_feed_name(report_name, licensor), date, STAGING_RAW_TASK_ID) def _get_feed_name(report_name, licensor): """Generate feed_name for specified licensor and report. Args: report_name (str): Name of report. licensor (str): Name of the licensor to ingest. Returns: str: feed_name. """ return '_'.join([config.feed_name, licensor, report_name]) def get_vendors_config(licensor, vendor_accounts): """Map vendor ids and property files for all vendors. Args: licensor (str): The licensor to ingest. vendor_accounts (list): The list of vendor_account the licensor. Returns: dict: with vendor ids and names of property files. """ vendors = defaultdict(dict) for vendor_id in vendor_accounts: vendor_account = config.vendor_account_mapping.get( vendor_id, config.vendor_account_mapping.get(licensor)) vendors[vendor_id] = { 'VENDOR_ID': vendor_id, 'ACCOUNT': vendor_account } return vendors def get_temp_table_name(date, report_name, reporter_account, licensor): """Generate temp_table_name for specified licensor and report. Args: date (str): Reporting date (YYYY-MM-DD). report_name (str): Name of report. reporter_account (str): A vendor account of the licensor. licensor (str): Name of the licensor to ingest. Returns: str: temp_table_name. """ date_obj = datetime.strptime(date, '%Y-%m-%d') return config.temp_table_name.format( report=report_name, reporter_account=reporter_account, date=date_obj, licensor=licensor)