"""set-product-to-complete.""" from ddex_ingester_common.constants.status import LABEL_PROCESSING from ddex_ingester_common.constants.swb_deal_types import FOR_DISTRIBUTION from ddex_ingester_common.helpers.rds import run_rds_query import config from config import secrets_manager_client from constants import constants, graphql_queries, sql_queries logger = config.get_current_logger(config.app_logger) def handler(event, context): """Begin main script execution function.""" logger.info( f'Begin Set Product to Complete. Event: {event} Context: {context}') nr_products_per_run = \ event.get('nr_products_per_run') or constants.NR_PRODUCTS_PER_RUN logger.info(f'nr_products_per_run: {nr_products_per_run}') try: # Get UPCs from RDS logger.info('Reading list of UPCs from RDS') rows_to_process = run_rds_query( logger, config.RDS_HOST, config.RDS_DB_NAME, config.RDS_RW_USER, secrets_manager_client.get_cred('rds_read_write_password'), sql_queries.SELECT_UPC_N_ROWS, nr_products_per_run, ) except Exception as err: err_string = str(err) logger.info(f'Failed to execute DB query: {err_string}') return if not rows_to_process: logger.info('No products to process. Ending set product to complete.') return all_upcs = [row.get('upc') for row in rows_to_process] logger.info(f'Found UPCs: {all_upcs}') upc_to_product_id_audio = {} upc_to_product_id_video = {} failed_upcs = {} # Retrieve and store productId for each upc for upc in all_upcs: try: result = config.graphql_gateway.execute( graphql_queries.GET_PRODUCT_BY_UPC, {'upc': upc} )['data']['productByUpc'] logger.info( f'Ran productByUpc for {upc} and got the result: {result}') if result: product_id = result.get('productId') vendor_id = result.get('vendorId') subaccount_id = result.get('subaccountId') if result.get('videoProduct'): logger.info(f'Running video approval logic for {upc}') upc_to_product_id_video[upc] = product_id # The submitProduct mutation sets a video to complete # Approve is only used if the video is in the submitted # state or we want to skip validation for NFD != N try: nfd = result.get('notForDistribution') if nfd and nfd != FOR_DISTRIBUTION: # If not set you get the error: # Context creation failed: Client name must be specified. # noqa: E501 config.graphql_gateway.set_headers({ 'apollographql-client-name': 'mobile-orchard' }) result = config.graphql_gateway.execute( graphql_queries.APPROVE_VIDEO_PRODUCT, {'data': { 'productId': product_id, 'approvalType': 'FINAL', 'bypassValidation': True }} )['data']['approveVideoSingleProduct'] logger.info( f'Ran approveVideoSingleProduct with bypassValidation ' # noqa: E501 f'for {upc} with ID {product_id} ' f'and got the result: {result}') elif result.get('status') == LABEL_PROCESSING: payload = { 'productId': product_id, 'vendorId': vendor_id, 'subaccountId': subaccount_id } result = config.graphql_gateway.execute( graphql_queries.SUBMIT_PRODUCT, payload )['data']['submitProduct'] logger.info( f'Ran submitProduct for {upc} with payload' f' {payload} and got the result: {result}') else: result = config.graphql_gateway.execute( graphql_queries.APPROVE_VIDEO_PRODUCT, {'data': { 'productId': product_id, 'approvalType': 'FINAL', }} )['data']['approveVideoSingleProduct'] logger.info( f'Ran approveVideoSingleProduct for {upc} with ID' # noqa: E501 f' {product_id} and got the result: {result}') except Exception as err: err_string = str(err) failed_upcs[upc] = err_string logger.info(f'approveVideo failed for UPC: {upc}: {err_string}') # noqa: E501 else: upc_to_product_id_audio[upc] = product_id else: logger.info( f'Product was not found. Setting {upc} date to 1971-01-01.') run_rds_query( logger, config.RDS_HOST, config.RDS_DB_NAME, config.RDS_RW_USER, secrets_manager_client.get_cred('rds_read_write_password'), sql_queries.UPDATE_ROW_PRODUCT_SET_TO_NOT_FOUND, int(upc), ) except Exception as err: err_string = str(err) logger.info(f'Error retrieving product: {upc}: {err_string}') if not upc_to_product_id_audio and not upc_to_product_id_video: logger.info( 'No product IDs found. Ending set product to complete lambda.') return for upc in upc_to_product_id_audio.keys(): try: product_id = int(upc_to_product_id_audio[upc]) result = config.graphql_gateway.execute( graphql_queries.APPROVE_AUDIO_PRODUCT, {'productId': product_id} )['data']['approveProduct'] logger.info(f'Ran approveProduct for {upc} with ID' f' {product_id} and got the result: {result}') upc_to_product_id_audio[upc] = result.get('productId') except Exception as err: err_string = str(err) failed_upcs[upc] = err_string logger.info(f'approveProduc failed for UPC: {upc}: {err_string}') audio_and_video_upcs = ( list(upc_to_product_id_audio.keys()) + list(upc_to_product_id_video.keys())) for upc in audio_and_video_upcs: try: if upc not in failed_upcs: logger.info(f'Updating RDS row for UPC {upc}') run_rds_query( logger, config.RDS_HOST, config.RDS_DB_NAME, config.RDS_RW_USER, secrets_manager_client.get_cred('rds_read_write_password'), sql_queries.UPDATE_ROW_PRODUCT_SET_TO_COMPLETE, int(upc), ) except Exception as err: err_string = str(err) logger.info(f'Failed to update DB rows: {err_string}') for upc, error_message in failed_upcs.items(): try: logger.info(f'Updating RDS row for failed UPC {upc}') run_rds_query( logger, config.RDS_HOST, config.RDS_DB_NAME, config.RDS_RW_USER, secrets_manager_client.get_cred('rds_read_write_password'), sql_queries.UPDATE_ROW_PRODUCT_FAILED_SET_TO_COMPLETE, (error_message, int(upc)), ) except Exception as err: err_string = str(err) logger.info(f'Failed to update DB rows: {err_string}')