import pymysql import json import multiprocessing from math import ceil from helpers.general_use import file_len from helpers.general_use import var_to_bool from mapper.mapper import process_file from mapper.mapper import consumer from mapper.mapper import map_file from db.map_assets_to_tracks_sql import truncate_target_table from db.map_assets_to_tracks_sql import create_map_table_query from db.map_assets_to_tracks_sql import check_table_query from db.map_assets_to_tracks_sql import delete_rows_by_isrc_and_filename from db.map_assets_to_tracks_sql \ import get_list_from_target_table_by_isrcs_and_filename from db.map_assets_to_tracks_sql import get_dupe_isrcs_by_filename from config import log_fields from helpers.general_use import prep_insert_sql def multiprocess_file( filename, config, log, table_appendix=None, dedupe=True): # Init queues inQ = multiprocessing.SimpleQueue() outQ = multiprocessing.SimpleQueue() # Create workers workers = [multiprocessing.Process( target=consumer, args=(inQ, outQ, map_file)) for i in range(config['num_threads'])] # Start workers for w in workers: w.start() # Get file length print("Getting input file length.") full = file_len(filename) step = int(config['ROW_STEP']) # Get per thread line count thread_boundary = ceil(full / config['num_threads']) # Target mysql connection target_conn = pymysql.connect(host=config['TARGET_HOST'], user=config['TARGET_USER'], password=config['TARGET_PASSWORD'], db=config['TARGET_DATABASE'], port=int(config['TARGET_PORT']), charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) # Target Cursor t_cursor = target_conn.cursor() log_table = config['LOG_TABLE'] # Use table_appendix if necessary if table_appendix: target_table = config['TARGET_TABLE']+"_"+str(table_appendix) else: target_table = config['TARGET_TABLE'] # Check for table existence t_cursor.execute(check_table_query, target_table) check_table = t_cursor.fetchone() # Make table if necessary if not check_table: t_cursor.execute(create_map_table_query.format(target_table)) log.log(25, 'Table `{}` created.'.format(target_table)) target_truncate = var_to_bool(config['TRUNCATE_TARGET']) # Truncate table, IFF there is a table_appendix if target_truncate and table_appendix: target_truncate = False t_cursor.execute(truncate_target_table.format(target_table)) log.log(25, 'Table `{}` truncated.'.format(target_table)) # Init parameter list var param_list = [] log.log(25, "Map Multiprocessing beginning.") # Generate params for i in range(config['num_threads']): start = i * thread_boundary stop = ((i + 1) * thread_boundary)-1 param_list.append( {'start': start, 'stop': stop, 'step': step, 'full': full, 'config': config, 'name': "Map: {} of {} (rows {}-{} of {})".format( i, config['num_threads'], start, stop, full), 'filename': filename, 'target_table': target_table }) # Load the Queues process_file(param_list, inQ, outQ) # Tell all workers, no more data (one msg for each) for i in range(config['num_threads']): inQ.put(None) # Join on the workers for w in workers: w.join() # RE-OPEN cause it takes so long # Target mysql connection target_conn = pymysql.connect(host=config['TARGET_HOST'], user=config['TARGET_USER'], password=config['TARGET_PASSWORD'], db=config['TARGET_DATABASE'], port=int(config['TARGET_PORT']), charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) # Target Cursor t_cursor = target_conn.cursor() # Print out final results (i*16) for i, msg in enumerate(param_list): for j, line in enumerate(msg[1]): line_dict = line.toDict() line_dict['reason'] = "Processing Error" line_dict['filename'] = filename line_dict['id'] = None sql, line_dict = prep_insert_sql(line_dict, log_fields, log_table) t_cursor.execute(sql, list(line_dict.values())) if j%5000 == 0: # Write errors to db target_conn.commit() # with open(filename + '_' + # target_table + # '-process_error.log', "a") as f: # f.writelines(msg[1]) # Write errors to db target_conn.commit() print(i, msg[0], (msg[2] if msg[2] else "")) # # Handle Dupe ISRC's ------------------------------ if (dedupe): # Refresh Connection # Target mysql connection target_conn = pymysql.connect(host=config['TARGET_HOST'], user=config['TARGET_USER'], password=config['TARGET_PASSWORD'], db=config['TARGET_DATABASE'], port=int(config['TARGET_PORT']), charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) # Target Cursor t_cursor = target_conn.cursor() # Get isrc list t_cursor.execute( get_dupe_isrcs_by_filename.format(target_table, filename)) dupe_isrc_dict = t_cursor.fetchall() # Convert to list dupe_isrc_list = [x['isrc'] for x in dupe_isrc_dict] # Get all metadata for isrc's t_cursor.execute( get_list_from_target_table_by_isrcs_and_filename.format( target_table, '\', \''.join(dupe_isrc_list), filename)) dupe_isrc_full_rows = t_cursor.fetchall() if (dupe_isrc_full_rows): for line_dict in dupe_isrc_full_rows: #line_dict = line.toDict() line_dict['reason'] = "Duplicate ISRC" line_dict['filename'] = filename line_dict['id'] = None sql, line_dict = prep_insert_sql(line_dict, log_fields, log_table) t_cursor.execute(sql, list(line_dict.values())) # Write to file # with open(filename + '_' + config[ # 'TARGET_TABLE'] + '-dupe_isrc.log', "a") as f: # f.writelines( # json.dumps( # dupe_isrc_full_rows, # sort_keys=True)+"\n") # Delete them dupes t_cursor.execute(delete_rows_by_isrc_and_filename.format( target_table, '\', \''.join(dupe_isrc_list), filename)) target_conn.commit() return full