hiffi-artist-intelligence/collectors/kworb_spotify_collector.py

"""
collectors/kworb_spotify_collector.py
Audited public streaming telemetry collector.
Extracts verified Spotify Monthly Listeners and Daily Change (+/-) from Kworb.net tables.
Enforces strict null governance (unlisted artists remain None/NaN).
Includes stylized hip-hop stage name normalization and alias resolution.
"""

import re
from typing import Dict, Optional, Tuple
from bs4 import BeautifulSoup
import requests

KWORB_LISTENERS_URLS = [
    "https://kworb.net/spotify/listeners.html",
    "https://kworb.net/spotify/listeners2.html"
]

HEADERS = {
    "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 KworbStreamingCollector:
    # Curated canonical alias map for stylized hip-hop stage names
    STAGE_ALIASES: Dict[str, str] = {
        "a$ap rocky": "asap rocky",
        "a$ap ferg": "asap ferg",
        "a$ap mob": "asap mob",
        "joey bada$$": "joey badass",
        "6ix9ine": "sixnine",
        "curren$y": "currensy",
        "ty dolla $ign": "ty dolla sign",
        "kool g rap": "kool g rap",
        "t-pain": "tpain",
        "jay-z": "jay z",
        "e-40": "e40",
    }

    def __init__(self):
        # Maps normalized artist_name -> (monthly_listeners, daily_change)
        self.cached_metrics: Dict[str, Tuple[int, Optional[int]]] = {}
        self.is_loaded = False

    @classmethod
    def normalize_artist_name(cls, name: str) -> str:
        """
        Normalizes artist names by resolving currency symbols, stripping
        extraneous punctuation, and mapping known stage aliases.
        """
        if not name:
            return ""
        norm = name.strip().lower()
        if norm in cls.STAGE_ALIASES:
            norm = cls.STAGE_ALIASES[norm]

        # Standardize common hip-hop typography substitutions
        norm = norm.replace("$", "s").replace("&", "and")
        norm = re.sub(r"[^\w\s]", "", norm)
        return re.sub(r"\s+", " ", norm).strip()

    def load_aggregator_tables(self, max_pages: int = 2):
        """Scrapes and indexes artist listener counts and daily velocity from public tables."""
        print(f"[KworbCollector] Indexing public streaming metrics across {max_pages} tables...")
        for url in KWORB_LISTENERS_URLS[:max_pages]:
            try:
                res = requests.get(url, headers=HEADERS, timeout=12)
                if res.status_code == 200:
                    soup = BeautifulSoup(res.text, "html.parser")
                    rows = soup.find_all("tr")
                    for row in rows:
                        cols = row.find_all("td")
                        if len(cols) >= 4:
                            # Col 1: Raw Artist Name
                            raw_artist = cols[1].get_text(strip=True)
                            artist_norm = self.normalize_artist_name(raw_artist)

                            # Col 2: Monthly Listeners
                            raw_listeners = cols[2].get_text(strip=True).replace(",", "")
                            try:
                                listeners = int(raw_listeners)
                            except ValueError:
                                continue

                            # Col 3: Daily +/- Change
                            raw_daily = cols[3].get_text(strip=True).replace(",", "").replace("+", "")
                            try:
                                daily_change = int(raw_daily)
                            except ValueError:
                                daily_change = None

                            if artist_norm:
                                self.cached_metrics[artist_norm] = (listeners, daily_change)
            except Exception as e:
                print(f"[KworbCollector] Error reading {url}: {e}")
                continue

        self.is_loaded = True
        print(f"[KworbCollector] Loaded {len(self.cached_metrics):,} artist telemetry records.")

    def get_artist_metrics(self, artist_name: str) -> Tuple[Optional[int], Optional[int]]:
        """
        Returns (monthly_listeners, daily_change) or (None, None).
        Searches both the direct normalization and alias substitutions.
        """
        if not self.is_loaded:
            self.load_aggregator_tables(max_pages=2)
        norm_key = self.normalize_artist_name(artist_name)
        return self.cached_metrics.get(norm_key, (None, None))

    def get_monthly_listeners(self, artist_name: str) -> Optional[int]:
        return self.get_artist_metrics(artist_name)[0]

    def get_daily_change(self, artist_name: str) -> Optional[int]:
        return self.get_artist_metrics(artist_name)[1]


if __name__ == "__main__":
    collector = KworbStreamingCollector()
    collector.load_aggregator_tables(max_pages=2)
    test_cohort = [
        "Drake", "Kendrick Lamar", "SZA", "Future", 
        "Cardi B", "A$AP Rocky", "Joey Bada$$", "6ix9ine", "Ty Dolla $ign"
    ]
    print("\nValidating Kworb Stream & Daily Velocity Extraction with Aliases:")
    print(f"{'Artist':20} | {'Monthly Listeners':>18} | {'Daily +/- Trend':>16}")
    print("-" * 60)
    for artist in test_cohort:
        listeners, daily = collector.get_artist_metrics(artist)
        l_str = f"{listeners:,}" if listeners is not None else "Not in Top 5000"
        d_str = f"{daily:+,}" if daily is not None else "N/A"
        print(f"{artist:20} | {l_str:>18} | {d_str:>16}")