import collections import itertools import operator from datetime import date, datetime, timedelta from typing import Dict, List from server.legacy.analytics.constants import TWO_WEEKS_DAYS from server.utils.combine import get_combined_data def generate_missing_day(str_date: str) -> Dict: """Generate empty data for a date. Args: str_date (str): yyyy-mm-dd string to generate data for. Returns: Dict """ return { "date": str_date, "metrics": { "traffic_source_types": None, "views": 0, }, } def fill_missing_dates(data: List, latest_date: date, days_required: int) -> List: """Fill data with missing dates and sort by date. Args: data (List): performance data for videos. latest_date (date): latest date. days_required (int): number of days required in data Returns: List. """ existing_dates = [i["date"] for i in data] for i in range(days_required): str_date = (latest_date - timedelta(days=i)).strftime("%Y-%m-%d") if str_date not in existing_dates: data.append(generate_missing_day(str_date)) data.sort(key=lambda item: item["date"], reverse=True) return data def define_biggest_source(data: Dict) -> Dict: """Get biggest traffic sources. Args: data (Dict): performance data for video. Returns: Dict. """ traffic_source_data = [ i["metrics"]["traffic_source_types"] for i in data if i["metrics"]["traffic_source_types"] is not None ] total_views = sum([i["metrics"]["views"] for i in data]) counter = collections.Counter() for i in traffic_source_data: counter.update(i) traffic_source_data = collections.defaultdict(list) for source, views in dict(counter).items(): traffic_source_data[views].append(source) result = { "sources": None, "percentage": 0, } if traffic_source_data: result["sources"] = traffic_source_data[max(traffic_source_data.keys())] if total_views != 0: result["percentage"] = round(100 * max(traffic_source_data.keys()) / total_views, 0) return result def get_trend(latest: int, previous: int) -> int: """ Get trend. Args: latest (int): latest period views. previous (int): previous period views. Returns: Int. """ if latest == previous == 0: return 0 if previous == 0: return 100 return round((latest - previous) / previous * 100) def calculate_trends(data: Dict) -> Dict: """Get daily and weekly trends numbers. Args: data (Dict): performance data for video. Returns: Dict. """ daily_data = {i["date"]: i["metrics"] for i in data} transformed_data = {datetime.strptime(k, "%Y-%m-%d"): v for k, v in daily_data.items()} transformed_data = collections.OrderedDict(sorted(transformed_data.items(), key=operator.itemgetter(0))) latest_date = max(transformed_data.keys()) latest_date_views = transformed_data[latest_date]["views"] previous_date_views = transformed_data[latest_date - timedelta(days=1)]["views"] latest_week_data = list(transformed_data.values())[TWO_WEEKS_DAYS:] weekly_views = sum([i["views"] for i in latest_week_data]) data_iterator = iter(transformed_data.items()) previous_weeks = dict(itertools.islice(data_iterator, TWO_WEEKS_DAYS)) previous_weeks_views = sum([i["views"] for i in previous_weeks.values()]) latest_weeks = dict(data_iterator) latest_weeks_views = sum([i["views"] for i in latest_weeks.values()]) result = { "weekly_views": weekly_views, "latest_date_views": latest_date_views, "daily_trend": get_trend(latest_date_views, previous_date_views), "weekly_trend": get_trend(latest_weeks_views, previous_weeks_views), } return result def define_demographics_popularity(data: Dict) -> Dict: """Get biggest demographics groups. Args: data (Dict): all demographics data for video. Returns: Dict. """ age_band_views_percentages, gender_views_percentages = ( data["age_band_views_percentages"], data["gender_views_percentages"], ) gender_views_percentages = {v: k for k, v in gender_views_percentages.items()} max_gender_band = max(gender_views_percentages.keys()) gender_most_popular = gender_views_percentages[max_gender_band] age_band_views_percentages = { v: k for k, v in age_band_views_percentages.items() if k.startswith(gender_most_popular) } max_age_band = max(age_band_views_percentages.keys()) age_band_most_popular = age_band_views_percentages[max_age_band] return { "gender_most_popular": gender_most_popular, "gender_percentage": max_gender_band, "age_band_most_popular": age_band_most_popular, "age_band_percentage": max_age_band, } def map_and_group_analytics_data_by_date(data: List) -> List: performance_data = [{"metrics": row["metrics"], "date": row["dimensions"]["date"]} for row in data] return get_combined_data(performance_data, group_keys=("date",)) def map_and_group_demographics_percent_data(data: List) -> Dict: demographics_data = [{"group_key": "default", **row["metrics"]} for row in data] cnt_rows = len(demographics_data) combined_data = get_combined_data(demographics_data, group_keys=("group_key",)) demographics_dict = combined_data[0] for k, v in demographics_dict["age_band_views_percentages"].items(): demographics_dict["age_band_views_percentages"][k] = v / cnt_rows for k, v in demographics_dict["gender_views_percentages"].items(): demographics_dict["gender_views_percentages"][k] = v / cnt_rows demographics_dict.pop("group_key") return demographics_dict