import os import logging import spotipy from spotipy.oauth2 import SpotifyClientCredentials from typing import Dict, List, Optional, Union, Any logger = logging.getLogger(__name__) class SpotifyService: """Service for interacting with the Spotify API.""" def __init__(self, config): """Initialize the Spotify service with configuration. Args: config: Configuration service instance """ self.config = config self.client_id = config.spotify_client_id self.client_secret = config.spotify_client_secret self.spotify = self._initialize_spotify() def _initialize_spotify(self) -> spotipy.Spotify: """Initialize and return a Spotify client.""" try: auth_manager = SpotifyClientCredentials( client_id=self.client_id, client_secret=self.client_secret ) return spotipy.Spotify(auth_manager=auth_manager) except Exception as e: logger.error(f"Failed to initialize Spotify client: {e}") raise def search(self, query: str, search_type: str = 'track', limit: int = 10) -> List[Dict[str, Any]]: """Search for tracks, albums, artists, or playlists on Spotify. Args: query: Search query string search_type: Type of search ('track', 'album', 'artist', 'playlist', or 'all') limit: Maximum number of results to return Returns: List of search results """ try: logger.info(f"Searching Spotify with query: '{query}', type: '{search_type}', limit: {limit}") logger.debug(f"Using client_id: {self.client_id[:5]}...{self.client_id[-3:] if self.client_id else None}") if not query or not query.strip(): logger.warning("Empty search query provided") return [] if search_type.lower() == 'all': search_types = ['track', 'album', 'artist', 'playlist'] results = {} for st in search_types: try: logger.debug(f"Searching for {st} with query: {query}") search_result = self.spotify.search( q=query, type=st, limit=min(limit, 20) # Limit to 20 max per type ) logger.debug(f"Raw {st} search result: {search_result}") results[st] = search_result.get(f'{st}s', {}).get('items', []) logger.info(f"Found {len(results[st])} {st} results") except Exception as e: logger.error(f"Error searching for {st}: {str(e)}", exc_info=True) results[st] = [] return self._format_search_results(results, 'all') else: logger.debug(f"Searching for {search_type} with query: {query}") search_result = self.spotify.search( q=query, type=search_type, limit=min(limit, 50) # Limit to 50 max for single type ) logger.debug(f"Raw search result: {search_result}") formatted = self._format_search_results(search_result, search_type) logger.info(f"Found {len(formatted)} {search_type} results") return formatted except Exception as e: logger.error(f"Search error: {str(e)}", exc_info=True) logger.error(f"Search parameters - Query: '{query}', Type: '{search_type}', Limit: {limit}") return [] def _format_search_results(self, results: Any, result_type: str) -> List[Dict[str, Any]]: """Format search results into a consistent format. Args: results: Raw search results from Spotify result_type: Type of results ('track', 'album', 'artist', 'playlist', or 'all') Returns: List of formatted result items """ formatted_results = [] if result_type == 'all': # Handle combined results from multiple search types for item_type, items in results.items(): for item in items: formatted_item = self._format_item(item, item_type) if formatted_item: formatted_results.append(formatted_item) else: # Handle single search type items = results.get(f'{result_type}s', {}).get('items', []) for item in items: formatted_item = self._format_item(item, result_type) if formatted_item: formatted_results.append(formatted_item) return formatted_results def _format_item(self, item: Dict[str, Any], item_type: str) -> Optional[Dict[str, Any]]: """Format a single search result item. Args: item: Raw item from Spotify item_type: Type of the item Returns: Formatted item or None if invalid """ try: if item_type == 'track': return { 'type': 'track', 'id': item.get('id'), 'name': item.get('name', 'Unknown Track'), 'artist': item.get('artists', [{}])[0].get('name', 'Unknown Artist'), 'album': item.get('album', {}).get('name', 'Unknown Album'), 'year': item.get('album', {}).get('release_date', '').split('-')[0], 'duration_ms': item.get('duration_ms', 0), 'image': self._get_image_url(item.get('album', {}).get('images', [])), 'url': item.get('external_urls', {}).get('spotify', ''), 'uri': item.get('uri', '') } elif item_type == 'album': return { 'type': 'album', 'id': item.get('id'), 'name': item.get('name', 'Unknown Album'), 'artist': item.get('artists', [{}])[0].get('name', 'Unknown Artist'), 'year': item.get('release_date', '').split('-')[0], 'tracks': item.get('total_tracks', 0), 'image': self._get_image_url(item.get('images', [])), 'url': item.get('external_urls', {}).get('spotify', ''), 'uri': item.get('uri', '') } elif item_type == 'artist': return { 'type': 'artist', 'id': item.get('id'), 'name': item.get('name', 'Unknown Artist'), 'genres': ', '.join(item.get('genres', [])), 'followers': item.get('followers', {}).get('total', 0), 'image': self._get_image_url(item.get('images', [])), 'url': item.get('external_urls', {}).get('spotify', ''), 'uri': item.get('uri', '') } elif item_type == 'playlist': return { 'type': 'playlist', 'id': item.get('id'), 'name': item.get('name', 'Unknown Playlist'), 'owner': item.get('owner', {}).get('display_name', 'Unknown'), 'tracks': item.get('tracks', {}).get('total', 0), 'image': self._get_image_url(item.get('images', [])), 'url': item.get('external_urls', {}).get('spotify', ''), 'uri': item.get('uri', '') } return None except Exception as e: logger.error(f"Error formatting {item_type} item: {e}") return None @staticmethod def _get_image_url(images: List[Dict[str, Any]]) -> str: """Get the most appropriate image URL from a list of images. Args: images: List of image objects from Spotify Returns: URL of the best image or empty string if none found """ if not images: return '' # Try to get medium size (index 1) or fall back to first available if len(images) > 1: return images[1].get('url', images[0].get('url', '')) return images[0].get('url', '') def perform_spotify_search(self, query_req: Dict[str, Any]) -> Dict[str, Any]: """Perform a Spotify search and return formatted results for the WebSocket API. This method is specifically designed to be called from the WebSocket handler. Args: query_req: Dictionary containing search parameters: - query: The search query string - type: Type of search ('track', 'album', 'artist', 'playlist', or 'all') - limit: Maximum number of results to return per type Returns: Dictionary with search results in the format expected by the WebSocket client """ try: query = query_req.get('query', '').strip() search_type = query_req.get('type', 'track') limit = min(int(query_req.get('limit', 5)), 50) # Cap at 50 results per type if not query: logger.warning("Empty search query received") return {'error': 'Empty search query'} logger.info(f"Performing Spotify search: query='{query}', type='{search_type}', limit={limit}") # Perform the search using the existing search method results = self.search(query, search_type, limit) # Format results for the WebSocket client formatted_results = {} # Group results by type for item in results: item_type = item.get('type') if not item_type: continue # Initialize the list for this type if it doesn't exist if item_type not in formatted_results: formatted_results[item_type] = [] # Add the item to the appropriate list formatted_results[item_type].append(item) logger.info(f"Search completed, found {sum(len(v) for v in formatted_results.values())} results") logger.debug(f"Formatted results: {formatted_results}") return formatted_results except Exception as e: error_msg = f"Error in perform_spotify_search: {str(e)}" logger.error(error_msg, exc_info=True) return {'error': error_msg}