from typing import List from server.constants.delphi.chart import ChartAnalyticsGroupBy from server.utils.delphi.converters import str_to_date from server.utils.delphi.misc import nullable_max, nullable_min from server.utils.delphi.summarize.tiktok import sum_lists def sum_chart_analytics_items(items_list: List[dict]) -> dict: """Sum chart analytics tracks data per ISRC. Args: items_list: Chart analytics data per track ID. Returns: Summary for ISRC. """ items_list = [i for i in items_list if i["min_date"]] min_dates = [str_to_date(i["min_date"]) for i in items_list] max_dates = [str_to_date(i["max_date"]) for i in items_list] min_date = nullable_min(*min_dates) max_date = nullable_max(*max_dates) positions = [] if min_date and max_date: days_count = (max_date - min_date).days + 1 positions = [None] * days_count for index, item in enumerate(items_list): positions = sum_lists( positions, item["metrics"]["positions"], min_dates[index], max_dates[index], min_date, max_date, nullable_min, ) return { "track_id": None, "min_date": min_date.isoformat() if min_date else None, "max_date": max_date.isoformat() if min_date else None, "min_position": nullable_min(*positions), "max_position": nullable_max(*positions), "metrics": {"positions": positions}, } def sum_chart_analytics(responses: List[List[dict]], *args, **kwargs) -> List[dict]: """Sum charts analytics data. Args: responses: All responses. Returns: Concatenated response. """ group_by_isrc = kwargs["group_by"] == ChartAnalyticsGroupBy.ISRC isrc_list = kwargs["isrc_list"] result = [] for isrc, items_list in zip(isrc_list, responses): if not items_list: continue if group_by_isrc: if len(items_list) > 1: response_item = sum_chart_analytics_items(items_list) else: response_item = items_list[0] response_item["isrc"] = isrc result.append(response_item) else: for item in items_list: item["isrc"] = isrc result.extend(items_list) return result