from datetime import date, datetime, timedelta from typing import List from server.constants import DSP from server.constants.charts import ChartBreakdown, ChartType def get_chart_id( market: str, chart_type: ChartType or str = ChartType.REGIONAL, chart_breakdown: ChartBreakdown or str = ChartBreakdown.DAILY, dsp: DSP = DSP.SPOTIFY, ): """Build chart ID. Args: chart_breakdown: Daily or weekly. market: Market code. chart_type: Regional or viral. dsp: DSP. Returns: Chart ID. """ if dsp.value == DSP.APPLE.value: chart_type = "charts" elif not isinstance(chart_type, str): chart_type = chart_type.value if not isinstance(chart_breakdown, str): chart_breakdown = chart_breakdown.value return f"{chart_type}_{chart_breakdown}_{market}" def add_days(date_value: date or str, days_count: int = 1) -> date: """Add days to date in str or date format. Args: date_value: Date. days_count: Days to add. Returns: Updated date. """ if isinstance(date_value, str): date_value = datetime.fromisoformat(date_value).date() return date_value + timedelta(days=days_count) def fix_weekly(data: List[dict]): """Change thursday to friday for weekly charts. Args: data: List of charts items data. """ lifetime_fields = ("earliest_position_date", "min_position_date", "latest_position_date") for item in data: if item["chart_meta"]["breakdown"] == ChartBreakdown.WEEKLY.value: if "lifetime_metrics" in item: for field in lifetime_fields: item["lifetime_metrics"][field] = add_days(item["lifetime_metrics"][field]) if "metrics" in item: item["metrics"]["date"] = add_days(item["metrics"]["date"])