import argparse import boto3 import json from sqlalchemy.sql import text import sys from fpsweeper import config from fpsweeper.connector import sentry from fpsweeper.connector.mysql import ar_session from fpsweeper.connector.mysql import dd_session from fpsweeper.connector.mysql import fpc_session from fpsweeper.connector.mysql import db_session_scope from fpsweeper.context import context from fpsweeper.model import fp_message from fpsweeper.model import sweeper_persister from fpsweeper.model.query.direct_delivery import QUERY_AUDIO_ASSET_UPC_FILENAMES from fpsweeper.model.query.track import (QUERY_TRACK_TUID_BACKFILL, QUERY_TRACK_TUID_DETAILS, QUERY_TRACK_TUID_SWEEPER) from fpsweeper.model.query.track import QUERY_TRACK_TUID_VENDOR from fpsweeper.model.query.track_fp import QUERY_TRACK_FP_TUID from fpsweeper.logic import util from fpsweeper.logic import parse_folder_structure def get_track_filename(db_session_direct_delivery, track_details): """Get filename for each tuid and create a dictionary of tuids that contains their upc, cd, track_id and filename Args: db_session_direct_delivery (sqlalchemy.orm.session.Session): db session object pointed to direct delivery db tuid_details (dict): tuid and its associated upc, cd and track_id Returns: asset_path (string): full path to the asset in direct delivery """ tuid, tuid_details = track_details upc = tuid_details['upc'] cd = tuid_details['cd'] track_id = tuid_details['track_id'] # get filenames for all tracks in upc, since we can't map filename directly to a track without parsing filename res = util.try_execute_query( db_session=db_session_direct_delivery, clause=QUERY_AUDIO_ASSET_UPC_FILENAMES, params={'physical_location_id': config.PHYSICAL_LOCATION_ID, 'upc': upc} ) for asset_details in res: if asset_details['filename'] is None: # catch and log error that we were getting when running backfill from 2016-01-01 to 2016-01-27 error_msg = "Null filename retrieved from direct_delivery for tuid={};upc={};cd={},track_id={}" error_msg = error_msg.format(tuid, upc, cd, track_id) context.logger.error(error_msg) else: filename_short = asset_details['filename'].partition('.')[0] # first check if file is in content, should have filename format UPC_CD_TRACKID # or if file is not in content (status L), it should have filename format UPC_TUID if ((filename_short == str(upc) + '_' + str(cd) + '_' + str(track_id)) or (filename_short == str(upc) + '_' + str(tuid))): asset_path = get_track_asset_path(upc, cd, track_id, asset_details) return asset_path context.logger.error("Could not find filename for tuid: {}".format(tuid)) return None def get_track_asset_path(upc, cd, track_id, asset_details): """Construct asset path for a track. Args: upc (string): upc of a track cd (string): cd of a track track_id (string): track_id of a track asset_details (dict): asset info needed to construct the path Returns: asset_path (string) """ path_variables_dict = { 'asset_id': asset_details['asset_id'], 'upc': upc, 'cd': cd, 'track_id': track_id, 'clip_number': asset_details['clip_number'] } asset_path = '\\\\' + asset_details['ip'] + '\\' + asset_details['initial_folder'] asset_path += parse_folder_structure.process_string( asset_details['folder_structure'], path_variables_dict) asset_path += '\\' + str(asset_details['filename']) asset_path = asset_path.replace('/', '\\') return asset_path def get_missing_tuids(db_session_fingerprint_capture, expected_tuids): """Get tuids that are in expected_tuids but NOT in the fingerprint capture database. Args: db_session_fingerprint_capture (sqlalchemy.orm.session.Session): db session object pointed to fingerprint_capture db expected_tuids (frozenset): tuids that are expected to be in fingerprint_capture db Returns: missing_tuids (frozenset): set of missing tuids """ # if empty set passed in, return an empty set if len(expected_tuids) == 0: return frozenset() expected_tuid_str = ",".join([str(x) for x in expected_tuids]) # get tuids that are in track_fp table AND expected_tuids res = util.try_execute_query(db_session_fingerprint_capture, QUERY_TRACK_FP_TUID.format(expected_tuid_str)) track_fp_tuids = frozenset([row['tuid'] for row in res]) # return tuids that are in expected_tuids but NOT in track_fingerprint database missing_tuids = expected_tuids - track_fp_tuids return missing_tuids def get_tuid_population( db_session_art_relations, daterange_start_days, daterange_end_seconds, is_backfill, vendor_id_csv=None): """ Get the population of tuids from art_relations.track for given time range Args: db_session_art_relations (sqlalchemy.orm.session.Session): db session object pointed to art_relations db daterange_start_days (int): Number of days ago to start at daterange_end_seconds (int): Number of seconds ago to end at is_backfill (bool): if this is backfill process then True else False vendor_id_csv (str): single vendor_id or csv string of vendor_ids Returns: tuids (list of int): tuids in given time range """ if vendor_id_csv: # get tracks for vendor_id(s) tuids = [] for vendor_id in(vendor_id_csv.split(',')): params = { 'daterange_start_days': daterange_start_days, 'daterange_end_seconds': daterange_end_seconds, 'vendor_id': vendor_id} res = util.try_execute_query( db_session_art_relations, text(QUERY_TRACK_TUID_VENDOR), params) tuids.extend([row['tuid'] for row in res]) return tuids else: # backfill or sweeper params = {'daterange_start_days': daterange_start_days, 'daterange_end_seconds': daterange_end_seconds} query = QUERY_TRACK_TUID_BACKFILL if is_backfill else QUERY_TRACK_TUID_SWEEPER res = util.try_execute_query(db_session_art_relations, text(query), params) return [row['tuid'] for row in res] def get_tuid_details_dict(tuids, db_session_art_relations): """Map tuid to upc, cd, and track_id Args: tuids (frozenset): tuids to map to upc's db_session_art_relations (sqlalchemy.orm.session.Session): db session object pointed to art_relations db Returns: tuid_details_dict (dict): mapping of tuid to upc, cd and track_id """ tuid_str = ",".join([str(x) for x in tuids]) res = util.try_execute_query(db_session_art_relations, QUERY_TRACK_TUID_DETAILS.format(tuid_str)) tuid_details_dict = {} for row in res: tuid = row['tuid'] tuid_details_dict[tuid] = {'upc': row['upc'], 'cd': row['cd'], 'track_id': row['track_id']} return tuid_details_dict def send_messages_to_sqs(message_list): """Send messages to sqs in batches of SQS_MAX_BATCH_SIZE Args: message_list (list of dict): list of dictionaries to send """ sqs = boto3.resource('sqs') queue = sqs.get_queue_by_name(QueueName=config.sqs_queue_name) for sublist in util.yield_sublists(message_list, config.SQS_MAX_BATCH_SIZE): try: res = queue.send_messages(Entries=sublist) # according to boto3 documentation, sqs can contain failed items if 'Failed' in res: msg = "Failures detected in SQS response: {}".format(json.dumps(res)) raise Exception(msg) except: # catch all exceptions sentry.sentry_client.captureException() context.logger.error("Error sending to SQS: {}".format(sys.exc_info()[1])) raise def process_missing_tuids(tuids, db_session_art_relations, db_session_direct_delivery, track_source): """Build messages and send them to fingerprint queue Args: tuids (frozenset): tuids that were not fingerprinted db_session_art_relations (sqlalchemy.orm.session.Session): db session object pointed to art_relations db db_session_direct_delivery (sqlalchemy.orm.session.Session): db session object pointed to direct_delivery db track_source (str): either 'backfill' or 'sweeper' Returns: int: number of messages sent to sqs """ tuid_details_dict = get_tuid_details_dict(tuids, db_session_art_relations) context.logger.info("Retrieved {} upc's for {} tuids".format(len(tuid_details_dict), len(tuids))) message_list = [] for track_details in tuid_details_dict.items(): tuid, tuid_details = track_details upc = tuid_details['upc'] filename = get_track_filename(db_session_direct_delivery, track_details) # try to create this message, if you can't create it then log and skip if filename is not None: msg = fp_message.FpMessage( filename=filename, tuid=tuid, upc=upc, correlation_id=context.get_correlation_id(), track_source=track_source) message_list.append(msg.to_sqs_json()) send_messages_to_sqs(message_list) context.logger.info("Sent {} messages to SQS".format(len(message_list))) return len(message_list) def init_logger(): context.reset() context.set('correlation_id', util.create_correlation_id()) def main(): """Main entry point into sweeper process. """ logging_start_time = util.get_timestamp_utc_iso8601(include_zone_designator=False) # for logging args = parse_cl_args() init_logger() track_source = fp_message.TRACK_SOURCE_BACKFILL if args.backfill else fp_message.TRACK_SOURCE_SWEEPER context.logger.info("Running fp sweeper with daterange_start_days={};daterange_end_seconds={};backfill={}" .format(args.daterange_start_days, args.daterange_end_seconds, args.backfill)) # initialize some variables for logging this run to fingerprint_capture.fp_sweeper_log logging_daterange_start = util.get_date_utc_iso8601_days_ago(days_ago=args.daterange_start_days, include_zone_designator=False) logging_daterange_end = util.get_datetime_utc_iso8601_seconds_ago(seconds_ago=args.daterange_end_seconds, include_zone_designator=False) logging_num_tracks_added = 0 with db_session_scope(ar_session) as db_session_art_relations: expected_tuids = get_tuid_population(db_session_art_relations=db_session_art_relations, daterange_start_days=args.daterange_start_days, daterange_end_seconds=args.daterange_end_seconds, is_backfill=args.backfill, vendor_id_csv=args.vendor_id_csv) context.logger.info("Found {} tuids in given time range".format(len(expected_tuids))) for sublist in util.yield_sublists(expected_tuids, config.TUID_CHUNK_SIZE): expected_tuids_chunk = frozenset(sublist) with db_session_scope(fpc_session) as db_session_fingerprint_capture: missing_tuids = get_missing_tuids(db_session_fingerprint_capture=db_session_fingerprint_capture, expected_tuids=expected_tuids_chunk) context.logger.info("Found {} missing tuids out of {} in sublist" .format(len(missing_tuids), len(expected_tuids_chunk))) if len(missing_tuids) == 0: continue with db_session_scope(ar_session) as db_session_art_relations,\ db_session_scope(dd_session) as db_session_direct_delivery: logging_num_tracks_added += process_missing_tuids(tuids=missing_tuids, db_session_art_relations=db_session_art_relations, db_session_direct_delivery=db_session_direct_delivery, track_source=track_source) with db_session_scope(fpc_session) as db_session_fingerprint_capture: sweeper_persister.insert_fp_sweeper_log(db_session_fingerprint_capture=db_session_fingerprint_capture, datetime_start=logging_start_time, datetime_end=util.get_timestamp_utc_iso8601(include_zone_designator=False), daterange_start=logging_daterange_start, daterange_end=logging_daterange_end, num_tracks_checked=len(expected_tuids), num_tracks_queued=logging_num_tracks_added, status='complete', created_by='sweeper.py') def parse_cl_args(): """Parse the command-line arguments and return them Returns: args (argparse.Namespace): namespace containing command-line arguments """ desc = """Fingerprint Sweeper and Backfill Process. Finds un-fingerprinted tuids and pushes them to queue to be fingerprinted by downstream fingerprint-capture process. Example usage: python fpsweeper/logic/sweeper.py --daterange_start_days=15000 --daterange_end_seconds=3600 --backfill . Help: python fpsweeper/logic/sweeper.py -h""" parser = argparse.ArgumentParser(description=desc) parser.add_argument('--daterange_start_days', type=int, help='Number of days ago to start at in art_relations.track.last_updated. ' + 'For example, if now is 2016-01-14T11:18:00Z and daterange_start_days is 5, ' + 'this process will start at 2016-01-09T00:00:00Z.', required=True) parser.add_argument('--daterange_end_seconds', type=int, help='Number of seconds ago to end at in art_relations.track.last_updated. ' + 'For example, if now is 2016-01-14T11:18:00Z and daterange_end_seconds is 3600, ' + 'this process will start at 2016-01-14T10:18:00Z.', required=True) parser.add_argument('--backfill', action='store_true', required=False, help='Run the fingerprint backfill process, which has slightly different logic than ' + 'the regular sweeper process') parser.add_argument('--vendor_id_csv', type=str, required=False, help='optional single vendor_id or csv string of vendor_ids with no spaces (e.g. 1,2,3)') return parser.parse_args() if __name__ == "__main__": main()