import collections import itertools import operator from datetime import date, datetime, timedelta from typing import Dict, List 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 { "dimensions": { "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["dimensions"]["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["dimensions"]["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["dimensions"]["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())[14:] weekly_views = sum([i["views"] for i in latest_week_data]) data_iterator = iter(transformed_data.items()) previous_weeks = dict(itertools.islice(data_iterator, 14)) 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