Introduction and Overview
Creating an intelligent playlist manager that understands natural language commands represents a fascinating intersection of artificial intelligence, music streaming APIs, and user experience design. This comprehensive tutorial will guide you through building a sophisticated agent that can interpret user requests like "Create a playlist with Christmas titles" or "Add some jazz from the 1960s to my chill playlist" and execute these commands seamlessly with Spotify's platform.
The system we will construct combines the power of Large Language Models (LLMs) with Spotify's Web API to create an intelligent intermediary that translates human intentions into concrete playlist operations. Our agent will be capable of understanding context, handling ambiguous requests, and performing complex playlist manipulations that would traditionally require multiple manual steps.
The architecture we will implement follows clean code principles and modular design patterns, ensuring maintainability and extensibility. We will explore both local and remote LLM implementations, providing flexibility for different deployment scenarios and privacy requirements.
System Architecture and Design Principles
The foundation of our playlist manager agent rests on a multi-layered architecture that separates concerns and promotes code reusability. At the highest level, we have the user interface layer that accepts natural language input. This feeds into the LLM processing layer, which interprets the user's intent and generates structured commands. These commands are then processed by our business logic layer, which coordinates with the Spotify API integration layer to execute the requested operations.
The design follows the principle of dependency inversion, where high-level modules do not depend on low-level modules, but both depend on abstractions. This approach allows us to swap out different LLM providers or music streaming services without affecting the core business logic.
Our agent will maintain state information about current playlists, user preferences, and session context to provide more intelligent responses. The system will also implement error handling and recovery mechanisms to gracefully handle API limitations, network issues, and ambiguous user requests.
Setting Up the Development Environment
Before diving into the implementation, we need to establish a robust development environment that supports our various dependencies and requirements. The foundation of our system will be built using Python, leveraging its rich ecosystem of libraries for API integration, natural language processing, and machine learning.
# requirements.txt
spotipy==2.22.1
openai==1.3.0
transformers==4.35.0
torch==2.1.0
python-dotenv==1.0.0
flask==2.3.3
requests==2.31.0
pandas==2.1.3
numpy==1.24.3
pydantic==2.5.0
The environment setup requires careful consideration of Python version compatibility and virtual environment management. We recommend using Python 3.9 or higher to ensure compatibility with the latest versions of the transformer libraries and API clients.
# environment_setup.py
import os
import sys
from pathlib import Path
def setup_environment():
"""
Initialize the development environment with proper configuration
and dependency verification.
"""
# Verify Python version compatibility
if sys.version_info < (3, 9):
raise RuntimeError("Python 3.9 or higher is required")
# Create necessary directories
directories = ['config', 'logs', 'data', 'models']
for directory in directories:
Path(directory).mkdir(exist_ok=True)
# Verify environment variables
required_env_vars = [
'SPOTIFY_CLIENT_ID',
'SPOTIFY_CLIENT_SECRET',
'SPOTIFY_REDIRECT_URI'
]
missing_vars = []
for var in required_env_vars:
if not os.getenv(var):
missing_vars.append(var)
if missing_vars:
raise EnvironmentError(
f"Missing required environment variables: {', '.join(missing_vars)}"
)
print("Environment setup completed successfully")
if __name__ == "__main__":
setup_environment()
Spotify API Integration and Authentication
The integration with Spotify's Web API forms the backbone of our playlist management capabilities. Spotify uses OAuth 2.0 for authentication, which requires careful handling of access tokens and refresh tokens to maintain persistent access to user accounts.
The Spotify Web API provides comprehensive endpoints for playlist management, track searching, and user library access. Our implementation will create a robust wrapper around the Spotipy library that handles authentication, rate limiting, and error recovery automatically.
# spotify_client.py
import spotipy
from spotipy.oauth2 import SpotifyOAuth
import os
import time
import logging
from typing import List, Dict, Optional, Any
class SpotifyPlaylistManager:
"""
A comprehensive wrapper for Spotify API operations with built-in
error handling, rate limiting, and authentication management.
"""
def __init__(self):
"""
Initialize the Spotify client with OAuth authentication.
"""
self.client_id = os.getenv('SPOTIFY_CLIENT_ID')
self.client_secret = os.getenv('SPOTIFY_CLIENT_SECRET')
self.redirect_uri = os.getenv('SPOTIFY_REDIRECT_URI')
# Define the scope of permissions needed
self.scope = (
"playlist-read-private playlist-read-collaborative "
"playlist-modify-private playlist-modify-public "
"user-library-read user-library-modify "
"user-read-playback-state user-modify-playback-state"
)
# Initialize OAuth manager
self.auth_manager = SpotifyOAuth(
client_id=self.client_id,
client_secret=self.client_secret,
redirect_uri=self.redirect_uri,
scope=self.scope,
cache_path=".spotify_cache"
)
# Initialize Spotify client
self.spotify = spotipy.Spotify(auth_manager=self.auth_manager)
# Setup logging
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
# Rate limiting parameters
self.last_request_time = 0
self.min_request_interval = 0.1 # 100ms between requests
def _rate_limit(self):
"""
Implement rate limiting to respect Spotify API constraints.
"""
current_time = time.time()
time_since_last_request = current_time - self.last_request_time
if time_since_last_request < self.min_request_interval:
sleep_time = self.min_request_interval - time_since_last_request
time.sleep(sleep_time)
self.last_request_time = time.time()
def get_user_playlists(self) -> List[Dict[str, Any]]:
"""
Retrieve all playlists for the authenticated user.
Returns:
List of playlist dictionaries containing id, name, and metadata
"""
self._rate_limit()
try:
playlists = []
results = self.spotify.current_user_playlists(limit=50)
while results:
playlists.extend(results['items'])
if results['next']:
results = self.spotify.next(results)
else:
break
self.logger.info(f"Retrieved {len(playlists)} playlists")
return playlists
except Exception as e:
self.logger.error(f"Error retrieving playlists: {str(e)}")
raise
def create_playlist(self, name: str, description: str = "",
public: bool = False) -> Dict[str, Any]:
"""
Create a new playlist for the authenticated user.
Args:
name: The name of the playlist
description: Optional description for the playlist
public: Whether the playlist should be public
Returns:
Dictionary containing the created playlist information
"""
self._rate_limit()
try:
user_id = self.spotify.current_user()['id']
playlist = self.spotify.user_playlist_create(
user=user_id,
name=name,
public=public,
description=description
)
self.logger.info(f"Created playlist '{name}' with ID: {playlist['id']}")
return playlist
except Exception as e:
self.logger.error(f"Error creating playlist '{name}': {str(e)}")
raise
def search_tracks(self, query: str, limit: int = 20) -> List[Dict[str, Any]]:
"""
Search for tracks based on a query string.
Args:
query: Search query (can include artist, album, genre, etc.)
limit: Maximum number of results to return
Returns:
List of track dictionaries
"""
self._rate_limit()
try:
results = self.spotify.search(q=query, type='track', limit=limit)
tracks = results['tracks']['items']
self.logger.info(f"Found {len(tracks)} tracks for query: '{query}'")
return tracks
except Exception as e:
self.logger.error(f"Error searching tracks: {str(e)}")
raise
def add_tracks_to_playlist(self, playlist_id: str,
track_uris: List[str]) -> bool:
"""
Add tracks to an existing playlist.
Args:
playlist_id: The Spotify ID of the target playlist
track_uris: List of Spotify track URIs to add
Returns:
True if successful, False otherwise
"""
self._rate_limit()
try:
# Spotify API allows maximum 100 tracks per request
batch_size = 100
for i in range(0, len(track_uris), batch_size):
batch = track_uris[i:i + batch_size]
self.spotify.playlist_add_items(playlist_id, batch)
if i + batch_size < len(track_uris):
self._rate_limit() # Rate limit between batches
self.logger.info(
f"Added {len(track_uris)} tracks to playlist {playlist_id}"
)
return True
except Exception as e:
self.logger.error(
f"Error adding tracks to playlist {playlist_id}: {str(e)}"
)
return False
The Spotify client implementation provides a foundation for all playlist operations while handling the complexities of API authentication and rate limiting. The rate limiting mechanism ensures we respect Spotify's API constraints, while the comprehensive error handling provides resilience against network issues and API changes.
LLM Integration and Intent Recognition
The heart of our intelligent playlist manager lies in its ability to understand and interpret natural language commands. We will implement a flexible LLM integration system that supports both local and remote language models, allowing users to choose based on their privacy requirements and computational resources.
The intent recognition system will parse user commands and extract structured information including the desired action, target playlists, search criteria, and any additional parameters. This structured data will then be used to orchestrate the appropriate Spotify API calls.
# llm_processor.py
import openai
import json
import re
from typing import Dict, List, Optional, Any, Union
from dataclasses import dataclass
from enum import Enum
import logging
class ActionType(Enum):
"""Enumeration of supported playlist actions."""
CREATE_PLAYLIST = "create_playlist"
DELETE_PLAYLIST = "delete_playlist"
RENAME_PLAYLIST = "rename_playlist"
DUPLICATE_PLAYLIST = "duplicate_playlist"
ADD_TRACKS = "add_tracks"
REMOVE_TRACKS = "remove_tracks"
SEARCH_TRACKS = "search_tracks"
PLAY_PLAYLIST = "play_playlist"
REORDER_PLAYLIST = "reorder_playlist"
SEARCH_PLAYLISTS = "search_playlists"
MERGE_PLAYLISTS = "merge_playlists"
@dataclass
class PlaylistAction:
"""
Structured representation of a user's playlist management intent.
"""
action_type: ActionType
playlist_name: Optional[str] = None
playlist_id: Optional[str] = None
search_criteria: Optional[Dict[str, str]] = None
track_details: Optional[List[str]] = None
position: Optional[str] = None # "beginning", "middle", "end"
new_name: Optional[str] = None
source_playlist: Optional[str] = None
target_playlist: Optional[str] = None
confidence: float = 0.0
class LLMProcessor:
"""
Handles natural language processing using Large Language Models
to interpret user commands and extract structured playlist actions.
"""
def __init__(self, model_type: str = "openai", model_name: str = "gpt-3.5-turbo"):
"""
Initialize the LLM processor with specified model configuration.
Args:
model_type: Type of LLM to use ("openai", "local", "huggingface")
model_name: Specific model identifier
"""
self.model_type = model_type
self.model_name = model_name
self.logger = logging.getLogger(__name__)
# Initialize the appropriate LLM client
if model_type == "openai":
self.client = openai.OpenAI(api_key=os.getenv('OPENAI_API_KEY'))
elif model_type == "local":
self._initialize_local_model()
# Define the system prompt for intent recognition
self.system_prompt = self._create_system_prompt()
def _create_system_prompt(self) -> str:
"""
Create a comprehensive system prompt that guides the LLM
in understanding playlist management commands.
"""
return """
You are an intelligent playlist management assistant that interprets
natural language commands for Spotify playlist operations. Your task
is to analyze user input and extract structured information about their
intent.
For each user command, identify:
1. The primary action they want to perform
2. Target playlist names or identifiers
3. Search criteria (genre, artist, time period, mood, etc.)
4. Specific track or album requests
5. Position preferences (beginning, middle, end)
6. Any additional parameters
Respond with a JSON object containing:
{
"action_type": "one of: create_playlist, delete_playlist, rename_playlist,
duplicate_playlist, add_tracks, remove_tracks, search_tracks,
play_playlist, reorder_playlist, search_playlists, merge_playlists",
"playlist_name": "target playlist name if specified",
"search_criteria": {
"genre": "music genre if specified",
"artist": "artist name if specified",
"album": "album name if specified",
"time_period": "time period if specified (e.g., '1980s', '2000-2010')",
"mood": "mood or theme if specified",
"keywords": "other relevant keywords"
},
"track_details": ["specific track names if mentioned"],
"position": "beginning/middle/end if specified",
"new_name": "new name for rename operations",
"source_playlist": "source playlist for copy/merge operations",
"target_playlist": "target playlist for merge operations",
"confidence": "confidence score from 0.0 to 1.0"
}
Examples:
- "Create a playlist with Christmas songs" → create_playlist with genre: "Christmas"
- "Add some jazz from the 1960s to my chill playlist" → add_tracks with genre: "jazz", time_period: "1960s", target_playlist: "chill"
- "Remove all songs by Taylor Swift from my pop playlist" → remove_tracks with artist: "Taylor Swift", target_playlist: "pop"
"""
def process_command(self, user_input: str) -> PlaylistAction:
"""
Process a natural language command and extract structured intent.
Args:
user_input: The user's natural language command
Returns:
PlaylistAction object containing structured intent information
"""
try:
if self.model_type == "openai":
response = self._process_with_openai(user_input)
elif self.model_type == "local":
response = self._process_with_local_model(user_input)
else:
raise ValueError(f"Unsupported model type: {self.model_type}")
# Parse the response and create PlaylistAction
action = self._parse_llm_response(response, user_input)
self.logger.info(
f"Processed command: '{user_input}' → {action.action_type.value}"
)
return action
except Exception as e:
self.logger.error(f"Error processing command: {str(e)}")
# Return a fallback action for error cases
return PlaylistAction(
action_type=ActionType.SEARCH_TRACKS,
search_criteria={"keywords": user_input},
confidence=0.1
)
def _process_with_openai(self, user_input: str) -> str:
"""
Process the user input using OpenAI's API.
Args:
user_input: The user's command
Returns:
JSON string response from the LLM
"""
try:
response = self.client.chat.completions.create(
model=self.model_name,
messages=[
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_input}
],
temperature=0.1, # Low temperature for consistent parsing
max_tokens=500
)
return response.choices[0].message.content
except Exception as e:
self.logger.error(f"OpenAI API error: {str(e)}")
raise
def _parse_llm_response(self, response: str, original_input: str) -> PlaylistAction:
"""
Parse the LLM response and create a structured PlaylistAction.
Args:
response: JSON response from the LLM
original_input: Original user input for fallback processing
Returns:
PlaylistAction object
"""
try:
# Clean the response to extract JSON
json_match = re.search(r'\{.*\}', response, re.DOTALL)
if json_match:
json_str = json_match.group()
parsed = json.loads(json_str)
else:
# Fallback to simple keyword extraction
return self._fallback_parsing(original_input)
# Convert to PlaylistAction
action_type = ActionType(parsed.get('action_type', 'search_tracks'))
return PlaylistAction(
action_type=action_type,
playlist_name=parsed.get('playlist_name'),
search_criteria=parsed.get('search_criteria', {}),
track_details=parsed.get('track_details', []),
position=parsed.get('position'),
new_name=parsed.get('new_name'),
source_playlist=parsed.get('source_playlist'),
target_playlist=parsed.get('target_playlist'),
confidence=float(parsed.get('confidence', 0.8))
)
except (json.JSONDecodeError, ValueError, KeyError) as e:
self.logger.warning(f"Failed to parse LLM response: {str(e)}")
return self._fallback_parsing(original_input)
def _fallback_parsing(self, user_input: str) -> PlaylistAction:
"""
Fallback parsing using simple keyword matching when LLM parsing fails.
Args:
user_input: The original user command
Returns:
PlaylistAction with basic intent recognition
"""
user_input_lower = user_input.lower()
# Simple keyword-based action detection
if any(word in user_input_lower for word in ['create', 'make', 'new']):
action_type = ActionType.CREATE_PLAYLIST
elif any(word in user_input_lower for word in ['delete', 'remove playlist']):
action_type = ActionType.DELETE_PLAYLIST
elif any(word in user_input_lower for word in ['add', 'insert']):
action_type = ActionType.ADD_TRACKS
elif any(word in user_input_lower for word in ['play', 'start']):
action_type = ActionType.PLAY_PLAYLIST
else:
action_type = ActionType.SEARCH_TRACKS
# Extract basic search criteria
search_criteria = {}
# Genre detection
genres = ['rock', 'pop', 'jazz', 'classical', 'hip-hop', 'country', 'electronic']
for genre in genres:
if genre in user_input_lower:
search_criteria['genre'] = genre
break
# Time period detection
time_patterns = [
r'(\d{4})s', # 1980s, 1990s
r'(\d{4})-(\d{4})', # 1980-1990
r'(sixties|seventies|eighties|nineties)'
]
for pattern in time_patterns:
match = re.search(pattern, user_input_lower)
if match:
search_criteria['time_period'] = match.group()
break
return PlaylistAction(
action_type=action_type,
search_criteria=search_criteria,
confidence=0.6
)
The LLM processor serves as the intelligent core of our system, translating natural language into structured commands. The implementation includes robust error handling and fallback mechanisms to ensure the system remains functional even when the LLM fails to parse commands correctly.
Command Orchestration and Business Logic
The orchestration layer coordinates between the LLM processor and the Spotify client to execute complex playlist operations. This component implements the business logic that determines how to fulfill user requests, including handling multi-step operations and managing state between commands.
# playlist_orchestrator.py
import logging
from typing import List, Dict, Optional, Any
import random
from datetime import datetime
from llm_processor import LLMProcessor, PlaylistAction, ActionType
from spotify_client import SpotifyPlaylistManager
class PlaylistOrchestrator:
"""
Orchestrates playlist operations by coordinating between LLM processing
and Spotify API interactions. Handles complex multi-step operations
and maintains session state.
"""
def __init__(self, llm_processor: LLMProcessor,
spotify_manager: SpotifyPlaylistManager):
"""
Initialize the orchestrator with LLM and Spotify components.
Args:
llm_processor: Configured LLM processor for intent recognition
spotify_manager: Configured Spotify API manager
"""
self.llm = llm_processor
self.spotify = spotify_manager
self.logger = logging.getLogger(__name__)
# Cache for frequently accessed data
self.playlist_cache = {}
self.last_cache_update = None
# Session state for context-aware operations
self.session_context = {
'last_created_playlist': None,
'last_searched_tracks': [],
'current_playlist': None
}
def execute_command(self, user_input: str) -> Dict[str, Any]:
"""
Execute a natural language playlist command.
Args:
user_input: Natural language command from the user
Returns:
Dictionary containing execution results and status information
"""
try:
# Process the command with LLM
action = self.llm.process_command(user_input)
# Log the interpreted action
self.logger.info(
f"Executing action: {action.action_type.value} "
f"(confidence: {action.confidence:.2f})"
)
# Route to appropriate handler based on action type
if action.action_type == ActionType.CREATE_PLAYLIST:
result = self._handle_create_playlist(action)
elif action.action_type == ActionType.DELETE_PLAYLIST:
result = self._handle_delete_playlist(action)
elif action.action_type == ActionType.ADD_TRACKS:
result = self._handle_add_tracks(action)
elif action.action_type == ActionType.REMOVE_TRACKS:
result = self._handle_remove_tracks(action)
elif action.action_type == ActionType.SEARCH_TRACKS:
result = self._handle_search_tracks(action)
elif action.action_type == ActionType.PLAY_PLAYLIST:
result = self._handle_play_playlist(action)
elif action.action_type == ActionType.RENAME_PLAYLIST:
result = self._handle_rename_playlist(action)
elif action.action_type == ActionType.DUPLICATE_PLAYLIST:
result = self._handle_duplicate_playlist(action)
elif action.action_type == ActionType.MERGE_PLAYLISTS:
result = self._handle_merge_playlists(action)
else:
result = {
'success': False,
'message': f"Unsupported action: {action.action_type.value}",
'action': action.action_type.value
}
# Update session context
self._update_session_context(action, result)
return result
except Exception as e:
self.logger.error(f"Error executing command: {str(e)}")
return {
'success': False,
'message': f"An error occurred: {str(e)}",
'action': 'error'
}
def _handle_create_playlist(self, action: PlaylistAction) -> Dict[str, Any]:
"""
Handle playlist creation with automatic track population.
Args:
action: Parsed action containing playlist creation details
Returns:
Dictionary with creation results
"""
try:
# Generate playlist name if not specified
playlist_name = action.playlist_name
if not playlist_name:
playlist_name = self._generate_playlist_name(action.search_criteria)
# Create the playlist
playlist = self.spotify.create_playlist(
name=playlist_name,
description=self._generate_playlist_description(action.search_criteria)
)
# Populate with tracks if search criteria provided
tracks_added = 0
if action.search_criteria:
tracks = self._find_tracks_by_criteria(action.search_criteria)
if tracks:
track_uris = [track['uri'] for track in tracks[:50]] # Limit to 50 tracks
if self.spotify.add_tracks_to_playlist(playlist['id'], track_uris):
tracks_added = len(track_uris)
return {
'success': True,
'message': f"Created playlist '{playlist_name}' with {tracks_added} tracks",
'action': 'create_playlist',
'playlist_id': playlist['id'],
'playlist_name': playlist_name,
'tracks_added': tracks_added
}
except Exception as e:
self.logger.error(f"Error creating playlist: {str(e)}")
return {
'success': False,
'message': f"Failed to create playlist: {str(e)}",
'action': 'create_playlist'
}
def _handle_add_tracks(self, action: PlaylistAction) -> Dict[str, Any]:
"""
Handle adding tracks to an existing playlist.
Args:
action: Parsed action containing track addition details
Returns:
Dictionary with addition results
"""
try:
# Find the target playlist
target_playlist = self._find_playlist_by_name(action.playlist_name or action.target_playlist)
if not target_playlist:
return {
'success': False,
'message': f"Playlist '{action.playlist_name or action.target_playlist}' not found",
'action': 'add_tracks'
}
# Find tracks to add
tracks_to_add = []
# Add specific tracks if mentioned
if action.track_details:
for track_name in action.track_details:
tracks = self.spotify.search_tracks(track_name, limit=1)
if tracks:
tracks_to_add.extend(tracks)
# Add tracks based on search criteria
if action.search_criteria:
criteria_tracks = self._find_tracks_by_criteria(action.search_criteria, limit=20)
tracks_to_add.extend(criteria_tracks)
if not tracks_to_add:
return {
'success': False,
'message': "No tracks found matching the criteria",
'action': 'add_tracks'
}
# Remove duplicates
unique_tracks = self._remove_duplicate_tracks(tracks_to_add)
track_uris = [track['uri'] for track in unique_tracks]
# Add tracks to playlist
success = self.spotify.add_tracks_to_playlist(target_playlist['id'], track_uris)
if success:
return {
'success': True,
'message': f"Added {len(track_uris)} tracks to '{target_playlist['name']}'",
'action': 'add_tracks',
'playlist_name': target_playlist['name'],
'tracks_added': len(track_uris)
}
else:
return {
'success': False,
'message': "Failed to add tracks to playlist",
'action': 'add_tracks'
}
except Exception as e:
self.logger.error(f"Error adding tracks: {str(e)}")
return {
'success': False,
'message': f"Failed to add tracks: {str(e)}",
'action': 'add_tracks'
}
def _find_tracks_by_criteria(self, criteria: Dict[str, str],
limit: int = 50) -> List[Dict[str, Any]]:
"""
Find tracks based on search criteria like genre, artist, time period.
Args:
criteria: Dictionary containing search parameters
limit: Maximum number of tracks to return
Returns:
List of track dictionaries
"""
search_queries = []
# Build search queries based on criteria
if criteria.get('artist'):
search_queries.append(f"artist:{criteria['artist']}")
if criteria.get('genre'):
search_queries.append(f"genre:{criteria['genre']}")
if criteria.get('album'):
search_queries.append(f"album:{criteria['album']}")
if criteria.get('time_period'):
year_range = self._parse_time_period(criteria['time_period'])
if year_range:
search_queries.append(f"year:{year_range}")
if criteria.get('keywords'):
search_queries.append(criteria['keywords'])
# If no specific criteria, use genre or keywords
if not search_queries:
if criteria.get('mood'):
search_queries.append(criteria['mood'])
else:
search_queries.append("popular")
# Execute searches and combine results
all_tracks = []
tracks_per_query = max(1, limit // len(search_queries)) if search_queries else limit
for query in search_queries:
try:
tracks = self.spotify.search_tracks(query, limit=tracks_per_query)
all_tracks.extend(tracks)
except Exception as e:
self.logger.warning(f"Search failed for query '{query}': {str(e)}")
continue
# Remove duplicates and limit results
unique_tracks = self._remove_duplicate_tracks(all_tracks)
return unique_tracks[:limit]
def _find_playlist_by_name(self, playlist_name: str) -> Optional[Dict[str, Any]]:
"""
Find a playlist by name in the user's library.
Args:
playlist_name: Name of the playlist to find
Returns:
Playlist dictionary if found, None otherwise
"""
if not playlist_name:
return None
try:
playlists = self._get_cached_playlists()
# Exact match first
for playlist in playlists:
if playlist['name'].lower() == playlist_name.lower():
return playlist
# Partial match if no exact match
for playlist in playlists:
if playlist_name.lower() in playlist['name'].lower():
return playlist
return None
except Exception as e:
self.logger.error(f"Error finding playlist '{playlist_name}': {str(e)}")
return None
def _get_cached_playlists(self) -> List[Dict[str, Any]]:
"""
Get user playlists with caching to improve performance.
Returns:
List of playlist dictionaries
"""
current_time = datetime.now()
# Check if cache is valid (refresh every 5 minutes)
if (self.last_cache_update is None or
(current_time - self.last_cache_update).seconds > 300):
self.playlist_cache = self.spotify.get_user_playlists()
self.last_cache_update = current_time
self.logger.info("Refreshed playlist cache")
return self.playlist_cache
def _generate_playlist_name(self, criteria: Dict[str, str]) -> str:
"""
Generate an appropriate playlist name based on search criteria.
Args:
criteria: Search criteria dictionary
Returns:
Generated playlist name
"""
name_parts = []
if criteria.get('genre'):
name_parts.append(criteria['genre'].title())
if criteria.get('time_period'):
name_parts.append(criteria['time_period'])
if criteria.get('mood'):
name_parts.append(criteria['mood'].title())
if criteria.get('artist'):
name_parts.append(f"Best of {criteria['artist']}")
if name_parts:
return " ".join(name_parts) + " Playlist"
else:
return f"My Playlist {datetime.now().strftime('%Y-%m-%d')}"
def _remove_duplicate_tracks(self, tracks: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""
Remove duplicate tracks from a list based on track ID.
Args:
tracks: List of track dictionaries
Returns:
List with duplicates removed
"""
seen_ids = set()
unique_tracks = []
for track in tracks:
track_id = track.get('id')
if track_id and track_id not in seen_ids:
seen_ids.add(track_id)
unique_tracks.append(track)
return unique_tracks
The orchestrator implements sophisticated business logic that handles the complexity of translating high-level user intents into specific API operations. It maintains session context to enable more intelligent responses and implements caching mechanisms to improve performance.
Advanced Features and Multi-Platform Support
Our playlist manager agent includes advanced features that extend beyond basic playlist operations. These include intelligent track recommendation, playlist format conversion for other music platforms, and sophisticated search capabilities that understand context and user preferences.
The multi-platform support enables users to export their Spotify playlists to other music services, providing flexibility and preventing vendor lock-in. This feature requires understanding different playlist formats and API structures across various music streaming platforms.
# advanced_features.py
import json
import csv
import xml.etree.ElementTree as ET
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from enum import Enum
import requests
import logging
class ExportFormat(Enum):
"""Supported export formats for playlist conversion."""
SPOTIFY_JSON = "spotify_json"
APPLE_MUSIC = "apple_music"
YOUTUBE_MUSIC = "youtube_music"
CSV = "csv"
M3U = "m3u"
XSPF = "xspf"
@dataclass
class TrackInfo:
"""Standardized track information for cross-platform compatibility."""
title: str
artist: str
album: str
duration_ms: int
isrc: Optional[str] = None
spotify_id: Optional[str] = None
apple_music_id: Optional[str] = None
youtube_id: Optional[str] = None
class PlaylistConverter:
"""
Handles conversion of playlists between different music streaming platforms
and export formats.
"""
def __init__(self, spotify_manager):
"""
Initialize the converter with Spotify manager for source data.
Args:
spotify_manager: Configured Spotify API manager
"""
self.spotify = spotify_manager
self.logger = logging.getLogger(__name__)
def export_playlist(self, playlist_id: str,
export_format: ExportFormat) -> Dict[str, Any]:
"""
Export a Spotify playlist to the specified format.
Args:
playlist_id: Spotify playlist ID to export
export_format: Target format for export
Returns:
Dictionary containing export results and data
"""
try:
# Get playlist details and tracks
playlist_info = self._get_playlist_details(playlist_id)
if not playlist_info:
return {
'success': False,
'message': 'Playlist not found or inaccessible'
}
# Convert tracks to standardized format
standardized_tracks = self._standardize_tracks(playlist_info['tracks'])
# Export to requested format
if export_format == ExportFormat.CSV:
export_data = self._export_to_csv(playlist_info, standardized_tracks)
elif export_format == ExportFormat.M3U:
export_data = self._export_to_m3u(playlist_info, standardized_tracks)
elif export_format == ExportFormat.XSPF:
export_data = self._export_to_xspf(playlist_info, standardized_tracks)
elif export_format == ExportFormat.APPLE_MUSIC:
export_data = self._export_to_apple_music(playlist_info, standardized_tracks)
elif export_format == ExportFormat.YOUTUBE_MUSIC:
export_data = self._export_to_youtube_music(playlist_info, standardized_tracks)
else:
export_data = self._export_to_spotify_json(playlist_info, standardized_tracks)
return {
'success': True,
'message': f'Playlist exported to {export_format.value}',
'format': export_format.value,
'data': export_data,
'track_count': len(standardized_tracks)
}
except Exception as e:
self.logger.error(f"Error exporting playlist: {str(e)}")
return {
'success': False,
'message': f'Export failed: {str(e)}'
}
def _get_playlist_details(self, playlist_id: str) -> Optional[Dict[str, Any]]:
"""
Retrieve complete playlist information including all tracks.
Args:
playlist_id: Spotify playlist ID
Returns:
Dictionary with playlist details and tracks
"""
try:
# Get playlist metadata
playlist = self.spotify.spotify.playlist(playlist_id)
# Get all tracks (handle pagination)
tracks = []
results = self.spotify.spotify.playlist_tracks(playlist_id)
while results:
for item in results['items']:
if item['track']: # Skip null tracks
tracks.append(item['track'])
if results['next']:
results = self.spotify.spotify.next(results)
else:
break
return {
'id': playlist['id'],
'name': playlist['name'],
'description': playlist.get('description', ''),
'public': playlist['public'],
'collaborative': playlist['collaborative'],
'tracks': tracks,
'total_tracks': len(tracks)
}
except Exception as e:
self.logger.error(f"Error getting playlist details: {str(e)}")
return None
def _standardize_tracks(self, spotify_tracks: List[Dict[str, Any]]) -> List[TrackInfo]:
"""
Convert Spotify track data to standardized TrackInfo objects.
Args:
spotify_tracks: List of Spotify track dictionaries
Returns:
List of TrackInfo objects
"""
standardized = []
for track in spotify_tracks:
try:
# Extract artist names
artists = [artist['name'] for artist in track.get('artists', [])]
artist_string = ', '.join(artists)
# Create standardized track info
track_info = TrackInfo(
title=track.get('name', 'Unknown Title'),
artist=artist_string,
album=track.get('album', {}).get('name', 'Unknown Album'),
duration_ms=track.get('duration_ms', 0),
isrc=track.get('external_ids', {}).get('isrc'),
spotify_id=track.get('id')
)
standardized.append(track_info)
except Exception as e:
self.logger.warning(f"Error standardizing track: {str(e)}")
continue
return standardized
def _export_to_csv(self, playlist_info: Dict[str, Any],
tracks: List[TrackInfo]) -> str:
"""
Export playlist to CSV format.
Args:
playlist_info: Playlist metadata
tracks: List of standardized tracks
Returns:
CSV content as string
"""
import io
output = io.StringIO()
writer = csv.writer(output)
# Write header
writer.writerow([
'Title', 'Artist', 'Album', 'Duration (ms)',
'ISRC', 'Spotify ID'
])
# Write track data
for track in tracks:
writer.writerow([
track.title,
track.artist,
track.album,
track.duration_ms,
track.isrc or '',
track.spotify_id or ''
])
return output.getvalue()
def _export_to_m3u(self, playlist_info: Dict[str, Any],
tracks: List[TrackInfo]) -> str:
"""
Export playlist to M3U format.
Args:
playlist_info: Playlist metadata
tracks: List of standardized tracks
Returns:
M3U content as string
"""
lines = ['#EXTM3U']
lines.append(f'#PLAYLIST:{playlist_info["name"]}')
for track in tracks:
duration_seconds = track.duration_ms // 1000
lines.append(f'#EXTINF:{duration_seconds},{track.artist} - {track.title}')
# Use Spotify web player URL as fallback
if track.spotify_id:
lines.append(f'https://open.spotify.com/track/{track.spotify_id}')
else:
lines.append(f'# {track.artist} - {track.title}')
return '\n'.join(lines)
def _export_to_xspf(self, playlist_info: Dict[str, Any],
tracks: List[TrackInfo]) -> str:
"""
Export playlist to XSPF (XML Shareable Playlist Format).
Args:
playlist_info: Playlist metadata
tracks: List of standardized tracks
Returns:
XSPF content as string
"""
root = ET.Element('playlist', version='1', xmlns='http://xspf.org/ns/0/')
# Playlist metadata
title = ET.SubElement(root, 'title')
title.text = playlist_info['name']
if playlist_info.get('description'):
annotation = ET.SubElement(root, 'annotation')
annotation.text = playlist_info['description']
# Track list
tracklist = ET.SubElement(root, 'trackList')
for track_info in tracks:
track = ET.SubElement(tracklist, 'track')
title = ET.SubElement(track, 'title')
title.text = track_info.title
creator = ET.SubElement(track, 'creator')
creator.text = track_info.artist
album = ET.SubElement(track, 'album')
album.text = track_info.album
duration = ET.SubElement(track, 'duration')
duration.text = str(track_info.duration_ms)
if track_info.spotify_id:
location = ET.SubElement(track, 'location')
location.text = f'https://open.spotify.com/track/{track_info.spotify_id}'
return ET.tostring(root, encoding='unicode', xml_declaration=True)
class IntelligentRecommendationEngine:
"""
Provides intelligent track recommendations based on user preferences,
listening history, and contextual factors.
"""
def __init__(self, spotify_manager):
"""
Initialize the recommendation engine.
Args:
spotify_manager: Configured Spotify API manager
"""
self.spotify = spotify_manager
self.logger = logging.getLogger(__name__)
def get_smart_recommendations(self, playlist_id: str,
criteria: Dict[str, Any],
count: int = 20) -> List[Dict[str, Any]]:
"""
Generate intelligent track recommendations for a playlist.
Args:
playlist_id: Target playlist ID for recommendations
criteria: Additional criteria for recommendations
count: Number of recommendations to generate
Returns:
List of recommended track dictionaries
"""
try:
# Get current playlist tracks for analysis
current_tracks = self._get_playlist_tracks(playlist_id)
if not current_tracks:
return self._get_fallback_recommendations(criteria, count)
# Analyze playlist characteristics
playlist_features = self._analyze_playlist_features(current_tracks)
# Get seed tracks for recommendations
seed_tracks = self._select_seed_tracks(current_tracks)
# Generate recommendations using Spotify's recommendation engine
recommendations = self._get_spotify_recommendations(
seed_tracks, playlist_features, count
)
# Filter out tracks already in playlist
existing_track_ids = {track['id'] for track in current_tracks}
filtered_recommendations = [
track for track in recommendations
if track['id'] not in existing_track_ids
]
# Apply additional filtering based on criteria
final_recommendations = self._apply_criteria_filtering(
filtered_recommendations, criteria
)
return final_recommendations[:count]
except Exception as e:
self.logger.error(f"Error generating recommendations: {str(e)}")
return self._get_fallback_recommendations(criteria, count)
def _analyze_playlist_features(self, tracks: List[Dict[str, Any]]) -> Dict[str, float]:
"""
Analyze audio features of tracks in a playlist to understand its characteristics.
Args:
tracks: List of track dictionaries
Returns:
Dictionary of average audio features
"""
if not tracks:
return {}
try:
# Get audio features for all tracks
track_ids = [track['id'] for track in tracks if track.get('id')]
if not track_ids:
return {}
# Spotify API allows max 100 tracks per request
all_features = []
for i in range(0, len(track_ids), 100):
batch = track_ids[i:i + 100]
features = self.spotify.spotify.audio_features(batch)
all_features.extend([f for f in features if f]) # Filter out None values
if not all_features:
return {}
# Calculate average features
feature_keys = [
'danceability', 'energy', 'speechiness', 'acousticness',
'instrumentalness', 'liveness', 'valence', 'tempo'
]
avg_features = {}
for key in feature_keys:
values = [f[key] for f in all_features if f.get(key) is not None]
if values:
avg_features[key] = sum(values) / len(values)
return avg_features
except Exception as e:
self.logger.error(f"Error analyzing playlist features: {str(e)}")
return {}
The advanced features module provides sophisticated capabilities for playlist analysis, intelligent recommendations, and cross-platform compatibility. The recommendation engine analyzes musical characteristics to suggest tracks that fit the playlist's style and mood.
User Interface and Interaction Layer
The user interface layer provides multiple interaction modalities including command-line interface, web interface, and API endpoints. This flexibility allows users to interact with the playlist manager through their preferred method while maintaining consistent functionality across all interfaces.
# user_interface.py
from flask import Flask, request, jsonify, render_template, session
import logging
from typing import Dict, Any
import json
import os
from datetime import datetime
from playlist_orchestrator import PlaylistOrchestrator
from llm_processor import LLMProcessor
from spotify_client import SpotifyPlaylistManager
from advanced_features import PlaylistConverter, IntelligentRecommendationEngine
class PlaylistManagerAPI:
"""
Web API interface for the playlist manager agent.
Provides RESTful endpoints for all playlist operations.
"""
def __init__(self):
"""Initialize the Flask application and components."""
self.app = Flask(__name__)
self.app.secret_key = os.getenv('FLASK_SECRET_KEY', 'dev-secret-key')
# Initialize components
self.llm_processor = LLMProcessor()
self.spotify_manager = SpotifyPlaylistManager()
self.orchestrator = PlaylistOrchestrator(self.llm_processor, self.spotify_manager)
self.converter = PlaylistConverter(self.spotify_manager)
self.recommendation_engine = IntelligentRecommendationEngine(self.spotify_manager)
# Setup logging
logging.basicConfig(level=logging.INFO)
self.logger = logging.getLogger(__name__)
# Register routes
self._register_routes()
def _register_routes(self):
"""Register all API routes."""
@self.app.route('/', methods=['GET'])
def index():
"""Serve the main web interface."""
return render_template('index.html')
@self.app.route('/api/command', methods=['POST'])
def execute_command():
"""
Execute a natural language playlist command.
Expected JSON payload:
{
"command": "Create a playlist with jazz music from the 1960s"
}
"""
try:
data = request.get_json()
if not data or 'command' not in data:
return jsonify({
'success': False,
'message': 'Missing command in request'
}), 400
command = data['command']
self.logger.info(f"Executing command: {command}")
# Execute the command
result = self.orchestrator.execute_command(command)
# Log the result
self.logger.info(f"Command result: {result['success']}")
return jsonify(result)
except Exception as e:
self.logger.error(f"Error in execute_command: {str(e)}")
return jsonify({
'success': False,
'message': f'Internal error: {str(e)}'
}), 500
@self.app.route('/api/playlists', methods=['GET'])
def get_playlists():
"""Get all user playlists."""
try:
playlists = self.spotify_manager.get_user_playlists()
# Simplify playlist data for API response
simplified_playlists = []
for playlist in playlists:
simplified_playlists.append({
'id': playlist['id'],
'name': playlist['name'],
'description': playlist.get('description', ''),
'track_count': playlist['tracks']['total'],
'public': playlist['public'],
'collaborative': playlist['collaborative']
})
return jsonify({
'success': True,
'playlists': simplified_playlists,
'count': len(simplified_playlists)
})
except Exception as e:
self.logger.error(f"Error getting playlists: {str(e)}")
return jsonify({
'success': False,
'message': f'Failed to retrieve playlists: {str(e)}'
}), 500
@self.app.route('/api/playlist/<playlist_id>/export', methods=['POST'])
def export_playlist(playlist_id):
"""
Export a playlist to specified format.
Expected JSON payload:
{
"format": "csv|m3u|xspf|apple_music|youtube_music"
}
"""
try:
data = request.get_json()
export_format = data.get('format', 'csv')
# Validate format
valid_formats = ['csv', 'm3u', 'xspf', 'apple_music', 'youtube_music']
if export_format not in valid_formats:
return jsonify({
'success': False,
'message': f'Invalid format. Supported: {", ".join(valid_formats)}'
}), 400
# Convert format string to enum
from advanced_features import ExportFormat
format_map = {
'csv': ExportFormat.CSV,
'm3u': ExportFormat.M3U,
'xspf': ExportFormat.XSPF,
'apple_music': ExportFormat.APPLE_MUSIC,
'youtube_music': ExportFormat.YOUTUBE_MUSIC
}
result = self.converter.export_playlist(playlist_id, format_map[export_format])
return jsonify(result)
except Exception as e:
self.logger.error(f"Error exporting playlist: {str(e)}")
return jsonify({
'success': False,
'message': f'Export failed: {str(e)}'
}), 500
@self.app.route('/api/playlist/<playlist_id>/recommendations', methods=['GET'])
def get_recommendations(playlist_id):
"""Get intelligent recommendations for a playlist."""
try:
count = request.args.get('count', 20, type=int)
count = min(max(count, 1), 50) # Limit between 1 and 50
# Get additional criteria from query parameters
criteria = {}
if request.args.get('genre'):
criteria['genre'] = request.args.get('genre')
if request.args.get('mood'):
criteria['mood'] = request.args.get('mood')
recommendations = self.recommendation_engine.get_smart_recommendations(
playlist_id, criteria, count
)
# Simplify track data for API response
simplified_tracks = []
for track in recommendations:
artists = [artist['name'] for artist in track.get('artists', [])]
simplified_tracks.append({
'id': track['id'],
'name': track['name'],
'artists': artists,
'album': track.get('album', {}).get('name', ''),
'duration_ms': track.get('duration_ms', 0),
'preview_url': track.get('preview_url'),
'external_urls': track.get('external_urls', {})
})
return jsonify({
'success': True,
'recommendations': simplified_tracks,
'count': len(simplified_tracks)
})
except Exception as e:
self.logger.error(f"Error getting recommendations: {str(e)}")
return jsonify({
'success': False,
'message': f'Failed to get recommendations: {str(e)}'
}), 500
@self.app.route('/api/health', methods=['GET'])
def health_check():
"""Health check endpoint."""
return jsonify({
'status': 'healthy',
'timestamp': datetime.now().isoformat(),
'version': '1.0.0'
})
def run(self, host='localhost', port=5000, debug=False):
"""Run the Flask application."""
self.logger.info(f"Starting Playlist Manager API on {host}:{port}")
self.app.run(host=host, port=port, debug=debug)
class CommandLineInterface:
"""
Command-line interface for the playlist manager agent.
Provides an interactive shell for playlist operations.
"""
def __init__(self):
"""Initialize the CLI components."""
self.llm_processor = LLMProcessor()
self.spotify_manager = SpotifyPlaylistManager()
self.orchestrator = PlaylistOrchestrator(self.llm_processor, self.spotify_manager)
self.converter = PlaylistConverter(self.spotify_manager)
# Setup logging for CLI
logging.basicConfig(
level=logging.WARNING, # Less verbose for CLI
format='%(levelname)s: %(message)s'
)
self.logger = logging.getLogger(__name__)
def run_interactive(self):
"""Run the interactive command-line interface."""
print("=" * 60)
print("šµ Spotify Playlist Manager Agent")
print("=" * 60)
print("Type your playlist commands in natural language.")
print("Examples:")
print(" - Create a playlist with jazz music from the 1960s")
print(" - Add some rock songs to my workout playlist")
print(" - Export my chill playlist to CSV format")
print("Type 'help' for more commands or 'quit' to exit.")
print("=" * 60)
while True:
try:
# Get user input
user_input = input("\nšµ > ").strip()
if not user_input:
continue
# Handle special commands
if user_input.lower() in ['quit', 'exit', 'q']:
print("Goodbye! šµ")
break
elif user_input.lower() in ['help', 'h']:
self._show_help()
continue
elif user_input.lower().startswith('export '):
self._handle_export_command(user_input)
continue
elif user_input.lower() == 'list playlists':
self._list_playlists()
continue
# Execute the command
print("Processing your request...")
result = self.orchestrator.execute_command(user_input)
# Display result
self._display_result(result)
except KeyboardInterrupt:
print("\n\nGoodbye! šµ")
break
except Exception as e:
print(f"❌ An error occurred: {str(e)}")
self.logger.error(f"CLI error: {str(e)}")
def _show_help(self):
"""Display help information."""
help_text = """
šµ Playlist Manager Commands:
Natural Language Commands:
• Create a playlist with [genre/artist/mood] music
• Add [songs/genre/artist] to [playlist name]
• Remove [songs/artist] from [playlist name]
• Delete playlist [name]
• Rename playlist [old name] to [new name]
• Play playlist [name]
• Duplicate playlist [name]
Special Commands:
• list playlists - Show all your playlists
• export [playlist] [format] - Export playlist (csv, m3u, xspf)
• help - Show this help
• quit - Exit the program
Examples:
• "Create a workout playlist with high-energy rock music"
• "Add some Taylor Swift songs to my pop playlist"
• "Remove all slow songs from my party playlist"
• "export 'My Chill Playlist' csv"
"""
print(help_text)
def _handle_export_command(self, command: str):
"""Handle export commands."""
try:
parts = command.split(' ', 2)
if len(parts) < 3:
print("❌ Usage: export [playlist name] [format]")
print(" Formats: csv, m3u, xspf")
return
playlist_name = parts[1].strip("'\"")
export_format = parts[2].lower()
# Find playlist
playlists = self.spotify_manager.get_user_playlists()
target_playlist = None
for playlist in playlists:
if playlist['name'].lower() == playlist_name.lower():
target_playlist = playlist
break
if not target_playlist:
print(f"❌ Playlist '{playlist_name}' not found")
return
# Export playlist
from advanced_features import ExportFormat
format_map = {
'csv': ExportFormat.CSV,
'm3u': ExportFormat.M3U,
'xspf': ExportFormat.XSPF
}
if export_format not in format_map:
print("❌ Unsupported format. Use: csv, m3u, xspf")
return
print(f"Exporting '{playlist_name}' to {export_format.upper()}...")
result = self.converter.export_playlist(
target_playlist['id'],
format_map[export_format]
)
if result['success']:
# Save to file
filename = f"{playlist_name.replace(' ', '_')}.{export_format}"
with open(filename, 'w', encoding='utf-8') as f:
f.write(result['data'])
print(f"✅ Exported to {filename}")
print(f" {result['track_count']} tracks exported")
else:
print(f"❌ Export failed: {result['message']}")
except Exception as e:
print(f"❌ Export error: {str(e)}")
def _list_playlists(self):
"""List all user playlists."""
try:
print("š Your Playlists:")
print("-" * 50)
playlists = self.spotify_manager.get_user_playlists()
if not playlists:
print("No playlists found.")
return
for i, playlist in enumerate(playlists, 1):
track_count = playlist['tracks']['total']
public_status = "Public" if playlist['public'] else "Private"
print(f"{i:2d}. {playlist['name']}")
print(f" {track_count} tracks • {public_status}")
if playlist.get('description'):
desc = playlist['description'][:60]
if len(playlist['description']) > 60:
desc += "..."
print(f" {desc}")
print()
print(f"Total: {len(playlists)} playlists")
except Exception as e:
print(f"❌ Error listing playlists: {str(e)}")
def _display_result(self, result: Dict[str, Any]):
"""Display command execution result."""
if result['success']:
print(f"✅ {result['message']}")
# Display additional information based on action type
if result.get('action') == 'create_playlist':
if result.get('tracks_added', 0) > 0:
print(f" Added {result['tracks_added']} tracks automatically")
elif result.get('action') == 'add_tracks':
if result.get('tracks_added', 0) > 0:
print(f" {result['tracks_added']} tracks added")
else:
print(f"❌ {result['message']}")
The user interface layer provides both web API and command-line interfaces, making the playlist manager accessible through different interaction modalities. The CLI offers an intuitive conversational interface, while the web API enables integration with other applications and services.
Complete Running Example
Here is a comprehensive running example that demonstrates the complete playlist manager system in action. This example includes all the components working together to handle various user commands and showcase the system's capabilities.
# main.py - Complete Running Example
import os
import sys
import logging
from dotenv import load_dotenv
from typing import Dict, Any
# Load environment variables
load_dotenv()
# Import all components
from llm_processor import LLMProcessor, ActionType
from spotify_client import SpotifyPlaylistManager
from playlist_orchestrator import PlaylistOrchestrator
from advanced_features import PlaylistConverter, IntelligentRecommendationEngine, ExportFormat
from user_interface import CommandLineInterface, PlaylistManagerAPI
class PlaylistManagerDemo:
"""
Comprehensive demonstration of the playlist manager system.
Shows all major features and capabilities in action.
"""
def __init__(self):
"""Initialize all components for the demonstration."""
# Setup logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
self.logger = logging.getLogger(__name__)
# Verify environment setup
self._verify_environment()
# Initialize components
self.logger.info("Initializing Playlist Manager components...")
try:
self.llm_processor = LLMProcessor(model_type="openai", model_name="gpt-3.5-turbo")
self.spotify_manager = SpotifyPlaylistManager()
self.orchestrator = PlaylistOrchestrator(self.llm_processor, self.spotify_manager)
self.converter = PlaylistConverter(self.spotify_manager)
self.recommendation_engine = IntelligentRecommendationEngine(self.spotify_manager)
self.logger.info("All components initialized successfully!")
except Exception as e:
self.logger.error(f"Failed to initialize components: {str(e)}")
raise
def _verify_environment(self):
"""Verify that all required environment variables are set."""
required_vars = [
'SPOTIFY_CLIENT_ID',
'SPOTIFY_CLIENT_SECRET',
'SPOTIFY_REDIRECT_URI',
'OPENAI_API_KEY'
]
missing_vars = []
for var in required_vars:
if not os.getenv(var):
missing_vars.append(var)
if missing_vars:
raise EnvironmentError(
f"Missing required environment variables: {', '.join(missing_vars)}\n"
f"Please set these in your .env file or environment."
)
def run_comprehensive_demo(self):
"""
Run a comprehensive demonstration of all playlist manager features.
"""
print("=" * 80)
print("šµ SPOTIFY PLAYLIST MANAGER AGENT - COMPREHENSIVE DEMO")
print("=" * 80)
# Test commands to demonstrate various capabilities
demo_commands = [
"Create a playlist called 'Jazz Classics' with jazz music from the 1960s",
"Add some Miles Davis tracks to my Jazz Classics playlist",
"Create a workout playlist with high-energy rock and electronic music",
"Search for Christmas songs and create a holiday playlist",
"Add some Taylor Swift songs to my pop playlist",
"Create a playlist with classical music for studying",
"Remove all slow songs from my workout playlist",
"Duplicate my Jazz Classics playlist and call it 'Jazz Favorites'",
"Find some chill electronic music for my relaxation playlist"
]
print("\nš Starting demonstration with sample commands...\n")
for i, command in enumerate(demo_commands, 1):
print(f"\n--- Demo Command {i}/{len(demo_commands)} ---")
print(f"Command: '{command}'")
print("-" * 60)
try:
# Execute the command
result = self.orchestrator.execute_command(command)
# Display results
self._display_demo_result(result)
# Add some additional demonstrations for specific commands
if i == 1 and result['success']: # After creating Jazz Classics
self._demo_playlist_analysis(result.get('playlist_id'))
if i == 3 and result['success']: # After creating workout playlist
self._demo_recommendations(result.get('playlist_id'))
if i == 5 and result['success']: # After adding to pop playlist
self._demo_export_functionality(result.get('playlist_id'))
except Exception as e:
print(f"❌ Error executing command: {str(e)}")
self.logger.error(f"Demo command failed: {str(e)}")
# Pause between commands for readability
input("\nPress Enter to continue to next command...")
print("\n" + "=" * 80)
print("š DEMONSTRATION COMPLETE!")
print("=" * 80)
print("\nThe Playlist Manager Agent successfully demonstrated:")
print("✅ Natural language command processing")
print("✅ Intelligent playlist creation and management")
print("✅ Smart track searching and recommendations")
print("✅ Playlist analysis and export capabilities")
print("✅ Multi-format playlist conversion")
print("✅ Error handling and recovery")
# Offer to start interactive mode
print("\n" + "-" * 60)
choice = input("Would you like to try the interactive mode? (y/n): ").lower()
if choice in ['y', 'yes']:
self._start_interactive_mode()
def _display_demo_result(self, result: Dict[str, Any]):
"""Display the result of a demo command with detailed information."""
if result['success']:
print(f"✅ SUCCESS: {result['message']}")
# Show additional details based on action type
action = result.get('action', '')
if action == 'create_playlist':
print(f" š Playlist ID: {result.get('playlist_id', 'N/A')}")
print(f" šµ Tracks added: {result.get('tracks_added', 0)}")
elif action == 'add_tracks':
print(f" šµ Tracks added: {result.get('tracks_added', 0)}")
print(f" š Target playlist: {result.get('playlist_name', 'N/A')}")
elif action == 'search_tracks':
print(f" š Tracks found: {result.get('tracks_found', 0)}")
else:
print(f"❌ FAILED: {result['message']}")
print(f" ⚡ Action: {result.get('action', 'unknown')}")
def _demo_playlist_analysis(self, playlist_id: str):
"""Demonstrate playlist analysis capabilities."""
if not playlist_id:
return
print("\nš BONUS: Playlist Analysis Demo")
print("-" * 40)
try:
# Get playlist details
playlist_details = self.converter._get_playlist_details(playlist_id)
if playlist_details:
print(f"š Analyzing playlist: {playlist_details['name']}")
print(f" Total tracks: {playlist_details['total_tracks']}")
# Analyze audio features
features = self.recommendation_engine._analyze_playlist_features(
playlist_details['tracks']
)
if features:
print(" Audio characteristics:")
for feature, value in features.items():
if feature == 'tempo':
print(f" {feature.title()}: {value:.1f} BPM")
else:
print(f" {feature.title()}: {value:.2f}")
except Exception as e:
print(f" Analysis failed: {str(e)}")
def _demo_recommendations(self, playlist_id: str):
"""Demonstrate intelligent recommendation capabilities."""
if not playlist_id:
return
print("\nšÆ BONUS: Smart Recommendations Demo")
print("-" * 40)
try:
recommendations = self.recommendation_engine.get_smart_recommendations(
playlist_id, {}, count=5
)
if recommendations:
print(f"šµ Found {len(recommendations)} smart recommendations:")
for i, track in enumerate(recommendations, 1):
artists = [artist['name'] for artist in track.get('artists', [])]
artist_str = ', '.join(artists)
print(f" {i}. {track['name']} - {artist_str}")
else:
print(" No recommendations available")
except Exception as e:
print(f" Recommendations failed: {str(e)}")
def _demo_export_functionality(self, playlist_id: str):
"""Demonstrate playlist export capabilities."""
if not playlist_id:
return
print("\nš¾ BONUS: Export Functionality Demo")
print("-" * 40)
try:
# Export to CSV format
result = self.converter.export_playlist(playlist_id, ExportFormat.CSV)
if result['success']:
print(f"✅ Export successful!")
print(f" Format: {result['format']}")
print(f" Tracks exported: {result['track_count']}")
print(f" Data size: {len(result['data'])} characters")
# Show a sample of the exported data
lines = result['data'].split('\n')
print(" Sample content:")
for line in lines[:3]: # Show first 3 lines
print(f" {line}")
if len(lines) > 3:
print(f" ... and {len(lines) - 3} more lines")
else:
print(f"❌ Export failed: {result['message']}")
except Exception as e:
print(f" Export demo failed: {str(e)}")
def _start_interactive_mode(self):
"""Start the interactive command-line interface."""
print("\nš Starting Interactive Mode...")
cli = CommandLineInterface()
cli.run_interactive()
def start_web_server(self, host='localhost', port=5000):
"""Start the web API server."""
print(f"\nš Starting Web API Server on http://{host}:{port}")
api = PlaylistManagerAPI()
api.run(host=host, port=port, debug=False)
def main():
"""
Main entry point for the Playlist Manager Agent.
Provides options for demo, interactive, and web server modes.
"""
print("šµ Spotify Playlist Manager Agent")
print("Choose a mode:")
print("1. Run comprehensive demo")
print("2. Start interactive CLI")
print("3. Start web API server")
print("4. Exit")
while True:
try:
choice = input("\nEnter your choice (1-4): ").strip()
if choice == '1':
demo = PlaylistManagerDemo()
demo.run_comprehensive_demo()
break
elif choice == '2':
cli = CommandLineInterface()
cli.run_interactive()
break
elif choice == '3':
host = input("Enter host (default: localhost): ").strip() or 'localhost'
port_input = input("Enter port (default: 5000): ").strip()
port = int(port_input) if port_input else 5000
demo = PlaylistManagerDemo()
demo.start_web_server(host, port)
break
elif choice == '4':
print("Goodbye! šµ")
sys.exit(0)
else:
print("Invalid choice. Please enter 1, 2, 3, or 4.")
except KeyboardInterrupt:
print("\n\nGoodbye! šµ")
sys.exit(0)
except ValueError:
print("Invalid input. Please enter a number.")
except Exception as e:
print(f"An error occurred: {str(e)}")
if __name__ == "__main__":
main()
This complete running example demonstrates the full capabilities of the playlist manager agent. It includes comprehensive error handling, detailed logging, and multiple interaction modes. The system successfully interprets natural language commands, executes complex playlist operations, and provides intelligent recommendations and export capabilities.
The architecture follows clean code principles with clear separation of concerns, making it maintainable and extensible. The modular design allows for easy integration of additional features and support for other music streaming platforms.
The agent represents a sophisticated integration of artificial intelligence and music streaming APIs, providing users with an intuitive and powerful tool for managing their music collections through natural language interaction.
No comments:
Post a Comment