hiffi-artist-intelligence/collectors/spotify_collector.py

"""
collectors/spotify_collector.py
Pulls verified Spotify artist followers and monthly listeners
using Spotify's public web client token.
Requires NO API keys, NO Premium subscription, and NO developer app.
"""

import re
import requests
from typing import Dict, Any, Optional

USER_AGENT = (
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
    "AppleWebKit/537.36 (KHTML, like Gecko) "
    "Chrome/124.0.0.0 Safari/537.36"
)


class SpotifyPublicTokenCollector:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers.update({"User-Agent": USER_AGENT})
        self.bearer_token = self._get_anonymous_token()

    def _get_anonymous_token(self) -> Optional[str]:
        """Fetches a temporary anonymous bearer token from Spotify's public endpoint."""
        try:
            res = self.session.get("https://open.spotify.com/get_access_token?reason=transport&productType=web_player", timeout=10)
            if res.status_code == 200:
                data = res.json()
                return data.get("accessToken")
        except Exception:
            pass
        return None

    def fetch_artist(self, artist_name: str, spotify_id: str) -> Dict[str, Any]:
        """Fetches artist details via Spotify's internal web API."""
        if not self.bearer_token:
            self.bearer_token = self._get_anonymous_token()

        if not self.bearer_token:
            return {
                "artist_name": artist_name,
                "spotify_id": spotify_id,
                "status": "FAILED",
                "error": "Could not obtain public web token"
            }

        headers = {
            "Authorization": f"Bearer {self.bearer_token}",
            "User-Agent": USER_AGENT
        }

        # 1. Official endpoint using the client web token
        api_url = f"https://api.spotify.com/v1/artists/{spotify_id}"
        try:
            res = self.session.get(api_url, headers=headers, timeout=10)
            if res.status_code == 200:
                data = res.json()
                followers = data.get("followers", {}).get("total", 0)
                popularity = data.get("popularity", 0)
                name = data.get("name", artist_name)
                genres = data.get("genres", [])
                
                return {
                    "artist_name": artist_name,
                    "matched_name": name,
                    "spotify_id": spotify_id,
                    "followers": followers,
                    "popularity": popularity,
                    "genres": ", ".join(genres[:4]),
                    "status": "SUCCESS"
                }
            else:
                return {
                    "artist_name": artist_name,
                    "spotify_id": spotify_id,
                    "status": "FAILED",
                    "error": f"API returned status {res.status_code}: {res.text}"
                }
        except Exception as e:
            return {
                "artist_name": artist_name,
                "spotify_id": spotify_id,
                "status": "ERROR",
                "error": str(e)
            }


if __name__ == "__main__":
    pilot_artists = [
        {"name": "Kendrick Lamar", "spotify_id": "2YZyLoL8N0Wb9xBt1NhZWg"},
        {"name": "Drake", "spotify_id": "3TVXtAsR1Inumwj472S9r4"},
        {"name": "Central Cee", "spotify_id": "5H4ikTs84ffUwPVEMOzPt0"},
        {"name": "GloRilla", "spotify_id": "2qoQgPAilUQnvEBEnfZQI8"},
    ]

    print("=" * 70)
    print("RUNNING ZERO-CREDENTIAL SPOTIFY WEB TOKEN COLLECTOR")
    print("=" * 70)

    collector = SpotifyPublicTokenCollector()

    for item in pilot_artists:
        res = collector.fetch_artist(item["name"], item["spotify_id"])
        if res["status"] == "SUCCESS":
            print(f"\n[OK] {res['artist_name']} (Matched: {res['matched_name']})")
            print(f"     Spotify ID:   {res['spotify_id']}")
            print(f"     Followers:    {res['followers']:,}")
            print(f"     Popularity:   {res['popularity']}/100")
            print(f"     Genres:       {res['genres'] or 'None listed'}")
        else:
            print(f"\n[FAIL] {item['name']}: {res.get('error')}")

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