Spaces:
Running
Running
| """Bulk-resolve Spotify IDs and collect tracks for all artists missing Spotify data. | |
| Two-phase approach: | |
| Phase 1: Link — resolve spotify_id for all unlinked artists (1 API call each) | |
| Phase 2: Collect — get albums + tracks (3-7 API calls per artist) | |
| Uses a JSON cache file to skip already-processed artists on re-runs. | |
| """ | |
| import hashlib | |
| import json | |
| import os | |
| import sys | |
| import time | |
| sys.path.insert(0, os.path.dirname(__file__)) | |
| from dotenv import load_dotenv | |
| load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env")) | |
| from db import get_connection, init_db, upsert_song, get_previous_snapshot, insert_snapshot, insert_viral_alert | |
| from spotify_client import SpotifyClient | |
| SPOTIFY_MARKET = os.getenv("SPOTIFY_MARKET", "IN") | |
| SPOTIFY_VIRAL_POPULARITY_JUMP = int(os.getenv("SPOTIFY_VIRAL_POPULARITY_JUMP", "20")) | |
| CACHE_FILE = os.path.join(os.path.dirname(__file__), ".bulk_spotify_cache.json") | |
| def generate_song_id(platform: str, platform_id: str) -> str: | |
| return hashlib.sha256(f"{platform}:{platform_id}".encode()).hexdigest()[:16] | |
| def load_cache(): | |
| if os.path.exists(CACHE_FILE): | |
| with open(CACHE_FILE, "r") as f: | |
| data = json.load(f) | |
| return set(data.get("linked", [])), set(data.get("collected", [])) | |
| return set(), set() | |
| def save_cache(linked, collected): | |
| with open(CACHE_FILE, "w") as f: | |
| json.dump({"linked": list(linked), "collected": list(collected)}, f) | |
| def collect_via_albums(conn, client, artist): | |
| """Collect tracks via albums endpoint (works with Client Credentials). | |
| API calls per artist: | |
| 1x get_artist_albums (limit=50, single page) | |
| ~5x get_album_tracks (one per album, limit=50) | |
| ~1x get_tracks (batch enrich up to 50 tracks) | |
| Total: ~7 calls per artist | |
| """ | |
| spotify_id = artist["spotify_id"] | |
| artist_id = artist["id"] | |
| # 1 API call: get up to 50 albums+singles in one request | |
| albums = client.get_artist_albums( | |
| spotify_id, include_groups="album,single", limit=50, market=SPOTIFY_MARKET | |
| ) | |
| if not albums: | |
| return {"songs_processed": 0, "viral_alerts": 0} | |
| # Sort by release date descending, take top 5 albums to keep API usage low | |
| albums = sorted(albums, key=lambda a: (a.get("release_date") or ""), reverse=True)[:5] | |
| # Collect track IDs from albums (~5 API calls) | |
| track_ids = [] | |
| album_meta = {} # track_id -> album metadata | |
| seen = set() | |
| for album in albums: | |
| album_id = album.get("id") | |
| if not album_id: | |
| continue | |
| try: | |
| album_tracks = client.get_album_tracks(album_id, limit=50) | |
| for track in album_tracks: | |
| tid = track.get("id") | |
| if tid and tid not in seen: | |
| seen.add(tid) | |
| track_ids.append(tid) | |
| album_meta[tid] = { | |
| "name": album.get("name"), | |
| "release_date": album.get("release_date", ""), | |
| "images": album.get("images") or [], | |
| } | |
| except Exception as e: | |
| if "403" in str(e): | |
| continue # skip forbidden albums | |
| raise | |
| if not track_ids: | |
| return {"songs_processed": 0, "viral_alerts": 0} | |
| # 1 API call per 50 tracks: batch get full track details with popularity | |
| full_tracks = client.get_tracks(track_ids[:100]) # cap at 100 tracks | |
| tracks_by_id = {t.get("id"): t for t in full_tracks if t} | |
| processed = 0 | |
| viral = 0 | |
| for track_id in track_ids[:100]: | |
| track = tracks_by_id.get(track_id) | |
| if not track: | |
| continue | |
| album_info = album_meta.get(track_id, {}) | |
| album = track.get("album") or {} | |
| images = album.get("images") or album_info.get("images") or [] | |
| release_date = album.get("release_date") or album_info.get("release_date", "") | |
| if release_date and len(release_date) == 4: | |
| release_date = f"{release_date}-01-01" | |
| elif release_date and len(release_date) == 7: | |
| release_date = f"{release_date}-01" | |
| song_id = generate_song_id("spotify", track_id) | |
| upsert_song(conn, { | |
| "id": song_id, | |
| "artist_id": artist_id, | |
| "title": track.get("name", "Unknown"), | |
| "platform": "spotify", | |
| "platform_id": track_id, | |
| "album_name": album.get("name") or album_info.get("name"), | |
| "release_date": release_date[:10] if release_date else "", | |
| "thumbnail_url": images[0].get("url", "") if images else "", | |
| }) | |
| popularity = int(track.get("popularity") or 0) | |
| prev = get_previous_snapshot(conn, song_id, "spotify") | |
| insert_snapshot(conn, song_id, popularity, 0, 0, "spotify") | |
| processed += 1 | |
| if prev and prev.get("play_count") is not None: | |
| delta = popularity - int(prev["play_count"]) | |
| if delta >= SPOTIFY_VIRAL_POPULARITY_JUMP: | |
| insert_viral_alert(conn, { | |
| "song_id": song_id, | |
| "previous_count": int(prev["play_count"]), | |
| "current_count": popularity, | |
| "growth_factor": float(delta), | |
| "platform": "spotify", | |
| }) | |
| viral += 1 | |
| return {"songs_processed": processed, "viral_alerts": viral} | |
| def phase1_link(conn, client): | |
| """Phase 1: Resolve Spotify IDs for artists missing them.""" | |
| linked_cache, collected_cache = load_cache() | |
| missing = conn.execute(""" | |
| SELECT id, name FROM artists WHERE spotify_id IS NULL ORDER BY name | |
| """).fetchall() | |
| todo = [a for a in missing if a["id"] not in linked_cache] | |
| if not todo: | |
| print(f"Phase 1: All artists already linked ✅ ({len(linked_cache)} cached)") | |
| return 0 | |
| print(f"\n{'='*70}") | |
| print(f"Phase 1: Linking {len(todo)} artists ({len(linked_cache)} cached/skipped)") | |
| print(f"{'='*70}\n") | |
| resolved = 0 | |
| skipped_no_match = 0 | |
| for i, artist in enumerate(todo, start=1): | |
| name = artist["name"] | |
| artist_id = artist["id"] | |
| print(f"[{i}/{len(todo)}] {name:35s}", end=" ", flush=True) | |
| try: | |
| matches = client.search_artist(name, limit=5) | |
| if not matches: | |
| print("— not found") | |
| linked_cache.add(artist_id) | |
| save_cache(linked_cache, collected_cache) | |
| time.sleep(0.3) | |
| continue | |
| # STRICT matching: require exact case-insensitive name match | |
| # Never fall back to a different artist name | |
| best = None | |
| name_lower = name.strip().lower() | |
| for m in matches: | |
| m_name = (m.get("name") or "").strip().lower() | |
| if m_name == name_lower: | |
| best = m | |
| break | |
| # Also accept minor punctuation diffs (e.g. KR$NA vs KRSNA) | |
| m_cleaned = ''.join(c for c in m_name if c.isalnum() or c == ' ') | |
| n_cleaned = ''.join(c for c in name_lower if c.isalnum() or c == ' ') | |
| if m_cleaned == n_cleaned and len(n_cleaned) > 2: | |
| best = m | |
| break | |
| if not best: | |
| # No exact match — do NOT blindly take first result | |
| top_name = (matches[0].get("name") or "???") | |
| print(f"⚠️ skipped (top result: {top_name})") | |
| skipped_no_match += 1 | |
| linked_cache.add(artist_id) | |
| save_cache(linked_cache, collected_cache) | |
| time.sleep(0.3) | |
| continue | |
| spotify_id = best.get("id") | |
| spotify_name = best.get("name", name) | |
| images = best.get("images") or [] | |
| image_url = images[0].get("url") if images else None | |
| conn.execute(""" | |
| UPDATE artists SET spotify_id = ?, image_url = COALESCE(image_url, ?) WHERE id = ? | |
| """, (spotify_id, image_url, artist_id)) | |
| conn.commit() | |
| resolved += 1 | |
| linked_cache.add(artist_id) | |
| save_cache(linked_cache, collected_cache) | |
| match_note = "" if spotify_name.lower() == name.lower() else f" (→ {spotify_name})" | |
| print(f"✅{match_note}") | |
| time.sleep(0.3) | |
| except Exception as e: | |
| err = str(e) | |
| if "429" in err: | |
| print(f"⏳ rate limited, waiting 30s...") | |
| time.sleep(30) | |
| else: | |
| print(f"❌ {e}") | |
| print(f"\nPhase 1 done: linked {resolved}/{len(todo)}\n") | |
| return resolved | |
| def phase2_collect(conn, client): | |
| """Phase 2: Collect tracks for artists without Spotify songs.""" | |
| linked_cache, collected_cache = load_cache() | |
| need_collection = conn.execute(""" | |
| SELECT a.id, a.name, a.spotify_id | |
| FROM artists a | |
| WHERE a.spotify_id IS NOT NULL | |
| AND (SELECT COUNT(*) FROM songs WHERE artist_id = a.id AND platform = 'spotify') = 0 | |
| ORDER BY a.name | |
| """).fetchall() | |
| todo = [a for a in need_collection if a["id"] not in collected_cache] | |
| if not todo: | |
| print(f"Phase 2: All artists collected ✅ ({len(collected_cache)} cached)") | |
| return | |
| print(f"\n{'='*70}") | |
| print(f"Phase 2: Collecting tracks for {len(todo)} artists ({len(collected_cache)} cached/skipped)") | |
| print(f"{'='*70}\n") | |
| total_songs = 0 | |
| for i, artist in enumerate(todo, start=1): | |
| name = artist["name"] | |
| print(f"[{i}/{len(todo)}] {name:35s}", end=" ", flush=True) | |
| try: | |
| result = collect_via_albums(conn, client, artist) | |
| conn.commit() | |
| songs = result.get("songs_processed", 0) | |
| total_songs += songs | |
| collected_cache.add(artist["id"]) | |
| save_cache(linked_cache, collected_cache) | |
| print(f"✅ {songs} tracks") | |
| time.sleep(1) # gentle rate limiting | |
| except Exception as e: | |
| conn.rollback() | |
| err = str(e) | |
| if "429" in err: | |
| print(f"⏳ rate limited, waiting 60s...") | |
| time.sleep(60) | |
| try: | |
| result = collect_via_albums(conn, client, artist) | |
| conn.commit() | |
| songs = result.get("songs_processed", 0) | |
| total_songs += songs | |
| collected_cache.add(artist["id"]) | |
| save_cache(linked_cache, collected_cache) | |
| print(f" ✅ retry: {songs} tracks") | |
| except Exception as e2: | |
| conn.rollback() | |
| print(f" ❌ retry failed: {e2}") | |
| elif "403" in err: | |
| print(f"⛔ forbidden (skipping)") | |
| collected_cache.add(artist["id"]) | |
| save_cache(linked_cache, collected_cache) | |
| else: | |
| print(f"❌ {e}") | |
| print(f"\nPhase 2 done: collected {total_songs} total tracks\n") | |
| def main(): | |
| init_db() | |
| conn = get_connection() | |
| client = SpotifyClient() | |
| phase1_link(conn, client) | |
| phase2_collect(conn, client) | |
| stats = conn.execute(""" | |
| SELECT | |
| (SELECT COUNT(*) FROM artists) as total, | |
| (SELECT COUNT(*) FROM artists WHERE spotify_id IS NOT NULL) as linked, | |
| (SELECT COUNT(*) FROM songs WHERE platform = 'spotify') as tracks | |
| """).fetchone() | |
| print(f"\n{'='*70}") | |
| print(f"Final: {stats['linked']}/{stats['total']} artists linked, {stats['tracks']} Spotify tracks") | |
| print(f"{'='*70}") | |
| conn.close() | |
| if os.path.exists(CACHE_FILE): | |
| os.remove(CACHE_FILE) | |
| print("(session cache cleared)") | |
| if __name__ == "__main__": | |
| main() | |