"""Tasks of Source of Stream pre-aggregation workflow.""" import datetime from datetime import timedelta from garcon import task from garcon.contrib.dynamo_feed_status import \ feed_status_ingestion as feed_status from analytics_aggregation import base_config from analytics_aggregation.flows.spotify_sos import utils from analytics_aggregation.flows.spotify_sos.snowflake_executor \ import SpotifySOSExecutor from analytics_aggregation.util import common as common_utils @task.decorate(timeout=600) def extract_date( activity, feed_name, context_date_range=None, reload=False, labelids=None): """Extract date either from passed context, either from ETLs' statuses. Args: activity (ActivityWorker): The activity worker. feed_name (str): Name of the feed. context_date_range (str|None): Date range from context (context_date). reload (bool|str): Flag to force reload and ignore all the statuses. labelids (list | str | int): List of label ids. Optional. If is not None then the whole flow will be run only for these label ids. Can be list of integers or comma-separated ids or single int value. """ reload = common_utils.get_bool_from_flag(reload) if not context_date_range and reload: return common_utils.exit_message( "'reload' flag passed but date range is not specified.") if context_date_range and not reload: return common_utils.exit_message( "'explicit_date_range' specified but 'reload' flag omitted.") if context_date_range: start_date, end_date = context_date_range.split('_') else: range_result = _find_range_to_process(feed_name) if not range_result['found_days']: return common_utils.exit_message( 'There is no data to aggregate for {} range'.format( range_result['week_range'])) start_date = range_result['start_date'] end_date = range_result['end_date'] activity.logger.info( 'Start the workflow for {start_date} - {end_date}'.format( start_date=start_date, end_date=end_date)) context = { 'reload': reload, 'date_range_as_str': start_date + '_' + end_date, 'date_range': { 'start_date': start_date, 'end_date': end_date}} if labelids: try: labelids_validated = common_utils.format_id_list(labelids) except common_utils.InvalidIntListException: raise ValueError('Incorrect label ids. Must be list of integers.') context['labelids'] = labelids_validated return context def _find_range_to_process(feed_name): """Determine range of dates that need processing within past week. Args: feed_name (str): Name of the feed. Returns: dict: { 'found_days': Boolean, 'week_range': String date range for past week, 'start_date': First date that needs processing, 'end_date': Last date that needs processing } """ result = {'found_days': False} today = datetime.date.today().strftime('%Y-%m-%d') today_minus_seven_days = ( datetime.date.today() - timedelta(days=7)).strftime('%Y-%m-%d') result['week_range'] = today_minus_seven_days + '_' + today day_range = [] for day in common_utils.list_of_dates(today_minus_seven_days, today): is_spotify_ingested = utils.is_spotify_ingested(day) is_day_processed = feed_status.get_overall_status( feed_name, day) == feed_status.STATUS_INGESTED if is_spotify_ingested and not is_day_processed: result['found_days'] = True day_range.append(day) if result['found_days']: result['start_date'] = day_range[0] result['end_date'] = day_range[-1] return result @task.decorate(timeout=600) def check_spotify_ingested(activity, date_range, reload): """Check if Spotify report is already ingested. This is required for reload runs only (when context date was explicitly passed). (To fact_analytics, which means it's also was ingested to staging_raw_spotify_v2). Args: activity (ActivityWorker): The activity worker. date_range (dict): Dict with 2 keys start_date and end_date. reload (bool): Flag to force reload and ignore all the statuses. Returns: dict: If data is missing, return {'stop': True}. """ if reload: activity.logger.info( 'Checking if raw Spotify data available for Start the workflow for' ' {start_date} - {end_date}'.format( start_date=date_range['start_date'], end_date=date_range['end_date'])) dates_list = common_utils.list_of_dates( date_range['start_date'], date_range['end_date']) for day in dates_list: if not utils.is_spotify_ingested(day): return common_utils.exit_message( 'There is no Spotify data for {}'.format(day)) activity.logger.info('Raw Spotify data is available, proceed to run') @task.decorate(timeout=12600) def cleanup_staging_sos(activity, date_range, labelids=None): """Delete data from staging_sos for the given period. Args: activity (ActivityWorker): The activity worker. date_range (dict): Dict with 2 keys start_date and end_date. labelids (list[int]): List of label ids. Optional. """ activity.logger.info( 'Delete from staging_sos for {start_date} - {end_date}'.format( start_date=date_range['start_date'], end_date=date_range['end_date'])) labelids_clause, labelids_values = common_utils.sos_labelid_filter( labelids) with SpotifySOSExecutor( sf_config=base_config.SNOWFLAKE_CONFIG) as executor: executor.cleanup_staging_sos( params={ 'db': base_config.SNOWFLAKE_CONFIG['db'], 'schema': base_config.SNOWFLAKE_CONFIG['schema'], 'start_date': date_range['start_date'], 'end_date': date_range['end_date'], 'labelids': labelids_values}, labelids_clause=labelids_clause) activity.logger.info('Cleanup of staging_sos complete') @task.decorate(timeout=21600) def populate_staging_with_spotify_data(activity, date_range, labelids=None): """Populate staging_sos table with Spotify data. Args: activity (ActivityWorker): The activity worker. date_range (dict): Dict with 2 keys start_date and end_date. labelids (list[int]): List of label ids. Optional. """ labelids_clause, labelids_values = common_utils.sos_labelid_filter( labelids, 'fai') with SpotifySOSExecutor( sf_config=base_config.SNOWFLAKE_CONFIG) as executor: executor.populate_staging_sos( params={ 'db': base_config.SNOWFLAKE_CONFIG['db'], 'schema': base_config.SNOWFLAKE_CONFIG['schema'], 'start_date': date_range['start_date'], 'end_date': date_range['end_date'], 'labelids': labelids_values}, labelids_clause=labelids_clause) activity.logger.info('Spotify data ingested to staging_sos.') @task.decorate(timeout=12600) def sanity_check_populate_staging_sos(activity, date_range, reload=False): """Sanity checks for the populate_staging_sos activity. Basically we compare overall number of streams for the current day to the overall number of streams week ago. Threshold is 35%: if the week difference is more than 35%, it's a good reason to check the data manually. Args: activity (ActivityWorker): The activity worker. date_range (dict): Dict with 2 keys start_date and end_date. reload (bool): Flag to force reload and ignore all the statuses. """ if reload: # skip if we're loading data manually return # skip if we're loading data for more than one day if date_range['start_date'] != date_range['end_date']: return with SpotifySOSExecutor( sf_config=base_config.SNOWFLAKE_CONFIG) as executor: current_row_count = executor.get_row_count_from_staging_sos( params={ 'db': base_config.SNOWFLAKE_CONFIG['db'], 'schema': base_config.SNOWFLAKE_CONFIG['schema'], 'date': date_range['end_date'], 'storeid': 286}) end_date_obj = datetime.datetime.strptime( date_range['end_date'], '%Y-%m-%d') end_date_minus_week = ( end_date_obj - timedelta(days=6)).strftime('%Y-%m-%d') with SpotifySOSExecutor( sf_config=base_config.SNOWFLAKE_CONFIG) as executor: week_ago_row_count = executor.get_row_count_from_staging_sos( params={ 'db': base_config.SNOWFLAKE_CONFIG['db'], 'schema': base_config.SNOWFLAKE_CONFIG['schema'], 'date': end_date_minus_week, 'storeid': 286}) abs_difference = abs(current_row_count - week_ago_row_count) if abs_difference < current_row_count * 0.35: activity.logger.info( 'Sanity check for the populate_staging_sos performed, ' 'everything is fine, abs_difference for Spotify is {}.'.format( abs_difference)) else: message = ( 'Sanity check for the populate_staging_sos performed, ' 'the number of streams is suspicious, please check manually! ' 'abs_difference for Spotify is {}.'.format(abs_difference)) utils.capture_warning(message, activity)