Transfer spotify playlist to Youtube Music with an export csv.

spotify-to-ytmusic.md

transfer.py

Spotify CSV - YouTube Music playlist transfer. Uses browser auth, no Spotify API keys needed. Caches videoIDs, prevents duplicates.

Setup

pip install ytmusicapi
ytmusicapi browser --file browser.json   # paste YTM request headers

Usage

python transfer.py browser.json playlist.csv --playlist "My Playlist"

Flags

Flag	What it does
-p, --playlist	Playlist name on YT Music (required)
-d, --description	Playlist description
--delay 2.0	Seconds between API calls (default: 1.5)
--dry-run	Search + cache only, don't touch playlists
--skip 62	Skip first N songs in CSV
--resume	Auto-resume from checkpoint.json
--cache-only	Skip search, add only songs already in song_cache.json

Files it creates

• song_cache.json — persistent search results, reused across runs
• checkpoint.json — auto-resume state, deleted on completion
• transfer_*.log — full debug log
• report_*.json — summary with skipped songs list

transfer.py

#!/usr/bin/env python3
"""
Spotify CSV → YouTube Music Playlist Transfer
Uses ytmusicapi browser auth + Exportify CSV
"""

import csv
import sys
import time
import json
import logging
import argparse
from datetime import datetime
from pathlib import Path

try:
    from ytmusicapi import YTMusic
except ImportError:
    print("Missing ytmusicapi. Install with: pip install ytmusicapi")
    sys.exit(1)

# ── Logging Setup ──────────────────────────────────────────────
LOG_FILE = f"transfer_{datetime.now().strftime('%Y%m%d_%H%M%S')}.log"

logger = logging.getLogger("transfer")
logger.setLevel(logging.DEBUG)

# console: INFO+ with colors
class ColorFormatter(logging.Formatter):
    COLORS = {
        "DEBUG": "\033[90m",     # gray
        "INFO": "\033[36m",      # cyan
        "WARNING": "\033[33m",   # yellow
        "ERROR": "\033[31m",     # red
        "SUCCESS": "\033[32m",   # green
    }
    RESET = "\033[0m"

    def format(self, record):
        color = self.COLORS.get(record.levelname, self.RESET)
        timestamp = datetime.now().strftime("%H:%M:%S")
        return f"{color}[{timestamp}] {record.levelname:<7}{self.RESET} {record.getMessage()}"

console = logging.StreamHandler(sys.stdout)
console.setLevel(logging.INFO)
console.setFormatter(ColorFormatter())
logger.addHandler(console)

# file: DEBUG+ (everything)
file_handler = logging.FileHandler(LOG_FILE, encoding="utf-8")
file_handler.setLevel(logging.DEBUG)
file_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)-7s | %(message)s"))
logger.addHandler(file_handler)

# custom SUCCESS level
SUCCESS = 25
logging.addLevelName(SUCCESS, "SUCCESS")
def log_success(msg, *args, **kwargs):
    logger.log(SUCCESS, msg, *args, **kwargs)
logger.success = log_success


# ── CSV Parsing ────────────────────────────────────────────────
def load_csv(filepath):
    """
    Parse Exportify CSV. Tries common column names.
    Returns list of dicts: {track, artist, album}
    """
    songs = []
    path = Path(filepath)
    if not path.exists():
        logger.error(f"CSV not found: {filepath}")
        sys.exit(1)

    with open(path, encoding="utf-8") as f:
        reader = csv.DictReader(f)
        headers = reader.fieldnames
        logger.debug(f"CSV columns: {headers}")

        # map exportify columns (case-insensitive)
        col_map = {}
        for h in headers:
            hl = h.strip().lower()
            if hl in ("track name", "name", "song", "title", "track"):
                col_map["track"] = h
            elif hl in ("artist name(s)", "artist", "artists", "artist name"):
                col_map["artist"] = h
            elif hl in ("album name", "album"):
                col_map["album"] = h

        if "track" not in col_map or "artist" not in col_map:
            # fallback: use column indices (row[1] = track, row[3] = artist)
            logger.warning("Could not auto-detect columns, falling back to index-based parsing")
            f.seek(0)
            raw_reader = csv.reader(f)
            next(raw_reader)  # skip header
            for row in raw_reader:
                if len(row) >= 4:
                    songs.append({
                        "track": row[1].strip(),
                        "artist": row[3].strip(),
                        "album": row[5].strip() if len(row) > 5 else "",
                    })
            return songs

        logger.info(f"Detected columns → track: '{col_map['track']}', artist: '{col_map['artist']}', album: '{col_map.get('album', 'N/A')}'")

        for row in reader:
            track = row.get(col_map["track"], "").strip()
            artist = row.get(col_map["artist"], "").strip()
            album = row.get(col_map.get("album", ""), "").strip()
            if track and artist:
                songs.append({"track": track, "artist": artist, "album": album})

    return songs


# ── Song Cache ─────────────────────────────────────────────────
CACHE_FILE = "song_cache.json"

def cache_key(song):
    """Consistent key for a song: 'artist||track' lowercased."""
    return f"{song['artist'].lower().strip()}||{song['track'].lower().strip()}"

def load_cache():
    p = Path(CACHE_FILE)
    if not p.exists():
        return {}
    with open(p, encoding="utf-8") as f:
        return json.load(f)

def save_cache(cache):
    with open(CACHE_FILE, "w", encoding="utf-8") as f:
        json.dump(cache, f, indent=2, ensure_ascii=False)


# ── YTMusic Search ─────────────────────────────────────────────
def search_song(ytmusic, song, retry=2):
    """
    Search YT Music for a song. Tries multiple query strategies.
    Returns videoId or None.
    """
    queries = [
        f"{song['track']} {song['artist']}",                    # standard
        f"{song['track']} {song['artist']} {song['album']}",    # with album
        f"{song['artist']} {song['track']}",                    # flipped
    ]

    for attempt, query in enumerate(queries):
        try:
            logger.debug(f"Search query (strategy {attempt+1}): '{query}'")
            results = ytmusic.search(query, filter="songs", limit=5)

            if results:
                top = results[0]
                vid = top.get("videoId")
                title = top.get("title", "?")
                artists = ", ".join(a["name"] for a in top.get("artists", []))
                logger.debug(f"  → Match: '{title}' by {artists} (videoId: {vid})")
                return vid

            # try without filter
            results = ytmusic.search(query, limit=5)
            for r in results:
                if r.get("videoId"):
                    vid = r["videoId"]
                    title = r.get("title", "?")
                    logger.debug(f"  → Unfiltered match: '{title}' (videoId: {vid})")
                    return vid

        except Exception as e:
            err = str(e)
            if "400" in err or "timeout" in err.lower():
                logger.warning(f"  → Request failed (attempt {attempt+1}): {err}")
                if attempt < len(queries) - 1:
                    time.sleep(3)
                    continue
                # retry with backoff
                for r in range(retry):
                    wait = (r + 1) * 5
                    logger.warning(f"  → Retrying in {wait}s... ({r+1}/{retry})")
                    time.sleep(wait)
                    try:
                        results = ytmusic.search(queries[0], filter="songs", limit=3)
                        if results and results[0].get("videoId"):
                            return results[0]["videoId"]
                    except:
                        continue
            else:
                logger.error(f"  → Unexpected error: {err}")
                return None

    return None


def search_all_songs(ytmusic, songs, delay=1.5):
    """Search for all songs with progress tracking."""
    total = len(songs)
    found = []
    skipped = []

    for i, song in enumerate(songs):
        label = f"{song['artist']} - {song['track']}"
        logger.info(f"[{i+1}/{total}] Searching: {label}")

        vid = search_song(ytmusic, song)

        if vid:
            found.append({"song": song, "videoId": vid})
            logger.success(f"  ✓ Found")
        else:
            skipped.append(song)
            logger.warning(f"  ✗ Not found → SKIPPED")

        # progress summary every 25 songs
        if (i + 1) % 25 == 0:
            logger.info(f"── Progress: {len(found)} found, {len(skipped)} skipped out of {i+1} ──")

        time.sleep(delay)

    return found, skipped


# ── Playlist Creation ──────────────────────────────────────────
def create_or_get_playlist(ytmusic, name, description=""):
    """Create a new playlist or find existing one."""
    try:
        existing = ytmusic.search(name, filter="playlists", scope="library")
        if existing:
            pid = existing[0].get("browseId", "").replace("VL", "")
            logger.info(f"Found existing playlist '{name}' (id: {pid})")
            return pid
    except:
        pass

    logger.info(f"Creating new playlist: '{name}'")
    pid = ytmusic.create_playlist(name, description or f"Imported from Spotify on {datetime.now().strftime('%Y-%m-%d')}")
    logger.info(f"Created playlist (id: {pid})")
    return pid


def get_playlist_video_ids(ytmusic, playlist_id):
    """Fetch all videoIds already in a playlist."""
    try:
        playlist = ytmusic.get_playlist(playlist_id, limit=5000)
        existing = set()
        for track in playlist.get("tracks", []):
            vid = track.get("videoId")
            if vid:
                existing.add(vid)
        logger.info(f"Playlist has {len(existing)} existing tracks")
        return existing
    except Exception as e:
        logger.warning(f"Could not fetch playlist contents: {e}")
        return set()


def add_songs_to_playlist(ytmusic, playlist_id, found_songs, delay=1.0):
    """Add songs to playlist one by one, skipping duplicates."""
    existing = get_playlist_video_ids(ytmusic, playlist_id)

    total = len(found_songs)
    added = 0
    failed = 0
    dupes = 0

    for i, item in enumerate(found_songs):
        label = f"{item['song']['artist']} - {item['song']['track']}"

        if item["videoId"] in existing:
            dupes += 1
            logger.info(f"[{i+1}/{total}] Already in playlist, skipping: {label}")
            continue

        try:
            ytmusic.add_playlist_items(playlist_id, [item["videoId"]])
            added += 1
            logger.info(f"[{i+1}/{total}] Added: {label}")
        except Exception as e:
            failed += 1
            logger.error(f"[{i+1}/{total}] Failed to add: {label} → {e}")
        time.sleep(delay)

    if dupes:
        logger.info(f"Skipped {dupes} duplicates already in playlist")
    return added, failed


# ── Report ─────────────────────────────────────────────────────
def save_report(found, skipped, added, failed, playlist_name):
    """Save a JSON report of the transfer."""
    report = {
        "timestamp": datetime.now().isoformat(),
        "playlist": playlist_name,
        "total_in_csv": len(found) + len(skipped),
        "found_on_ytm": len(found),
        "skipped": len(skipped),
        "added_to_playlist": added,
        "failed_to_add": failed,
        "skipped_songs": [f"{s['artist']} - {s['track']}" for s in skipped],
    }

    report_file = f"report_{playlist_name.replace(' ', '_')}_{datetime.now().strftime('%H%M%S')}.json"
    with open(report_file, "w", encoding="utf-8") as f:
        json.dump(report, f, indent=2, ensure_ascii=False)

    logger.info(f"Report saved to {report_file}")
    return report


# ── Checkpoint ─────────────────────────────────────────────────
CHECKPOINT_FILE = "checkpoint.json"

def save_checkpoint(data):
    """Save progress after every song."""
    with open(CHECKPOINT_FILE, "w", encoding="utf-8") as f:
        json.dump(data, f, indent=2, ensure_ascii=False)

def load_checkpoint():
    """Load checkpoint if it exists."""
    p = Path(CHECKPOINT_FILE)
    if not p.exists():
        return None
    with open(p, encoding="utf-8") as f:
        return json.load(f)

def clear_checkpoint():
    p = Path(CHECKPOINT_FILE)
    if p.exists():
        p.unlink()
        logger.info("Checkpoint cleared")


# ── Main ───────────────────────────────────────────────────────
def main():
    parser = argparse.ArgumentParser(
        description="Transfer Spotify CSV playlist to YouTube Music",
        formatter_class=argparse.RawTextHelpFormatter,
    )
    parser.add_argument("auth", help="browser.json auth file from ytmusicapi")
    parser.add_argument("csv", help="Exportify CSV file")
    parser.add_argument("-p", "--playlist", required=True, help="Playlist name on YT Music")
    parser.add_argument("-d", "--description", default="", help="Playlist description")
    parser.add_argument("--delay", type=float, default=1.5, help="Delay between requests in seconds (default: 1.5)")
    parser.add_argument("--dry-run", action="store_true", help="Search only, don't create playlist")
    parser.add_argument("--skip", type=int, default=0, help="Skip first N songs in CSV (resume from N+1)")
    parser.add_argument("--resume", action="store_true", help="Auto-resume from last checkpoint file")
    parser.add_argument("--cache-only", action="store_true", help="Skip search, add songs using cached videoIds only")

    args = parser.parse_args()

    # auth
    auth_path = Path(args.auth)
    if not auth_path.exists():
        logger.error(f"Auth file not found: {args.auth}")
        sys.exit(1)

    logger.info(f"Logging to {LOG_FILE}")
    logger.info(f"Auth file: {args.auth}")
    logger.info(f"CSV file: {args.csv}")
    logger.info(f"Playlist: {args.playlist}")
    logger.info(f"Delay: {args.delay}s")
    if args.dry_run:
        logger.info("DRY RUN mode — will not create or modify playlists")

    # connect
    logger.info("Connecting to YouTube Music...")
    try:
        ytmusic = YTMusic(args.auth)
    except Exception as e:
        logger.error(f"Failed to connect: {e}")
        sys.exit(1)
    logger.success("Connected to YouTube Music")

    # load csv
    logger.info(f"Loading CSV: {args.csv}")
    songs = load_csv(args.csv)
    logger.info(f"Loaded {len(songs)} songs from CSV")

    if not songs:
        logger.error("No songs found in CSV. Check the file format.")
        sys.exit(1)

    # ── Resume / Skip logic ──
    skip_n = args.skip
    prev_found = []
    prev_skipped = []
    playlist_id = None

    if args.resume:
        ckpt = load_checkpoint()
        if ckpt:
            skip_n = ckpt.get("searched_up_to", 0)
            prev_found = ckpt.get("found", [])
            prev_skipped = ckpt.get("skipped", [])
            playlist_id = ckpt.get("playlist_id")
            logger.info(f"Resuming from checkpoint — skipping first {skip_n} songs")
            logger.info(f"  Previously found: {len(prev_found)}, skipped: {len(prev_skipped)}")
            if playlist_id:
                logger.info(f"  Existing playlist ID: {playlist_id}")
        else:
            logger.warning("No checkpoint file found, starting fresh")

    if skip_n > 0:
        logger.info(f"Skipping first {skip_n} songs")
        songs = songs[skip_n:]
        logger.info(f"Remaining: {len(songs)} songs")

    # search
    logger.info("Starting search...")
    start = time.time()

    # load song cache
    cache = load_cache()
    cache_hits = 0
    logger.info(f"Song cache: {len(cache)} entries loaded from {CACHE_FILE}")

    # modified search loop with checkpointing + cache
    total = len(songs)
    found = list(prev_found)
    skipped = list(prev_skipped)
    search_offset = skip_n

    for i, song in enumerate(songs):
        label = f"{song['artist']} - {song['track']}"
        key = cache_key(song)

        # check cache first
        if key in cache:
            vid = cache[key]
            found.append({"song": song, "videoId": vid})
            cache_hits += 1
            logger.info(f"[{search_offset + i + 1}/{search_offset + total}] CACHED: {label}")
            continue

        # cache-only mode: skip search entirely
        if args.cache_only:
            skipped.append(song)
            logger.warning(f"[{search_offset + i + 1}/{search_offset + total}] Not in cache → SKIPPED: {label}")
            continue

        logger.info(f"[{search_offset + i + 1}/{search_offset + total}] Searching: {label}")
        vid = search_song(ytmusic, song)

        if vid:
            found.append({"song": song, "videoId": vid})
            cache[key] = vid
            save_cache(cache)  # persist after every find
            logger.success(f"  ✓ Found (cached for next time)")
        else:
            skipped.append(song)
            logger.warning(f"  ✗ Not found → SKIPPED")

        # save checkpoint after every song
        save_checkpoint({
            "searched_up_to": search_offset + i + 1,
            "found": found,
            "skipped": skipped,
            "playlist_id": playlist_id,
            "playlist_name": args.playlist,
        })

        if (search_offset + i + 1) % 25 == 0:
            logger.info(f"── Progress: {len(found)} found, {len(skipped)} skipped, {cache_hits} from cache ──")

        time.sleep(args.delay)

    elapsed = time.time() - start
    logger.info(f"Search complete in {elapsed:.0f}s — {len(found)} found ({cache_hits} from cache), {len(skipped)} skipped")

    if not found:
        logger.error("No songs matched. Nothing to add.")
        sys.exit(1)

    # figure out which songs still need to be added
    already_added = set()
    if args.resume and load_checkpoint():
        ckpt = load_checkpoint()
        already_added = set(ckpt.get("added_ids", []))

    to_add = [f for f in found if f["videoId"] not in already_added]

    # add to playlist
    added, failed = len(already_added), 0
    if not args.dry_run:
        if not playlist_id:
            playlist_id = create_or_get_playlist(ytmusic, args.playlist, args.description)

        # save playlist_id to checkpoint
        save_checkpoint({
            "searched_up_to": search_offset + total,
            "found": found,
            "skipped": skipped,
            "playlist_id": playlist_id,
            "playlist_name": args.playlist,
            "added_ids": list(already_added),
        })

        # fetch existing playlist tracks to avoid duplicates
        existing_vids = get_playlist_video_ids(ytmusic, playlist_id)
        to_add = [f for f in to_add if f["videoId"] not in existing_vids]
        logger.info(f"After dedup: {len(to_add)} new songs to add")

        if not to_add:
            logger.info("All songs already in playlist, nothing to add!")
        else:
            logger.info(f"Adding {len(to_add)} songs to playlist...")

        added_ids = list(already_added)
        for i, item in enumerate(to_add):
            label = f"{item['song']['artist']} - {item['song']['track']}"
            try:
                ytmusic.add_playlist_items(playlist_id, [item["videoId"]])
                added += 1
                added_ids.append(item["videoId"])
                logger.info(f"[{i+1}/{len(to_add)}] Added: {label}")
            except Exception as e:
                failed += 1
                logger.error(f"[{i+1}/{len(to_add)}] Failed to add: {label} → {e}")

            # checkpoint after each add
            save_checkpoint({
                "searched_up_to": search_offset + total,
                "found": found,
                "skipped": skipped,
                "playlist_id": playlist_id,
                "playlist_name": args.playlist,
                "added_ids": added_ids,
            })
            time.sleep(args.delay)
    else:
        logger.info("Dry run — skipping playlist creation")

    # summary
    total_songs = search_offset + total
    logger.info("=" * 50)
    logger.info(f"  Total in CSV:      {total_songs}")
    logger.success(f"  Found on YTM:      {len(found)} ({cache_hits} from cache)")
    logger.warning(f"  Not found:         {len(skipped)}")
    if not args.dry_run:
        logger.success(f"  Added to playlist: {added}")
        if failed:
            logger.error(f"  Failed to add:     {failed}")
    logger.info("=" * 50)

    # report
    report = save_report(found, skipped, added, failed, args.playlist)

    if skipped:
        logger.info("Songs not found on YT Music:")
        for s in skipped:
            logger.warning(f"  • {s['artist']} - {s['track']}")

    # clear checkpoint on success
    clear_checkpoint()
    logger.info("Done!")


if __name__ == "__main__":
    main()