import argparse import pymysql import csv import config import webbrowser from random import randint from db.map_assets_to_tracks_sql import check_table_query from db.map_assets_to_tracks_sql import get_column_names from db.map_assets_to_tracks_sql import get_row_from_target_table_by_field from db.map_assets_to_tracks_sql import get_row_from_target_table_by_id from db.map_assets_to_tracks_sql import get_max_id_from_target_table from helpers.search_modules import display_general_file from helpers.search_modules import display_general_db from helpers.search_modules import display_ownership_db from helpers.search_modules import display_ownership_file from sys import exit as sysexit if __name__ == '__main__': # Setup commandline args parser = argparse.ArgumentParser( description='Spot check YouTube mapping exercise', add_help=False) parser.add_argument( '-?', '--help', action='help', help='Show this help message and exit.') parser.add_argument( '-f', '--filename', help='The file source to check.', type=str) parser.add_argument( '-s', '--search_file', help='An optional file containing a field, ' 'and a list of values to search for.', type=str) parser.add_argument( '-n', '--num_checks', help='The count of checks to make.', type=str) # parser.add_argument( # '-e', '--errors', help='The count of checks to make.', # action='store_true') parser.add_argument( '-o', '--open_url', help='Automatically opens OA urls.', action='store_true') # parser.add_argument( # '-s', '--source', help='The DB source to check.', # type=str) # Parse args args = parser.parse_args() 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) test_table = config.TEST_TABLE test_host = config.TEST_HOST test_user = config.TEST_USER test_password = config.TEST_PASSWORD test_database = config.TEST_DATABASE test_port = int(config.TEST_PORT) # Target mysql connection test_conn = pymysql.connect(host=test_host, user=test_user, password=test_password, db=test_database, port=test_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 = test_conn.cursor() # Art Relations Cursor ar_cursor = ar_conn.cursor() t_cursor.execute(check_table_query, test_table) check_table = t_cursor.fetchone() # Make table if necessary if not check_table: sysexit("\nThe target table was not found.\n") # Get length of file print("\nGetting highest id from target table.\n") t_cursor.execute(get_max_id_from_target_table.format(test_table)) row = t_cursor.fetchone() max_row = int(row['max_id']) # # Choose match fields # # # get all fields from the DB t_cursor.execute( get_column_names.format(test_database, test_table)) db_header = t_cursor.fetchall() db_header = [i['COLUMN_NAME'] for i in db_header] # get all fields from the file with open(args.filename, 'r', encoding='utf8') as f: # create reader reader = csv.reader(f) # pull the header row file_header = next(reader) # lower and space substitute all fields db_header_formatted = [i.lower().replace(" ", "_") for i in sorted(db_header)] file_header_formatted = [i.lower().replace(" ", "_") for i in sorted(file_header)] # show user all choices with likely choices marked print('DB:\n', ' | '.join(db_header_formatted), '\n\nFile:\n', ' | '.join(file_header)) # List file fields # ask which field to compare on file_match_field = input( 'Please choose a field from the file (\'Asset ID\'): ').strip() \ or 'Asset ID' # List db fields # ask which field to compare on db_match_field = input( 'Please choose a field from the db which contains the same values as ' 'the field in the file you selected (\'asset_id\'): ', ).strip() \ or 'asset_id' file_compare_field = input( "Please choose the field (\'Ownership\'): ").strip() or 'Ownership' db_inspection = input( 'Please choose the field you are comparing (\'derived_ownership\'):')\ .strip() or 'derived_ownership' # ask which collateral fields should be displayed ## begin loop # init vars searches = 0 if (args.search_file): with open(args.search_file, 'r', encoding='utf8') as f: search_items = f.readlines() loop_count = len(search_items) else: loop_count = int(args.num_checks) # start loop for i in range (0, loop_count): # Increment searches searches += 1 # init val file_row = {} found = False response = "" if (args.search_file): item = search_items[i].strip('\n') t_cursor.execute( get_row_from_target_table_by_field.format( test_table, db_inspection, item)) db_row = t_cursor.fetchone() print("\nSearching for {}: {}\n".format(file_compare_field, item)) else: # Create random rand_row = (randint(1, max_row - 1)) # Handle potentially missing rows from deletions db_row = True while not db_row: t_cursor.execute( get_row_from_target_table_by_id.format( test_table, rand_row)) db_row = t_cursor.fetchone() ## ------- DB Processing Modules (Add more here)---- # 1) masters registry backfill if (db_inspection == 'derived_ownership'): # Pass the db_row for processing OA_link, response = display_ownership_db(db_row) # open url if necessary if args.open_url: webbrowser.open(OA_link, autoraise=False) else: response = display_general_db(db_row) # Print the formatted string print(response) # Search file for string with open(args.filename, 'r', encoding='utf8') as f: reader = csv.reader(f) response = "" # init row count line_num = 1 # pull the header row file_header = next(reader) # Message user print( "Searching {} for key ({}): {}\n".format( args.filename, file_match_field, db_row[db_match_field])) # loop through the file for line in reader: if not found: # Increment line number line_num += 1 # increase line if db_row[db_match_field] in line: # Create a dict file_row = dict(zip(file_header, line)) # if match if file_row[file_match_field] \ == db_row[db_match_field]: found = True break if found: ## ------- File Processing Modules (Add more here)---- # 1) masters registry backfill if (db_inspection == 'derived_ownership'): response = display_ownership_file(file_row, line_num) else: response = display_general_file(file_row, line_num) # Print the formatted string print(response) else: print('Asset not found in file \'{}\''.format(args.filename)) user_input = input("Continue? (yes[any key]/[n]o)") if user_input == 'n': break else: print("\n\n--------------------------------------------") print("Searched {} times.".format(searches))