import pymysql import logging import json import psycopg2.extras from timeit import default_timer from multiprocessing import current_process from multiprocessing import util from db.map_assets_to_tracks_sql import insert_tuid_query from db.map_assets_to_tracks_sql import update_ownership_by_row_id_query from db.map_assets_to_tracks_sql import get_recent_fact_sales_rows_by_isrc from db.map_assets_to_tracks_sql import get_unlocked_list_with_tuid_from_target_table_by_isrcs from helpers.general_use import remap_transitional_territories from sys import stdout from sys import stderr from transform_YT_ownership import clean_territories def fs_fix_up (start, stop, full, config, name, chunk, target_table, file): count_updated = 0 error_msg = None name = current_process().name = name 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)) 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.log(25, "Connecting to DB's. Please wait....") def compare_rows(fs_row, rows_to_fix): count = 0 # No rows if not len(rows_to_fix): # Write error to file with open( target_table+'_fs_isrc_territory_missing.log', 'a', encoding='utf8') as error_file: error_file.write('Fact_sales isrc not found in mapping:\n') json.dump(fs_row, error_file) return 0 # Add territory if missing to ALL matching rows # (should only be one after de-duping) # Dump multiples to file # if len(rows_to_fix) > 1: # use the territory mapper country = remap_transitional_territories([fs_row['country_code']])[0] # Init search success existing_tuid_row = False # Figure out if there is an existing target # short_list = [row['derived_tuid'] for row in rows_to_fix] # List of all territories all_territories = set() # Aggregate territories for look-ahead for row in rows_to_fix: all_territories = all_territories.union(filter(None, set(row['derived_ownership'].split(',')))) # Check if the territory is a no-op if country not in all_territories: # Not in Youtube mapping, then don't add it return 0 # Check and fix tuid targets for row in rows_to_fix: new_ownership = set() # Get current ownership profile current_ownership = row['derived_ownership'] \ .split(',') current_ownership_set = set(current_ownership) # TUID match if int(fs_row['track_unique_id']) \ == int(row['derived_tuid']) and not existing_tuid_row: existing_tuid_row = True # Check territory if country not in current_ownership: # If this is a territory not seen in the asset report # Don't add it! Report it if (len(rows_to_fix) == 1): # Write error to file with open( target_table + '_fs_isrc_territory_missing.log', 'a', encoding='utf8') as error_file: error_file.write( 'Fact_sales territory matching sole tuid not ' 'found in mapping:\n') json.dump(fs_row, error_file) json.dump(row, error_file) return 0 # Add territory - Fix ownership new_ownership = set(filter( None, current_ownership_set.union({country}))) new_ownership = clean_territories(str(sorted( new_ownership)).strip('[]').replace(" ", "")) # TUID mismatch else: # Check territory if country in current_ownership: # Remove territory - Fix ownership new_ownership = current_ownership_set.difference( {country}) new_ownership = clean_territories(str(sorted( new_ownership)).strip('[]').replace(" ", "")) # # Replace the territory list # t_cursor.execute( # update_ownership_by_row_id_query.format( # target_table), (new_ownership, row['id'])) # If a new territory profile has been created if new_ownership != set(): # Replace the territory list t_cursor.execute( update_ownership_by_row_id_query.format( target_table), (new_ownership, row['id'])) count += 1 if not existing_tuid_row: # Add the new row t_cursor.execute( insert_tuid_query.format(target_table), (row['custom_id'], 'fact_sales', row['asset_id'], fs_row['upc'], fs_row['upc'], row['isrc'], fs_row['track_unique_id'], row['reason'], country, row['last_delivery_date'] ) ) return count if count else 0 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) # Target Cursor t_cursor = target_conn.cursor() # yt_rows = chunk # csv.DictReader(f, delimiter=',') # 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: {} to {} of {}'.format( name, len(chunk), start+1, stop+1, full)) conn_string = \ "dbname='{}' port='{}' user='{}' password='{}' host='{}'".format( config['FS_DATABASE'], config['FS_PORT'], config['FS_USER'], config['FS_PASSWORD'], config['FS_HOST'] ) conn = psycopg2.connect(conn_string) fs_cursor = conn.cursor(cursor_factory=psycopg2.extras.DictCursor) # Get all ISRC / YT revenue per territory with tuid fs_cursor.execute( get_recent_fact_sales_rows_by_isrc.format( '\', \''.join(chunk))) fs_rows = fs_cursor.fetchall() count_lines = 0 # Handle per-file target tables # target_table = config['TARGET_TABLE'] + "_" + str(i + 1) last_period = 0 last_country = '' last_isrc = '' # Process chunk row by row for count_lines, fs_row in enumerate(fs_rows): if last_isrc == fs_row['isrc'] \ and fs_row['country_code'] == last_country \ and fs_row['period'] <= last_period: continue else: last_period = fs_row['period'] last_country = fs_row['country_code'] last_isrc = fs_row['isrc'] # fail_message = [] # count_lines += 1 if count_lines == 0 or not count_lines % (stop-start / 10): log.log( 25, "{}\nProcessing row {} of {} " "in {} before commit".format( name, count_lines+1, stop-start, file)) # # Search for isrc / territory combo # t_cursor.execute( # get_rows_by_isrc_and_territory.format( # target_table, # fs_row['isrc'], fs_row['country_code'])) # Check for territory / tuid for a given isrc in a given # file t_cursor.execute( get_unlocked_list_with_tuid_from_target_table_by_isrcs .format(target_table, fs_row['isrc'] )) # ISRC/tuid match rows_to_fix = t_cursor.fetchall() # Found any? if (rows_to_fix): # fix the rows count_updated += compare_rows(fs_row, rows_to_fix) # Commit the result per file target_conn.commit() loop_stop = default_timer() log.log(25, "{}:\nProgram executed {} lines in {}\n".format( name, len(chunk), str(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, str(count_updated)