#!/usr/bin/env python3 """ Check which Spotify playlists are missing name or artwork at source. Usage: # Credentials are loaded from .env (see .env.shadow) python check_playlist_metadata.py --file playlists.txt python check_playlist_metadata.py --ids 04aEvbcWRqLW8YgYZcAuAw,5jaigTmF4Zoy7MQlgz1GL6 python check_playlist_metadata.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_playlist_metadata_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 REQUEST_TIMEOUT = 10 MAX_RETRIES = 4 PROGRESS_FILE = "check_playlist_metadata_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_metadata(token: str, playlist_id: str) -> dict | None: """ Returns {"has_name": bool, "has_artwork": bool} or None if inaccessible. Raises TokenExpired on 401. """ params = urllib.parse.urlencode({"fields": "id,name,images"}) 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 { "has_name": bool(data.get("name")), "has_artwork": bool(data.get("images")), } except urllib.error.HTTPError as e: if e.code == 404: return None 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 if e.code >= 500: time.sleep(2 ** attempt) continue raise except (urllib.error.URLError, TimeoutError, OSError): if attempt < MAX_RETRIES - 1: time.sleep(2 ** attempt) continue 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 classify(result: dict | None) -> str: if result is None: return "inaccessible" if result["has_name"] and result["has_artwork"]: return "fully_renderable" if result["has_name"]: return "missing_artwork" if result["has_artwork"]: return "missing_name" return "missing_both" def main(): parser = argparse.ArgumentParser( description="Check Spotify playlists for missing name or artwork" ) 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") 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_playlist_metadata_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): 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: result = get_metadata(token, pid) except TokenExpired: print("\n Token expired — refreshing...", flush=True) token = get_token() result = get_metadata(token, pid) progress[pid] = classify(result) save_progress(progress) time.sleep(RATE_LIMIT_PAUSE) print() fully_renderable = [pid for pid, v in progress.items() if v == "fully_renderable"] missing_artwork = [pid for pid, v in progress.items() if v == "missing_artwork"] missing_name = [pid for pid, v in progress.items() if v == "missing_name"] missing_both = [pid for pid, v in progress.items() if v == "missing_both"] inaccessible = [pid for pid, v in progress.items() if v == "inaccessible"] print(f"\n{'='*55}") print(f" Total checked : {len(progress)}") print(f" Fully renderable : {len(fully_renderable)}") print(f" Missing artwork : {len(missing_artwork)}") print(f" Missing name : {len(missing_name)}") print(f" Missing both : {len(missing_both)}") print(f" Inaccessible : {len(inaccessible)}") print(f"{'='*55}") if missing_artwork: print(f"\nMISSING ARTWORK ({len(missing_artwork)}):") for pid in missing_artwork: print(f" https://open.spotify.com/playlist/{pid}") if missing_name: print(f"\nMISSING NAME ({len(missing_name)}):") for pid in missing_name: print(f" https://open.spotify.com/playlist/{pid}") if missing_both: print(f"\nMISSING BOTH ({len(missing_both)}):") for pid in missing_both: 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()