import argparse import csv import pexpect import subprocess as sp import sys import config from connectors import mysql def load_meat(full=False, download=True, backfill=False): MEAT_RAW_FILENAME = config.MEAT_FILE # unprocessed file from YouTube MEAT_RAW_OLD_FILENAME = 'MEAT-00000-of-00001-old.csv' MEAT_DIFFED_FILENAME = 'MEAT_diff.csv' MEAT_NEW_SORTED_FILENAME = 'MEAT_new_sorted.csv' MEAT_OLD_SORTED_FILENAME = 'MEAT_old_sorted.csv' MEAT_NEWLINE_FIX_FILENAME = 'MEAT_newline_fix.csv' # embedded newlines removed MEAT_CLEAN_FILENAME = 'MEAT_clean.csv' # final cleansed file truncate_mapping_table_query = """ truncate table youtube_asset_map """ load_delivery_history_query = """ insert ignore into masters_delivery_history (isrc, tuid, upc, partner_id, delivery_date) select yam.isrc, yam.derived_tuid, yam.upc, 453, if(last_delivery_date = 'ERROR', '1970-01-01', last_delivery_date) from youtube_asset_map yam where yam.isrc is not null and yam.isrc != '' and yam.derived_tuid is not null and yam.derived_tuid != '' and yam.upc is not null and yam.upc != '' """ def get_meat_file(): child = pexpect.spawnu('sftp {}@{}'.format( config.FTP_MEAT_USER, config.FTP_MEAT_URL)) i = child.expect( ['The authenticity', '(?i)password']) if i == 0: child.sendline('yes') child.expect('(?i)password', timeout=config.FTP_TIMEOUT) child.sendline(config.FTP_MEAT_PASSWORD) else: child.sendline(config.FTP_MEAT_PASSWORD) child.expect('sftp> ', timeout=config.FTP_TIMEOUT) child.sendline('cd /') child.expect('sftp> ', timeout=config.FTP_TIMEOUT) child.sendline('get -f {}'.format(MEAT_RAW_FILENAME)) child.expect('sftp> ', timeout=config.FTP_TIMEOUT) child.sendline('cd /old') child.expect('sftp> ', timeout=config.FTP_TIMEOUT) child.sendline('get -f {}'.format(MEAT_RAW_OLD_FILENAME)) child.expect('sftp> ', timeout=config.FTP_TIMEOUT) child.sendline('rename /{} /old/{}'.format(MEAT_RAW_FILENAME, MEAT_RAW_OLD_FILENAME)) child.expect('sftp> ', timeout=config.FTP_TIMEOUT) child.sendline('exit') sys.stdout.write(child.after) sys.stdout.flush() child.interact() if download: print('downloading MEAT file') get_meat_file() print('finished downloading MEAT file') if not backfill: sort_new_meat = 'sort {} > {}'.format(MEAT_RAW_FILENAME, MEAT_NEW_SORTED_FILENAME) sort_old_meat = 'sort {} > {}'.format(MEAT_RAW_OLD_FILENAME, MEAT_OLD_SORTED_FILENAME) # sort files print('sorting files') sort_new_meat_process = sp.Popen(sort_new_meat, shell=True, stdout=sp.PIPE) sort_new_meat_process.wait() sort_old_meat_process = sp.Popen(sort_old_meat, shell=True, stdout=sp.PIPE) sort_old_meat_process.wait() print('finished sorting files') with open(MEAT_DIFFED_FILENAME, 'w') as diff_result: wtr = csv.writer(diff_result) wtr.writerow( ['asset_id', 'owner_name', 'asset_type', 'title', 'artist', 'isrc', 'video_isrc', 'album', 'upc', 'grid', 'label', 'custom_id']) # take diff of files diff_command = "comm -2 -3 {} {} >> {}".format( MEAT_NEW_SORTED_FILENAME, MEAT_OLD_SORTED_FILENAME, MEAT_DIFFED_FILENAME) print('creating diff file with only new lines in MEAT') diff_meat_process = sp.Popen(diff_command, shell=True, stdout=sp.PIPE) diff_meat_process.wait() print('finished creating diff file with only new lines in MEAT') # strip new lines (should move this to python later) prep_meat_command = 'awk -f prep_MEAT.awk {} > {}'.format( MEAT_DIFFED_FILENAME, MEAT_NEWLINE_FIX_FILENAME) print('stripping new lines from MEAT') prep_meat_process = sp.Popen(prep_meat_command, shell=True, stdout=sp.PIPE) prep_meat_process.wait() print('finished stripping new lines from MEAT') # remove unneccessary columns from MEAT file (including columns with bad data) print('removing columns from MEAT') with open(MEAT_NEWLINE_FIX_FILENAME, 'r') as source: rdr = csv.reader(source) with open(MEAT_CLEAN_FILENAME, 'w') as result: wtr = csv.writer(result) for r in rdr: if(len(r) == 12): wtr.writerow((r[0], r[2], r[5], r[8], r[11])) print('finished removing columns from MEAT') else: # strip new lines (should move this to python later) prep_meat_command = 'awk -f prep_MEAT.awk {} > {}'.format( MEAT_RAW_OLD_FILENAME, MEAT_NEWLINE_FIX_FILENAME) print('stripping new lines from MEAT') prep_meat_process = sp.Popen(prep_meat_command, shell=True, stdout=sp.PIPE) prep_meat_process.wait() print('finished stripping new lines from MEAT') # remove unneccessary columns from MEAT file (including columns with bad data) print('removing columns from MEAT') with open(MEAT_NEWLINE_FIX_FILENAME, 'r') as source: rdr = csv.reader(source) with open(MEAT_CLEAN_FILENAME, 'w') as result: wtr = csv.writer(result) for r in rdr: if (len(r) == 12): wtr.writerow((r[0], r[2], r[5], r[8], r[11])) print('finished removing columns from MEAT') # final meat prepping (to remove sound recordings, etc) print('prep MEAT for the mapper') prep_files_command = "/bin/sh prep_files.sh {}".format(MEAT_CLEAN_FILENAME) prep_files_process = sp.Popen(prep_files_command, shell=True, stdout=sp.PIPE) prep_files_process.wait() print('finished prepping MEAT') if full and not backfill: print('truncating mapping table') with mysql.mr_session_scope() as session: session.execute(truncate_mapping_table_query) print('feeding prepped MEAT to the mapper') map_meat_command = "python map_assets_to_tracks.py -f MEAT_clean.csv" map_meat_process = sp.Popen(map_meat_command, shell=True, stdout=sp.PIPE) map_meat_process.wait() print('finished mapping MEAT') # load mapped meat into delivery history table print('loading mapped MEAT into delivery history table') session.execute(load_delivery_history_query) print('finished loading MEAT') return MEAT_CLEAN_FILENAME if __name__ == '__main__': parser = argparse.ArgumentParser() parser.add_argument('-b', '--backfill', default=0) args = parser.parse_args() load_meat(full=True, backfill=bool(args.backfill))