85 lines
2.9 KiB
Python
85 lines
2.9 KiB
Python
"""
|
|
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) |