Spotify to YouTube Playlist Sync

yt_playlist_sync.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "playwright",
#     "google-api-python-client",
#     "google-auth-oauthlib",
#     "google-auth-httplib2",
#     "keyring",
#     "rich",
# ]
# ///

import os
import json
import sys
import argparse
import keyring
import time
import logging
from playwright.sync_api import sync_playwright
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
from googleapiclient.discovery import build
from rich.logging import RichHandler
from rich.console import Console

# --- Setup Logging ---
stderr_console = Console(stderr=True)
logging.basicConfig(
    level="INFO",
    format="%(message)s",
    datefmt="[%X]",
    handlers=[RichHandler(rich_tracebacks=True, console=stderr_console)],
)
log = logging.getLogger("rich")

# --- Configuration ---
SCOPES = ["https://www.googleapis.com/auth/youtube.force-ssl"]
KEYRING_SERVICE = "youtube_playlist_sync"
KEYRING_USER = "oauth_token"
AUTH_FILE = "spotify_auth.json"


def get_spotify_tracks(playlist_url):
    """Scrapes Spotify tracks using a saved auth state and strict recommendation filtering."""
    with sync_playwright() as p:
        log.info("Launching isolated Chrome instance...")
        browser = p.chromium.launch(headless=False, channel="chrome")

        context_args = {}
        if os.path.exists(AUTH_FILE):
            log.info(f"Loading saved session from {AUTH_FILE}")
            context_args["storage_state"] = AUTH_FILE

        context = browser.new_context(**context_args)
        page = context.new_page()
        log.info(f"Navigating to: {playlist_url}")
        page.goto(playlist_url)

        try:
            # Check for login requirement
            if (
                page.locator('[data-testid="login-button"]').is_visible()
                or "https://api.spotify.com/v1/albums/4aawyAB9vmqN3uQ7FjRGTy1"
                in page.url
            ):
                log.warning(
                    "Not logged in. Please log in to Spotify in the browser window now."
                )
                page.wait_for_selector('[data-testid="tracklist-row"]', timeout=0)
                context.storage_state(path=AUTH_FILE)
                log.info(f"Login successful! Session saved to {AUTH_FILE}")
            else:
                page.wait_for_selector('[data-testid="tracklist-row"]', timeout=30000)

        except Exception as e:
            log.error(f"Error during navigation/login: {e}")
            browser.close()
            return []

        log.info("Scrolling to capture all tracks...")
        for i in range(10):
            page.keyboard.press("PageDown")
            time.sleep(0.5)

        tracks = page.evaluate("""
            () => {
                const rows = document.querySelectorAll('[data-testid="tracklist-row"]');
                const results = [];
                rows.forEach(row => {
                    // Filter 1: Check if inside a section labeled 'Recommended'
                    const isRecSection = row.closest('[aria-label="Recommended"]') || 
                                        row.closest('section')?.querySelector('h2')?.innerText.includes('Recommended');
                    
                    // Filter 2: Explicitly check for the recommended-track testid
                    const isRecTestId = row.querySelector('[data-testid="recommended-track"]') || 
                                       row.closest('[data-testid="recommended-track"]');

                    if (!isRecSection && !isRecTestId) {
                        const titleEl = row.querySelector('div[dir="auto"]') || row.querySelector('a[href^="/track/"]');
                        const artistEls = row.querySelectorAll('a[href^="/artist/"]');
                        if (titleEl) {
                            results.push({
                                title: titleEl.innerText,
                                artist: Array.from(artistEls).map(a => a.innerText).join(", ")
                            });
                        }
                    }
                });
                return results;
            }
        """)

        log.info(f"Extracted {len(tracks)} tracks.")
        browser.close()
        return tracks


def get_authenticated_service(client_secrets_file):
    creds = None
    saved_token = keyring.get_password(KEYRING_SERVICE, KEYRING_USER)
    if saved_token:
        try:
            creds = Credentials.from_authorized_user_info(
                json.loads(saved_token), SCOPES
            )
        except:
            log.warning("Saved credentials invalid, re-authenticating.")

    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            flow = InstalledAppFlow.from_client_secrets_file(
                client_secrets_file, SCOPES
            )
            creds = flow.run_local_server(port=0)
        keyring.set_password(KEYRING_SERVICE, KEYRING_USER, creds.to_json())

    return build("youtube", "v3", credentials=creds)


def get_or_create_playlist(youtube, title):
    request = youtube.playlists().list(part="snippet,status", mine=True, maxResults=50)
    response = request.execute()
    for item in response.get("items", []):
        if item["snippet"]["title"] == title:
            log.info(f"Found existing YouTube playlist: {title}")
            return item["id"]

    log.info(f"Creating new PUBLIC YouTube playlist: {title}")
    request = youtube.playlists().insert(
        part="snippet,status",
        body={
            "snippet": {"title": title, "description": "Synced from Spotify"},
            "status": {"privacyStatus": "public"},  # Set to Public
        },
    )
    res = request.execute()
    return res["id"]


def find_and_add_track(youtube, playlist_id, title, artist):
    query = f"{artist} {title} official audio"
    search_res = (
        youtube.search()
        .list(q=query, part="id", maxResults=1, type="video", videoCategoryId="10")
        .execute()
    )

    items = search_res.get("items", [])
    if items:
        video_id = items[0]["id"]["videoId"]
        youtube.playlistItems().insert(
            part="snippet",
            body={
                "snippet": {
                    "playlistId": playlist_id,
                    "resourceId": {"kind": "youtube#video", "videoId": video_id},
                }
            },
        ).execute()
        return True
    return False


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--url", required=True, help="Spotify Playlist URL")
    parser.add_argument("--name", required=True, help="New YouTube Playlist Name")
    parser.add_argument("--secrets", default="client_secret.json")
    args = parser.parse_args()

    if not os.path.exists(args.secrets):
        log.error(f"Secrets file {args.secrets} not found.")
        return

    tracks = get_spotify_tracks(args.url)

    if not tracks:
        log.error("No tracks found. Did the playlist load correctly?")
        return

    youtube = get_authenticated_service(args.secrets)
    playlist_id = get_or_create_playlist(youtube, args.name)

    for track in tracks:
        log.info(f"Syncing: {track['artist']} - {track['title']}")
        try:
            find_and_add_track(youtube, playlist_id, track["title"], track["artist"])
        except Exception as e:
            log.error(f"Failed to add track {track['title']}: {e}")

    # Output the final URL
    playlist_url = f"https:///youtube.com/playlist?list={playlist_id}"
    print(playlist_url)


if __name__ == "__main__":
    main()