"""Check Spotify provider API for the latest available daily chart date.""" import base64 from datetime import date, timedelta import boto3 import requests import config logger = config.logger _secrets = boto3.client('secretsmanager') # How many days back to probe (today, yesterday, day before) _PROBE_DAYS = 3 # Statuses that mean "no chart yet for this date" — try an earlier date _NOT_FOUND_STATUSES = {403, 404} def get_latest_chart_date() -> str | None: """Return the most recent date (YYYY-MM-DD) for which Spotify has daily chart data. Probes today → yesterday → 2 days ago. Returns the first hit, or None if the Spotify API is unreachable / returning errors (so we don't misattribute a Spotify outage as a Chartmetric lag). """ try: token = _get_oauth_token() except Exception: logger.exception('Failed to obtain Spotify OAuth token') return None today = date.today() for days_back in range(_PROBE_DAYS): candidate = today - timedelta(days=days_back) status = _probe(token, candidate) if status == 200: logger.info(f'Spotify chart available for {candidate}') return candidate.isoformat() elif status is None: # Network error — treat as Spotify API down logger.warning(f'Network error probing Spotify chart for {candidate}') return None elif status not in _NOT_FOUND_STATUSES: # 401 / 429 / 5xx — real API error, not a missing date logger.error( f'Spotify API returned unexpected status {status} for {candidate}' ) return None # 404 / 403 = not yet available for this date, try earlier logger.warning(f'No daily chart found in last {_PROBE_DAYS} days') return None def _probe(token: str, chart_date: date) -> int | None: """GET the chart URL with stream=True (no body downloaded). Returns HTTP status code, or None on network error. """ url = config.SPOTIFY_CHART_URL.format( year=chart_date.year, month=f'{chart_date.month:02d}', day=f'{chart_date.day:02d}', ) try: # stream=True + immediate close avoids downloading the full chart file resp = requests.get( url, headers={'Authorization': f'Bearer {token}'}, stream=True, timeout=10, ) resp.close() return resp.status_code except requests.RequestException as e: logger.warning(f'Request error for {chart_date}: {e}') return None def _get_oauth_token() -> str: client_id = _get_secret('SPOTIFY_API_CLIENT_ID') client_secret = _get_secret('SPOTIFY_API_CLIENT_SECRET') credentials = base64.b64encode(f'{client_id}:{client_secret}'.encode()).decode() resp = requests.post( config.SPOTIFY_OAUTH_URL, data={'grant_type': 'client_credentials'}, headers={'Authorization': f'Basic {credentials}'}, timeout=10, ) resp.raise_for_status() return resp.json()['access_token'] def _get_secret(key: str) -> str: resp = _secrets.get_secret_value(SecretId=f'{config.SPOTIFY_SECRETS_PATH}/{key}') return resp['SecretString']