hiffi-artist-intelligence/collectors/test_spotify_endpoints.py

"""
collectors/test_spotify_endpoints.py
Direct diagnostic to test Spotify OAuth2 Client Credentials flow
and probe endpoint permissions for Artist data, Top Tracks, and Search.
"""

import base64
import os
from pathlib import Path
import sys
import requests
from dotenv import load_dotenv

PROJECT_ROOT = Path(__file__).resolve().parent.parent
if str(PROJECT_ROOT) not in sys.path:
    sys.path.insert(0, str(PROJECT_ROOT))

load_dotenv(PROJECT_ROOT / ".env")

CLIENT_ID = os.getenv("SPOTIFY_CLIENT_ID", "").strip()
CLIENT_SECRET = os.getenv("SPOTIFY_CLIENT_SECRET", "").strip()


def get_token(client_id: str, client_secret: str):
    """Requests a bearer token via standard OAuth2 Client Credentials."""
    if not client_id or not client_secret:
        return 0, {"error": "SPOTIFY_CLIENT_ID or SPOTIFY_CLIENT_SECRET not configured"}

    auth_header = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
    url = "https://accounts.spotify.com/api/token"
    res = requests.post(
        url,
        data={"grant_type": "client_credentials"},
        headers={"Authorization": f"Basic {auth_header}"},
        timeout=10,
    )
    return res.status_code, res.json()


def test_endpoints(token: str):
    headers = {"Authorization": f"Bearer {token}"}
    pilot_artist_id = "3TVXtAsR1Inumwj472S9r4"  # Drake

    endpoints = {
        "Artist Metadata (GET /v1/artists/{id})": f"https://api.spotify.com/v1/artists/{pilot_artist_id}",
        "Artist Top Tracks (GET /v1/artists/{id}/top-tracks)": f"https://api.spotify.com/v1/artists/{pilot_artist_id}/top-tracks?market=US",
        "Search Endpoint (GET /v1/search)": "https://api.spotify.com/v1/search?q=Drake&type=artist&limit=1",
    }

    results = {}
    for name, url in endpoints.items():
        res = requests.get(url, headers=headers, timeout=10)
        results[name] = {
            "status_code": res.status_code,
            "response_preview": res.text[:300],
        }
    return results


if __name__ == "__main__":
    print("=" * 80)
    print("SPOTIFY API AUTHENTICATION & ENDPOINT PERMISSION DIAGNOSTIC")
    print("=" * 80)

    if not CLIENT_ID or not CLIENT_SECRET:
        print("Error: Missing credentials in environment. Set SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET in .env.")
        sys.exit(1)

    status, token_resp = get_token(CLIENT_ID, CLIENT_SECRET)
    print(f"Token Endpoint Status: {status}")

    if status == 200:
        token = token_resp.get("access_token")
        print(f"Bearer Token Acquired: {token[:15]}... [expires_in: {token_resp.get('expires_in')}s]")
        print("-" * 80)

        diagnostics = test_endpoints(token)
        for ep, out in diagnostics.items():
            print(f"\nEndpoint: {ep}")
            print(f"Status:   {out['status_code']}")
            print(f"Payload:  {out['response_preview']}")
    else:
        print(f"Failed to obtain token: {token_resp}")

    print("\n" + "=" * 80)