""" Deezer Snowflake-only Data Ingestion Workflow tasks. Tasks to ingest data from Deezer feed into staging_raw_deezer_v2 and fact tables in Snowflake. """ from datetime import date as date_module, datetime from datetime import timedelta import errno import gzip import os import shutil import boto3 from garcon import task from garcon_contrib.aws.utils import garcon_s3 from garcon_contrib.dynamo_feed_status import garcon_feed_status from feed_ingestion.conf.config import BOTO3_CONFIG from feed_ingestion.flows.deezer import config from feed_ingestion.flows.deezer.config import get_version from feed_ingestion.tasks import check_status, s3_tasks from feed_ingestion.util import os_tools from feed_ingestion.util import task_status from feed_ingestion.util.aws import s3 as s3utils from feed_ingestion.util.context_util import strtobool import feed_ingestion.util.deezer_zephir_utils as zephir @task.decorate(timeout=600) def bootstrap(activity, date, licensor, reload=None, use_s3='False', date_format='DD-MM-YYYY', fraud_report_backfill=None): """Bootstrap workflow by getting the correct configurations. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). licensor (str): one of config.licensors. reload (str or None): If 'True' then clear all feed statuses. use_s3 (str): If flow should backfill data from S3 only. date_format (str): Format of date in files. fraud_report_backfill (str or None): If 'True', backfill fraud report from the separate fraudulent_reports folder. Returns: dict: Initial context of the workflow. """ activity.logger.info( 'Bootstrapping {feed_name}...'.format(feed_name=config.feed_name)) # date is the date passed in or yesterday's date if not date: date = (date_module.today() - timedelta(days=1)).strftime('%Y-%m-%d') if not licensor: raise ValueError('no licensor in the context') assert licensor in config.licensors, f'unsupported licensor "{licensor}"' feed_name = '_'.join([config.feed_name, licensor]) is_fraud_backfill = ( fraud_report_backfill is not None and bool(strtobool(fraud_report_backfill)) ) 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 and not is_fraud_backfill): message = 'already ingested for {}'.format(date) activity.logger.info(message) return { 'message': message, 'stop': True } date_format = date_format if date_format is not None else 'DD-MM-YYYY' use_s3 = use_s3 if use_s3 is not None else 'False' use_s3 = bool(strtobool(use_s3)) date_dt = datetime.strptime(date, '%Y-%m-%d').date() month_prefix = '0' if date_dt.month < 10 else '' day_prefix = '0' if date_dt.day < 10 else '' spec_version = get_version( config.spec_version[licensor], date_dt ) drop_file_name = \ config.drop_file_name[f'{licensor}_v{spec_version}'].format( datestamp_YYYYMMDD=date.replace('-', '') ) if licensor == 'theorchard' and spec_version == 1: source_path = config.zephir.get('path') elif licensor == 'sme' and spec_version == 2: source_path = config.sme_drop_path elif licensor == 'altafonte': source_path = config.altafonte_drop_path.format( date=date_dt) else: # orchard instead of theorchard is used s3_licensor = licensor.replace('the', '') # cut PG45/ from drop pathfor v3 source_path = ( config.sme_drop_path[:-5] + f'{s3_licensor}/{date_dt.year}/{month_prefix}{date_dt.month}/' + f'{day_prefix}{date_dt.day}/' ) # for theorchard_v1 sample: TheOrchard_20201111_20201111 # for sme_v2 sample: sony_20201113_20201113 # for licensor_v3 file names are in file_suffixes completely, no prefix filename_prefix_common = \ '' if spec_version == 3 else drop_file_name.replace('.zip', '') if licensor == 'altafonte': filename_prefix_common = config.altafonte_reports_prefix.format( date=date_dt) files = [f'{filename_prefix_common}{suffix}' for suffix in config.file_suffixes[ config.names_alias[f'{licensor}_v{spec_version}'] ]] staging_raw_table = config.snowflake_table_names[ 'staging_raw'][config.names_alias[f'{licensor}_v{spec_version}']] s3_archive_bucket, s3_temp_staging_raw_bucket = \ config.s3.get('archive_bucket').format( datestamp=date, licensor=licensor, spec_version=spec_version, s3_bucket=config.s3_bucket), \ config.s3.get( 'temp_staging_raw_bucket').format( datestamp=date, licensor=licensor, spec_version=spec_version, s3_bucket=config.s3_bucket) if licensor == 'altafonte': s3_drop_bucket = config.altafonte_drop_bucket else: s3_drop_bucket = config.sme_drop_bucket if is_fraud_backfill: files = list(config.v3_files_fraud) staging_raw_table = [ config.snowflake_table_names['staging_raw']['v3'][-1] ] spec_version = 3 return dict( date=date, source_files_dict={'files': files}, date_as_in_uuid=date.replace('-', ''), feed_name=feed_name, licensor=licensor, spec_version=spec_version, secrets_path=config.secrets_path, source_path=source_path, s3_drop_bucket=s3_drop_bucket, s3_archive_bucket=s3_archive_bucket, s3_temp_staging_raw_bucket=s3_temp_staging_raw_bucket, drop_file_name=drop_file_name, staging_raw_date_col='download_date', staging_raw_table=staging_raw_table, fact_analytics_table='fact_analytics', fact_analytics_error_table='fact_analytics_error', dimension_tables=config.dimension_tables, common_kwargs={'licensor': licensor}, use_s3=use_s3, date_format=date_format, jenkins_config=config.jenkins_config, fraud_report_backfill=is_fraud_backfill, ) @task.decorate(timeout=1 * 60 * 60) @check_status() def fetch_from_drop_location( activity, date, feed_name, source_path, target_s3_path, drop_file_name): """Download a feed file from the drop location into the archive folder. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). feed_name (str): Name of the feed. source_path (str): remote zephir route & params to the source file. target_s3_path (str): Destination S3 path to the archive location. drop_file_name (str): The dropped file name. """ task_id = 'fetch_from_drop_location' # Short circuit flow if overall status is already INGESTED if garcon_feed_status.get_overall_status( feed_name, date) == garcon_feed_status.STATUS_INGESTED: activity.logger.info('Feed already ingested for {}'.format(date)) return { 'stop': True, 'message': '{feed_name} is already ingested for {date}'.format( feed_name=feed_name, date=date)} # Short circuit flow if task status is already Completed if task_status.is_completed_task(feed_name, date, task_id): activity.logger.info( '{file} was already copied from Zephir.'.format( file=drop_file_name)) return s3_bucket_name, s3_dir_key = garcon_s3.extract_bucket_path(target_s3_path) s3_file_key = os.path.join(s3_dir_key, drop_file_name) # delete from S3 if exists s3utils.delete_s3_obj(s3_bucket_name, s3_file_key) activity.logger.info("'{}' was succesfully deleted from '{}'".format( s3_file_key, s3_bucket_name)) # setup Zephir settings zephir_settings = dict( username=config.zephir.get('username'), password=config.zephir.get('password'), host=config.zephir.get('host'), path=source_path ) try: local_dir = os_tools.create_temp_dir(feed_name=feed_name, date=date) local_file_path = zephir.download_daily_report_from_zephir( zephir_settings, drop_file_name, local_dir) if local_file_path != 'empty_file': activity.logger.info( '{file} was downloaded from drop location.'.format( file=drop_file_name)) else: raise FileNotFoundError( errno.ENODATA, 'Empty file was downloaded from drop location', drop_file_name ) except FileNotFoundError as err: garcon_feed_status.set_overall_status( feed_name, date, garcon_feed_status.STATUS_NOT_AVAILABLE) garcon_feed_status.set_missing_files( feed_name, date, [s3_file_key]) activity.logger.error(str(err)) return {'stop': True, 'message': str(err)} # Upload downloaded file to s3 file_size = s3utils.upload_to_s3( local_file_path, s3_bucket_name, s3_file_key) if file_size is None: garcon_feed_status.set_overall_status( feed_name, date, garcon_feed_status.STATUS_NOT_INGESTED) return {'stop': True, 'message': "Upload to '{}' has failed".format(s3_file_key)} # Remove file from local dir when done uploading os.remove(local_file_path) task_status.mark_completed_task(feed_name, date, task_id) activity.logger.info( '{file} ({file_size}) was uploaded to S3 archives.'.format( file_size=file_size, file=drop_file_name)) @task.decorate(timeout=3000) @check_status() def grab_drop_files_sme( activity, feed_name, date, drop_file_name, source_bucket_name, source_path, destination_full_path): """Copy drop file from sme S3 bucket to theorchard s3.""" source_key_name = '{dir}{file}'.format( dir=source_path, file=drop_file_name) archive_file = '{dir}{file}'.format( dir=destination_full_path, file=drop_file_name) destination_key_name = garcon_s3.extract_bucket_path(archive_file)[1] destination_bucket_name = garcon_s3.extract_bucket_path(archive_file)[0] result = s3_tasks.copy_file_from_sme_s3_to_theocrhard( activity, config.secrets_path, source_bucket_name, source_key_name, destination_bucket_name, destination_key_name, replace=True) file_name = destination_key_name.split('/')[-1] file_name_exists = result.get(file_name) if not file_name_exists: garcon_feed_status.set_overall_status( feed_name, date, garcon_feed_status.STATUS_NOT_AVAILABLE) garcon_feed_status.set_missing_files(feed_name, date, [file_name]) return {'stop': True} @task.decorate(timeout=3600) @check_status() def grab_drop_files_altafonte( activity, feed_name, date, drop_file_name, source_bucket_name, source_path, destination_full_path): """Copy a feed files from the drop location into the archive folder. Args: activity (ActivityWorker): The activity worker. feed_name (str): The name of the feed. date (str): Reporting date (YYYY-MM-DD). drop_path (str): s3 path to raw files. archive_path (str): s3 path to drop raw files. licensor (str): The licensor to ingest. """ source_key_name = '{dir}{file}'.format( dir=source_path, file=drop_file_name) archive_file = '{dir}{file}'.format( dir=destination_full_path, file=drop_file_name) destination_key_name = garcon_s3.extract_bucket_path(archive_file)[1] destination_bucket_name = garcon_s3.extract_bucket_path(archive_file)[0] result = s3_tasks.copy_file( activity=activity, source_bucket_name=source_bucket_name, source_key_name=source_key_name, destination_bucket_name=destination_bucket_name, destination_key_name=destination_key_name, replace=True) file_name = destination_key_name.split('/')[-1] file_name_exists = result.get(file_name) if not file_name_exists: garcon_feed_status.set_overall_status( feed_name, date, garcon_feed_status.STATUS_NOT_AVAILABLE) garcon_feed_status.set_missing_files(feed_name, date, [file_name]) return {'stop': True} @task.decorate(timeout=3000) @check_status() def grab_fraud_report_backfill( activity, feed_name, date, source_bucket_name, destination_full_path, licensor): """Copy fraud report from the separate fraudulent_reports folder. Downloads the txt file from the fraudulent_reports S3 path, gzips it (the Snowflake stage expects COMPRESSION='GZIP'), and uploads it to the temp staging raw location with the canonical filename so the stage_loader can find it. Args: activity (ActivityWorker): The activity worker. feed_name (str): Name of the feed. date (str): Reporting date (YYYY-MM-DD). source_bucket_name (str): SME S3 bucket name. destination_full_path (str): S3 path to temp staging raw. licensor (str): one of config.licensors. """ date_compact = date.replace('-', '') fraud_name = config.fraud_report_licensor_names.get(licensor, licensor) s3_licensor = licensor.replace('the', '') source_path = ( f'deezer/in/{s3_licensor}/fraudulent_reports/{date_compact}/' ) source_file = ( f'{date_compact}_fraud_report_{fraud_name}-{date_compact}.txt' ) source_key = source_path + source_file local_dir = os_tools.create_temp_dir(feed_name=feed_name, date=date) local_txt = os.path.join(local_dir, source_file) sme_s3_client = s3_tasks._get_sme_s3_client(config.secrets_path) try: sme_s3_client.download_file( source_bucket_name, source_key, local_txt) except Exception as err: activity.logger.error( 'Failed to download %s: %s', source_key, err) garcon_feed_status.set_overall_status( feed_name, date, garcon_feed_status.STATUS_NOT_AVAILABLE) garcon_feed_status.set_missing_files( feed_name, date, [source_file]) return {'stop': True} target_file = f'fraud_report_{fraud_name}-{date_compact}.txt' local_gz = os.path.join(local_dir, f'{target_file}.gz') with open(local_txt, 'rb') as f_in: with gzip.open(local_gz, 'wb') as f_out: shutil.copyfileobj(f_in, f_out) target_gz = f'{target_file}.gz' target_full = destination_full_path + target_gz dest_bucket, dest_key = garcon_s3.extract_bucket_path(target_full) s3utils.delete_s3_obj(dest_bucket, dest_key) file_size = s3utils.upload_to_s3(local_gz, dest_bucket, dest_key) os.remove(local_txt) os.remove(local_gz) if file_size is None: garcon_feed_status.set_overall_status( feed_name, date, garcon_feed_status.STATUS_NOT_INGESTED) return { 'stop': True, 'message': f"Upload to '{dest_key}' has failed"} activity.logger.info( '%s (%s) uploaded to %s', target_gz, file_size, dest_key) @task.decorate(timeout=60) def sns_publish_message(activity, feed_name, date, topic, message, subject): """Send SNS messages to specific topic w/subject & message. Assumes AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY are set as env vars Args: activity (ActivityWorker): The swf activity worker. feed_name (str): Feed name of workflow execution for status updates. date (str): Reporting date (YYYY-MM-DD). topic (str): Topic ARN (ex.arn:aws:sns:us-east-1:103233932089:dev_test) message (str): The message you want to send to the topic. Messages must be UTF-8 encoded strings and be at most 4KB in size. subject (str): Optional parameter to be used as the "Subject" line of the email notifications. """ task_id = 'sns_publish_message' if task_status.is_completed_task(feed_name, date, task_id): 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=feed_name, date=date)) return if subject: client = boto3.client('sns', config=BOTO3_CONFIG) client.publish(TopicArn=topic, Message=message, Subject=subject) activity.logger.info('SNS report about update_dim_tables sent') task_status.mark_completed_task(feed_name, date, task_id) activity.logger.info( 'Task {task_id} of {feed_name} for {date} completed'.format( task_id=task_id, feed_name=feed_name, date=date))