"""Logic for retrieving streams.""" import itertools from collections import defaultdict from ddtrace import tracer from oto import response as oto_response from sound_recordings.logic import data_availability, permissions from sound_recordings.logic.parallel import parallel from sound_recordings.logic.store_outages import add_outage_error_to_stores from sound_recordings.models import streams as streams_model from sound_recordings.schemas.streams import ( AggregateStreamsSchema, StreamsAllSchema, StreamsSchema, ) from sound_recordings.utils import date as date_utils from sound_recordings.utils import store_availability from sound_recordings.utils import streams as streams_utils from sound_recordings.validation.schema import schema_dump def _get_total_streams_for_period(streams_list, start_date, end_date): """Get total streams for given date range. Args: streams_list (list): List of dictionaries with keys date (str): Date in YYYY-MM-DD format streams (int | None): Total streams for date start_date (datetime.date): Start date end_date (datetime.date): End date Returns: Integer representing total streams for range """ start_date_str = start_date.strftime("%Y-%m-%d") end_date_str = end_date.strftime("%Y-%m-%d") dates = [record["date"] for record in streams_list] streams = [record["streams"] if record["streams"] else 0 for record in streams_list] start_idx = dates.index(start_date_str) end_idx = dates.index(end_date_str) return sum(streams[start_idx : end_idx + 1]) def _get_time_period(start_date=None, end_date=None): """Determine time_period based on max_date or start and end_date. Args: start_date (datetime.date): Start date end_date (datetime.date): End date Returns: Time period """ if not (start_date and end_date): time_period, *_ = date_utils.get_prior_date_intervals( data_availability.get_max_available_date(), count=1, length=28 ) else: time_period, *_ = date_utils.get_prior_date_intervals( end_date, count=1, length=(end_date - start_date).days + 1 ) return time_period def _calc_total_skip_rate(streams_by_store): """Calculate skip rate from all stores. Args: streams_by_store (list): list of stores with total_skips Returns: Skip rate or None """ stores_with_skips = list( filter(lambda store: store["total_skips"] is not None, streams_by_store) ) if len(stores_with_skips) > 0: total_streams = sum( list( map(lambda store: store["total_streams_with_skips"], stores_with_skips) ) ) total_skips = sum( list(map(lambda store: store["total_skips"], stores_with_skips)) ) return total_skips / (total_skips + total_streams) if total_skips else 0 return None def _calc_total_store_skips(items): """Calculate total skips for store. Args: items (list): daily streams from store Returns: Skips or None """ items_with_skips = list( filter(lambda item: item and item["skips"] is not None, items) ) if len(items_with_skips) > 0: total_skips = sum(list(map(lambda item: item["skips"], items_with_skips))) return total_skips return None def _calc_total_store_streams_with_skips(items): """Calculate total streams with for store. Args: items (list): daily streams from store Returns: Streams where skips not null or None """ items_with_skips = list( filter(lambda item: item and item["skips"] is not None, items) ) if len(items_with_skips) > 0: total_streams = sum( list(map(lambda item: item["streams_with_skips"], items_with_skips)) ) return total_streams return None def _get_streams_by_store(streams_all_time, streams, time_period): """Format streams by store. Args: streams_all_time (list): all time streams result streams (list): streams result time_period (dict): time period Returns: Formatted streams by store """ store_names = store_availability.get_store_names() streams_all_time_by_store = {e["store_id"]: e for e in streams_all_time} all_store_ids = list(set([record["store_id"] for record in streams_all_time])) active_store_ids = set(filter(lambda e: e in store_names, all_store_ids)) streams_by_store = [] for store_id in active_store_ids: at = streams_all_time_by_store.get(store_id, {}) items = streams_utils.get_streams_for_store( streams, store_id, time_period["start_date"], time_period["end_date"] ) total_streams = sum(list(map(lambda item: item["streams"], items))) total_streams_with_skips = _calc_total_store_streams_with_skips(items) total_skips = _calc_total_store_skips(items) streams_by_store.append( { "id": store_id, "name": store_names.get(store_id), "all_time": at.get("all_time"), "growth_percentage": at.get("growth_percentage"), "skip_rate": total_skips / (total_skips + total_streams_with_skips) if total_skips else None, "total_streams": total_streams, "total_streams_with_skips": total_streams_with_skips, "total_skips": total_skips, "items": items, } ) return streams_by_store def _get_streams_totals(streams, time_period): """Format streams totals. Args: streams (list): streams result time_period (dict): time period Returns: Formatted streams totals """ streams_totals_by_date = { k.strftime("%Y-%m-%d"): { "streams": 0, "skips": None, "saves": None, "streams_with_skips": None, } for k, v in itertools.groupby(streams, lambda e: e["date"]) } for item in streams: date_key = item["date"].strftime("%Y-%m-%d") streams_totals_by_date[date_key]["streams"] += item["streams"] if streams_totals_by_date[date_key]["streams_with_skips"] is None: streams_totals_by_date[date_key]["streams_with_skips"] = item[ "streams_with_skips" ] else: streams_totals_by_date[date_key]["streams_with_skips"] += ( 0 if item["streams_with_skips"] is None else item["streams_with_skips"] ) if streams_totals_by_date[date_key]["skips"] is None: streams_totals_by_date[date_key]["skips"] = item["skips"] else: streams_totals_by_date[date_key]["skips"] += ( 0 if item["skips"] is None else item["skips"] ) if streams_totals_by_date[date_key]["saves"] is None: streams_totals_by_date[date_key]["saves"] = item["saves"] else: streams_totals_by_date[date_key]["saves"] += ( 0 if item["saves"] is None else item["saves"] ) streams_totals = streams_utils.get_empty_streams_for_period( time_period["start_date"], time_period["end_date"] ) for item in streams_totals: item["streams"] = ( streams_totals_by_date.get(item["date"])["streams"] if streams_totals_by_date.get(item["date"]) else 0 ) item["skips"] = ( streams_totals_by_date.get(item["date"])["skips"] if streams_totals_by_date.get(item["date"]) else None ) item["saves"] = ( streams_totals_by_date.get(item["date"])["saves"] if streams_totals_by_date.get(item["date"]) else None ) item["skip_rate"] = streams_utils.calc_date_skip_rate( streams_totals_by_date.get(item["date"]) ) return streams_totals def _add_stream_all_time_data( isrc, start_date, end_date, streams, streams_all_time, response_body ): if not streams_all_time: return response_body response_body["aggregate"]["all_time"] = sum( map(lambda e: e["all_time"], streams_all_time) ) time_period = _get_time_period(start_date, end_date) streams_by_store = _get_streams_by_store(streams_all_time, streams, time_period) response_body["stores"] = streams_by_store streams_totals = _get_streams_totals(streams, time_period) curr_week = sum(map(lambda e: e["streams"], streams_totals[-7:])) prev_week = sum(map(lambda e: e["streams"], streams_totals[-14:-7])) old_enough = response_body["aggregate"]["all_time"] > curr_week + prev_week show_growth = prev_week > 0 and old_enough response_body["aggregate"].update( { "items": streams_totals, "skip_rate": _calc_total_skip_rate(streams_by_store), "growth_percentage": round((curr_week - prev_week) / prev_week, 4) if show_growth else None, } ) return response_body @tracer.wrap(name="get_streams_bulk") def get_streams_bulk( request_context, isrcs, distributors, countries=[], store_ids=[], start_date=None, end_date=None, ): """Return streams for ISRC for given account type and ID. Args: request_context (RequestContext): RequestContext class isrc (str): ISRC of track to fetch streams for distributors (str[]): List of distributors names countries (str[]): List of country codes to filter by store_ids (list): List of store ids to filter by start_date (datetime.date): Start date end_date (datetime.date): End date Returns: oto.response.Response with streams payload """ permissions_filter = permissions.get_permissions_filter(request_context) if not (start_date and end_date): start_date, end_date = data_availability.get_date_range( data_availability.HIGHWATERMARK_DATE, days=28 ) requests = { "streams": { "func": streams_model.get_streams_bulk, "args": ( permissions_filter, isrcs, distributors, countries, store_ids, start_date, end_date, ), }, "streams_all_time": { "func": streams_model.get_streams_all_time_bulk, "args": (permissions_filter, isrcs, distributors, countries, store_ids), }, "sources": { "func": add_outage_error_to_stores, "args": (store_availability.get_sources(),), }, } result = parallel(requests) streams = result.message["streams"] streams_all_time = result.message["streams_all_time"] sources = result.message["sources"] streams_by_isrc = defaultdict(list) for s in streams: streams_by_isrc[s["isrc"]].append(s) streams_all_time_by_isrc = defaultdict(list) for s in streams_all_time: streams_all_time_by_isrc[s["isrc"]].append(s) result = {} for isrc in isrcs: # an empty response, but with sources empty_response_body = { "isrc": isrc, "stores": [], "sources": sources, "aggregate": {"items": [], "all_time": 0}, } response_body = _add_stream_all_time_data( isrc, start_date, end_date, streams_by_isrc[isrc], streams_all_time_by_isrc[isrc], empty_response_body, ) schema = StreamsSchema() result[isrc] = schema_dump(schema, response_body) return oto_response.Response(result) @tracer.wrap(name="get_streams_all") def get_streams_all( request_context, isrc, distributors, countries=[], store_ids=[], start_date=None, end_date=None, ): """Return aggregate streams timeseries for ISRC, given account type and ID. Args: request_context (RequestContext): RequestContext class isrc (str): ISRC of track to fetch streams for distributors (list): List of distributors names countries (list): List of country codes to filter by store_ids (list): List of store ids to filter by start_date (datetime.date): Start date end_date (datetime.date): End date Returns: oto.response.Response with aggregate streams """ # an empty response response_body = {"isrc": isrc, "items": []} permissions_filter = permissions.get_permissions_filter(request_context) if not (start_date and end_date): start_date, end_date = data_availability.get_date_range( data_availability.HIGHWATERMARK_DATE, days=28 ) streams = streams_model.get_streams_all( permissions_filter, isrc, distributors, countries, store_ids, start_date, end_date, ) streams_all = streams_utils.get_streams_all(streams, start_date, end_date) if streams_all: response_body["items"] = streams_all schema = StreamsAllSchema() return oto_response.Response(schema_dump(schema, response_body)) @tracer.wrap(name="get_aggregate_streams") def get_aggregate_streams( request_context, isrcs, distributors, countries=[], store_ids=[] ): """Return aggregate streams for ISRCs, given account type and ID. Args: request_context (RequestContext): RequestContext class isrcs ([str]): List of ISRCs to fetch streams for distributors (list): List of distributors names countries (list): List of country codes to filter by store_ids (list): List of store ids to filter by Returns: oto.response.Response with aggregate streams """ permissions_filter = permissions.get_permissions_filter(request_context) streams = streams_model.get_aggregate_streams( permissions_filter, isrcs, distributors, countries, store_ids ) streams_by_isrc = defaultdict(list) for s in streams: streams_by_isrc[s["isrc"]].append(s) result = {} schema = AggregateStreamsSchema() for isrc in isrcs: # an empty response response_body = {"streams_all_time": 0, "growth_percentage_7_days": None} if not streams_by_isrc[isrc]: result[isrc] = schema_dump(schema, response_body) else: response_body["streams_all_time"] = sum( map(lambda e: e["streams_all_time"], streams_by_isrc[isrc]) ) response_body["growth_percentage_7_days"] = sum( map(lambda e: e["streams_7_day_growth"], streams_by_isrc[isrc]) ) result[isrc] = schema_dump(schema, response_body) return oto_response.Response(result)