#!/usr/bin/env python3 """ Check which Spotify playlists have zero tracks. Usage: # Credentials are loaded from .env (see .env.shadow) python check_empty_playlists.py --file playlists.txt python check_empty_playlists.py --ids 04aEvbcWRqLW8YgYZcAuAw,5jaigTmF4Zoy7MQlgz1GL6,610AzeIyxFgR53LmJDBQfZ,6Q0maR6S9ZmWJk2OB8xmh7,6viuBxd0MwmFQMlbonXn8Z,7uhq0B6Goct4uLBoH1KKnJ,0jNxDq4UCKlcS56v5aOI6i,37i9dQZF1DWSFDWzEZlALC,37i9dQZF1DWSTqUqJcxFk6,37i9dQZF1DWSWyJydK4fTU python check_empty_playlists.py --file playlists.csv --column STORE_PLAYLIST_ID Input file formats: .txt — one Spotify playlist ID per line (# lines are ignored) .csv — CSV with a column containing playlist IDs (default column name: "STORE_PLAYLIST_ID") Progress is saved to check_empty_playlists_progress.json after each request. Re-running will resume from where it left off. Delete that file to start fresh. """ from __future__ import annotations import argparse import base64 import csv import json import os import sys import time import urllib.error import urllib.parse import urllib.request TOKEN_URL = "https://accounts.spotify.com/api/token" API_BASE = "https://api.spotify.com/v1" RATE_LIMIT_PAUSE = 0.1 # seconds between requests REQUEST_TIMEOUT = 10 # seconds before a stalled request is abandoned MAX_RETRIES = 4 PROGRESS_FILE = "check_empty_playlists_progress.json" class TokenExpired(Exception): pass def get_token() -> str: client_id = os.environ.get("SPOTIFY_CLIENT_ID") client_secret = os.environ.get("SPOTIFY_CLIENT_SECRET") if not client_id or not client_secret: sys.exit("Error: set SPOTIFY_CLIENT_ID and SPOTIFY_CLIENT_SECRET env vars") credentials = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode() data = urllib.parse.urlencode({"grant_type": "client_credentials"}).encode() req = urllib.request.Request( TOKEN_URL, data=data, headers={"Authorization": f"Basic {credentials}"}, ) with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp: return json.loads(resp.read())["access_token"] def get_track_count(token: str, playlist_id: str) -> int | None: """ Returns track count, None if inaccessible, raises TokenExpired on 401. Retries automatically on rate limits (429) and server errors (5xx). """ params = urllib.parse.urlencode({"fields": "id,name,tracks.total"}) url = f"{API_BASE}/playlists/{playlist_id}?{params}" for attempt in range(MAX_RETRIES): req = urllib.request.Request(url, headers={"Authorization": f"Bearer {token}"}) try: with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT) as resp: data = json.loads(resp.read()) return data.get("tracks", {}).get("total") except urllib.error.HTTPError as e: if e.code == 404: return None # deleted or private if e.code == 401: raise TokenExpired() if e.code == 429: wait = int(e.headers.get("Retry-After", 5)) print(f"\n Rate limited — waiting {wait}s...", flush=True) time.sleep(wait) continue # retry if e.code >= 500: time.sleep(2 ** attempt) continue # retry with backoff raise # unexpected HTTP error — surface it except (urllib.error.URLError, TimeoutError, OSError): if attempt < MAX_RETRIES - 1: time.sleep(2 ** attempt) continue # Gave up after retries — treat as inaccessible rather than crashing print(f"\n Network error on {playlist_id} after {MAX_RETRIES} attempts, skipping.", flush=True) return None print(f"\n Gave up on {playlist_id} after {MAX_RETRIES} retries.", flush=True) return None def load_progress() -> dict: if os.path.exists(PROGRESS_FILE): with open(PROGRESS_FILE) as f: return json.load(f) return {} def save_progress(progress: dict) -> None: with open(PROGRESS_FILE, "w") as f: json.dump(progress, f) def load_ids_from_file(path: str, column: str) -> list[str]: if path.endswith(".csv"): with open(path, newline="") as f: reader = csv.DictReader(f) if column not in (reader.fieldnames or []): available = ", ".join(reader.fieldnames or []) sys.exit(f"Error: column '{column}' not found. Available: {available}") return [row[column].strip() for row in reader if row[column].strip()] else: with open(path) as f: return [ line.strip() for line in f if line.strip() and not line.startswith("#") ] def fmt_eta(seconds: float) -> str: m, s = divmod(int(seconds), 60) h, m = divmod(m, 60) if h: return f"{h}h{m:02d}m" if m: return f"{m}m{s:02d}s" return f"{s}s" def main(): parser = argparse.ArgumentParser( description="Find Spotify playlists with zero tracks" ) source = parser.add_mutually_exclusive_group(required=True) source.add_argument("--file", help="Path to .txt or .csv file of playlist IDs") source.add_argument("--ids", help="Comma-separated playlist IDs") parser.add_argument( "--column", default="STORE_PLAYLIST_ID", help="CSV column name containing playlist IDs (default: STORE_PLAYLIST_ID)", ) args = parser.parse_args() if args.file: playlist_ids = load_ids_from_file(args.file, args.column) else: playlist_ids = [p.strip() for p in args.ids.split(",") if p.strip()] if not playlist_ids: sys.exit("Error: no playlist IDs found") # Resume from previous run if progress file exists progress = load_progress() already_done = len(progress) if already_done: print(f"Resuming — {already_done} already checked, {len(playlist_ids) - already_done} remaining.") remaining = [pid for pid in playlist_ids if pid not in progress] if not remaining: print("All playlists already checked. Delete check_empty_playlists_progress.json to start fresh.") else: print(f"Checking {len(remaining)} playlists (of {len(playlist_ids)} total)...\n") token = get_token() start = time.time() for i, pid in enumerate(remaining, 1): # Progress line with ETA elapsed = time.time() - start rate = i / elapsed if elapsed > 0 else 0 eta = (len(remaining) - i) / rate if rate > 0 else 0 print( f" [{i + already_done:>4}/{len(playlist_ids)}] {pid} ETA {fmt_eta(eta)} ", end="\r", flush=True, ) try: count = get_track_count(token, pid) except TokenExpired: print("\n Token expired — refreshing...", flush=True) token = get_token() count = get_track_count(token, pid) # retry with fresh token if count is None: progress[pid] = "inaccessible" elif count == 0: progress[pid] = "empty" else: progress[pid] = "has_tracks" save_progress(progress) time.sleep(RATE_LIMIT_PAUSE) print() # clear progress line # Summarise results from full progress dict (includes prior runs) empty = [pid for pid, v in progress.items() if v == "empty"] inaccessible = [pid for pid, v in progress.items() if v == "inaccessible"] has_tracks = [pid for pid, v in progress.items() if v == "has_tracks"] print(f"\n{'='*55}") print(f" Total checked : {len(progress)}") print(f" Empty (0 tracks) : {len(empty)}") print(f" Inaccessible : {len(inaccessible)}") print(f" Has tracks : {len(has_tracks)}") print(f"{'='*55}") if empty: print(f"\nEMPTY PLAYLISTS ({len(empty)}):") for pid in empty: print(f" https://open.spotify.com/playlist/{pid}") if inaccessible: print(f"\nINACCESSIBLE / NOT FOUND ({len(inaccessible)}):") for pid in inaccessible: print(f" {pid}") if __name__ == "__main__": main()