import inspect from datetime import date from typing import Callable, Dict, List, Optional from server.constants.delphi.tiktok import TiktokTrackAnalyticsBreakdown from server.utils.args.getter_setter import get_arg from server.utils.delphi.converters import str_to_date from server.utils.delphi.misc import fix_position_list, nullable_max, nullable_min def sum_lists( values1: List[int or None], values2: List[int or None], response_min_date: date, response_max_date: date, result_min_date: date, result_max_date: date, sum_func: Optional[Callable] = None, ) -> List[int or None]: """Add some None values to a list (tiktok grouped by day values). Args: values1: First list of day by day metric values. values2: Second list of day by day metric values. response_min_date: Min date of the current response. response_max_date: Max date of the current response. result_min_date: Min date of a common result. result_max_date: Max date of a common result. sum_func: Function to sum items. Returns: Modified list. """ return [ ( None if item1 is None and item2 is None else (sum_func(item1, item2) if sum_func else (item1 or 0) + (item2 or 0)) ) for item1, item2 in zip( values1, fix_position_list(values2, response_min_date, response_max_date, result_min_date, result_max_date) ) ] def sum_tiktok_totals(values1: Dict[str, int or None], values2: Dict[str, int or None]) -> Dict[str, int or None]: """Sum tiktok totals breakdown or country totals single country data. Args: values1: First breakdown or country values. values2: Second breakdown or country values. Returns: Sum data as modified first values. """ for metric_name, metric_value in values2.items(): value = values1.get(metric_name) values1[metric_name] = None if value is None and metric_value is None else (value or 0) + (metric_value or 0) return values1 def sum_tiktok_country_totals( breakdown_values1: Dict[str, Dict[str, int or None]], breakdown_values2: Dict[str, Dict[str, int or None]], ) -> Dict[str, Dict[str, int or None]]: """Sum country totals breakdown data. Args: breakdown_values1: First values. breakdown_values2: Second values. Returns: Sum data as modified first values """ for country_code, country_values in breakdown_values2.items(): breakdown_values1[country_code] = sum_tiktok_totals(breakdown_values1.get(country_code, {}), country_values) return breakdown_values1 def sum_tiktok_tracks_analytics(responses: List[dict], args_spec: inspect.FullArgSpec, *args, **kwargs) -> dict: """Sum tiktok data for many ISRC. Args: responses: Response per each ISRC. args_spec: Args spec. Returns: Common response. """ response1 = responses[0] responses = [i for i in responses if i["min_date"] and i["max_date"]] if not responses: return response1 min_dates = [str_to_date(i["min_date"]) for i in responses] max_dates = [str_to_date(i["max_date"]) for i in responses] use_args_dates = kwargs.get("use_args_dates", False) if use_args_dates: min_date = get_arg(args, kwargs, args_spec, "start_date") max_date = get_arg(args, kwargs, args_spec, "end_date") else: min_date = nullable_min(*min_dates) max_date = nullable_max(*max_dates) days_count = ((max_date - min_date).days + 1) if min_date and max_date else None country_code_list = get_arg(args, kwargs, args_spec, "country_code_list") result_breakdowns = {} for index, response in enumerate(responses): breakdowns = response["breakdowns"] if TiktokTrackAnalyticsBreakdown.CONTENT_TYPE_COUNTRY in breakdowns: breakdown_values = breakdowns[TiktokTrackAnalyticsBreakdown.CONTENT_TYPE_COUNTRY] result_item = result_breakdowns.get(TiktokTrackAnalyticsBreakdown.CONTENT_TYPE_COUNTRY, {}) for content_type_name, content_type_values in breakdown_values.items(): if content_type_name not in result_item: result_item[content_type_name] = {} for country_code_name, country_code_values in content_type_values.items(): result_item[content_type_name][country_code_name] = sum_tiktok_daily( result_item[content_type_name].get(country_code_name, {}), country_code_values, min_dates[index], max_dates[index], min_date, max_date, days_count, ) result_breakdowns[TiktokTrackAnalyticsBreakdown.CONTENT_TYPE_COUNTRY] = result_item if TiktokTrackAnalyticsBreakdown.DAILY in breakdowns: result_breakdowns[TiktokTrackAnalyticsBreakdown.DAILY] = sum_tiktok_daily( result_breakdowns.get(TiktokTrackAnalyticsBreakdown.DAILY, {}), breakdowns[TiktokTrackAnalyticsBreakdown.DAILY], min_dates[index], max_dates[index], min_date, max_date, days_count, ) if TiktokTrackAnalyticsBreakdown.COUNTRY_TOTALS in breakdowns: result_breakdowns[TiktokTrackAnalyticsBreakdown.COUNTRY_TOTALS] = sum_tiktok_country_totals( result_breakdowns.get(TiktokTrackAnalyticsBreakdown.COUNTRY_TOTALS, {}), breakdowns[TiktokTrackAnalyticsBreakdown.COUNTRY_TOTALS], ) if TiktokTrackAnalyticsBreakdown.TOTALS in breakdowns: result_breakdowns[TiktokTrackAnalyticsBreakdown.TOTALS] = ( sum_tiktok_country_totals if country_code_list else sum_tiktok_totals )( result_breakdowns.get(TiktokTrackAnalyticsBreakdown.TOTALS, {}), breakdowns[TiktokTrackAnalyticsBreakdown.TOTALS], ) return { "min_date": min_date.isoformat(), "max_date": max_date.isoformat(), "days_count": days_count, "breakdowns": result_breakdowns, } def sum_tiktok_daily( values1: Dict[str, List[int or None]], values2: Dict[str, List[int or None]], response_min_date: date, response_max_date: date, result_min_date: date, result_max_date: date, days_count: int, ) -> Dict[str, List[int or None]]: """Sum tiktok daily or content type country values. Args: values1: First metrics to lists of day by day metric values. values2: Second metrics to lists of day by day metric values. response_min_date: Min date of the current response. response_max_date: Max date of the current response. result_min_date: Min date of a common result. result_max_date: Max date of a common result. days_count: Total days count. """ for metric_name, metric_values in values2.items(): values1[metric_name] = sum_lists( values1.get(metric_name, [None] * days_count), metric_values, response_min_date, response_max_date, result_min_date, result_max_date, ) return values1