import argparse import csv import logging import multiprocessing import timeit import pymysql import config from sys import stderr from helpers.general_use import var_to_bool from db.map_assets_to_tracks_sql import truncate_target_table from db.map_assets_to_tracks_sql \ import get_unique_unlocked_isrcs_with_tuids_from_target_table from db.map_assets_to_tracks_sql import update_target_locks_by_filename from db.map_assets_to_tracks_sql import update_target_locks_by_isrc from db.map_assets_to_tracks_sql import get_max_of_table from db.map_assets_to_tracks_sql import check_table_query from db.map_assets_to_tracks_sql import create_map_table_query from db.map_assets_to_tracks_sql import create_log_table_query from db.map_assets_to_tracks_sql import get_dupe_isrcs_by_exclude_filename from db.map_assets_to_tracks_sql import update_ioda_category_by_isrc from db.map_assets_to_tracks_sql import check_table_for_ioda_col from db.map_assets_to_tracks_sql import update_table_add_ioda_col from helpers.general_use import primitive_dict from multiprocess.multiprocess_file import multiprocess_file from multiprocess.multiprocess_fs import multiprocess_fs from config import log_fields from helpers.general_use import prep_insert_sql from helpers.IODA_comparisons import compare_isrc_rows from helpers.call_ows_territories import call_ows_territories # pydevd.settrace('localhost', port=12345, stdoutToServer=True, stderrToServer=True) if __name__ == '__main__': t_start = timeit.default_timer() # Setup commandline args parser = argparse.ArgumentParser( description='Map YouTube assets to Orchard tracks', add_help=False) parser.add_argument( '-?', '--help', action='help', help='Show this help message and exit.') parser.add_argument( '-f', '--file', help='A formatted \'TheOrchardMusic\' YouTube Asset ' 'reports.', type=str) parser.add_argument( '-i', '--ioda_file', help='A formatted \'IODA\' YouTube Asset ' 'reports.', type=str) parser.add_argument( '-m', '--meat_file', help='A formatted list of MEAT tracks.', type=str) parser.add_argument( '-c', '--count', help='The count of concurrent processes to run.', type=str) # parser.add_argument( # '-s', '--multi_table', help='Use a separate target table for ' # 'each file.', action='store_true', # default=False) parser.add_argument( '--debug', help='Enable debug mode - Equivalent to enabling all ' '--skip-? toggles', action='store_true', default= False) parser.add_argument( '--skip_orch', help='Disable processing of the OrchardMusic file(s)', action='store_true', default=False) parser.add_argument( '--skip_ioda', help='Disable processing of the IODA file(s)', action='store_true', default=False) parser.add_argument( '--skip_meat', help='Disable processing of the MEAT file(s)', action='store_true', default=False) parser.add_argument( '--skip_fact_sales', help='Disable fact_sales tuid adjustments', action='store_true', default=False) # parser.add_argument( # '--skip-dedupe', help='Disable per account de-duping of ISRC\'s', # action='store_true', default=False) parser.add_argument( '--skip_categories', help='Disable processing of IODA/Orchard ' 'comparison categories', action='store_true', default=False) # parser.add_argument( # '--fact_sales', help='Enable fact_sales pass', action='store_true', # default=False) # Parse commandline args main_args = parser.parse_args() # Jobs list jobs = [] # Number of lines processed: full = 0 meat_full = 0 # Create a dict w configs = primitive_dict(config) # Debug reassign debug = True if var_to_bool(main_args.debug) \ or var_to_bool(configs['DEBUG']) else False # Add number of threads to config configs['num_threads'] = int(main_args.count)if main_args.count else 2 # Add multi_table check to configs #configs['multi_table'] = main_args.multi_table # Create dict from files to ensure stability of enumeration # configs['files'] = {i: f for i, f in enumerate(main_args.files)} # Setup the logger log = multiprocessing.util.get_logger() # log.getEffectiveLevel() # You can setup a file instead of stdout/stderr in the StreamHandler ch = logging.StreamHandler(stderr) ch.setLevel(25) ch.setFormatter( logging.Formatter('[%(levelname)s] [%(processName)s]\n%(message)s')) log.addHandler(ch) log.setLevel(25) # with open('error_rows.txt', 'w') as f: # log.log(25, "Error file truncated.") # f.close() # Choose territory standard for requesting all territories and grab # the territory list configs['ISO_3166_1_2016_list_2_chr'] = \ call_ows_territories('ISO_3166_1_2016') # Init master_isrc_list master_isrc_list = set() if debug: with open('debug_file.log', 'w', encoding='utf8') as f: f.write('Debug File:\n') # Target mysql connection target_conn = pymysql.connect(host=configs['TARGET_HOST'], user=configs['TARGET_USER'], password=configs['TARGET_PASSWORD'], db=configs['TARGET_DATABASE'], port=int(configs['TARGET_PORT']), charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) # Target Cursor t_cursor = target_conn.cursor() # Check for map table existence t_cursor.execute(check_table_query, configs['TARGET_TABLE']) check_table = t_cursor.fetchone() # Make table if necessary if not check_table: t_cursor.execute(create_map_table_query.format(configs['TARGET_TABLE'])) log.log(25, 'Table `{}` created.'.format(configs['TARGET_TABLE'])) else: t_cursor.execute( check_table_for_ioda_col.format(configs['TARGET_DATABASE'], configs['TARGET_TABLE'])) check_ioda = t_cursor.fetchone() if not check_ioda: t_cursor.execute( update_table_add_ioda_col.format(configs['TARGET_TABLE'])) target_conn.commit() # Check for log table existence t_cursor.execute(check_table_query, configs['LOG_TABLE']) check_table = t_cursor.fetchone() # Make table if necessary if not check_table: t_cursor.execute( create_log_table_query.format(configs['LOG_TABLE'])) log.log(25, 'Table `{}` created.'.format(configs['LOG_TABLE'])) # Truncate - Check if truncate bit is set target_truncate = var_to_bool(configs['TRUNCATE_TARGET']) # Truncate table, IFF there's only one table if target_truncate: target_truncate = False t_cursor.execute( truncate_target_table.format(configs['TARGET_TABLE'])) t_cursor.execute( truncate_target_table.format(configs['LOG_TABLE'])) log.log( 25, 'Table `{}` truncated.'.format(configs['TARGET_TABLE'])) log.log( 25, 'Table `{}` truncated.'.format(configs['LOG_TABLE'])) # Map each asset file if not (debug or main_args.skip_orch): #for i, f in configs['file'].items(): # Handle per-file target tables table_suffix = None #"orch" if main_args.multi_table else None # i + 1 # Run the multiprocess full += multiprocess_file(main_args.file, configs, log, table_suffix) if not (debug or main_args.skip_ioda): # Map each asset file #for i, f in configs['ioda_file'].items(): # Handle per-file target tables table_suffix = None # 'IODA' if main_args.multi_table else None # i + 1 # Run the multiprocess full += multiprocess_file( main_args.ioda_file, configs, log, table_suffix) # get Meat tracks if main_args.meat_file: # Report to user print('Processing MEAT file \'{}\''.format(main_args.meat_file)) # Target mysql connection target_conn = pymysql.connect(host=configs['TARGET_HOST'], user=configs['TARGET_USER'], password=configs['TARGET_PASSWORD'], db=configs['TARGET_DATABASE'], port=int(configs['TARGET_PORT']), charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) # Target Cursor t_cursor = target_conn.cursor() # Mark meat as uningested # meat_ingested = False # Init unique isrc list vars table_isrc_list = set() target_table = "" if not (debug or main_args.skip_meat or main_args.skip_fact_sales): # Report to user print('Collecting unique ISRC\'s') # Get all unique isrc's across all tables # if main_args.multi_table: # for i, f in configs['files'].items(): # # Handle per-file target tables # target_table = configs['TARGET_TABLE'] + "_" + str(i + 1) # # # Get all unique UNLOCKED isrcs from table # t_cursor.execute( # get_unique_unlocked_isrcs_with_tuids_from_target_table # .format(target_table)) # # # Union in the set of claimedisrc's # table_isrc_list = table_isrc_list.union( # set([v['isrc'] for v in t_cursor.fetchall()])) # # if debug: # # Get max row for debubg reporting # t_cursor.execute(get_max_of_table.format(target_table)) # # max_id = t_cursor.fetchone() # # print("Table `{}` max row: {}".format(target_table, # max_id['max'])) # else: # # Get all unique UNLOCKED isrcs from table t_cursor.execute( get_unique_unlocked_isrcs_with_tuids_from_target_table.format( configs['TARGET_TABLE'])) # Union in the set of claimed isrc's table_isrc_list = set([v['isrc'] for v in t_cursor.fetchall()]) if debug: # Get max row for debubg reporting t_cursor.execute( get_max_of_table.format(configs['TARGET_TABLE'])) max_id = t_cursor.fetchone() print("Table `{}` max row: {}".format(configs['TARGET_TABLE'], max_id['max'])) # Add to Master ISRC List master_isrc_list = master_isrc_list.union(table_isrc_list) # Report to user print('Analyzing MEAT file...') # meat_table_name = configs['TARGET_TABLE'] + '_meat' \ # if main_args.multi_table \ # else configs['TARGET_TABLE'] meat_table_name = configs['TARGET_TABLE'] # Write a new CSV meat_filename = 'meat_insert.csv' if not (debug or main_args.skip_meat): # Open MEAT file with open(main_args.meat_file, encoding='utf8') as f: # Init default lists lock_meat_isrc_list = set() insert_meat_isrcs = [] # assume first line is header cf = csv.DictReader(f, delimiter=',') # Loop through rows, and bank ISRC's for row in cf: if row['isrc']: if row['isrc'] in table_isrc_list: # Rows to lock lock_meat_isrc_list.add(row['isrc']) else: # New rows to insert insert_meat_isrcs.append(row) # Report to user print('Locking MEAT ISRC\'s') # Create a string from the lock meat isrc's lock_isrc_str = '\', \''.join(list(lock_meat_isrc_list)) # Get all unique rows Set all rows in each table with isrc's in # meat to is_locked = 1 # if main_args.multi_table: # for i, f in configs['files'].items(): # # Handle per-file target tables # target_table = configs['TARGET_TABLE'] + "_" + str(i + 1) # # # Lock the rows # t_cursor.execute( # update_target_locks_by_isrc.format( # target_table, lock_isrc_str)) # else: t_cursor.execute( update_target_locks_by_isrc.format( configs['TARGET_TABLE'], lock_isrc_str)) target_conn.commit() print('\nMEAT tracks in \'{}\' locked.\nAdding new MEAT'.format( main_args.meat_file)) # TODO: Replace this hackneyed file-write nonsense with open(meat_filename, 'w', encoding='utf-8') as output_file: keys = insert_meat_isrcs[0].keys() dict_writer = csv.DictWriter(output_file, keys) dict_writer.writeheader() dict_writer.writerows(insert_meat_isrcs) # Report to user print('Mapping MEAT tracks...') # Handle multi-table #meat_table_appendix = 'meat' if main_args.multi_table else None meat_table_appendix = None # call the multiprocess again! meat_full += multiprocess_file( meat_filename, configs, log, meat_table_appendix, dedupe=False) # Lock the rows t_cursor.execute( update_target_locks_by_filename.format( meat_table_name, meat_filename)) target_conn.commit() # Report to user print('MEAT tracks added.') # Get unique unlocked isrc's from meat table # t_cursor.execute( # get_unique_unlocked_isrcs_with_tuids_from_target_table.format( # meat_table_name)) # # # add them to the master set # master_isrc_list = master_isrc_list.union( # set([v['isrc'] for v in t_cursor.fetchall()])) if not (debug or main_args.skip_fact_sales): # Notify user print("Getting fact_sales data for {} isrc's".format( len(master_isrc_list))) # Begin Fact Sales pass # if main_args.fact_sales: # Map each asset file # for i, f in configs['file'].items(): # Handle per-file target tables #table_suffix = 'orch' if main_args.multi_table else None #i + 1 table_suffix = None # Run the multiprocess full += multiprocess_fs( master_isrc_list, configs, log, main_args.file, table_suffix) # Map each asset file # for i, f in configs['ioda_file'].items(): # Handle per-file target tables #table_suffix = 'IODA' if main_args.multi_table else None # i + 1 table_suffix = None # Run the multiprocess full += multiprocess_fs( master_isrc_list, configs, log, main_args.ioda_file, table_suffix) # REMOVED - MEAT doesn't need this # if main_args.meat_file: # # Run the multiprocess # full += multiprocess_fs( # master_isrc_list, configs, log, meat_filename, 'meat') ## ABCD IODA sorting algorithm psuedo-code ## if not (debug or main_args.skip_categories): # Report to user print('Beginning IODA territory comparisons.') # Target mysql connection target_conn = pymysql.connect(host=configs['TARGET_HOST'], user=configs['TARGET_USER'], password=configs['TARGET_PASSWORD'], db=configs['TARGET_DATABASE'], port=int(configs['TARGET_PORT']), charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) # Target Cursor t_cursor = target_conn.cursor() # Get the duplicate ISRC's across all non-meat files - `meat_filename` variable t_cursor.execute( get_dupe_isrcs_by_exclude_filename.format( configs['TARGET_TABLE'], meat_filename)) dupe_isrc_list = t_cursor.fetchall() # Call distributed conflict sorting on chunks # Chunk responses for i, row in enumerate(dupe_isrc_list): # if > 2 - Can't automatically make a decision if row['c'] > 2: # write to error log log_row = {} log_row['reason'] = "More than 2 dupes across files." log_row['filename'] = "IODA Compare" log_row['id'] = None sql, log_row = prep_insert_sql(log_row, log_fields, configs['LOG_TABLE']) t_cursor.execute(sql, list(log_row.values())) else: # Check and compare teritories ioda_category = compare_isrc_rows(t_cursor, row['isrc']) # SQL to update table with category t_cursor.execute( update_ioda_category_by_isrc.format( configs['TARGET_TABLE'], ioda_category, row['isrc'])) target_conn.commit() # Report to user print('IODA comparisons complete.') # Stop the timer t_stop = timeit.default_timer() log.log(25, "Program executed {} lines in {}".format( full, t_stop - t_start)) log.log(25, " Done.")