"""Reads assets from S3 bucket. Rename and move them to the ripper.""" from datetime import datetime import os # basic os operations like paths import sys import glob import shutil from timeit import default_timer from integration_scripts.config import SNOWFLAKE_DATABASE, SNOWFLAKE_SCHEMA, \ S3_BUCKET from integration_scripts.db_utils import get_snowflake_results, \ get_snowflake_bind_results from integration_scripts.file_utils import create_path_with_todays_date, \ get_s3_file_key, copy_file, get_files_at_path, on_same_filesystem from integration_scripts import logger from integration_scripts.connectors import sentry from integration_scripts.connectors.s3 import s3_bucket_name import integration_scripts.s3_backoff_utils as s3_util from constants import paths as asset_paths from constants import asset_types from util import logic_utils from util import wav_utils from util.exceptions import FlowControlBackupFound, FlowControlDupeFound from util.file_utils import create_target_paths from util.transform import transform_audio, transform_covers from sql.select_eligible_audio_for_ripper \ import SELECT_ELIGIBLE_AUDIO_FOR_RIPPER from sql.select_eligible_covers_for_ripper \ import SELECT_ELIGIBLE_COVERS_FOR_RIPPER import config UNDO = config.UNDO_ACTION COPY_LOCAL = config.COPY_LOCAL BACKUP = config.MAKE_BACKUP DELETE = config.DELETE_SOURCE CHECK_WAV = config.CHECK_WAV SKIP_EXISTING = config.SKIP_EXISTING def preprocess_track_list(actions, release_count, db_track_count_dict, results, source_relid, table_name, use_local=False): """Perform preliminary checks on tracks.""" do_continue = False dupe_name_list = list() process_list = list() release_completed_track_count = 0 # Tally release release_count += 1 # Generate release-level S3 key from asset list since all assets in any # given list share the same foreign_rel_id release_level_key = results[0].release_level_key # Local files if use_local: track_list_files_only, source_path, dupe_name_list = \ find_local_files_by_rel_id(release_level_key) else: # S3 Files track_list_files_only, source_path = find_s3_files_by_rel_id( release_level_key) source_track_count = len(track_list_files_only) # TODO - Check for vol, track as well as full count. # TODO - Check file order is consecutive, and matches the DB # Get track counts db_track_count = db_track_count_dict[source_relid] # Reports logger.info('Track List: {}', track_list_files_only) logger.info('Source track count: {}', source_track_count) logger.info('DB track count: {}', db_track_count) # Duplicate UPC found if len(dupe_name_list): msg = '[{} - {}] Exception: Duplicate UPC\'s found. Skipping. ' \ '\n{}'.format(table_name, release_count, dupe_name_list) logger.warning(msg) actions, release_completed_track_count = mark_tracks_collision( release_count, table_name, actions, results) do_continue = True # next results - source_relid # Nothing in the original file folder elif source_track_count == 0 and not config.UNDO_ACTION: # Update user msg = '[{} - {}] Exception: Tracks missing on Source. Skipping. - ' \ '{} vs. {}'.format(table_name, release_count, db_track_count, source_track_count) logger.warning(msg) actions, release_completed_track_count = mark_tracks_skipped( release_count, table_name, actions, results) do_continue = True # next results - source_relid # Compare S3 track_list to db_track_list: # - Different number of tracks - MISMATCH elif db_track_count != source_track_count and \ not config.IGNORE_FILE_COUNT_MISMATCH: # Update user msg = '[{} - {}] Exception: S3 track release_count and DB track ' \ 'release_count do not match. \n{} vs. {}' \ .format(table_name, release_count, db_track_count, source_track_count) logger.warning(msg) actions, release_completed_track_count = mark_tracks_mismatched( release_count, table_name, actions, results) do_continue = True else: # Matching number of tracks - GOOD if use_local: # Check for backups on local path # Get backup file key source_file_backup_root_key = os.path.join( config.SOURCE_ROOT_FOLDER_PATH, asset_paths.completed, source_relid) logger.info( 'Checking for backups in local path {}.', source_file_backup_root_key) # Get list of files in backup folder backup_list_files_only = get_files_at_path( source_file_backup_root_key) if not backup_list_files_only: backup_list_files_only = list() else: # Check for backups on S3 # Make backup target object key to check if backups exists source_file_backup_root_key = get_s3_file_key( asset_paths.source_audio, asset_paths.completed, source_relid, source_relid + '_') logger.info( 'Checking for backups on S3, using path {}.', source_file_backup_root_key) # Get list of backup files # Get list of all objects matching key in S3 bucket obj_list = s3_util.filter_backoff( source_file_backup_root_key, s3_bucket_name) backup_list_files_only = [os.path.split(obj.key)[1] for obj in obj_list] # Report # of backups to user. logger.info('{} backups found', len(backup_list_files_only)) # If there are ANY files in the backup folder if len(backup_list_files_only): logger.info('Determining which files are not backed up.') actions, release_completed_track_count, not_skipped, \ skipped_count, update_tracks = mark_tracks_original( release_count, table_name, actions, results, backup_list_files_only, track_list_files_only) # if there are updates, load the new tracks for delivery if not_skipped: logger.info( '{} New tracks queued for processing.', len(update_tracks)) process_list = update_tracks.copy() # if skipped_count == len(results): elif skipped_count == len(results): do_continue = True # else (implicit) # empty process list is returned else: process_list = results # Slice off leaf from source_path release_path = source_path.replace(asset_paths.source_audio, '')[1:] return actions, process_list, release_completed_track_count, \ release_count, release_path, do_continue def find_s3_files_by_rel_id(release_level_key): """ Find a list of files in an S3 folder which begins with the passed str releases_level_key . Args: release_level_key: the UPC-like value by which to filter folders. Returns: (list, str, list, bool) """ source_path = get_s3_file_key(asset_paths.source_audio, release_level_key, release_level_key + '_') # Get list of all objects matching key in S3 bucket obj_list = s3_util.filter_backoff(source_path, s3_bucket_name) # Create list of all tracks on S3 track_list_files_only = [os.path.split(obj.key)[1] for obj in obj_list] return track_list_files_only, source_path def find_local_files_by_rel_id(release_level_key): """ Find a list of files in a local folder which begins with the passed str releases_level_key. Args: release_level_key: the UPC-like value by which to filter folders. Returns: (list, str, list, bool) """ dupe_names = list() # Get potential mathc folders based on release_id local_source_glob = os.path.join(asset_paths.source_audio, release_level_key) source_path = os.getcwd() globs = glob.glob(local_source_glob + '*') # Error Dupe UPC if len(globs) > 1: dupe_names = [name for name in globs] source_path = globs[0] track_list_files_only = [] else: for name in globs: source_path = name source_path = os.path.join( asset_paths.source_audio, source_path) local_file_list = get_files_at_path(source_path) if not local_file_list: local_file_list = list() local_track_list = [file for file in local_file_list if os.path.splitext(file)[1][1:] in asset_types.audio_types] track_list_files_only = [os.path.split(file)[1] for file in local_track_list] return track_list_files_only, source_path, dupe_names def write_stopfile(log_path, tablenum): """Write stopfile file to log dir""" stopfile = os.path.join( log_path, 'asset_ingest_{}.stop'.format(tablenum)) with open(stopfile, 'w') as file: file.write('REMOVE THIS FILE TO RESUME PROCESSING.') def add_stopfiles(tablenum, log_path): """Add stop files to prevent further processing.""" message = 'All files skipped. Locking process. Remove ' \ '`asset_ingest_{}.stop` to resume processing at next ' \ 'interval.'.format(tablenum) logger.info(message) # Write stopfile file to log dir write_stopfile(log_path, tablenum) def process_audio_folder(arguments, log_path, log_date): """Process the file(s) in a folder as audio files. Args: arguments: (object) The arguments object as passed via commandline. log_path: (str) The path where the logfile resides. log_date: (str) The date to be assigned to the log file. Returns: bool: True if files in the folder have been omitted from processing (skipped). Otherwise False. """ # Use table num from commandline, or use table from config (None) if config.SOURCE_AUDIO_VIEW: audio_table_name = config.SOURCE_AUDIO_VIEW elif arguments.tablenum: audio_table_name = config.SNOWFLAKE_AUDIO_TABLE_NUMBERED.format( arguments.tablenum) elif arguments.tablename: audio_table_name = arguments.tablename.strip('\'\"') else: raise EnvironmentError('\'SOURCE_AUDIO_VIEW\' must be set to ' 'process audio.') # Start timer request_start_time = default_timer() try: # TODO: Splice in Assets V2 here? audio_count, skipped_all, log_file_name = \ transfer_audio_assets(arguments, audio_table_name, log_path, log_date) except Exception as e: default_timer() # End timer msg = 'Audio transfer failed: {}.'.format(str(e)) logger.error(msg) return False # End timer and calculate time spent request_end_time = default_timer() request_time = request_end_time - request_start_time msg = '[{} - {}] Audio transfer completed: {} secs.'.format( audio_table_name, audio_count, request_time) # Update user logger.info(msg) if skipped_all: if log_file_name: os.remove(log_file_name) return skipped_all def process_cover_folder(arguments, log_path, log_date): """Process the file(s) in a folder as covers. Args: arguments: (object) The arguments object as passed via commandline. log_path: (str) The path where the logfile resides. log_date: (str) The date to be assigned to the log file. Returns: bool: True if files in the folder have been omitted from processing (skipped). Otherwise False. """ # Use table num or name from commandline, or use table from environment if config.SOURCE_COVER_VIEW: cover_table_name = config.SOURCE_COVER_VIEW elif arguments.tablenum: cover_table_name = config.SNOWFLAKE_COVER_TABLE_NUMBERED.format( arguments.tablenum) elif arguments.tablename: cover_table_name = arguments.tablename.strip('\'\"') else: raise EnvironmentError('\'SOURCE_COVER_VIEW\' must be set to ' 'process audio.') # Start timer request_start_time = default_timer() cover_count, skipped_all, log_file_name = \ transfer_covers(arguments, cover_table_name, log_path, log_date) # End timer, calculate duration request_end_time = default_timer() request_time = request_end_time - request_start_time msg = '[{} - {}] Cover transfer completed: {} secs.'.format( cover_table_name, cover_count, request_time) logger.info(msg) if skipped_all: if log_file_name: os.remove(log_file_name) return skipped_all # TODO - Fix docstring def transfer_audio_assets(args, table_name, log_file_path, date): """Read a DB table, pull and transfer the audio files listed therein. Args: args: (object) The arguments object as passed via commandline. table_name: (str) The name of the table containing the file_names to be processed. log_file_path: (str) The path where the logfile resides. date: (str) The date to be assigned to the log file. Returns: (release_count, skipped_all_files, audio_log_f_name): (tuple) The name of the """ skipped_all_files = True # Update user msg = 'Fetching audio asset list from {}.'.format(table_name) logger.info(msg) # TODO: Copy assets from source bucket to raw bucket for ingestion. # Get all DB rows, and create tree structures that have the data we need audio_results, db_track_count_list = transform_audio( get_audio_assets(table_name), config.FILENAME_TRANSFORMER) # Init release_count release_count = 0 audio_log_f_name = logic_utils.start_log(table_name, date, log_file_path, 'audio') # TODO - ASSETS V2 Variable # audio_result_list = [(k, v) for k, v in audio_results.items()] # # for i in range(0, len(audio_result_list), config.MAX_REQUEST_PER_LOOP): # # # number of items per loop # batch_countdown = config.MAX_REQUEST_PER_LOOP long_orch_upc = '' # Loop per release for source_relid, results in audio_results.items(): # Reset loop vars has_audio = False delete_list = [] actions = {} # TERMINAL CONDITIONS terminate_process, msg = logic_utils.check_terminal_conditions( args.tablenum) if terminate_process: logger.info(msg) sys.exit(msg) # Pre-process track list: find dupes and other issues actions, process_list, release_track_count, release_count, \ release_path, skip_release = preprocess_track_list( actions, release_count, db_track_count_list, results, source_relid, table_name, config.USE_LOCAL_FOR_SOURCE) # If Pre-processing determines the release should be skipped: if skip_release: # Write to log update_log(audio_log_f_name, actions, table_name, source_relid, None, 'wav', release_track_count) continue # next results - source_relid # TODO: NEW ASSETS V2 CHANGES # asset_requests = list() # Loop per track for result in process_list: # TODO: NEW ASSETS V2 CHANGES # batch_countdown -= 1 # # if batch_countdown == 0: # batch_countdown = config.MAX_REQUEST_PER_LOOP # # # Send upload-token-bulk request # filenames = deque() # # # Copy files to S3 in bulk # # # Loop and send requests # for req in asset_requests: # filenames.popleft() # # Send asset V1 request # pass # # grass_id = result.vendor_id # orchard_id = get_access_token_alw(label_id=result.vendor_id) # content_type = result.asset_type # # # Info to get from DB # # tuid # # # # # Slice off # # vendor_id (as grass_id) # # upc / product_id # # generated file name # # input_file_name # # tuid # # staging s3 bucket # # target s3 bucket # # s3_filename_key release_track_count += 1 # Get pieces needed for path source_filename = result.source_audio_file source_flac_filename = result.source_audio_file_flac # CONTINUE - Still splitting the logic between S3 and local # CONTINUE - Add branch for local files from here down. if config.USE_LOCAL_FOR_SOURCE: source_file_key = os.path.join( asset_paths.source_audio, release_path, source_filename) source_file_flac_key = os.path.join( asset_paths.source_audio, release_path, source_flac_filename) else: source_file_key = get_s3_file_key( asset_paths.source_audio, release_path, source_filename) source_file_flac_key = get_s3_file_key( asset_paths.source_audio, release_path, source_flac_filename) # Get orchard file names # orch_file_name = result.orchard_audio_file # orch_flac_file_name = result.orchard_audio_file_flac long_orch_file_name = result.long_orchard_audio_file long_orch_flac_file_name = result.long_orchard_audio_file_flac # orch_upc = result.orch_upc long_orch_upc = result.long_orch_upc # Make backup target object key to check if it exists source_file_backup_key = get_s3_file_key( asset_paths.source_audio, asset_paths.completed, release_path, source_filename) source_flac_file_backup_key = get_s3_file_key( asset_paths.source_audio, asset_paths.completed, release_path, source_flac_filename) target_path, target_path_folder, \ tmp_target_path, asset_target_path, flac_target_path = \ create_target_paths( asset_paths, long_orch_file_name, long_orch_flac_file_name, long_orch_upc, COPY_LOCAL) # Skip existing files if SKIP_EXISTING and os.path.exists(target_path): msg = '{} found. Skipping....'.format(long_orch_file_name) logger.info(msg) continue file_exists = True if config.USE_LOCAL_FOR_SOURCE: if not os.path.isfile(source_file_key): if not os.path.isfile(source_file_flac_key): file_exists = False else: source_file_key = source_file_flac_key source_file_backup_key = \ source_flac_file_backup_key target_path = flac_target_path else: # Check key exists if not s3_util.s3_key_exists(source_file_key, s3_bucket_name): # WAV if not s3_util.s3_key_exists(source_file_flac_key, # FLAC s3_bucket_name): file_exists = False else: source_file_key = source_file_flac_key source_file_backup_key = \ source_flac_file_backup_key target_path = flac_target_path if not file_exists: # Update user msg = '[{} - {}] Stat {} failed. ' \ 'Key does not exist.'.format( table_name, release_count, source_file_key) logger.error(msg) # Update action for log actions[source_filename] = 'missing' continue # next result else: # KEY EXISTS! # Update action actions[source_filename] = 'original found' # Create if not exist target path folder os.makedirs(target_path_folder, exist_ok=True) # User update - downloading begin msg = '[{} - {}] Downloading audio file `{}` to `{}`...' \ .format(table_name, release_count, source_file_key, target_path) logger.info(msg) if config.USE_LOCAL_FOR_SOURCE: try: if config.MOVE_LOCAL_FILES: if not on_same_filesystem( source_file_key, os.path.split( target_path)[0]): logger.warning('Attempting to MOVE files ' 'located on different physical ' 'drives. This can lead ' 'to permanent data loss. ' 'Please consider ' 'setting MOVE_LOCAL_FILES=False' ' in your environment to ' 'enable COPY instead. This ' 'will be safer.') shutil.move(source_file_key, target_path) else: copy_file(source_file_key, target_path) action = 'downloaded' track_error = False err = None except Exception as e: action = 'failed' track_error = True err = e else: # TODO: RUN ASSET_V2 Here action, track_error, err = download_file( source_file_key, target_path) # Update action for log actions[source_filename] = action if track_error: # Update user msg = '[{} - {}] Exception: Download {} ' \ 'failed. \n{}'.format( table_name, release_count, source_file_key, str(err)) logger.error(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage( err, stack=True) continue # There is at least 1 audio file in this release has_audio = True skipped_all_files = False # TODO: FIX THIS TO WORK WITH NEW PATHS # Run the wav checker if CHECK_WAV: try: logger.info('Checking WAV file {} matches expected ' 'format.', os.path.split(target_path)[1]) wav_logger = wav_utils.setup_wav_check_logger( table_name, date, log_file_path) wav_utils.check_wav( target_path, wav_logger, overwrite=True) except FileNotFoundError as e: logger.info(str(e)) # TODO: FIX THIS TO WORK WITH NEW PATHS # Move from local tmp path to final target if COPY_LOCAL: # Move the file to the ripper msg = '\n[{} - {}] Moving `{}` to target `{}`...' \ .format(table_name, release_count, tmp_target_path, asset_target_path) logger.info(msg) os.makedirs(target_path, exist_ok=True) # This copy_file() uses a higher buffer for speed copy_file(src=tmp_target_path, dst=asset_target_path, preserve_file_date=False) # TODO: FIX THIS TO WORK WITH NEW PATHS # Move backup file to original directory if UNDO: # Update user msg = '[{} - {}] Restoring file from \'{}\' ' \ 'folder to {} on S3'.format( table_name, release_count, asset_paths.completed, source_file_key) logger.info(msg) action, track_error, err, delete_list = \ restore_backup( delete_list, source_file_backup_key, source_file_key) # Update action for log actions[source_filename] = action if track_error: # Update user msg = '[{} - {}] Exception: Restore {} ' \ 'backup failed. \n{}'.format( table_name, release_count, source_file_backup_key, str(err)) logger.error(msg) # ERROR to sentry if sentry.sentry_client: sentry.sentry_client.captureMessage( err, stack=True) continue # TODO: FIX THIS TO WORK WITH NEW PATHS elif BACKUP: # Move original to backup directory # Make orig source object key to backup s3_source_file_source_key = get_s3_file_key( S3_BUCKET, source_file_key) msg = '[{} - {}] Moving file {} to \'{}\'' \ ' folder on S3'.format(table_name, release_count, s3_source_file_source_key, asset_paths.completed) logger.info(msg) action, track_error, err, delete_list = \ backup_originals(delete_list, source_file_backup_key, source_file_key) # Update action for log actions[source_filename] = action if track_error: # Update user msg = '[{} - {}] Exception: Backup {} ' \ 'failed. \n{}'.format(table_name, release_count, s3_source_file_source_key, str(err)) logger.error(msg) # ERROR to sentry if sentry.sentry_client: sentry.sentry_client.captureMessage( err, stack=True) continue # If every file wasn't skipped if has_audio: key_array = logic_utils.create_donefile_keys( source_relid, long_orch_upc, config.USE_LOCAL_FOR_SOURCE) donefile_backup_key = key_array[0] donefile_key = key_array[1] # donefile_source_key = key_array[2] orch_donefile_key = key_array[2] release_folder_key = key_array[3] # Update user msg = '[{} - {}] Writing donefile: {}.'.format( table_name, release_count, orch_donefile_key) logger.info(msg) # Write .done file to asset target dir with open(orch_donefile_key, 'wb') as f: f.write(b'') # TODO: FIX THIS TO WORK WITH NEW PATHS # Restore backup pf.donefile to original position if UNDO and not config.USE_LOCAL_FOR_SOURCE: # Update user msg = '[{} - {}] Moving DONE file from \'{}\' ' \ 'folder to {} on S3'.format(table_name, release_count, asset_paths.completed, donefile_key) logger.info(msg) restore_donefile(release_count, table_name, donefile_backup_key, donefile_key) # TODO: FIX THIS TO WORK WITH NEW PATHS elif BACKUP and not config.USE_LOCAL_FOR_SOURCE: # Update user msg = '[{} - {}] Moving file {} to \'{}\' folder ' \ 'on S3'.format(table_name, release_count, donefile_key, asset_paths.completed) logger.info(msg) backup_donefile(release_count, table_name, donefile_backup_key, donefile_key) # TODO: FIX THIS TO WORK WITH NEW PATHS # Delete ALL files queued for deletion if DELETE: if delete_list: # Check for delete list # Update user msg = '[{} - {}] Deleting all files matching: {}' \ .format( table_name, release_count, release_folder_key) logger.info(msg) if not config.USE_LOCAL_FOR_SOURCE: delete_donefile(release_count, donefile_key, table_name) # Iterate through deletion list actions = delete_marked_files( actions, release_count, delete_list, table_name) # Check if folder is empty obj_list = s3_util.filter_backoff( release_folder_key, s3_bucket_name) # Create list of all tracks on S3 s3_track_list = [obj.key for obj in obj_list if obj.key[-1:] != '/'] # Delete folder if not s3_track_list: try: # Delete the marked path s3_util.delete_backoff(release_folder_key, s3_bucket_name) except Exception as e: # Update user msg = '[{} - {}] Exception: {} ' \ 'delete failed.\n{}'.format( table_name, release_count, release_folder_key, str(e)) logger.error(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage( e, stack=True) # Parse and write to log: update_log(audio_log_f_name, actions, table_name, source_relid, long_orch_upc, 'wav', release_track_count) return release_count, skipped_all_files, audio_log_f_name # TODO - Fix docstring def transfer_covers(args, table_name, log_file_path, date): """Read a DB table, pull and transfer the cover files listed therein. Args: args: (object) The arguments object as passed via commandline. table_name: (str) The name of the table containing the file_names to be processed. log_file_path: (str) The path where the logfile resides. date: (str) The date to be assigned to the log file. Returns: (): (tuple) """ # Update user msg = 'Fetching cover asset list from {}.'.format(table_name) logger.info(msg) skipped_all_files = True # Init default val cover_filename = None # Get all DB rows, and create tree structures that have the data we need cover_results = transform_covers(get_cover_assets(table_name), config.FILENAME_TRANSFORMER) # Init loop vars release_count = 0 # Generate log file name and header cover_log_file_name = logic_utils.start_log(table_name, date, log_file_path, 'cover') # Loop over covers (1 per release) for result in cover_results: # TERMINAL CONDITIONS terminate_process, msg = logic_utils.check_terminal_conditions( args.tablenum) if terminate_process: logger.info(msg) sys.exit(msg) # Reset loop vars has_cover = False already_moved = False source_path = None source_file_key = None delete_list = [] temp_list = [] actions = {} release_track_count = 1 # Always one cover release_count += 1 # Running tally # Get pieces needed for paths source_relid = result.source_relid long_orch_file_name = result.long_orchard_cover_file long_orch_upc = result.long_orch_upc # Update action for log actions[source_relid] = 'started' # Build a source key based on if it's a load or restore # TODO: Make sure that UNDO ACTION works with local source if config.UNDO_ACTION: # Backup is source source_root_file_key = get_s3_file_key( asset_paths.source_cover, asset_paths.completed, source_relid, source_relid) else: # Orig is source if config.USE_LOCAL_FOR_SOURCE: source_root_file_key = os.path.join( asset_paths.source_cover, source_relid) else: source_root_file_key = get_s3_file_key( asset_paths.source_cover, source_relid, source_relid) # Create if not exists tmp folder temp_file_path = create_path_with_todays_date( config.TEMP_FILE_PATH) # Build tmp path for downloads tmp_target_path = os.path.join( temp_file_path, asset_paths.ripper_tif_folder) # Make dir if needed os.makedirs(tmp_target_path, exist_ok=True) if config.USE_LOCAL_FOR_SOURCE: source_file_completed_cover_key = os.path.join( asset_paths.source_cover, asset_paths.completed, source_relid) else: # Make object keys source_file_completed_cover_key = get_s3_file_key( asset_paths.source_cover, asset_paths.completed, source_relid) # Loading new covers if not UNDO: # Update user msg = '[{} - {}] Checking if {} was processed previously.'.format( table_name, release_count, source_root_file_key) logger.info(msg) action, do_continue, backup_obj_key = mark_cover_original( source_root_file_key, source_file_completed_cover_key, config.USE_LOCAL_FOR_SOURCE) actions[source_relid] = action if do_continue: # Update user msg = '[{} - {}] Skipping: {}'.format( table_name, release_count, backup_obj_key) logger.info(msg) # Write to log update_log(cover_log_file_name, actions, table_name, source_relid, long_orch_upc, 'image', release_track_count) continue if config.USE_LOCAL_FOR_SOURCE: try: cover_keys, source_path = find_files_at_path_glob( source_root_file_key) except ValueError: cover_keys = list() else: cover_keys = s3_util.filter_backoff( source_root_file_key + '.', s3_bucket_name) cover_keys = [ obj.key for obj in cover_keys if os.path.splitext(obj.key)[1][1:] in asset_types.cover_types ] # Find all keys that match the relid (Some are Tifs, some jpgs) for source_file_key in cover_keys: # Cover exists has_cover = True # Some work performed skipped_all_files = False # Split into dir and file path, cover_filename = os.path.split(source_file_key) # Generate temp target path tmp_target_path = os.path.join( tmp_target_path, cover_filename) # Prepare target path asset_target_path = os.path.join( asset_paths.target, asset_paths.ripper_tif_folder) # Create if needed os.makedirs(asset_target_path, exist_ok=True) # Prepare target filename asset_target_file = os.path.join( asset_target_path, long_orch_file_name) # Skip existing files if SKIP_EXISTING and os.path.exists(asset_target_file): msg = '{} found. Skipping....'.format(long_orch_file_name) logger.info(msg) already_moved = True continue # If not restoring. We are loading new covers if not config.UNDO_ACTION: # Update user msg = '[{} - {}] Downloading cover file `{}` to `{}`...' \ .format(table_name, release_count, source_file_key, tmp_target_path) logger.info(msg) if config.USE_LOCAL_FOR_SOURCE: if not source_path: raise ValueError check_key = os.path.join(source_path, source_file_key) else: check_key = source_file_key action, do_continue, err = check_cover_key_exists( check_key, config.USE_LOCAL_FOR_SOURCE) actions[source_relid] = action if do_continue: # Update user msg = '[{} - {}] Exception: {} does not exist. \n' \ '{}'.format(table_name, release_count, source_file_key, str(err)) logger.info(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage( err, stack=True) # Write to log update_log(cover_log_file_name, actions, table_name, source_relid, long_orch_upc, 'image', release_track_count) continue if config.USE_LOCAL_FOR_SOURCE: download_key = os.path.join(source_path, source_file_key) else: download_key = source_file_key action, do_continue, err, temp_list = download_cover( download_key, temp_list, tmp_target_path, config.USE_LOCAL_FOR_SOURCE) actions[source_relid] = action if do_continue: # Update user msg = '[{} - {}] Exception: Download of {} failed. ' \ '\n{}'.format( table_name, release_count, source_file_key, str(err)) logger.error(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage( err, stack=True) # Write to log update_log(cover_log_file_name, actions, table_name, source_relid, long_orch_upc, 'image', release_track_count) continue # FILE DOWNLOADED! # Update user - msg = '[{} - {}] Opening file: {}'.format( table_name, release_count, tmp_target_path) logger.info(msg) image, do_continue, action, err = logic_utils.open_image( tmp_target_path) # Update action for log actions[source_relid] = action if do_continue: # Update user msg = '[{} - {}] Exception: Could not open image: ' \ '{}.\n{}'.format( table_name, release_count, tmp_target_path, str(err)) logger.info(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage( err, stack=True) # Write to log update_log(cover_log_file_name, actions, table_name, source_relid, long_orch_upc, 'image', release_track_count) continue # next result # Update user msg = '[{} - {}] Converting and moving cover file `{}` ' \ 'to `{}`...'.format( table_name, release_count, tmp_target_path, asset_target_file) logger.info(msg) action, do_continue, err = logic_utils.save_image( asset_target_file, image) actions[source_relid] = action if do_continue: # Update user msg = '[{} - {}] Exception: Could not convert image ' \ '{}\n{}'.format( table_name, release_count, tmp_target_path, str(err)) logger.info(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage( err, stack=True) # Write to log update_log(cover_log_file_name, actions, table_name, source_relid, long_orch_upc, 'image', release_track_count) continue # If no covers processed if not has_cover: # Update user msg = '[{} - {}] Skipping missing file {}.'.format( table_name, release_count, source_root_file_key) logger.info(msg) # Update action for log actions[source_relid] = 'missing' elif already_moved: # Update user msg = '[{} - {}] Skipping already moved file {}.'.format( table_name, release_count, source_root_file_key) logger.info(msg) # Update action for log actions[source_relid] = 'missing' else: # A cover was processed if config.USE_LOCAL_FOR_SOURCE: if not cover_filename: raise ValueError('Cover filename is not assigned.') # Make object keys for backup source_file_backup_key = os.path.join( source_path, asset_paths.completed, cover_filename) source_file_source_key = os.path.join( source_path, cover_filename) else: # Make object keys for backup source_file_backup_key = get_s3_file_key( asset_paths.source_cover, asset_paths.completed, cover_filename) source_file_source_key = get_s3_file_key( S3_BUCKET, source_file_key) # Move backup file to original directory if UNDO: # Make cover file key source_file_key = get_s3_file_key( asset_paths.source_cover, cover_filename) # Update user msg = '[{} - {}] Restoring file from {} folder to {} on ' 'S3'.format(table_name, release_count, source_file_backup_key, source_file_key) logger.info(msg) action, do_continue, msg, err = \ restore_cover_backup( source_file_backup_key, source_file_key) actions[source_relid] = action if do_continue: # Update user msg = msg.format( table_name, release_count, source_file_backup_key, source_file_key) logger.info(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage(err, stack=True) # Write to log update_log(cover_log_file_name, actions, table_name, source_relid, long_orch_upc, 'image', release_track_count) continue else: # Backup and delete if BACKUP: # Update user msg = '[{} - {}] Moving file {} to \'Completed\' folder ' \ 'on S3'.format( table_name, release_count, source_file_source_key) logger.info(msg) action, do_continue, err, delete_list = backup_keys( delete_list, source_file_backup_key, source_file_key) actions[source_relid] = action if do_continue: # Update user msg = '[{} - {}] S3 copy failed: Backing up file {} '\ 'to \'Completed\' folder'.format( table_name, release_count, source_file_source_key) logger.info(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage( err, stack=True) # Write to log update_log(cover_log_file_name, actions, table_name, source_relid, long_orch_upc, 'image', release_track_count) continue # Delete original cover if DELETE: # Update user msg = '[{} - {}] Deleting {}.'.format( table_name, release_count, source_file_key) logger.info(msg) actions = delete_covers( actions, release_count, delete_list, source_file_source_key, table_name, temp_list) # actions[source_relid] = action # Write to log update_log(cover_log_file_name, actions, table_name, source_relid, long_orch_upc, 'image', release_track_count) return release_count, skipped_all_files, cover_log_file_name def mark_tracks_original(release_count, table_name, actions, results, backup_list, originals_list): """Mark tracks as original.""" # Init counts release_track_count = 0 skipped_count = 0 not_skipped = 0 update_tracks = [] # Check for updates to original files for track in results: # Tally track release_track_count += 1 # Check if track is backed up has_backup = track.source_audio_file in backup_list # TODO - Handle Flac processing # has_backup = has_backup \ # or track.source_audio_file_flac in backup_list has_original = track.source_audio_file in originals_list # TODO - Handle Flac processing # has_original = has_original or \ # track.source_audio_file_flac in originals_list if has_backup: # Check if NOT NEW original if not has_original: # Update user msg = '[{} - {}] Skipping {}. No original file. ' \ 'Backup only.'.format(table_name, release_count, track.source_audio_file) logger.info(msg) # Increment skipped_count skipped_count += 1 # Update action for log actions[track.source_audio_file] = 'skipped' # File IS NEW original else: update_tracks.append(track) # Increment not_skipped release_count not_skipped += 1 # Update action for log actions[track.source_audio_file] = 'new original' elif has_original: update_tracks.append(track) not_skipped += 1 actions[track.source_audio_file] = 'original' else: skipped_count += 1 actions[track.source_audio_file] = 'file missing' return actions, release_track_count, not_skipped, \ skipped_count, update_tracks def mark_tracks_mismatched(release_count, table_name, actions, results): """Mark tracks as mismatched.""" release_track_count = 0 for track in results: # Tally asset release_track_count += 1 # release_count += 1 # Update action for log actions[track.source_audio_file] = 'file count mismatch' msg = '[{} - {}] Skipping {}: File count mismatch.'.format( table_name, release_count, track.source_audio_file) logger.warning(msg) return actions, release_track_count def mark_tracks_skipped(release_count, table_name, actions, results): """Mark all tracks in results as skipped.""" release_track_count = 0 for track in results: # Tally asset release_track_count += 1 # Update action for log actions[track.source_audio_file] = 'skipped' msg = '[{} - {}] Skipping {}: File missing.'.format( table_name, release_count, track.source_audio_file) logger.warning(msg) return actions, release_track_count def mark_tracks_collision(release_count, table_name, actions, results): """Mark all tracks in results as collisions.""" release_track_count = 0 for track in results: # Tally asset release_track_count += 1 # Update action for log actions[track.source_audio_file] = 'duplicate' msg = '[{} - {}] Skipping {}: Duplicate UPC.'.format( table_name, release_count, track.source_audio_file) logger.warning(msg) return actions, release_track_count def restore_backup(delete_list, backup_key, target_key): """Restore a backup file, and mark it for deletion.""" # Init return vals has_error = False err = None # Copy backup back to original position try: s3_util.copy_from_backoff( s3_bucket_name, backup_key, target_key) except Exception as e: # Restore fails # Update action for log action = 'restore failed' err = e has_error = True # continue # next result else: # Mark backup for deletion delete_list.append(backup_key) # Update action for log action = 'restored' return action, has_error, err, delete_list.copy() def download_file(s3_source_file_key, target_path): """Download a file from S3.""" # Init return val has_error = False err = None # Download file try: s3_util.download_fileobj_backoff(s3_source_file_key, target_path, s3_bucket_name) except Exception as e: # Download fails # Update action for log action = 'failed' err = e has_error = True # next result else: # DOWNLOADED! # Update action for log action = 'downloaded' return action, has_error, err def backup_donefile(release_count, table_name, backup_key, source_key): """Backup a donefile.""" bucket_name = s3_bucket_name try: # Make backup of donefile s3_util.copy_from_backoff(bucket_name, source_key, backup_key) except Exception as e: # Donefile backup fails # Update user msg = '[{} - {}] Exception: Donefile backup failed. ' \ '\n{}'.format(table_name, release_count, str(e)) logger.error(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage(e, stack=True) def backup_keys(delete_list, backup_key, source_key): """Backup keys and collect source keys for deletion.""" has_error = False err = None try: # Copy to backup s3_util.copy_from_backoff( s3_bucket_name, source_key, backup_key) except Exception as e: # Copy fails msg = 'Exception: Backup key {} failed. \n{}'.format( source_key, str(e)) logger.error(msg) # Update user action = 'backup failed' err = e has_error = True # next result else: # Update user action = 'backed up' # Mark S3 key for deletion delete_list.append(source_key) return action, has_error, err, delete_list.copy() def backup_originals(delete_list, backup_key, file_source_key): """Backup an original file to the backup folder.""" has_error = False err = None # Copy original to backup folder try: s3_util.copy_from_backoff( s3_bucket_name, file_source_key, backup_key) except Exception as e: # Backup fails has_error = True # next result err = e action = 'backup failed' else: # Mark for deletion delete_list.append(file_source_key) # Update action for log action = 'backed up' return action, has_error, err, delete_list.copy() def check_cover_key_exists(key, use_local=False): """Check if a cover key exists.""" has_error = False err = None action = 'exists' try: # Download file to temp folder if use_local: if not os.path.exists(key): raise FileNotFoundError('Key {} does not exist', key) else: # Check existence of key s3_util.head_object_backoff(key, s3_bucket_name) except Exception as e: # Download fails # Update action for log action = 'missing' err = e has_error = True return action, has_error, err def restore_donefile(table_name, release_count, backup_key, key): """Restore a donefile from the backup directory.""" try: # Restore donefile s3_util.copy_from_backoff(s3_bucket_name, backup_key, key) except Exception as e: # Donefile restore fails # Update user msg = '[{} - {}] Exception: Donefile restore failed. ' \ '\n{}'.format(table_name, release_count, str(e)) logger.error(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage(e, stack=True) def download_cover(key, temp_to_delete_list, tmp_target_path, use_local=False): """Download a cover based on passed key.""" has_error = False err = None try: if use_local: copy_file(key, tmp_target_path) else: s3_util.download_fileobj_backoff(key, tmp_target_path, s3_bucket_name) except Exception as e: # Download fails # Update action for log action = 'failed' err = e # continue # next result has_error = True else: # Update action for log action = 'downloaded' # Mark temp file for deletion temp_to_delete_list.append(tmp_target_path) return action, has_error, err, temp_to_delete_list.copy() def mark_cover_original(key, backup_key, use_local=False): """Mark cover as original.""" has_error = False new_orig_found = False found_backup_key = None backup_obj = None try: # Check if file is already processed - i.e. in backup if use_local: try: local_file_list, source_path = find_files_at_path_glob(key) except ValueError: raise FlowControlDupeFound() backup_path = os.path.join(source_path, asset_paths.completed) backup_file_list = get_files_at_path( backup_path, ext=asset_types.cover_types) if backup_file_list: for backup_obj in backup_file_list: for _ in local_file_list: new_orig_found = True if not new_orig_found: raise FlowControlBackupFound() else: for backup_obj in s3_util.filter_backoff( backup_key + '.', s3_bucket_name): # If any in backup - check if there is a new original file too for _ in s3_util.filter_backoff( key + '.', s3_bucket_name): # TODO: Add checks for size and date - See S3_cleaner.py # New original - Don't skip new_orig_found = True if not new_orig_found: raise FlowControlBackupFound() action = 'processing' # Image has already been processed except FlowControlBackupFound: # Update action for log action = 'skipped' if use_local: found_backup_key = backup_obj else: found_backup_key = backup_obj.key has_error = True # next result except FlowControlDupeFound: # Update action for log action = 'duplicate' if use_local: found_backup_key = backup_obj else: found_backup_key = backup_obj.key has_error = True # next result return action, has_error, found_backup_key def find_files_at_path_glob(key): """Get all files at a wildcard path.""" # TODO - Figure out if the key element works for covers as well as audio. local_source_glob = os.path.join(asset_paths.source_cover, key) source_path = os.getcwd() globs = glob.glob(local_source_glob + '*') if len(globs) > 1: raise ValueError( 'Multiple folders found for Foreign Release Id: {}', key) for name in globs: source_path = name local_file_path = os.path.join(source_path) local_file_list = get_files_at_path( local_file_path, ext=asset_types.cover_types) return local_file_list, source_path def delete_marked_files(actions, release_count, delete_list, table_name): """Delete the passed marked files.""" for key in delete_list: # Update user msg = '[{} - {}] Deleting: {}'.format(table_name, release_count, key) logger.info(msg) try: # Delete the marked file s3_util.delete_backoff(key, s3_bucket_name) except Exception as e: # Update user msg = '[{} - {}] Exception: {} delete failed.\n{}'.format( table_name, release_count, key, str(e)) logger.error(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage(e, stack=True) else: path, file = os.path.split(key) # Update action for log - USE key HERE due to loop actions[file] = 'completed' return actions def delete_donefile(release_count, key, table_name): """Delete donefile.""" try: # Delete donefile s3_util.delete_backoff(key, s3_bucket_name) except Exception as e: # Update user msg = '[{} - {}] Exception: Donefile delete ' \ 'failed. \n{}'.format(table_name, release_count, str(e)) logger.error(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage(e, stack=True) def delete_covers(actions, release_count, delete_list, key, table_name, temp_list): """Delete covers.""" # Loop through and delete marked S3 files for f_del in delete_list: action = 'begin delete' source_relid = os.path.splitext(f_del)[0] try: # Delete original file s3_util.delete_backoff(f_del, s3_bucket_name) except Exception as e: # Update user msg = '[{} - {}] S3 delete failed: {} '.format( table_name, release_count, key) logger.info(msg) # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage(e, stack=True) actions[source_relid] = action continue # next deletion else: # Update user action = 'completed' actions[source_relid] = action # Loop through and delete temp files for f_del_temp in temp_list: try: os.remove(f_del_temp) except Exception as e: # No user updates; its temp # ERROR to Sentry if sentry.sentry_client: sentry.sentry_client.captureMessage(e, stack=True) continue # next delete return actions def restore_cover_backup(backup_key, key): """Restore cover backup.""" has_error = False err = None msg = '' try: # Restore backup via copy s3_util.copy_from_backoff(s3_bucket_name, key, backup_key) except Exception as e: # Copy fails # Update action for log # Update user msg = '[{} - {}] S3 copy failed: Restoring {} folder to {} on S3' action = 'failed' err = e has_error = True else: # Update action for log # action = 'copied' try: # Delete backup s3_util.delete_backoff(backup_key, s3_bucket_name) except Exception as e: # Delete fails # Update user msg = '[{} - {}] S3 delete failed: {}.' # Update action for log action = 'failed' err = e has_error = True else: # Update action for log action = 'restored' return action, has_error, msg, err def update_log(log_file, action_list, t_name, source_relid, orchard_upc, asset_type, release_track_count_num): """Insert a formatted line into the log.""" path, file = os.path.split(log_file) os.makedirs(path, exist_ok=True) # Parse action log l_complete, l_skipped, l_restored, l_failed, l_missing, l_other, \ l_mismatch = parse_actions(action_list) # Assemble log line line = '{}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}, {}\n'.format( t_name, source_relid, orchard_upc, asset_type, release_track_count_num, l_complete, l_skipped, l_failed, l_missing, l_mismatch, l_other, l_restored, datetime.today().strftime('%Y%m%d%H%M%S') ) # Write log with open(log_file, 'a') as f: f.write(line) def parse_actions(actions): """Parse action log; determine how many tracks were copied, failed, etc.""" p_complete = 0 p_skipped = 0 p_restored = 0 p_failed = 0 p_missing = 0 p_other = 0 p_mismatch = 0 for key, action in actions.items(): if action == 'completed': p_complete += 1 elif action == 'skipped': p_skipped += 1 elif action == 'restored': p_restored += 1 elif action == 'failed': p_failed += 1 elif action == 'missing': p_missing += 1 elif action == 'file_count_mismatch': p_mismatch += 1 else: p_other += 1 return p_complete, p_skipped, p_restored, p_failed, p_missing, \ p_other, p_mismatch def get_audio_assets(table_name=None): """Get all audio assets in ripper audio view.""" if not table_name: raise EnvironmentError('Table name must be set to process audio.') sql = SELECT_ELIGIBLE_AUDIO_FOR_RIPPER params = { 'db': SNOWFLAKE_DATABASE, 'schema': SNOWFLAKE_SCHEMA, 'table_name': table_name, 'foreign_key_id': config.FOREIGN_KEY_ID } result = get_snowflake_results(sql=sql, params=params) return result def get_cover_assets(table_name=None): """Get all audio assets in ripper audio view.""" if not table_name: table_name = config.SOURCE_COVER_VIEW sql = SELECT_ELIGIBLE_COVERS_FOR_RIPPER params = { 'db': SNOWFLAKE_DATABASE, 'schema': SNOWFLAKE_SCHEMA, 'table_name': table_name, 'foreign_key_id': config.FOREIGN_KEY_ID } result = get_snowflake_results(sql=sql, params=params) return result def get_ingestion_table_row_by_upc(upc): table_name = config.RITV_TABLE sql = 'SELECT * FROM {db}.{schema}.{table_name} WHERE UPC = {upc}' params = { 'db': SNOWFLAKE_DATABASE, 'schema': SNOWFLAKE_SCHEMA, 'table_name': table_name, 'upc': upc } result = get_snowflake_results(sql=sql, params=params) return result def get_ingestion_table_row_by_upc_list(upc_list): table_name = config.RITV_TABLE sql = 'SELECT * FROM {}.{}.{} WHERE UPC in (:upc_list)' sql = sql.format(SNOWFLAKE_DATABASE, SNOWFLAKE_SCHEMA, table_name) params = {'upc_list': upc_list} result = get_snowflake_bind_results(sql=sql, params=params) return result