"""Logic for retrieving bulk streams per ISRC (POST /sound-recording/streams).""" import datetime import itertools from typing import Any, Mapping from ddtrace import tracer from oto import response as oto_response from analytics.constants import cache from analytics.constants.parameters import ALL_TIME from analytics.logic import data_availability from analytics.logic.parallel import parallel from analytics.logic.stores import add_outage_error_to_stores from analytics.logic.streams_helpers import ( build_bulk_query_input, group_by_isrc, rows_to_dicts, ) from analytics.queries.sound_recording_streams_bulk import ( StreamsBulkAllTime, StreamsBulkDaily, ) from analytics.schemas.streams import StreamsSchema from analytics.utils import date as date_utils from analytics.utils import store_availability from analytics.utils import streams as streams_utils from analytics.utils.cache import cache_in_redis from analytics.validation.schema import schema_dump def _get_time_period(start_date=None, end_date=None): 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): stores_with_skips = [ store for store in streams_by_store if store["total_skips"] is not None ] if not stores_with_skips: return None total_streams = sum( store["total_streams_with_skips"] for store in stores_with_skips ) total_skips = sum(store["total_skips"] for store in stores_with_skips) return total_skips / (total_skips + total_streams) if total_skips else 0 def _calc_total_store_skips(items): items_with_skips = [item for item in items if item and item["skips"] is not None] if not items_with_skips: return None return sum(item["skips"] for item in items_with_skips) def _calc_total_store_streams_with_skips(items): items_with_skips = [item for item in items if item and item["skips"] is not None] if not items_with_skips: return None return sum(item["streams_with_skips"] for item in items_with_skips) def _get_streams_by_store(streams_all_time, streams, time_period): 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({record["store_id"] for record in streams_all_time}) active_store_ids = {e for e in all_store_ids if e in store_names} 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(item["streams"] for item in 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): streams_totals_by_date = { k.strftime("%Y-%m-%d"): { "streams": 0, "skips": None, "saves": None, "streams_with_skips": None, } for k, _ in itertools.groupby(streams, lambda e: e["date"]) } for item in streams: date_key = item["date"].strftime("%Y-%m-%d") bucket = streams_totals_by_date[date_key] bucket["streams"] += item["streams"] if bucket["streams_with_skips"] is None: bucket["streams_with_skips"] = item["streams_with_skips"] else: bucket["streams_with_skips"] += ( 0 if item["streams_with_skips"] is None else item["streams_with_skips"] ) if bucket["skips"] is None: bucket["skips"] = item["skips"] else: bucket["skips"] += 0 if item["skips"] is None else item["skips"] if bucket["saves"] is None: bucket["saves"] = item["saves"] else: bucket["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: bucket = streams_totals_by_date.get(item["date"]) item["streams"] = bucket["streams"] if bucket else 0 item["skips"] = bucket["skips"] if bucket else None item["saves"] = bucket["saves"] if bucket else None item["skip_rate"] = streams_utils.calc_date_skip_rate(bucket) return streams_totals def _add_stream_all_time_data( start_date, end_date, streams, streams_all_time, response_body ): if not streams_all_time: return response_body response_body["aggregate"]["all_time"] = sum( e["all_time"] for e in 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(e["streams"] for e in streams_totals[-7:]) prev_week = sum(e["streams"] for e in 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 def _fetch_daily(query_input): return rows_to_dicts(StreamsBulkDaily(query_input).execute()) def _fetch_all_time(query_input): rows = rows_to_dicts(StreamsBulkAllTime(query_input).execute()) for item in rows: item["all_time"] = item["streams_all_time"] return rows @cache_in_redis(ttl=cache.ONE_DAY) @tracer.wrap(name="get_streams_bulk") def get_streams_bulk( query_params: Mapping[str, Any], permissions: Mapping[str, Any], ): """Return streams per ISRC for a list of ISRCs. Args: query_params: Dict with isrcs, distributors, country_ids, store_ids, start_date, end_date. permissions: Dict with permission_* keys. Returns: oto.response.Response keyed by ISRC with stream payloads. """ isrcs = query_params["isrcs"] start_date = query_params.get("start_date") end_date = query_params.get("end_date") if start_date == ALL_TIME: start_date = None if not (start_date and end_date): start_date, end_date = data_availability.get_date_range( data_availability.HIGHWATERMARK_DATE, days=28 ) store_ids = store_availability.get_query_store_ids( query_params.get("store_ids") or [] ) sources = add_outage_error_to_stores(store_availability.get_sources()) schema = StreamsSchema() def _empty_body(isrc): return { "isrc": isrc, "stores": [], "sources": sources, "aggregate": {"items": [], "all_time": 0}, } if not store_ids: return oto_response.Response( {isrc: schema_dump(schema, _empty_body(isrc)) for isrc in isrcs} ) end_date_str = ( end_date.strftime("%Y-%m-%d") if isinstance(end_date, datetime.date) else end_date ) start_date_str = ( start_date.strftime("%Y-%m-%d") if isinstance(start_date, datetime.date) else start_date ) common = build_bulk_query_input(isrcs, store_ids, query_params, permissions) daily_input = { **common, "start_date": start_date_str, "end_date": end_date_str, } parallel_result = parallel( { "streams": {"func": _fetch_daily, "args": (daily_input,)}, "streams_all_time": {"func": _fetch_all_time, "args": (common,)}, } ) streams_by_isrc = group_by_isrc(parallel_result.message["streams"]) streams_all_time_by_isrc = group_by_isrc( parallel_result.message["streams_all_time"] ) result = {} for isrc in isrcs: response_body = _add_stream_all_time_data( start_date, end_date, streams_by_isrc[isrc], streams_all_time_by_isrc[isrc], _empty_body(isrc), ) result[isrc] = schema_dump(schema, response_body) return oto_response.Response(result)