74 lines
3.0 KiB
Python
74 lines
3.0 KiB
Python
|
"""
|
||
|
|
collectors/kworb_spotify_collector.py
|
||
|
|
Audited alternative streaming telemetry collector.
|
||
|
|
Parses verified public Spotify Monthly Listeners from Kworb.net aggregator tables.
|
||
|
|
Enforces strict provenance attribution.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import requests
|
||
|
|
from bs4 import BeautifulSoup
|
||
|
|
from typing import Dict, Optional
|
||
|
|
|
||
|
|
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:
|
||
|
|
def __init__(self):
|
||
|
|
self.cached_listeners: Dict[str, int] = {}
|
||
|
|
self.is_loaded = False
|
||
|
|
|
||
|
|
def load_aggregator_tables(self, max_pages: int = 2):
|
||
|
|
"""Scrapes and indexes artist listener counts from public HTML tables using BeautifulSoup."""
|
||
|
|
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) >= 3:
|
||
|
|
# Col 0: Rank, Col 1: Artist (nested inside div > a), Col 2: Monthly Listeners
|
||
|
|
artist_cell = cols[1]
|
||
|
|
artist_name = artist_cell.get_text(strip=True).lower()
|
||
|
|
raw_listeners = cols[2].get_text(strip=True).replace(",", "")
|
||
|
|
try:
|
||
|
|
listeners = int(raw_listeners)
|
||
|
|
self.cached_listeners[artist_name] = listeners
|
||
|
|
except ValueError:
|
||
|
|
continue
|
||
|
|
except Exception as e:
|
||
|
|
print(f"[KworbCollector] Error reading {url}: {e}")
|
||
|
|
continue
|
||
|
|
|
||
|
|
self.is_loaded = True
|
||
|
|
print(f"[KworbCollector] Loaded {len(self.cached_listeners):,} artist records into memory.")
|
||
|
|
|
||
|
|
def get_monthly_listeners(self, artist_name: str) -> Optional[int]:
|
||
|
|
if not self.is_loaded:
|
||
|
|
self.load_aggregator_tables(max_pages=2)
|
||
|
|
return self.cached_listeners.get(artist_name.strip().lower())
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
collector = KworbStreamingCollector()
|
||
|
|
test_cohort = [
|
||
|
|
"Drake", "Kendrick Lamar", "SZA", "Future", "Latto",
|
||
|
|
"Russ Millions", "Cash Cobain", "Larry June", "Big Boogie", "Rylo Rodriguez"
|
||
|
|
]
|
||
|
|
print("\nTesting Kworb Streaming Telemetry Extraction:")
|
||
|
|
for artist in test_cohort:
|
||
|
|
listeners = collector.get_monthly_listeners(artist)
|
||
|
|
formatted = f"{listeners:,}" if listeners else "Not Listed (< Top 5000)"
|
||
|
|
print(f" {artist:20} -> {formatted}")
|