from typing import List from server.constants.charts import NEW_ENTRY_TREND, YoutubeVideoTrendChartFields, YoutubeVideoTrendField, \ YoutubeVideoTrendRealField, YoutubeVideoTrendSortBy, YoutubeVideoTrendSortByDefault from server.constants.common import SortOrder def calc_field_value(source: dict, field_name: str) -> int: """Calculate correct field value. Args: source: Source data. field_name: Field name. Returns: Calculated value. """ # if is_entry then it should be on the first place according to the sorting order, # so we use random big negative number, otherwise use trend with opposite sign. return ( (NEW_ENTRY_TREND if source.get("is_entry") or source.get("is_reentry") else source.get("trend", 0) * -1) if field_name == YoutubeVideoTrendField.TREND.value else source.get(field_name) ) def get_video_chart_data(chart: dict, summary: dict, field_list: List[YoutubeVideoTrendField]) -> dict or None: """Build youtube charts video data item. Args: chart: Day by day youtube charts data. summary: Summary youtube charts data. field_list: Fields to include. Returns: Video charts data or None. """ include_chart = set(field_list) & set(YoutubeVideoTrendChartFields) # data = None if 1) return markets only (not field_list) or 2) chart/positions data is empty # or 3) only summary was requested and it is empty if not field_list or (not chart and (not summary or include_chart)): return None result = {} for field in field_list: real_field = YoutubeVideoTrendRealField.by_name(field.name) source = chart if field in YoutubeVideoTrendChartFields else summary result[field.value] = calc_field_value(source, real_field.value) return result def sort_video_chart_data(data: List[dict], sort_by: YoutubeVideoTrendSortBy, sort_order: SortOrder) -> List[dict]: """Sort youtube charts video data. Args: data: Data to sort. sort_by: Sort by field. sort_order: Sort order (asc/desc). Returns: Sorted data. """ data.sort( key=lambda x: ( x[YoutubeVideoTrendSortBy.COUNTRY_CODE.value] if sort_by == YoutubeVideoTrendSortBy.COUNTRY_CODE else ( YoutubeVideoTrendSortByDefault[sort_by] if x["data"] is None or x["data"].get(sort_by.value) is None else x["data"][sort_by.value] ) ), reverse=sort_order == SortOrder.DESC, ) return data