"""Ingest Phonofile Bulk Upload files programmatically.""" from collections import OrderedDict, Counter from time import sleep from sys import exit as sysexit from concurrency import multiprocess_requests as mp from constants import derek from constants import s3 as s3_const from constants.data_headers import FORCED_TABLE from logic import graphql_logic import logic.bulk_upload_feeder_logic as bulk_logic from integration_scripts import logger as log from integration_scripts import orch_utils from integration_scripts.common_config import S3_BUCKET from util import db_utils from util.exceptions import NewArtistError, StopFileError import config from model.bulk_upload_tracking import \ create_bulk_upload_tracking_table, \ create_bulk_upload_tracking_row, \ update_bulk_upload_row # Begin main control if __name__ == '__main__': # Init vars full_request_dict = dict() # requests full_response_dict = dict() # responses standard_response_tuple = tuple() master_release_list = set() # release in Project_____Product format created_table_dict = dict() embargo_artist_list = list() all_artist_names_list = set() # Input file related table_name = None code = None new_artist_count = 0 # Session totals session_total_rows = 0 session_total_release_count = 0 # Multiprocessing jobs list jobs = list() # Number of lines processed: lines_processed = 0 session_id = bulk_logic.get_session_id() # Report to user log.info(derek.get_derek_small()) log.info('DEREK IS WATCHING....') log.info('Env is: {}'.format(config.ENVIRONMENT.upper())) log.info('Number of threads to run: {}'.format(config.NUM_THREADS)) log.info('Session ID: {}'.format(session_id)) # TODO: Convert CREATE to model # create log tables (if necessary). db_utils.create_rds_release_log_table() db_utils.create_rds_track_log_table() log.info('Creating tracking table.') create_bulk_upload_tracking_table() if config.S3_INGEST: try: # Check for stopfile, and quit if found. bulk_logic.check_for_stopfile() # Load new files in bulk_logic.load_s3_bucket(copy_file_count=1) except FileNotFoundError: sysexit() except StopFileError: sysexit() msg = 'Creating tables names based on files in {}/{}.'.format( S3_BUCKET, s3_const.to_ingest_key.format( config.S3_FOLDER, config.ENVIRONMENT.lower())) log.info(msg) table_list, start_time = bulk_logic.preprocess_files() else: table_list, start_time = bulk_logic.preprocess_table() if not len(table_list): log.error('There are no tables to process.') sysexit() elif table_list.get(FORCED_TABLE): log.info( 'The following existing table will be used: {}'.format( table_list[FORCED_TABLE]['table_name'])) else: log.info('The following tables will be created: ') for f, t in table_list.items(): log.info('{} will be read into {}'.format(f, t['table_name'])) for file_name, elements in table_list.items(): log.info('Processing \'{}\'.', file_name) table_name = elements['table_name'] code = elements['code'] # Check for a stopfile, and raise Error if found. bulk_logic.check_for_stopfile() if file_name != FORCED_TABLE: s3_stage_name = s3_const.s3_stage_template + '{}_{}'.format( code, start_time) bulk_logic.ingest_file( file_name, s3_stage_name, table_name) created_table_dict[file_name] = table_name log.info('Retrieving list of labels in file.') # Get labels in dataset label_list = bulk_logic.get_input_label_list(table_name) # Shape data into json for bulk upload endpoint # Loop through list of label id's for label_id in label_list: # Each label artists_failed = False # Insert Bulk Upload Tracking log and get result tracking = Counter(create_bulk_upload_tracking_row( filename=file_name, vendor_id=label_id, source_table=table_name, session_id=session_id, environment=config.ENVIRONMENT, code=code ).message) # update_bulk_upload_row(**tracking) # Set/Reset access_token access_token = None # Debug print log.info('Processing Label_ID: {}'.format(label_id)) # Init vars # label_request_list = list() # DEBUG - Print log.info('Fetching Rows from DB....') # Get rows to send to VAPI label_bulk_upload_rows = bulk_logic.get_bulk_upload_rows( session_id, table_name, label_id) file_total_rows = len(label_bulk_upload_rows) session_total_rows += file_total_rows # Update Tracking Log tracking['rows'] = file_total_rows # Debug Print log.info('{} rows retrieved.'.format(file_total_rows)) if not file_total_rows: msg = 'No rows retrieved for label {}.'.format(label_id) log.info(msg) # Update Tracking Log tracking['notes'].append(msg) tracking = update_bulk_upload_row(**tracking).message continue # -- GET ARTIST EMBARGO LIST TO PREVENT DUPLICATES ---------------- if config.CHECK_ARTISTS: log.info( 'Fetching all label artists for {} in {}.'.format( label_id, table_name)) # Get list of unique artists. all_artist_names_list = bulk_logic.get_all_input_artist_names( table_name, label_id, session_id) log.info( '{} artists found. Determining which are new artists.' .format(len(all_artist_names_list))) # Check presence of artist_name / label_id in artist_info table # for each artist not found in table, add to embargo list. if len(all_artist_names_list): embargo_artist_list = db_utils.get_new_input_artists( label_id=label_id, input_artist_list=all_artist_names_list) log.info( '{} new release-level artists found.'.format( len(embargo_artist_list))) # -- BEGIN CHUNKING PRE-PROCESS ----------------------------------- # Separate rows into request chunks per label (2 chunks in tuple) label_requests = bulk_logic.chunk_requests( label_bulk_upload_rows, embargo_artist_list.copy()) label_chunks = list() new_artist_chunks = list() # Separate tuples new_artist_req_dict, label_request_list = label_requests new_artist_len = sum( [len(x) for a, x in new_artist_req_dict.items()]) if len(new_artist_req_dict) and new_artist_len: # label_corpus = dict() # Debug Print log.info('Chunking new artist releases with access token.') # tmp_artist_dict = OrderedDict() transformed_dict = OrderedDict() # TODO: make sure the lambda count % 9 race condition never # occurs, by ensuring less than 9 requests occur in a batch. # TODO: Extract to method for rel_art_name, req_list in new_artist_req_dict.items(): if not access_token: # Fetch access token access_token = orch_utils.fetch_access_token( label_id=label_id) # Build Request Object for JSON new_artist_chunks = bulk_logic.chunk_with_token( access_token, req_list) # Embargo first request only for each primary artist for i, chunk in enumerate(new_artist_chunks): if i in transformed_dict.keys(): transformed_dict[i].append(chunk) elif i <= 1: transformed_dict[i] = [chunk] else: transformed_dict[1].append(chunk) # tmp_artist_dict[rel_art_name] = new_artist_chunks new_artist_req_dict = transformed_dict new_artist_len = sum( [len(x) for a, x in new_artist_req_dict.items()]) # END TODO if len(label_request_list) and len(label_request_list[0]): # Build Request Object for JSON label_corpus = dict() # Debug Print log.info('Chunking standard releases with access token.') access_token = orch_utils.fetch_access_token(label_id=label_id) label_chunks = bulk_logic.chunk_with_token( access_token, label_request_list) # Check sum of chunks if not len(label_chunks) and not new_artist_len: msg = 'No rows retrieved for label {}.'.format(label_id) log.info(msg) # Update Tracking Log tracking['notes'].append(msg) tracking = update_bulk_upload_row(**tracking).message continue else: new_artist_count = len(embargo_artist_list) new_req_count = sum( [len(x) for a, x in new_artist_req_dict.items()]) remaining_req_count = len(label_request_list) total_request_count = new_req_count + remaining_req_count session_total_release_count += total_request_count # Report summary of requests log.info( 'Label {} will make {} requests.\n'.format( label_id, total_request_count)) # Update Tracking Log tracking['total_releases'] = total_request_count if new_artist_count > 0: # Debug Print log.info( '{} new release-level artists found.'.format( new_artist_count)) # Debug Print log.info( '{} new artist requests will be run ahead of remaining' ' {}.'.format(new_req_count, remaining_req_count)) # -- END CHUNKING PRE-PROCESS ------------------------------------- # -- BEGIN MULTIPROCESSING OF REQUESTS ---------------------------- tracking = update_bulk_upload_row(**tracking).message new_artist_resp_dict = dict() # Run the new artist multiprocess if new_req_count: # pre_run_num = 1 for run_num, new_artist_chunks in new_artist_req_dict.items(): if len(new_artist_chunks): try: new_artist_resp_dict[run_num] = \ mp.multiprocess_requests( label_id, new_artist_chunks, session_id, file_name) # Update Tracking Log tracking.update(new_artist_resp_dict[run_num][3]) tracking = update_bulk_upload_row( **tracking).message except Exception as e: log.error(str(e)) log.info( 'Run #{}: Sleeping {} seconds for new ' 'inserts.'.format( run_num + 1, config.NEW_ARTIST_SLEEP )) # Sleep sleep(config.NEW_ARTIST_SLEEP) # Begin standard release multiprocess if len(label_chunks): # Check if new artists were properly written. if config.CHECK_ARTISTS: try: log.info( 'Checking if new artists were properly written.') # Check all artists were ingested. check_artist_list = db_utils.get_new_input_artists( label_id=label_id, input_artist_list=all_artist_names_list, except_if_found=True ) # If not all artists were found, catch exception except NewArtistError as f: # Report msg = 'First pass of artists failed to write all new' \ ' artists.The following artists did not ' \ 'appear: "{}"'.format('","'.join(list(f.result))) log.info(msg) # Report msg = 'Updating tracking table.' log.info(msg) # Update Tracking Log tracking['notes'].append(msg) tracking = update_bulk_upload_row(**tracking).message # Report msg = 'Tracking table updated.' log.info(msg) # Stop processing if there is an artist skew if not config.ALLOW_INCOMPLETE_ARTISTS: msg = ( "Execution terminated due to missing new " "artists. Set ALLOW_INCOMPLETE_ARTISTS=True " "to allow processing to continue in this " "event." ) log.error(msg) # Set flag to stop processing artists_failed = True # Continue if all artists were found or if incomplete artists # are allowed. if not artists_failed: if config.CHECK_ARTISTS: # Simple notification log.info('Continuing execution of remaining queued ' 'releases.') try: # Run the standard multiprocess standard_response_tuple = mp.multiprocess_requests( label_id, label_chunks, session_id, file_name) # Update Tracking Log tracking.update(standard_response_tuple[3]) tracking = update_bulk_upload_row(**tracking).message except Exception as e: log.error(str(e)) # -- END MULTIPROCESSING OF REQUESTS ----------------------------- new_artist_response_len = bulk_logic.get_response_length( new_artist_resp_dict) # Add new results to full result list if full_response_dict.get(label_id): full_response_dict[label_id] = [ *full_response_dict[label_id] ] else: full_response_dict[label_id] = list() if new_artist_response_len: for _, response_tuple in new_artist_resp_dict.items(): full_response_dict[label_id].append(response_tuple) full_response_dict[label_id] = [ *full_response_dict[label_id], *standard_response_tuple ] # Report if config.S3_OUTPUT: table_list[file_name]['label_tracking'][label_id] = tracking # Copy original file to output folder if config.S3_INGEST: bulk_logic.copy_ingested_file( file_name, config.ENVIRONMENT, session_id, code) if config.S3_MOVE_INGESTED: # Move ingested files to completed folder bulk_logic.move_ingested_file(file_name) # Per-file GraphQL Calls # Insert participants to graphQL backend if config.INPUT_FORMAT == 'SME_LABEL_COPY' and config.CREATE_PARTICIPANTS: # noqa participant_report = graphql_logic.create_file_participants( table_name) # Per session GraphQL calls # Set default track values if config.INPUT_FORMAT == 'SME_LABEL_COPY' and config.SET_DEFAULT_TRACKS: # noqa track_update_report = graphql_logic.set_default_track_fields( session_id) # Set correct SME project names. if config.INPUT_FORMAT == 'SME_LABEL_COPY' and config.SET_PROJECT_NAMES: # noqa participant_report = graphql_logic.set_project_names( session_id) # Set correct SME project names. if config.INPUT_FORMAT == 'SME_LABEL_COPY' and config.SET_NOT_FOR_DIST: # noqa participant_report = graphql_logic.set_not_for_distribution( session_id) # Set products to in_content if config.INPUT_FORMAT == 'SME_LABEL_COPY' and config.APPROVE_RELEASES: # noqa product_approval_report = graphql_logic.approve_products(session_id) # Handle output file generation if config.S3_OUTPUT: bulk_logic.output_files_with_polling( session_id, table_list, session_total_rows, session_total_release_count, start_time ) log.info('Writing Session ID to output file.') bulk_logic.put_session_id(session_id=session_id, env=config.ENVIRONMENT) log.info('Ingestion complete.') # TODO: Add table cleanup on Keyboard Interrupt or other terminal Exception # except (SystemExit, KeyboardInterrupt) as e: # for file_name, table_name in created_table_dict.items(): # db_utils.drop_source_table(config.SNOWFLAKE_SCHEMA, table_name)