""" YouTube Asset Conflict Snowflake-only Data Ingestion Workflow tasks. Tasks to ingest data from the YouTube Asset Conflict report into Snowflake. """ from datetime import date as date_module from garcon import task from garcon_contrib.aws.utils import garcon_s3 from feed_ingestion.flows import helpers from feed_ingestion.flows.helpers import get_sf_config from feed_ingestion.flows.youtube_asset_conflict import config from feed_ingestion.flows.youtube_asset_conflict.models import ows_territories from feed_ingestion.flows.youtube_asset_conflict.snowflake_executor import \ YouTubeAssetConflictSFExecutor from feed_ingestion.tasks import bootstrap as reload from feed_ingestion.util import task_status @task.decorate(timeout=600) @reload.block_non_sequential_then_reset_on_reload(config.feed_name) def bootstrap(activity, date, dw_config=None, stop=None): """Bootstrap workflow by getting the correct configurations. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). dw_config (str): dw_config stop (bool): passed in from the reload decorator. Returns: dict: Initial context of the workflow. """ activity.logger.info( 'Bootstrapping {feed_name}...'.format(feed_name=config.feed_name)) # date is the date passed in or today's date date = date or date_module.today().strftime('%Y-%m-%d') drop_file_names = [] staging_raw_temp_tables = [] for account in config.cms_dict.values(): file_name = config.source_filename_template.format(account=account) temp_table_name = config.snowflake_table_names.get( 'staging_raw_temp').format( account=account.lower(), datestamp=date.replace('-', '')) drop_file_names.append(file_name) staging_raw_temp_tables.append(( temp_table_name, file_name, account.lower())) return { 'date': date, 'date_as_in_uuid': date.replace('-', ''), 'feed_name': config.feed_name, 'report_name': config.youtube_report_full_name, 'report_date': date, 'credentials_path': config.credentials_path, 'api_service_name': config.youtube_reporting_api_service_name, 'api_version': config.youtube_reporting_api_version, 'jobs_meta_path': config.jobs_meta_path, 'cms_dict': config.cms_dict, 's3_dir_path': config.s3_dir_path.format(datestamp=date), 's3_feed_name': config.s3_feed_name, 's3_archive_bucket': config.s3_archive_bucket, 's3_archive_path': config.s3.get('archive').format( datestamp=date, s3_feed_name=config.s3_feed_name, s3_bucket=config.s3_archive_bucket), 's3_staging_raw_temp': config.s3.get( 'staging_raw_temp').format( datestamp=date, s3_feed_name=config.s3_feed_name, s3_bucket=config.s3_archive_bucket), 'drop_file_names': drop_file_names, 'art_relations_db': config.art_relations_db, 'art_relations_schema': config.art_relations_schema, 'asset_report_schema': config.asset_report_schema, 'registry_schema': config.registry_schema, 'staging_raw_temp_table_relationships': staging_raw_temp_tables, 'staging_raw_table': config.snowflake_table_names[ 'staging_raw'], 'fact_conflict_table': config.snowflake_table_names['fact_conflict'], 'youtube_asset_conflict_by_territory_temp_table': config.snowflake_table_names[ 'youtube_asset_conflict_by_territory_temp'].format( datestamp=date.replace('-', '')), 'fact_conflict_temp_table': config.snowflake_table_names.get( 'fact_conflict_temp').format(datestamp=date.replace('-', '')), 'staging_raw_youtube_asset_report_table': config.snowflake_table_names.get( 'staging_raw_youtube_asset_report'), 'registry_table': config.snowflake_table_names.get('registry'), 'track_table': config.snowflake_table_names.get('track'), 'releases_table': config.snowflake_table_names.get('releases'), 'artist_info_table': config.snowflake_table_names.get('artist_info'), 'territory_standard': config.territory_standard, 'sound_recording_asset_type': config.sound_recording_asset_type, 'time_zone': config.time_zone, 'orchard_account': config.accounts['ORCHARD']['file_label'], 'replace_archive_files': config.replace_archive_files, 'stop': stop, } @task.decorate(timeout=1800) def create_temp_staging_table(activity, date, staging_raw_temp_table): """Task to create temp staging table. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). staging_raw_temp_table (str): temp staging raw table name. """ task_name = 'create_temp_staging_table_{staging_raw_temp_table}'.format( staging_raw_temp_table=staging_raw_temp_table) # check task status if task_status.is_completed_task(config.feed_name, date, task_name): return activity.logger.info( 'Creating temp staging raw table {staging_raw_temp_table}'.format( staging_raw_temp_table=staging_raw_temp_table)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.create_staging_raw_temp_table(staging_raw_temp_table) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def load_temp_staging_table( activity, date, staging_raw_temp_table, key_dir, drop_file): """Task to load temp staging table from S3 file. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). staging_raw_temp_table (str): temp staging raw table name. key_dir (str): s3 path to load from. drop_file (str): file name of report on S3. """ task_name = 'load_temp_staging_table_{staging_raw_temp_table}'.format( staging_raw_temp_table=staging_raw_temp_table) # check task status if task_status.is_completed_task( config.feed_name, date, task_name): return drop_file_dir = '{key_dir}{drop_file}'.format( key_dir=key_dir, drop_file=drop_file) activity.logger.info( 'Loading temp staging raw table \ {staging_raw_temp_table} from {drop_file_dir}'.format( staging_raw_temp_table=staging_raw_temp_table, drop_file_dir=drop_file_dir)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.load_staging_raw_temp_table( staging_raw_temp_table, drop_file_dir) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def truncate_staging_raw_table(activity, date, staging_raw_table): """Truncate staging_raw_youtube_asset_conflict. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). staging_raw_table (str): temp staging raw table """ task_name = 'truncate_staging_raw_table' # check task status if task_status.is_completed_task( config.feed_name, date, task_name): return activity.logger.info( 'Truncate staging_raw_table from {staging_raw_table}'.format( staging_raw_table=staging_raw_table)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.truncate_table(staging_raw_table) task_status.mark_completed_task( config.feed_name, date, 'truncate_staging_raw_table') @task.decorate(timeout=1800) def insert_into_staging_raw_table( activity, date, staging_raw_temp_table, drop_file, s3_staging_raw_temp, staging_raw_table, content_owner): """Insert into staging_raw_youtube_asset_conflict. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). staging_raw_temp_table (str): temp staging raw table. drop_file (str): file name that was used to populate staging_raw_temp. s3_staging_raw_temp (str): S3 file location of reports. staging_raw_table (str): staging raw table name. content_owner (str): content owner of the report. """ task_name = 'insert_into_staging_raw_table_from_' \ '{staging_raw_temp_table}'.format( staging_raw_temp_table=staging_raw_temp_table) # check task status if task_status.is_completed_task( config.feed_name, date, task_name): return activity.logger.info( 'Inserting staging raw table from {staging_raw_temp_table}'.format( staging_raw_temp_table=staging_raw_temp_table)) s3_drop_file_location = '{dir}{file}'.format( dir=s3_staging_raw_temp, file=drop_file) source_bucket_name, source_key_name = garcon_s3.extract_bucket_path( s3_drop_file_location) file_size = helpers.get_filesize( source_bucket_name, source_key_name, config.expected_bucket_owner ) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.insert_into_staging_raw_table( staging_raw_temp_table, staging_raw_table, drop_file, str(file_size), date, content_owner) task_status.mark_completed_task( config.feed_name, date, task_name) @task.decorate(timeout=1800) def create_fact_conflict_temp_table(activity, date, fact_conflict_temp_table): """Task to create fact_conflict_temp table. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). fact_conflict_temp_table (str): fact_conflict_temp table name. """ task_name = 'create_{fact_conflict_temp_table}'.format( fact_conflict_temp_table=fact_conflict_temp_table) # check task status if task_status.is_completed_task(config.feed_name, date, task_name): return activity.logger.info( 'Creating fact_conflict_temp table {fact_conflict_temp_table}'.format( fact_conflict_temp_table=fact_conflict_temp_table)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.create_fact_conflict_temp_table(fact_conflict_temp_table) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def insert_overwrite_into_fact_conflict_temp_table( activity, date, art_relations_db, art_relations_schema, asset_report_schema, registry_schema, fact_conflict_temp_table, yact_table, staging_raw_youtube_asset_report_table, registry_table, track_table, releases_table, artist_info_table, territory_standard, asset_type, time_zone, orchard_account): """Insert overwrite into fact_conflict_temp table. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). art_relations_db (str): Name of art_relations database in Snowflake. art_relations_schema (str): Schema name used in art_relations database. asset_report_schema (str): Schema name used for asset report table. registry_schema (str): Schema name used for registry. fact_conflict_temp_table (str): Name of fact_conflict_temp table. yact_table (str): Name of youtube_asset_conflict_by_territory_temp table. asset_report_table (str): Name of staging_raw_youtube_asset_report table. registry_table (str): Name of registry table. track_table (str): Name of track table. releases_table (str): Name of releases table. artist_info_table (str): Name of artist info table territory_standard (str): Territory standard that the conflict is in. asset_type (str): YouTube asset type. time_zone (str): System specified time zone. orchard_account (str): YT CMS account name for The Orchard Music. """ task_name = 'insert overwrite into {fact_conflict_temp_table}'.format( fact_conflict_temp_table=fact_conflict_temp_table) # check task status if task_status.is_completed_task(config.feed_name, date, task_name): return activity.logger.info( 'Insert overwrite into {fact_conflict_temp_table}'.format( fact_conflict_temp_table=fact_conflict_temp_table)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.insert_overwrite_into_fact_conflict_temp_table( art_relations_db, art_relations_schema, asset_report_schema, registry_schema, fact_conflict_temp_table, yact_table, staging_raw_youtube_asset_report_table, registry_table, track_table, releases_table, artist_info_table, territory_standard, asset_type, time_zone, orchard_account) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def check_size_of_fact_conflict_temp_table( activity, date, fact_conflict_temp_table): """Task to check size of fact conflict temp table and exit if too small. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). fact_conflict_temp_table (str): Name of fact_conflict_temp table. """ task_name = 'check_size_of_fact_conflict_temp_table' if task_status.is_completed_task( config.feed_name, date, task_name): return activity.logger.info('Check size of fact conflict temp table') sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: num_rows = sf_executor.check_size_of_fact_conflict_temp_table( fact_conflict_temp_table) activity.logger.info( 'Number of rows in fact conflict temp: {num_rows}'.format( num_rows=num_rows)) if int(num_rows) <= config.fact_conflict_temp_size_exception_threshold: raise Exception( 'Not enough data in the fact conflict temp table, exiting') task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def check_number_of_unresolved_conflicts( activity, date, fact_conflict_table): """Task to check number of unresolved conflicts and exit if too small. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). fact_conflict_table (str): Name of fact_conflict table. """ task_name = 'check_number_of_unresolved_conflicts' if task_status.is_completed_task( config.feed_name, date, task_name): return activity.logger.info('Check number of unresolved conflicts') sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: num_rows = sf_executor.check_number_of_unresolved_conflicts( fact_conflict_table) activity.logger.info( 'Number of unresolved conflicts: {num_rows}'.format( num_rows=num_rows)) if int(num_rows) <= config.unresolved_conflicts_exception_threshold: raise Exception( 'Too few unresolved conflicts in fact conflict table, exiting') task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def drop_create_territories_temp_table(activity, date): """Task to drop and create temp territories table. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). """ task_name = 'drop_create_territories_temp_table' if task_status.is_completed_task( config.feed_name, date, task_name): return activity.logger.info('Drop and create territories_temp table') territories_temp_table = config.snowflake_table_names['territories_temp'] sf_config = get_sf_config(config.feed_name) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.drop_table(territories_temp_table) sf_executor.create_territories_temp_table(territories_temp_table) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def fill_territories_temp_table(activity, date, standard): """Task to fill temp territories table with data from ows-territories. Args: activity (ActivityWorker): The activity worker. date (str): Reporting date (YYYY-MM-DD). standard (str): Territories standard """ task_name = 'fill_territories_temp_table' if task_status.is_completed_task( config.feed_name, date, task_name): return activity.logger.info('Fill territories_temp table') territories_temp_table = config.snowflake_table_names['territories_temp'] territories_response = ows_territories.get_territories(standard) if not territories_response: activity.logger.error(territories_response.errors) raise Exception(str(territories_response.errors)) sf_config = get_sf_config(config.feed_name) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.fill_territories_temp_table( territories_temp_table, territories_response.message['items']) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def create_youtube_asset_conflict_by_territory_table( activity, date, yact_table): """Task to create youtube_asset_conflict_by_territory_temp table. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). yact_table (str): youtube_asset_conflict_by_territory_temp table name. """ task_name = 'create_youtube_asset_conflict_by_territory_table' # check task status if task_status.is_completed_task(config.feed_name, date, task_name): return activity.logger.info( 'Creating {yact_table} table'.format(yact_table=yact_table)) sf_config = get_sf_config(config.feed_name) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.create_youtube_asset_conflict_by_territory_table( yact_table) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def fill_youtube_asset_conflict_by_territory_table( activity, date, yact_table, staging_raw_table): """Task to create youtube_asset_conflict_by_territory_temp table. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). yact_table (str): youtube_asset_conflict_by_territory table name. staging_raw_table (str): staging raw table name. """ task_name = 'fill_youtube_asset_conflict_by_territory_table' # check task status if task_status.is_completed_task(config.feed_name, date, task_name): return activity.logger.info( 'Filling {yact_table} table'.format(yact_table=yact_table)) territories_temp_table = config.snowflake_table_names['territories_temp'] sf_config = get_sf_config(config.feed_name) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.fill_youtube_asset_conflict_by_territory_table( yact_table, staging_raw_table, territories_temp_table) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def insert_into_fact_conflict_table( activity, date, fact_conflict_table, fact_conflict_temp_table): """Task to insert into fact_conflict table. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). fact_conflict_table (str): fact_conflict table name. fact_conflict_temp_table (str): fact_conflict_temp table name. """ task_name = 'insert_into_fact_conflict_table' # check task status if task_status.is_completed_task(config.feed_name, date, task_name): return activity.logger.info( 'Inserting into {fact_conflict_table} table'.format( fact_conflict_table=fact_conflict_table)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.insert_into_fact_conflict_table( fact_conflict_table, fact_conflict_temp_table) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def update_fact_conflict_yt_recent_daily_average( activity, date, fact_conflict_table, fact_conflict_temp_table): """Task to update fact_conflict_table yt_recent_daily_average column. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). fact_conflict_table (str): fact_conflict table name. fact_conflict_temp_table (str): fact_conflict_temp table name. """ task_name = 'update_fact_conflict_yt_recent_daily_average' # check task status if task_status.is_completed_task(config.feed_name, date, task_name): return activity.logger.info( 'Updating yt_recent_daily_average in {fact_conflict_table} table' .format(fact_conflict_table=fact_conflict_table)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.update_fact_conflict_yt_recent_daily_average( fact_conflict_table, fact_conflict_temp_table) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def update_fact_conflict_views_in_conflict( activity, date, fact_conflict_table, fact_conflict_temp_table): """Task to update fact_conflict_table views_in_conflict column. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). fact_conflict_table (str): fact_conflict table name. fact_conflict_temp_table (str): fact_conflict_temp table name. """ task_name = 'update_fact_conflict_views_in_conflict' # check task status if task_status.is_completed_task(config.feed_name, date, task_name): return activity.logger.info( 'Updating views_in_conflict in {fact_conflict_table} table' .format(fact_conflict_table=fact_conflict_table)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.update_fact_conflict_views_in_conflict( fact_conflict_table, fact_conflict_temp_table) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1800) def update_fact_conflict_resolved_datetime( activity, date, fact_conflict_table, fact_conflict_temp_table, time_zone): """Task to update fact_conflict_table resolved_datetime column. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). fact_conflict_table (str): fact_conflict table name. fact_conflict_temp_table (str): fact_conflict_temp table name. time_zone (str): System specified time zone. """ task_name = 'update_fact_conflict_resolved_datetime' # check task status if task_status.is_completed_task(config.feed_name, date, task_name): return activity.logger.info( 'Updating resolved_datetime in {fact_conflict_table} table'.format( fact_conflict_table=fact_conflict_table)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.update_fact_conflict_resolved_datetime( fact_conflict_table, fact_conflict_temp_table, time_zone) task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=7200) def reset_es_indexed_for_partially_resolved_conflicts( activity, date, fact_conflict_table, time_zone): """Task to reset es_indexed to false for partially resolved conflicts. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). fact_conflict_table (str): fact_conflict table name. time_zone (str): System specified time zone. """ task_name = 'reset_es_indexed_for_partially_resolved_conflicts' # check task status if task_status.is_completed_task(config.feed_name, date, task_name): return activity.logger.info( 'Resetting es_indexed to false in for partially resolved' ' conflicts in {fact_conflict_table} table'.format( fact_conflict_table=fact_conflict_table)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.reset_es_indexed_for_partially_resolved_conflicts( fact_conflict_table, time_zone) # update task's status task_status.mark_completed_task(config.feed_name, date, task_name) @task.decorate(timeout=1200) def drop_temp_table(activity, date, table_name): """Drop temporary table. Args: activity (ActivityWorker): the activity worker. date (str): Reporting date (YYYY-MM-DD). table_name (str): Name of table to drop. """ task_name = 'drop_temp_table_{}'.format(table_name) activity.logger.info('Dropping temp table {}'.format(table_name)) sf_config = get_sf_config(config.secrets_path) with YouTubeAssetConflictSFExecutor(sf_config) as sf_executor: sf_executor.drop_table(table_name) task_status.mark_completed_task(config.feed_name, date, task_name)