import streamlit as st import requests import base64 import os import json from urllib.parse import urlencode, parse_qs, urlparse # Page configuration st.set_page_config( page_title="Spotify Playlist Duplicator", page_icon="🎵", layout="wide" ) # Constants SCOPE = 'playlist-modify-public playlist-modify-private playlist-read-private' AUTH_URL = 'https://accounts.spotify.com/authorize' TOKEN_URL = 'https://accounts.spotify.com/api/token' API_BASE_URL = 'https://api.spotify.com/v1' # Initialize session state for credentials if 'client_id' not in st.session_state: st.session_state.client_id = os.getenv('SPOTIPY_CLIENT_ID', '') if 'client_secret' not in st.session_state: st.session_state.client_secret = os.getenv('SPOTIPY_CLIENT_SECRET', '') if 'redirect_uri' not in st.session_state: st.session_state.redirect_uri = os.getenv('SPOTIPY_REDIRECT_URI', 'https://example.com/callback') # Initialize session state for authentication if 'access_token' not in st.session_state: st.session_state.access_token = None if 'refresh_token' not in st.session_state: st.session_state.refresh_token = None if 'user_info' not in st.session_state: st.session_state.user_info = None if 'oauth_state' not in st.session_state: import secrets st.session_state.oauth_state = secrets.token_urlsafe(16) if 'credentials_confirmed' not in st.session_state: st.session_state.credentials_confirmed = False def get_auth_url(): """Generate Spotify authorization URL""" params = { 'client_id': st.session_state.client_id, 'response_type': 'code', 'redirect_uri': st.session_state.redirect_uri, 'scope': SCOPE, 'state': st.session_state.oauth_state, 'show_dialog': False } return f"{AUTH_URL}?{urlencode(params)}" def get_token_from_code(code): """Exchange authorization code for access token""" auth_header = base64.b64encode(f"{st.session_state.client_id}:{st.session_state.client_secret}".encode()).decode() headers = { 'Authorization': f'Basic {auth_header}', 'Content-Type': 'application/x-www-form-urlencoded' } data = { 'grant_type': 'authorization_code', 'code': code, 'redirect_uri': st.session_state.redirect_uri } response = requests.post(TOKEN_URL, headers=headers, data=data) if response.status_code == 200: return response.json() else: raise Exception(f"Token exchange failed: {response.text}") def refresh_access_token(refresh_token): """Refresh the access token""" auth_header = base64.b64encode(f"{st.session_state.client_id}:{st.session_state.client_secret}".encode()).decode() headers = { 'Authorization': f'Basic {auth_header}', 'Content-Type': 'application/x-www-form-urlencoded' } data = { 'grant_type': 'refresh_token', 'refresh_token': refresh_token } response = requests.post(TOKEN_URL, headers=headers, data=data) if response.status_code == 200: return response.json() else: raise Exception(f"Token refresh failed: {response.text}") def make_spotify_request(endpoint, method='GET', data=None, params=None): """Make authenticated request to Spotify API""" if not st.session_state.access_token: raise Exception("No access token available") headers = { 'Authorization': f'Bearer {st.session_state.access_token}', 'Content-Type': 'application/json' } url = f"{API_BASE_URL}/{endpoint}" try: if method == 'GET': response = requests.get(url, headers=headers, params=params) elif method == 'POST': response = requests.post(url, headers=headers, json=data) elif method == 'DELETE': response = requests.delete(url, headers=headers, json=data) elif method == 'PUT': response = requests.put(url, headers=headers, json=data) else: raise Exception(f"Unsupported method: {method}") # Handle token expiration if response.status_code == 401 and st.session_state.refresh_token: # Try to refresh token token_data = refresh_access_token(st.session_state.refresh_token) st.session_state.access_token = token_data['access_token'] # Retry request headers['Authorization'] = f'Bearer {st.session_state.access_token}' if method == 'GET': response = requests.get(url, headers=headers, params=params) elif method == 'POST': response = requests.post(url, headers=headers, json=data) elif method == 'DELETE': response = requests.delete(url, headers=headers, json=data) elif method == 'PUT': response = requests.put(url, headers=headers, json=data) response.raise_for_status() # Some endpoints return no content if response.status_code == 204 or not response.content: return {} return response.json() except requests.exceptions.HTTPError as e: raise Exception(f"Spotify API error: {response.status_code} - {response.text}") def get_current_user(): """Get current user profile""" return make_spotify_request('me') def get_all_playlists(): """Fetch all playlists for the current user""" playlists = [] offset = 0 limit = 50 while True: params = {'limit': limit, 'offset': offset} results = make_spotify_request('me/playlists', params=params) playlists.extend(results['items']) if not results['next']: break offset += limit return playlists def get_playlist_tracks(playlist_id): """Fetch all tracks from a playlist""" tracks = [] offset = 0 limit = 100 while True: params = {'limit': limit, 'offset': offset} results = make_spotify_request(f'playlists/{playlist_id}/tracks', params=params) tracks.extend(results['items']) if not results['next']: break offset += limit return tracks def get_playlist(playlist_id): """Get playlist details""" return make_spotify_request(f'playlists/{playlist_id}') def clear_playlist(playlist_id, track_uris): """Remove all tracks from a playlist""" # Spotify API allows removing 100 tracks at a time for i in range(0, len(track_uris), 100): batch = track_uris[i:i+100] tracks_to_remove = [{'uri': uri} for uri in batch] data = {'tracks': tracks_to_remove} make_spotify_request(f'playlists/{playlist_id}/tracks', method='DELETE', data=data) def add_tracks_to_playlist(playlist_id, track_uris): """Add tracks to a playlist""" # Spotify API allows adding 100 tracks at a time for i in range(0, len(track_uris), 100): batch = track_uris[i:i+100] data = {'uris': batch} make_spotify_request(f'playlists/{playlist_id}/tracks', method='POST', data=data) def duplicate_playlist(source_playlist_id, destination_playlist_ids, progress_callback=None): """Duplicate tracks from source playlist to destination playlists""" results = [] # Get tracks from source playlist if progress_callback: progress_callback("Fetching tracks from source playlist...") track_items = get_playlist_tracks(source_playlist_id) track_uris = [item['track']['uri'] for item in track_items if item['track'] and item['track']['uri']] if not track_uris: return [{"success": False, "message": "No tracks found in source playlist"}] # Add tracks to each destination playlist for idx, dest_id in enumerate(destination_playlist_ids): try: if progress_callback: progress_callback(f"Copying to playlist {idx + 1} of {len(destination_playlist_ids)}...") # Clear destination playlist first dest_tracks = get_playlist_tracks(dest_id) if dest_tracks: dest_track_uris = [item['track']['uri'] for item in dest_tracks if item['track'] and item['track']['uri']] clear_playlist(dest_id, dest_track_uris) # Add tracks from source add_tracks_to_playlist(dest_id, track_uris) dest_playlist = get_playlist(dest_id) results.append({ "success": True, "playlist_name": dest_playlist['name'], "tracks_copied": len(track_uris) }) except Exception as e: results.append({ "success": False, "playlist_id": dest_id, "error": str(e) }) return results # Main app st.title("🎵 Spotify Playlist Duplicator") st.markdown("Duplicate a Spotify playlist to one or more destination playlists") # Step 1: Collect Spotify App Credentials if not st.session_state.credentials_confirmed: st.markdown("## 🔧 Step 1: Configure Your Spotify App") st.info("You need to create a Spotify Developer App to use this tool. This allows the app to access your Spotify account.") with st.expander("📚 How to create a Spotify App (click to expand)", expanded=True): st.markdown(""" 1. Go to [Spotify Developer Dashboard](https://developer.spotify.com/dashboard) 2. Log in with your Spotify account 3. Click **"Create app"** 4. Fill in the details: - **App name**: "Playlist Duplicator" (or any name) - **App description**: "Personal playlist management tool" - **Redirect URI**: `https://example.com/callback` - Check the Terms of Service box 5. Click **"Save"** 6. On your app page, click **"Settings"** 7. Note your **Client ID** and **Client Secret** (click "View client secret") 8. Add your Spotify email to **"User Management"** section """) st.markdown("### Enter Your Spotify App Credentials:") with st.form("credentials_form"): client_id = st.text_input( "Client ID", value=st.session_state.client_id, help="From your Spotify Developer Dashboard" ) client_secret = st.text_input( "Client Secret", value=st.session_state.client_secret, type="password", help="Click 'View client secret' in your Spotify app settings" ) redirect_uri = st.text_input( "Redirect URI", value=st.session_state.redirect_uri, help="Should be: https://example.com/callback" ) col1, col2 = st.columns([1, 3]) with col1: submitted = st.form_submit_button("✅ Save & Continue", type="primary", use_container_width=True) with col2: if st.session_state.client_id and st.session_state.client_secret: st.caption("✓ Credentials already saved. Click to update.") if submitted: if client_id and client_secret and redirect_uri: st.session_state.client_id = client_id.strip() st.session_state.client_secret = client_secret.strip() st.session_state.redirect_uri = redirect_uri.strip() st.session_state.credentials_confirmed = True st.success("✅ Credentials saved! Proceeding to authentication...") st.rerun() else: st.error("❌ Please fill in all fields") st.stop() # Add option to change credentials with st.sidebar: st.markdown("### ⚙️ Settings") if st.button("🔧 Change Credentials"): st.session_state.credentials_confirmed = False st.session_state.access_token = None st.session_state.refresh_token = None st.session_state.user_info = None st.rerun() # Try to refresh access token if we have a refresh token but no access token if not st.session_state.access_token and st.session_state.refresh_token: try: with st.spinner("🔄 Refreshing authentication..."): token_data = refresh_access_token(st.session_state.refresh_token) st.session_state.access_token = token_data['access_token'] if not st.session_state.user_info: st.session_state.user_info = get_current_user() except Exception as e: # Refresh token expired or invalid, need to re-authenticate st.session_state.access_token = None st.session_state.refresh_token = None st.session_state.user_info = None # Step 2: Authenticate with Spotify if not st.session_state.access_token: st.markdown("## 🎵 Step 2: Connect Your Spotify Account") st.info("You need to authorize this app to access your Spotify playlists.") # Generate auth URL auth_url = get_auth_url() # Create two-step process st.markdown("""

📝 Authorization Steps:

  1. Click the button below to open Spotify authorization
  2. Log in and approve the app permissions
  3. After approving, you'll see an error page - that's normal!
  4. Copy the code from the URL (the part after ?code=)
  5. Paste the code in the box below
""", unsafe_allow_html=True) # Authorization button st.markdown(f"""
""", unsafe_allow_html=True) st.markdown("---") st.markdown("### 📋 Paste Authorization Code") # Code input with visual example col1, col2 = st.columns([3, 1]) with col1: auth_code = st.text_input( "Paste the authorization code here:", placeholder="AQBq8uZ7X...", help="After approving on Spotify, copy everything after '?code=' in the URL", label_visibility="collapsed" ) with col2: submit_button = st.button("✅ Submit", type="primary", use_container_width=True) # Help section with visual with st.expander("❓ How do I find the code?", expanded=False): st.markdown(""" After clicking "Authorize with Spotify" and approving: 1. **You'll see an error page** - This is expected! Don't worry. 2. **Look at the URL in your browser's address bar**. It will look like: ``` https://example.com/callback?code=AQBq8uZ7X2k... ``` 3. **Copy only the code part** (everything after `?code=`): ``` AQBq8uZ7X2k... ``` 4. **Paste it in the box above** and click Submit. **Visual Example:** URL in address bar: `https://example.com/callback?code=`**`AQBq8uZ7X2k3L...`** ← Copy this part **Troubleshooting:** - Make sure you copy the ENTIRE code (it's usually quite long) - Don't include `?code=`, only the code itself - The code expires after a few minutes, so paste it promptly """) # Process the authorization code if submit_button and auth_code: auth_code = auth_code.strip() # Remove common mistakes (if user included the prefix) if '?code=' in auth_code: auth_code = auth_code.split('?code=')[1] if '&' in auth_code: auth_code = auth_code.split('&')[0] with st.spinner("🔄 Authenticating with Spotify..."): try: # Exchange code for tokens token_data = get_token_from_code(auth_code) st.session_state.access_token = token_data['access_token'] st.session_state.refresh_token = token_data.get('refresh_token') # Get user info st.session_state.user_info = get_current_user() st.success(f"✅ Successfully authenticated as **{st.session_state.user_info.get('display_name', 'Unknown')}**!") st.balloons() st.rerun() except Exception as e: st.error(f"❌ Authentication failed: {str(e)}") st.error("**Possible reasons:**") st.write("- The code may have expired (they only last a few minutes)") st.write("- The code may be incomplete or incorrect") st.write("- You may have already used this code") st.info("💡 **Solution:** Click the authorization button again to get a new code.") elif submit_button and not auth_code: st.warning("⚠️ Please paste the authorization code first.") st.stop() # Show authenticated user info with logout option col1, col2 = st.columns([3, 1]) with col1: if st.session_state.user_info: st.success(f"✅ Authenticated as: **{st.session_state.user_info.get('display_name', 'Unknown')}** ({st.session_state.user_info['id']})") else: try: st.session_state.user_info = get_current_user() st.success(f"✅ Authenticated as: **{st.session_state.user_info.get('display_name', 'Unknown')}** ({st.session_state.user_info['id']})") except Exception as e: st.error(f"Failed to get user info: {str(e)}") st.session_state.access_token = None st.session_state.refresh_token = None st.session_state.user_info = None if st.button("Re-authenticate"): st.rerun() st.stop() with col2: if st.button("🚪 Logout", use_container_width=True): st.session_state.access_token = None st.session_state.refresh_token = None st.session_state.user_info = None st.rerun() # Fetch user's playlists try: with st.spinner("Loading your playlists..."): playlists = get_all_playlists() st.info(f"Found {len(playlists)} playlists") if not playlists: st.warning("No playlists found in your account") st.stop() except Exception as e: st.error(f"Error fetching playlists: {str(e)}") st.error(f"Error type: {type(e).__name__}") st.stop() # Create playlist selection UI st.markdown("---") col1, col2 = st.columns(2) with col1: st.subheader("Source Playlist") source_options = {f"{p['name']} ({p['tracks']['total']} tracks)": p['id'] for p in playlists} source_selection = st.selectbox( "Select the playlist to copy from:", options=list(source_options.keys()), key="source" ) source_playlist_id = source_options[source_selection] with col2: st.subheader("Destination Playlist(s)") # Filter out the source playlist from destination options dest_options = {f"{p['name']} ({p['tracks']['total']} tracks)": p['id'] for p in playlists if p['id'] != source_playlist_id} dest_selections = st.multiselect( "Select one or more playlists to copy to:", options=list(dest_options.keys()), key="destinations" ) destination_playlist_ids = [dest_options[sel] for sel in dest_selections] # Preview if source_playlist_id and destination_playlist_ids: st.markdown("---") st.subheader("Preview") try: source_playlist = get_playlist(source_playlist_id) st.write(f"**Source:** {source_playlist['name']} ({source_playlist['tracks']['total']} tracks)") st.write(f"**Destinations:** {len(destination_playlist_ids)} playlist(s)") for dest_id in destination_playlist_ids: dest_playlist = get_playlist(dest_id) st.write(f" - {dest_playlist['name']}") except Exception as e: st.error(f"Error loading preview: {str(e)}") st.warning("⚠️ This will replace all tracks in the destination playlist(s) with tracks from the source playlist.") # Duplicate button st.markdown("---") if st.button("🎵 Duplicate Playlist", type="primary", disabled=not destination_playlist_ids): if destination_playlist_ids: progress_placeholder = st.empty() def update_progress(message): progress_placeholder.info(message) with st.spinner("Duplicating playlist..."): results = duplicate_playlist( source_playlist_id, destination_playlist_ids, progress_callback=update_progress ) progress_placeholder.empty() # Display results st.markdown("### Results") for result in results: if result['success']: st.success(f"✅ Successfully copied {result['tracks_copied']} tracks to **{result['playlist_name']}**") else: st.error(f"❌ Failed: {result.get('error', 'Unknown error')}") st.balloons() else: st.error("Please select at least one destination playlist")