""" Proper Music Distribution Ingestion Workflow. Tasks to download files from Proper SFTP to S3, validate them and ingest to MySQL. """ import csv from datetime import datetime import os import re import tempfile 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 garcon_contrib.ftp import garcon_ftp from garcon_contrib.mysql import garcon_mysql from garcon_contrib.snowflake import garcon_snowflake import pymysql # todo: the tested methods should be moved from helpers to proper_income.tasks from feed_ingestion.flows import helpers from feed_ingestion.flows.proper_incoming import config from feed_ingestion.flows.proper_incoming import json_schemas from feed_ingestion.util import task_status from feed_ingestion.util.aws import s3 as s3_feed_ingestion_utils from feed_ingestion.util.aws import s3 as s3utils @task.decorate(timeout=300) def bootstrap(activity, date, proper_feed_type): """Bootstrap workflow by getting the correct configurations. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). proper_feed_type (str): One of ['goodsin', 'sales', 'shortages', 'stock'] proper feed types. On this param value depends logic and sequence of executed tasks. Returns: dict: Context. """ activity.logger.info('Bootstrapping...') # set feed name, which depends on proper_feed_type feed_name = 'proper_daily_{feed_type}'.format(feed_type=proper_feed_type) s3_path = config.target_s3_path.format( s3_bucket=config.s3_bucket, date_YYYY_MM_DD=date) ftp_file_name = config.proper_feeds.get(proper_feed_type).get( 'ftp_file_name').format(date_YYYY_MM_DD=date) s3_file_name = config.proper_feeds.get(proper_feed_type).get( 's3_file_name').format(date_YYYY_MM_DD=date) s3_file_full_path = '{}{}'.format(s3_path, s3_file_name) return dict( date=date, proper_feed_type=proper_feed_type, feed_name=feed_name, s3_path=s3_path, ftp_file_name=ftp_file_name, s3_file_name=s3_file_name, s3_file_full_path=s3_file_full_path ) @task.decorate(timeout=300) def fetch_from_drop_location( activity, date, feed_name, ftp_path, s3_path, ftp_file_name, s3_file_name): """Download the feed files. (From the drop location on SFTP into the archive folder on S3). Stops if the flow has INGESTED status in DynamoDB. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). feed_name (str): Name of the feed, formatted for status service. ftp_path (str): FTP path to the source files. s3_path (str): Destination S3 path to the archive location. ftp_file_name (str): Source file name on FTP. s3_file_name (str): Target file name on S3. """ # Stop the flow if overall status is INGESTED. if (garcon_feed_status.get_overall_status(feed_name, date) == garcon_feed_status.STATUS_INGESTED): activity.logger.info( 'The flow was stopped since {file} was already ingested for ' '{date} date.'.format(file=ftp_file_name, date=date)) return {'stop': True} copy_response = garcon_ftp.copy_file_from_ftp_to_s3( activity, config.ftp, ftp_path, ftp_file_name, s3_path, s3_file_name) if copy_response.get('status') is True: garcon_feed_status.set_status( feed_name, date, copy_response.get('file'), status=garcon_feed_status.STATUS_DOWNLOADED) activity.logger.info( '{file} ({file_size}) was copied from FTP.'.format( file=copy_response.get('file'), file_size=copy_response.get('file_size'))) else: garcon_feed_status.set_overall_status( feed_name, date, garcon_feed_status.STATUS_NOT_AVAILABLE) activity.logger.info( "Couldn't load file {file} due error: {error}".format( file=copy_response.get('file'), error=copy_response.get('exception'))) return {'stop': True} @task.decorate(timeout=300) def validate_csv_file( activity, date, feed_name, proper_feed_type, s3_file_full_path, expected_bucket_owner='437795906767'): """Validate CSV file against the corresponding schema. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). feed_name (str): Name of the feed, formatted for status service. proper_feed_type (str): Proper feed type passed through run context. s3_file_full_path (str): Target S3 file path. expected_bucket_owner (str): Expected bucket owner. """ if proper_feed_type not in json_schemas.schemas['properties']: raise KeyError('There is no schema for {} feed type'.format( proper_feed_type)) s3_client = boto3.client('s3') bucket_name, path = garcon_s3.extract_bucket_path(s3_file_full_path) activity.logger.info( 'Connected to bucket: {bucket}'.format(bucket=bucket_name)) # save key content to local temp file with tempfile.TemporaryDirectory() as tmp_dir: filename = os.path.join(tmp_dir, 'tmp_file.csv') activity.logger.info( 'Downloading from {key} to file://{filename}'.format( key=path, filename=filename)) s3_client.download_file( bucket_name, path, filename, ExtraArgs={'ExpectedBucketOwner': expected_bucket_owner} ) with open(filename, mode='rt', newline='', encoding='iso-8859-1') as file_to_validate: # header validation fieldnames = json_schemas.schemas['properties'][proper_feed_type][ 'sql_options']['csv_file_header']['items'] csv_obj = csv.DictReader(file_to_validate) activity.logger.info('Validating the header of CSV file') helpers.validate_header(csv_obj.fieldnames, fieldnames) # rows values validation activity.logger.info('Validating the rows of CSV file') row_schema = json_schemas.schemas['properties'][proper_feed_type][ 'row_schema'] for row in csv_obj: helpers.validate_row(row_schema, row) task_status.mark_completed_task(feed_name, date, 'validate_csv_file') @task.decorate(timeout=300) def transform_csv_file( activity, proper_feed_type, feed_name, date, file_to_transform): """Extend passed CSV file with 3 additional columns. These columns are: 1. file_date - date in file name (file name must be presented with next format 'file_name_YYYY-MM-DD.csv, otherwise raise ValueError exception); 2. file_name - full S3 path; 3. ingestion_timestamp - timestamp of file transform execution (format: '%Y-%m-%d,%H:%M:%S.%f'). The transformed files then copied to /ProperIncoming/snowflake/. Args: activity (ActivityWorker): The activity worker. proper_feed_type (str): Type of the flow, which determines CSV file and table it works with. feed_name (str): Name of the feed, formatted for status service. date (str): Flow run date (YYYY-MM-DD). file_to_transform (str): S3 key of file to be transformed ( e.g., 's3://bucket/ProperIncoming/archives/2016-09-29/ GoodsIn_ESSN_2016-09-29.csv'). """ extend_csv_flag = json_schemas.schemas[ 'properties'][proper_feed_type]['sql_options'].get( 'extend_csv_before_ingestion_to_sf') if extend_csv_flag: activity.logger.info( 'Extending CSV file with file_name, file_date ' 'and ingestion_timestamp columns...') # Where to copy the transformed CSV file path_to_unload = config.path_to_unload.format(date=date) file_path_parts = file_to_transform.split('/') file_name = file_path_parts[-1] file_date = file_path_parts[-2] ingestion_timestamp = datetime.strftime( datetime.now(), '%Y-%m-%dT%H:%M:%S') columns_values = (file_name, file_date, ingestion_timestamp) columns_header = ('file_name', 'file_date', 'ingestion_timestamp') if proper_feed_type == config.FEED_TYPE_GOODSIN: line_rstrip_char = ',\r\n' else: line_rstrip_char = None s3_feed_ingestion_utils.expand_s3_csv_with_columns( file_to_transform, columns_header, columns_values, path_to_unload, line_rstrip_char=line_rstrip_char) activity.logger.info( '{file_to_transform} CSV file transformed and copied to ' '{path_to_unload}'.format( file_to_transform=file_to_transform, path_to_unload=path_to_unload)) task_status.mark_completed_task( feed_name, date, 'transform_csv_file') transormed_file_path = ('s3://{bucket}{path}/{file_name}'.format( bucket=config.s3_bucket, path=path_to_unload, file_name=file_name.rsplit('/')[-1])) return {'s3_file_path': transormed_file_path} @task.decorate(timeout=600) def transform_csv_file_stock( activity, feed_name, date, s3_file_full_path): """Create a new csv file without rows with incorrect label code. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). feed_name (str): Name of the feed, formatted for status service. s3_file_full_path (str): Target S3 file path. Returns: s3_file_full_path: The s3 path to the new file. """ activity.logger.info(f'Clean up stock data, file path {s3_file_full_path}') df = s3utils.read_csv(s3_file_full_path) df_filtered = df[df['LabelCode'] != 'FOC'] df_filtered['LabelCode'] = df_filtered['LabelCode'].astype(str) # after removing rows with incorrect label code # check the other rows if any(df_filtered['LabelCode'].str.len() == 3): raise ValueError('The LabelCode length should be at least 4.') s3_path = config.target_s3_stock_path.format( s3_bucket=config.s3_bucket, date_YYYY_MM_DD=date) filename = os.path.basename(s3_file_full_path) s3_file_full_path_new = f'{s3_path}{filename}' activity.logger.info( f'Creating new file with correct stock data, ' f'file path {s3_file_full_path_new}') s3utils.to_csv(df_filtered, s3_file_full_path_new) return {'s3_file_full_path': s3_file_full_path_new} @task.decorate(timeout=300) def ingest_stock_data_into_mysql_table( activity, feed_name, date, proper_feed_type, s3_file_full_path): """Ingest Stock_ESSN CSV file from S3 to MySQL table. We use masks to decouple ingesting task from validation task. This ingest_stock_data_into_mysql_table task takes as an argument all daily CSV files (files_to_ingest), but chooses which one to ingest to MySQL based on parameter 'ingest_to_mysql_mask' from JSON schema. For now it's a little bit redundant, but will be helpful if we'll want to ingest another file or files to MySQL. Args: activity (ActivityWorker): The activity worker. feed_name (str): Name of the feed, formatted for status service. date (str): Reporting date (YYYY-MM-DD). proper_feed_type (str): Proper feed type passed through run context. s3_file_full_path (str): Target S3 file path. """ if proper_feed_type not in json_schemas.schemas['properties']: raise KeyError('There is no schema for {} feed type'.format( proper_feed_type)) mysql_config = config.mysql_db_config activity.logger.info('Ingesting stock data from Stock_ESSN file...') table_name = json_schemas.schemas['properties'][proper_feed_type][ 'sql_options']['table_name'] columns_names = json_schemas.schemas['properties'][proper_feed_type][ 'sql_options']['columns_names']['items'] garcon_mysql.bulk_insert_from_csv_file_on_s3( activity, s3_file_full_path, mysql_config, table_name, columns_names, ignore_lines=1) task_status.mark_completed_task( feed_name, date, 'ingest_stock_data_into_mysql_table') activity.logger.info('Ingestion complete') return @task.decorate(timeout=300) def add_release_id_to_proper_stock_essn_table(activity, feed_name, date): """Update art_relations.proper_stock_essn.release_id. Update art_relations.proper_stock_essn.release_id with the corresponding release_id from art_relations.releases. The matching occurs as follows: 0) ALL release should have an entry in art_relations.product_physical table (can be determined by joining releases.release_id and product_physical.release_id). 1) Attempt to join proper_stock_essn and match proper_stock_essn.ean on releases.upc and update proper_stock_essn.release_id with the match release_id. 2) Attempt to join proper_stock_essn and match proper_stock_essn.ean on releases.display_upc and update proper_stock_essn.release_id with the match release_id. 2) If no match is found in step 1 or 2, attempt to match proper_stock_essn.ean on releases.manufacturer_upc and update proper_stock_essn.release_id with the match release_id. 3) If no match is found in step 1 or 2 or 3, attempt to match proper_stock_essn.catalogue_number on releases.vendor_catalog_number and update proper_stock_essn.release_id with the match release_id. 4) If no match with either step 1 or 2 or 3, leave release_id as NULL. Args: activity (ActivityWorker): The activity worker. feed_name (str): Name of the feed, formatted for status service. date (str): Reporting date (YYYY-MM-DD). """ mysql_config = config.mysql_db_config activity.logger.info('Adding proper_stock_essn.release_id column...') if not re.match(r'\d{4}-\d{2}-\d{2}', date): raise Exception('Feed date value incorrect') db_connection = pymysql.connect( host=mysql_config['host'], user=mysql_config['user'], password=mysql_config['password'], db=mysql_config['db_name'], charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) try: with db_connection.cursor() as cursor: cursor.execute(config.update_proper_stock_essn_by_upc_sql) cursor.execute( config.update_proper_stock_essn_by_display_upc_sql) cursor.execute( config.update_proper_stock_essn_by_manufacturer_upc_sql) cursor.execute( config.update_proper_stock_essn_by_catalog_number_sql) except: # noqa raise else: db_connection.commit() finally: db_connection.close() task_status.mark_completed_task( feed_name, date, 'add_release_id_to_proper_stock_essn_table') activity.logger.info( 'proper_stock_essn.release_id column and the matched values added') @task.decorate(timeout=300) def purge_old_data_from_snowflake( activity, proper_feed_type, feed_name, date, file_to_ingest_to_snowflake): """Purge the rows from Snowflake table. (If there are the rows with the same file_name (full S3 path) as in file to ingest). Args: activity (ActivityWorker): The activity worker. proper_feed_type (str): Type of the flow, which determines CSV file and table it works with. feed_name (str): Name of the feed, formatted for status service. date (str): Flow run date (YYYY-MM-DD). file_to_ingest_to_snowflake (str): S3 key of file to be ingested to SF """ table = json_schemas.schemas[ 'properties'][proper_feed_type]['sql_options'].get( 'snowflake_table_name') snowflake_config = config.snowflake_db_config if table: activity.logger.info( 'Purging old data for this file_date from Snowflake...') sql = ( 'DELETE FROM PROD.PRODUCTION.{table} ' "WHERE file_name='{file_name}';".format( file_name=file_to_ingest_to_snowflake, table=table)) with garcon_snowflake.connect( user=snowflake_config['user'], account=snowflake_config['account'], private_key=snowflake_config['key'], role=snowflake_config['role']) as conn: with garcon_snowflake.cursor(conn) as curs: curs.execute('USE WAREHOUSE {};'.format( snowflake_config['warehouse'])) curs.execute('USE DATABASE {};'.format( snowflake_config['db'])) curs.execute('USE SCHEMA {};'.format( snowflake_config['schema'])) curs.execute(sql) activity.logger.info( 'Purging from Snowflake table {table} complete'.format( table=table)) task_status.mark_completed_task( feed_name, date, 'purge_old_data_from_snowflake') return @task.decorate(timeout=7200) def load_data_into_snowflake( activity, feed_name, date, file_on_s3, proper_feed_type): """Load data into Snowflake. Args: activity (ActivityWorker): The activity worker. feed_name (str): Name of the feed, formatted for status service. date (str): Reporting date (YYYY-MM-DD). file_on_s3 (str): CSV file (S3 key of the renamed daily file). proper_feed_type (str): Type of the flow (one of 'Stock_ESSN', 'Shortages_ESSN', 'Sales_ESSN' or 'GoodsIn_ESSN'). """ table = json_schemas.schemas[ 'properties'][proper_feed_type]['sql_options'].get( 'snowflake_table_name') credentials = boto3.Session().get_credentials() snowflake_config = config.snowflake_db_config if table: sql = """COPY INTO PROD.PRODUCTION.{table} FROM '{file_on_s3}' FILE_FORMAT = ( TYPE=CSV RECORD_DELIMITER='\n' FIELD_DELIMITER=',' FIELD_OPTIONALLY_ENCLOSED_BY='"' DATE_FORMAT='YYYY-MM-DD' SKIP_HEADER=1 ) CREDENTIALS=( AWS_KEY_ID='{access_key}' AWS_SECRET_KEY='{access_secret}' AWS_TOKEN='{aws_token}' );""".format( table=table, file_on_s3=file_on_s3, access_key=credentials.access_key, access_secret=credentials.secret_key, aws_token=credentials.token ) with garcon_snowflake.connect( user=snowflake_config['user'], private_key=snowflake_config['key'], account=snowflake_config['account'], role=snowflake_config['role']) as conn: with garcon_snowflake.cursor(conn) as curs: curs.execute('USE WAREHOUSE {};'.format( snowflake_config['warehouse'])) curs.execute('USE DATABASE {};'.format( snowflake_config['db'])) curs.execute('USE SCHEMA {};'.format( snowflake_config['schema'])) curs.execute(sql) task_status.mark_completed_task( feed_name, date, 'load_data_into_snowflake')