import pymysql import csv import logging import json import re from timeit import default_timer from multiprocessing import current_process from multiprocessing import util from db.map_assets_to_tracks_sql import get_delivery_history from db.map_assets_to_tracks_sql import get_track_by_isrc_query from db.map_assets_to_tracks_sql import get_track_by_itm_query from db.map_assets_to_tracks_sql import get_track_by_tuid_query from db.map_assets_to_tracks_sql import insert_tuid_query from db.map_assets_to_tracks_sql import get_track_by_upc_and_isrc_query from sys import exit as sysexit from sys import stdout from sys import stderr from helpers import art_relations_helpers as arhelp from transform_YT_ownership import transform_ownership from dotmap import DotMap from itertools import islice from time import sleep from os.path import splitext from config import log_fields from helpers.general_use import prep_insert_sql def map_file (start, stop, step, full, config, name, filename, target_table): # , rw_lock): # prep utf-8 4-byte filter # try: # highpoints= re.compile(u'[\U00010000-\U0010ffff]') # except re.error: # # UCS-2 build # highpoints = re.compile(u'[\uD800-\uDBFF][\uDC00-\uDFFF]') error_msg = None name = current_process().name = name full_filename = filename filename = splitext(filename)[0] process_err = [] ww_list = config['ISO_3166_1_2016_list_2_chr'] log = util.get_logger() ch = logging.StreamHandler(stderr) ch.setLevel(25) ch.setFormatter( logging.Formatter( '[%(levelname)s] [%(processName)s]\n%(message)s')) log.addHandler(ch) log.setLevel(25) log.log(25, '{} starting'.format(name)) ar_host = config['AR_HOST'] ar_user = config['AR_USER'] ar_password = config['AR_PASSWORD'] ar_database = config['AR_DATABASE'] ar_port = int(config['AR_PORT']) #target_table = config['TARGET_TABLE'] target_host = config['TARGET_HOST'] target_user = config['TARGET_USER'] target_password = config['TARGET_PASSWORD'] target_database = config['TARGET_DATABASE'] target_port = int(config['TARGET_PORT']) log_table = config['LOG_TABLE'] log.log(25, "Connecting to DB's. Please wait....") try: # Target mysql connection target_conn = pymysql.connect(host=target_host, user=target_user, password=target_password, db=target_database, port=target_port, charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) # art_relations mysql connection ar_conn = pymysql.connect(host=ar_host, user=ar_user, password=ar_password, db=ar_database, port=ar_port, charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor) # Target Cursor t_cursor = target_conn.cursor() # Art Relations Cursor ar_cursor = ar_conn.cursor() # Open CSV try: f = open(full_filename, encoding='utf8') except: # parent of IOError, OSError *and* WindowsError where available sysexit( 'There was a problem opening \'{}\'.'.format(full_filename)) # assume first line is header yt_rows = csv.DictReader(f, delimiter=',') # Convert fieldnames to lower to match pg yt_rows.fieldnames = [name.lower() .replace(" ", "_") .replace("(yes/no)", "") .replace(".", "") .replace("%", "") for name in yt_rows.fieldnames] # This function makes use of GLOBAL namespace variables instantiated # in the main body - For legibility in the if/else block below def insert_tuid(track_unique_id, msg): # Check for delivery history if upc and arhelp.is_upc(upc): ar_cursor.execute(get_delivery_history, (upc)) date_results = ar_cursor.fetchone() else: date_results = None if not date_results: delivery = "ERROR" else: delivery = date_results['date_delivered'] t_cursor.execute( insert_tuid_query.format(target_table), (custom_id, filename, asset_id, display_upc, upc, isrc, track_unique_id, msg, derived_ownership, delivery )) # TODO: UPC/ISRC for missing combos - Done? for i in range(start, stop, step): loop_start = default_timer() # TODO: REMOVE ME. Profiling timer if (i+step) > stop: step = (stop - i) + 1 if (i == start): chunk = islice(yt_rows, i, i+step) else: chunk = islice(yt_rows, step) log.log(25, 'Now processing rows {0} to {1} of {2}'.format( i, i + step, full)) count_lines = 0 # Process chunk row by row for asset in chunk: fail_message = [] count_lines+=1 if count_lines == 1 or not count_lines%(step/10): log.log( 25, "{}\nProcessing row {} of {} before commit".format( name, count_lines, step)) # stdout.flush() # Handle convetring to dot access if full_filename: asset = DotMap(asset) # Skip if not in our special set if asset.asset_type != 'Sound Recording': # with rw_lock.write_lock(): # with open( # filename+'_'+target_table+'-type_error.log', # 'a', # encoding='utf8') as f: asset_dict = asset.toDict() asset_dict['reason'] = "Wrong Asset Type" asset_dict['filename'] = filename asset_dict['id'] = None sql, asset_dict = prep_insert_sql(asset_dict, log_fields, log_table) t_cursor.execute(sql, list(asset_dict.values())) # Write errors to db #target_conn.commit() # f.write( # "WRONG ASSET TYPE: "+json.dumps( # write_out, # sort_keys=True)+"\n") continue # Init some loop vars keep_searching = True fail_message = [] custom_id = asset.custom_id upc = asset.upc is_upc = arhelp.is_upc(upc, display=True) display_upc = None isrc = asset.isrc is_isrc = arhelp.is_isrc(isrc) # ydh_id = asset.row_id asset_id = asset.asset_id tuid = None filename = full_filename.split('\\')[-1] \ if full_filename else asset.filename custom_id_list = custom_id.split('_') \ if custom_id else [] # Transform ownership if asset.ownership: derived_ownership = transform_ownership( asset.ownership, ww_list) else: derived_ownership = '' if derived_ownership is None: # with open(filename+'_'+target_table+'-ownership_error.log', # 'a', # encoding='utf8') as f: # write_out = asset.toDict() asset_dict = asset.toDict() asset_dict['reason'] = "Ownership Error" asset_dict['filename'] = filename asset_dict['id'] = None sql, asset_dict = prep_insert_sql(asset_dict, log_fields, log_table) t_cursor.execute(sql, list(asset_dict.values())) # f.write( # "OWNERSHIP ERROR: "+json.dumps( # asset.toDict(), # sort_keys=True)+"\n") continue # DEPRECATED # derived_conflicts = transform_conflicts( # asset.conflicting_territories, # ISO_3166_1_2016_list_2_chr) # Catch non-processable row and forget it if (upc == '\\N' or upc is None) \ and (isrc == '\\N' or isrc is None) \ and (custom_id == '\\N' or custom_id is None): insert_tuid(None, 'No searchable fields given') keep_searching = False # GGL USR ID if keep_searching \ and custom_id is not None \ and custom_id != '\\N' \ and 'GGL_USR_ID' in custom_id: insert_tuid(None, 'GGL_USR_ID') keep_searching = False # ArtTrack if keep_searching \ and custom_id is not None \ and custom_id != '\\N' \ and 'ArtTrack' in custom_id \ and custom_id: insert_tuid(None, 'ArtTrack') keep_searching = False # if custom id has UPC/ISRC pair, use them if keep_searching \ and custom_id is not None \ and custom_id != '\\N' \ and len(custom_id_list) >= 2 \ and arhelp.is_upc(custom_id_list[0]) \ and arhelp.is_isrc(custom_id_list[1]): # Find track using sliced fields ar_cursor.execute( get_track_by_upc_and_isrc_query, dict(upc=custom_id_list[0], isrc=custom_id_list[1])) results = ar_cursor.fetchall() # No Matches if len(results) < 1: fail_message.append( 'NOT FOUND - custom_id UPC/ISRC pair') elif len(results) > 1: fail_message.append( 'TOO MANY - custom_id UPC/ISRC pair') else: # Juuuuust Right tuid = results[0].get('id') display_upc = results[0].get('display_upc') upc = results[0].get('upc') isrc = results[0].get('isrc') insert_tuid(tuid, 'TUID from custom_id UPC/ISRC pair') keep_searching = False # if upc and isrc fields are filled, use them if keep_searching \ and upc \ and isrc \ and is_upc \ and is_isrc: # Get tuid using filled fields ar_cursor.execute( get_track_by_upc_and_isrc_query, dict(upc=upc, isrc=isrc)) results = ar_cursor.fetchall() # No Matches if len(results) < 1: fail_message.append('NOT FOUND - ISRC and UPC') else: # Will be only one by design insert_tuid(results[0].get('id'), 'ISRC and UPC') keep_searching = False # If custom id has a tuid, use it if keep_searching \ and custom_id is not None \ and custom_id != '\\N' \ and len(custom_id_list) == 3: # Use the 3rd element to get full track info ar_cursor.execute( get_track_by_tuid_query, (custom_id_list[2],)) # Will be only one by design track = ar_cursor.fetchone() if track: tuid = track.get('id') display_upc = track.get('display_upc') upc = track.get('upc') isrc = track.get('isrc') insert_tuid(tuid, 'TUID from custom_id') keep_searching = False else: # No matches fail_message.append( 'NOT FOUND - TUID from custom_id is invalid.') # if custom id cannot be used for tuid, use isrc field if keep_searching \ and (upc is None or upc == '\\N') \ and (isrc is not None and isrc != '\\N' and is_isrc): # Get track list by ISRC ar_cursor.execute(get_track_by_isrc_query, (isrc)) results = ar_cursor.fetchall() if len(results) < 1: # No matches fail_message.append('NOT FOUND - ISRC but no UPC') elif len(results) > 1: # Too many matches fail_message.append('TOO MANY - ISRC but no UPC') else: # Juuuuust Right tuid = results[0].get('tuid') display_upc = results[0].get('display_upc') upc = results[0].get('upc') insert_tuid(tuid, 'ISRC but no UPC - One match') keep_searching = False # if there's no isrc field and custom id doesnt have tuid, # but custom_id has upc and isrc, use them if keep_searching \ and len(custom_id_list) == 2 \ and arhelp.is_upc(custom_id_list[0]) \ and arhelp.is_isrc(custom_id_list[1]): # TODO: Any NEED FOR SWAPPING MISMATCHED VALUES? # - UPC <-> ISRC? # Get a track from the sliced fields ar_cursor.execute( get_track_by_upc_and_isrc_query, dict(upc=custom_id_list[0], isrc=custom_id_list[1])) results = ar_cursor.fetchall() if len(results) < 1: # No matches fail_message.append( 'NOT FOUND - ISRC and UPC from custom_id') elif len(results) > 1: # Too many matches fail_message.append( 'TOO MANY - ISRC and UPC from custom_id') else: # Juuuuust Right tuid = results[0].get('tuid') display_upc = results[0].get('display_upc') upc = results[0].get('upc') isrc = results[0].get('isrc') insert_tuid( tuid, 'ISRC and UPC from custom_id - One match') keep_searching = False # Only ISRC in custom_id if keep_searching \ and len(custom_id_list) == 1 \ and arhelp.is_isrc(custom_id_list[0]): # Get track list by ISRC ar_cursor.execute( get_track_by_isrc_query, (custom_id_list[0])) results = ar_cursor.fetchall() if len(results) < 1: # No matches fail_message.append('NOT FOUND - ISRC but no UPC') elif len(results) > 1: # Too many matches fail_message.append('TOO MANY - ISRC but no UPC') else: # Juuuuust Right tuid = results[0].get('tuid') display_upc = results[0].get('display_upc') upc = results[0].get('upc') isrc = custom_id_list[0] insert_tuid( tuid, 'ISRC from custom_id but no UPC - One match') keep_searching = False # UPC from Custom_ID and no ISRC if keep_searching \ and len(custom_id_list) == 1 \ and arhelp.is_upc(custom_id_list[0]): insert_tuid(None, 'UPC from custom_id but no ISRC') keep_searching = False # if custom id is IODA id, use it if keep_searching \ and custom_id \ and custom_id != '\\N' \ and custom_id is not None \ and len(custom_id) < 9 \ and filename[18:22] == "IODA": # Query IODA table and get tuid ar_cursor.execute(get_track_by_itm_query, (custom_id,)) # Will be only one by design track = ar_cursor.fetchone() if track: tuid = track.get('id') display_upc = track.get('display_upc') upc = track.get('upc') isrc = track.get('isrc') insert_tuid(tuid, 'IODA id in custom_id') keep_searching = False else: # No matches fail_message.append( 'NOT FOUND - IODA ID not in IODA <-> ' 'TUID mapping table.') # Catch all if keep_searching: #with rw_lock.write_lock(): process_err.append(asset) insert_tuid(None, 'COULD NOT PROCESS: ' + ('; '.join(fail_message) if bool(fail_message) else 'UNKNOWN REASON')) target_conn.commit() loop_stop = default_timer() log.log(25, "{}:\nProgram executed {} lines in {}\n".format( name, step, loop_stop - loop_start)) stdout.flush() target_conn.close() ar_conn.close() msg ="{} done.".format(current_process().name) except Exception as e: error_msg = e msg = "{} failed.".format(current_process().name) return msg, process_err, error_msg, 0 def consumer(inQ, outQ, process): log = util.get_logger() while True: try: # get a new message val = inQ.get() # this is the 'TERM' signal if val is None: break # unpack the message pos = val[0] # its helpful to pass in/out the pos in the array data = val[1] # process the data ret, proc_err, err, count = process(**data) # send the response / results outQ.put((pos, ret, proc_err, err, count)) except Exception: log.log(25, "error!", Exception) break # def write(queue, fname, target_table): # with open(fname+'_'+target_table+'-process_error.log', "w") as f: # while True: # try: # msg, proc_err, err = queue.get() # f.writelines(proc_err) # except: # break def process_file(param_list, inQ, outQ): # send pos/data to workers for i, dat in enumerate(param_list): inQ.put((i, dat)) sleep(2) # process results for i in range(len(param_list)): pos, dat, proc_err, err, count = outQ.get() param_list[pos] = (dat, proc_err, err, count)