"""Reusable Garcon Tasks related to Deezer flows.""" from datetime import datetime import os import shutil from zipfile import BadZipFile 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.flows.deezer import config from feed_ingestion.flows.deezer.config import get_version from feed_ingestion.util import os_tools from feed_ingestion.util.aws import s3 as s3utils import feed_ingestion.util.deezer_zephir_utils as zephir def repack_zip_from_s3(activity, feed_name, date, source_name, s3_path, file_name, local_dir, one_file=False): """Repack file(s) to .zip.""" # Source config src_s3_bucket_name, src_s3_dir_key = garcon_s3.extract_bucket_path( s3_path) src_s3_file_key = os.path.join(src_s3_dir_key, file_name) local_file_path = os.path.join(local_dir, file_name) # Download '.zip' from s3 to local file_size = s3utils.download_from_s3( src_s3_bucket_name, src_s3_file_key, local_file_path) if file_size is None: garcon_feed_status.set_overall_status( feed_name, date, garcon_feed_status.STATUS_NOT_INGESTED) details = "'{file}' download from '{bucket}' bucket has failed".format( file=file_name, bucket=s3_path) activity.logger.error(details) return {'stop': True, 'message': str(details)} try: gzipped_files = zephir.repack_source_file( local_file_path, local_dir, source_name, one_file=one_file) except BadZipFile as err: garcon_feed_status.set_overall_status( feed_name, date, garcon_feed_status.STATUS_NOT_INGESTED) activity.logger.error( '{file} is not a valid zipfile'.format(file=file_name)) raise err for f in gzipped_files: activity.logger.info( "'{}' gzipped to '{}'".format( f['source_file_name'], f['gzip_file_name'])) return gzipped_files def upload_zipfile_to_s3( activity, feed_name, date, filedict, s3_path, s3_file_key=None): """Upload .zip file to S3.""" tgt_s3_bucket_name, tgt_s3_dir_key = garcon_s3.extract_bucket_path( s3_path) tgt_s3_file_key = os.path.join(tgt_s3_dir_key, filedict['gzip_file_name']) \ if not s3_file_key else s3_file_key # noqa # If exists delete '.tsv.gz' file from s3 s3utils.delete_s3_obj(tgt_s3_bucket_name, tgt_s3_file_key) activity.logger.info("'{}' was succesfully deleted from '{}'".format( tgt_s3_bucket_name, tgt_s3_file_key)) # Send gzipped file to s3 temp bucket file_size = s3utils.upload_to_s3( filedict['gzip_file'], tgt_s3_bucket_name, tgt_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( tgt_s3_file_key)} activity.logger.info( "'{file}' ({size}) was uploaded to '{key}'".format( file=filedict['gzip_file_name'], size=file_size, key=tgt_s3_file_key)) return { 'file_name': f's3://{tgt_s3_bucket_name}/{tgt_s3_file_key}', 'file_size': file_size, 'found': True } def cleanup_local_folder(activity, local_dir, drop_file_name): """Cleanup local unzipped folder and original .zip.""" shutil.rmtree(local_dir) activity.logger.info( '{file} deleted local unzipped folder and original .zip'.format( file=drop_file_name)) @task.decorate(timeout=10800) def unzip_and_clean( activity, date, feed_name, source_s3_path, target_s3_path, drop_file_name, licensor, source_files_dict, use_drop_file=False): """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_s3_path (str): Bucket from which to fetch .ZIP. target_s3_path (str): Destination bucket. drop_file_name (str): The file name. licensor (str): one of config.licensors. source_files_dict (dict): A dict with source files metadata. use_drop_file (bool): Use source files or a drop file. """ # Prepare downloads dir _result = [] local_dir = os_tools.create_temp_dir(feed_name=feed_name, date=date) tgt_s3_bucket_name, tgt_s3_dir_key = garcon_s3.extract_bucket_path( target_s3_path) if licensor is not None: spec_version = get_version( config.spec_version[licensor], datetime.strptime(date, '%Y-%m-%d').date() ) if (spec_version == 3 and isinstance(source_files_dict['files'], list)): fraud_licensor = config.fraud_report_licensor_names.get( licensor, licensor) source_files_dict['files'][-1] = \ source_files_dict['files'][-1].replace( '*', fraud_licensor + '-' + date.replace('-', '') ) if use_drop_file: for file in source_files_dict['files']: zip_file_name = file['file_name'] gzipped = repack_zip_from_s3( activity, feed_name, date, source_name=drop_file_name, s3_path=source_s3_path, file_name=zip_file_name, local_dir=local_dir, one_file=True) if isinstance(gzipped, dict) and gzipped.get('stop'): return gzipped for gzip in gzipped: tgt_s3_file_key = os.path.join( tgt_s3_dir_key, file['file_name'].replace('.zip', '_') + # noqa gzip['gzip_file_name']) # noqa uploaded = upload_zipfile_to_s3( activity, feed_name, date, filedict=gzip, s3_path=target_s3_path, s3_file_key=tgt_s3_file_key) if uploaded.get('stop'): return uploaded _result.append(uploaded) else: gzipped = repack_zip_from_s3( activity, feed_name, date, source_name=source_files_dict, s3_path=source_s3_path, file_name=drop_file_name, local_dir=local_dir) if isinstance(gzipped, dict) and gzipped.get('stop'): return gzipped for gzip in gzipped: upload_zipfile_to_s3(activity, feed_name, date, filedict=gzip, s3_path=target_s3_path) # noqa cleanup_local_folder(activity, local_dir, drop_file_name) return dict(source_files_dict={'files': _result})