from datetime import date from typing import Tuple from server import config from server.cache.utils import cached from server.client import services from server.client.utils import date_response from server.constants.charts import ChartGroup @date_response async def get_available_dates() -> Tuple[date, date]: (response,) = await services.dsp.get_chart_positions_dates(chart_group=ChartGroup.YOUTUBE_TRENDING.value) return response["earliest_data_date"], response["latest_data_date"] @date_response async def get_video_dates(video_id: str, market: str or None) -> Tuple[date, date]: """Get min and max available dates for video ID over all markets or for specific. Args: video_id: Video ID. market: Market code or None. Returns: Min and max available dates. """ response = await services.dsp.get_video_positions_charts_summary( video_id=video_id, chart_group=ChartGroup.YOUTUBE_TRENDING.value, country_code=market, ) min_date = None max_date = None for item in response: if not min_date or min_date > item["summary"]["earliest_position_date"]: min_date = item["summary"]["earliest_position_date"] if not max_date or max_date < item["summary"]["latest_position_date"]: max_date = item["summary"]["latest_position_date"] return min_date, max_date @cached(ttl=config.CHARTS_YOUTUBE_MARKETS_TTL) async def get_markets(): """Get all youtube charts markets. Returns: List of country codes. """ response = await services.dsp.get_charts(chart_group=ChartGroup.YOUTUBE_TRENDING.value) return [i["country_code"] for i in response]