"""Source of streams model.""" from datetime import date, datetime, timedelta from oto import response as oto_response from snowflake_connector import snowflake_conn from snowflake_connector.snowflake_conn import SQLLoader from analytics.consts import models as model_consts class Store: """Enumeration of supported stores.""" APPLE_MUSIC = '1' SPOTIFY = '286' AMAZON_UNLIMITED = '716' @classmethod def get_available_stores(cls): """Get available stores for current (sub)account. Returns: list[str]: available store ids. """ return [ cls.APPLE_MUSIC, cls.SPOTIFY, cls.AMAZON_UNLIMITED] STREAMS_VENDOR_TOTALS_KEY = '{account_type}:{vendor_id}' STREAMS_TRACK_TOTALS_KEY = '{account_type}:{vendor_id}:{isrc}' STREAMS_VENDOR_PLACEMENTS_KEY = ( '{account_type}:{vendor_id}:{transaction_type_id}:{store}:{period}') STREAMS_TRACK_PLACEMENTS_KEY = ( '{account_type}:{vendor_id}:{transaction_type_id}:{store}:{period}:{isrc}') STREAMS_VENDOR_PLACEMENTS_KEY_APPLE_MUSIC = ( '{account_type}:{vendor_id}:{transaction_type_id}:{period}') STREAMS_TRACK_PLACEMENTS_KEY_APPLE_MUSIC = ( '{account_type}:{vendor_id}:{transaction_type_id}:{period}:{isrc}') _sql_loader = SQLLoader('analytics/queries') def get_streams( *, start_date, end_date, labelid, subaccountid, storeids, isrcs, artistids): """Get the total streams based on parameters given. Using a snowflake connector, fetch the total streams for a given store, account, and date range. Args: start_date (str): start date (YYYY-MM-DD). end_date (str): end date (YYYY-MM-DD). labelid (int): labelid from dim_label table subaccountid (int) - subaccountid from dim_subaccount table storeids (list[str]): specific store ids. isrcs (list[str]): track ISRCs artistids (list[int]) - list of artist ids Returns: oto.Response: response containing list of streams data. """ query_params = { 'start_date': start_date, 'end_date': end_date, 'labelid': labelid, 'subaccountid': subaccountid, 'storeids': storeids, 'isrcs': isrcs, 'artistids': artistids} sql = _get_streams_sql(query_params) result_list = snowflake_conn.fetchall(sql, query_params) full_result_list = _add_missing_date_entries( result_list, start_date, end_date) return oto_response.Response( _format_query_result(full_result_list, model_consts.SOS_METRICS)) def get_streams_v2( *, start_date, end_date, labelid, subaccountid, storeids, isrcs, artistids): """Get the total streams based on parameters given. Using a snowflake connector, fetch the total streams for a given store, account, and date range. Args: start_date (str): start date (YYYY-MM-DD). end_date (str): end date (YYYY-MM-DD). labelid (int): labelid from dim_label table subaccountid (int) - subaccountid from dim_subaccount table storeids (list[str]): specific store ids. isrcs (list[str]): track ISRCs artistids (list[int]) - list of artist ids Returns: oto.Response: response containing list of streams data. """ query_params = { 'start_date': start_date, 'end_date': end_date, 'labelid': labelid, 'subaccountid': subaccountid, 'storeids': storeids, 'isrcs': isrcs, 'artistids': artistids} sql = _get_streams_sql_v2(query_params) result_list = snowflake_conn.fetchall(sql, query_params) full_result_list = _add_missing_date_entries_v2( result_list, start_date, end_date) return oto_response.Response( _format_query_result(full_result_list, model_consts.SOS_METRICS_V2)) def get_placements( *, start_date, end_date, labelid, subaccountid, storeids, isrcs, artistids, limit): """Get the top streams placements based on parameters given. Using a snowflake connector, fetch the total streams placements for a given store, account, and date range. Args: start_date (str): start date (YYYY-MM-DD). end_date (str): end date (YYYY-MM-DD). labelid (int): labelid from dim_label table subaccountid (int): subaccountid from dim_subaccount table. storeids (list[str]): specific store ids. isrcs (list[str]): track ISRCs. artistids (list[int]): list of artist ids. limit (int | None): number of placements to retrieve (value of None indicates to fetch all placements). Returns: oto.Response: response containing list of streams placements data. """ query_params = { 'start_date': start_date, 'end_date': end_date, 'labelid': labelid, 'subaccountid': subaccountid, 'storeids': storeids, 'isrcs': isrcs, 'artistids': artistids, 'limit': limit} sql_file = 'get_placements' column_names = model_consts.SOS_PLACEMENTS sql = build_placements_sql(sql_file, query_params) result_list = snowflake_conn.fetchall(sql, query_params) formatted_list = _format_query_result(result_list, column_names) for order_number, item in enumerate(formatted_list, 1): item['order_number'] = str(order_number) return oto_response.Response(formatted_list) def get_placement_totals( *, start_date, end_date, labelid, subaccountid, storeids, isrcs, artistids): """Get the total streams placements based on parameters given. Using a snowflake connector, fetch the total streams placements for a given store, account, and date range. Args: start_date (str): start date (YYYY-MM-DD). end_date (str): end date (YYYY-MM-DD). labelid (int): labelid from dim_label table subaccountid (int) - subaccountid from dim_subaccount table storeids (list[str]): specific store ids. isrcs (list[str]): track ISRCs artistids (list[int]) - list of artist ids Returns: oto.Response: response number of stream placements. """ query_params = { 'start_date': start_date, 'end_date': end_date, 'labelid': labelid, 'subaccountid': subaccountid, 'storeids': storeids, 'isrcs': isrcs, 'artistids': artistids} sql_file = 'get_placement_totals' sql = build_placements_sql(sql_file, query_params) result_item = snowflake_conn.fetchone(sql, query_params) if result_item is None or not all(result_item): return oto_response.Response(0) return oto_response.Response(result_item[0]) def _get_streams_sql(params=None): """Get streams SQL. Extrapolate SQL template with filtering conditions if required. Filter conditions contains SQL params which are filled in session.execute function call. Args: params (dict): should have the following keys: start_date (str): start date (YYYY-MM-DD) end_date (str): end date (YYYY-MM-DD) labelid (int): labelid from dim_label table subaccountid (int) - subaccountid from dim_subaccount table storeids (list[int]) - list of store_id from dim_store table isrcs (list[str]) - list of track isrc values artistids (list[int]) - list of artist ids Returns: str: streams SQL """ SUBACCOUNT_CLAUSE = 'AND subaccountid = :subaccountid' ISRCS_CLAUSE = 'AND isrc in (:isrcs)' STOREID_CLAUSE = 'AND storeid in (:storeids)' ARTISTID_CLAUSE = 'AND artistid in (:artistids)' raw_sql = _sql_loader.load_query('get_streams') return raw_sql.format( subaccount_clause=( SUBACCOUNT_CLAUSE if params.get('subaccountid') else ''), isrcs_clause=ISRCS_CLAUSE if params.get('isrcs') else '', storeid_clause=STOREID_CLAUSE if params.get('storeids') else '', artistids_clause=ARTISTID_CLAUSE if params.get('artistids') else '') def _get_streams_sql_v2(params=None): """Get streams SQL. Extrapolate SQL template with filtering conditions if required. Filter conditions contains SQL params which are filled in session.execute function call. Args: params (dict): should have the following keys: start_date (str): start date (YYYY-MM-DD) end_date (str): end date (YYYY-MM-DD) labelid (int): labelid from dim_label table subaccountid (int) - subaccountid from dim_subaccount table storeids (list[int]) - list of store_id from dim_store table isrcs (list[str]) - list of track isrc values artistids (list[int]) - list of artist ids Returns: str: streams SQL """ SUBACCOUNT_CLAUSE = 'AND subaccountid = :subaccountid' ISRCS_CLAUSE = 'AND isrc in (:isrcs)' STOREID_CLAUSE = 'AND storeid in (:storeids)' ARTISTID_CLAUSE = 'AND artistid in (:artistids)' raw_sql = _sql_loader.load_query('get_streams_v2') return raw_sql.format( subaccount_clause=( SUBACCOUNT_CLAUSE if params.get('subaccountid') else ''), isrcs_clause=ISRCS_CLAUSE if params.get('isrcs') else '', storeid_clause=STOREID_CLAUSE if params.get('storeids') else '', artistids_clause=ARTISTID_CLAUSE if params.get('artistids') else '') def build_placements_sql(sql_file, params=None): """Get streams placements SQL. Extrapolate SQL template with filtering conditions if required. Filter conditions contains SQL params which are filled in session.execute function call. Args: sql_file (str): the filename of the sql file. params (dict): should have the following keys: start_date (str): start date (YYYY-MM-DD) end_date (str): end date (YYYY-MM-DD) labelid (int): labelid from dim_label table subaccountid (int): subaccountid from dim_subaccount table storeids (list[int]): list of store_id from dim_store table isrcs (list[str]): list of track isrc values artistids (list[int]): list of artist ids limit (int | None): number of placements to retrieve (value of None indicates to fetch all placements) Returns: str: streams placements SQL """ SUBACCOUNT_CLAUSE = 'AND subaccountid = :subaccountid' ISRCS_CLAUSE = 'AND isrc in (:isrcs)' STOREID_CLAUSE = 'AND storeid in (:storeids)' ARTISTID_CLAUSE = 'AND artistid in (:artistids)' LIMIT_CLAUSE = 'LIMIT :limit' SUBACCOUNTID = 'subaccountid,' subaccountid = params.get('subaccountid') subaccount_clause_in_track_join = ( 'AND track_data.subaccountid = {subaccountid}'.format( subaccountid=subaccountid) if subaccountid else 'AND track_data.subaccountid IS NULL') raw_sql = _sql_loader.load_query(sql_file) return raw_sql.format( subaccountid=(SUBACCOUNTID if subaccountid else ''), subaccount_clause=(SUBACCOUNT_CLAUSE if subaccountid else ''), subaccount_clause_in_track_join=subaccount_clause_in_track_join, isrcs_clause=ISRCS_CLAUSE if params.get('isrcs') else '', storeid_clause=STOREID_CLAUSE if params.get('storeids') else '', artistids_clause=ARTISTID_CLAUSE if params.get('artistids') else '', limit_clause=LIMIT_CLAUSE if params.get('limit') is not None else '') def _format_query_result(query_result, dict_params): """Format query results into dictionary with properties. Args: query_result (list): A list containing tuples representing results. Returns: query_results: A formatted dictionary of results. """ results = [ dict(zip(dict_params, row)) for row in query_result] for data_row in results: for elem in data_row: if(isinstance(data_row[elem], date)): data_row[elem] = str(data_row[elem]) return results def _add_missing_date_entries(summary_list, start_date, end_date): """Add any missing dates from list of daily stream summaries. Return a list of daily summaries where any missing dates are included with zero for the values. Args: summary_list (list): list of daily summaries start_date (str): first date of query request end_date (str): last date of query request Returns: (list): list of daily summaries with missing dates added """ set_of_dates = [summary[0] for summary in summary_list] first_date = datetime.strptime(start_date, '%Y-%m-%d').date() last_date = datetime.strptime(end_date, '%Y-%m-%d').date() date_range = range((last_date - first_date).days + 1) full_date_set = set(first_date + timedelta(x) for x in date_range) missing_dates = sorted(full_date_set - set(set_of_dates)) missing_summaries = [(day, 0, 0, 0, 0) for day in missing_dates] return sorted(summary_list + missing_summaries) def _add_missing_date_entries_v2(summary_list, start_date, end_date): """Add any missing dates from list of daily stream summaries. Return a list of daily summaries where any missing dates are included with zero for the values. Args: summary_list (list): list of daily summaries start_date (str): first date of query request end_date (str): last date of query request Returns: (list): list of daily summaries with missing dates added """ set_of_dates = [summary[0] for summary in summary_list] first_date = datetime.strptime(start_date, '%Y-%m-%d').date() last_date = datetime.strptime(end_date, '%Y-%m-%d').date() date_range = range((last_date - first_date).days + 1) full_date_set = set(first_date + timedelta(x) for x in date_range) missing_dates = sorted(full_date_set - set(set_of_dates)) missing_summaries = [(day, 0, 0, 0, 0, 0, 0, 0) for day in missing_dates] return sorted(summary_list + missing_summaries)