from datetime import date import requests from typing import Dict, List, Optional from download_spotify_charts import config from download_spotify_charts.constants import ChartBreakdown, ChartType, GLOBAL, WORLDWIDE from download_spotify_charts.utils import retry class SpotifyCharts: def __init__(self, logger): """ Returns: object: """ self.logger = logger self.session = requests.Session() self.session.headers = {"Authorization": config.DSP_AUTH, "Content-Type": "application/json"} @retry() def send_request(self, url: str, **params): """Make http request. Args: url: Request URL. params: Request params. """ params = {k: v for k, v in params.items() if v} url = f"{config.DSP_URI}/{url}" response = self.session.get(url, params=params if params else None) response.raise_for_status() return response.json() def get_charts(self, chart_type: List[str], chart_breakdown: List[str]) -> List[dict]: """Get a list of all available spotify charts. Args: chart_type: Chart types filter. chart_breakdown: Chart breakdown filter. Returns: List of charts items. """ if chart_type and len(chart_type) > 1: chart_type = None if chart_breakdown and len(chart_breakdown) > 1: chart_breakdown = None result = self.send_request("delphi/spotify/charts", type=chart_type, breakdown=chart_breakdown) result = result["items"] for item in result: if item["country_code"] == WORLDWIDE: item["country_code"] = GLOBAL return result def get_latest_date_mapping(self) -> Dict[ChartBreakdown, Dict[ChartType, Dict[str, date]]]: """Get latest dates per breakdown, type and country code. Returns: Mapping of breakdown, type and country code to latest date. """ result = self.send_request("delphi/spotify/charts/data-health/status", status="complete,complete_volatile") return { ChartBreakdown(breakdown): { ChartType(chart_type): { (country_code if country_code != WORLDWIDE else GLOBAL): date.fromisoformat( country_item["max_date"] ) for country_code, country_item in type_items.items() } for chart_type, type_items in breakdown_items.items() } for breakdown, breakdown_items in result["breakdown_chart_type_country_code"].items() } def get_charts_tracks(self, chart_id: str, chart_date: Optional[date] = None) -> List[dict]: """Get charts tracks. Args: chart_id: Chart ID. chart_date: Chart date. Returns: List of tracks data. """ result = self.send_request( "delphi/spotify/charts/tracks", chart_id=chart_id, date=chart_date, track_state_includes="metrics,public_meta", ) return result["items"]