69 lines
2.4 KiB
Python
69 lines
2.4 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 os
|
||
|
|
import requests
|
||
|
|
import base64
|
||
|
|
|
||
|
|
CLIENT_ID = os.getenv("SPOTIFY_CLIENT_ID", "a4a008ec790d45fba6a37315a1a3d847")
|
||
|
|
CLIENT_SECRET = os.getenv("SPOTIFY_CLIENT_SECRET", "a5c67d9a63d340b6ae4a5bec83bca241")
|
||
|
|
|
||
|
|
|
||
|
|
def get_token(client_id: str, client_secret: str):
|
||
|
|
"""Requests a bearer token via standard OAuth2 Client Credentials."""
|
||
|
|
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)
|
||
|
|
|
||
|
|
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)
|