""" Universal Music Downloader - Minimal Working Version """ import os import time import asyncio import aiohttp from typing import Dict, List, Optional, Any from dataclasses import dataclass from enum import Enum from swingmusic.services.universal_url_parser import universal_url_parser, MusicService, ParsedURL import logging logger = logging.getLogger(__name__) class DownloadStatus(Enum): PENDING = "pending" DOWNLOADING = "downloading" COMPLETED = "completed" FAILED = "failed" class DownloadQuality(Enum): LOSSLESS = "lossless" HIGH = "high" MEDIUM = "medium" LOW = "low" @dataclass class UniversalMetadata: """Universal metadata structure for all music services""" service: MusicService service_id: str title: str artist: str album: Optional[str] = None duration_ms: Optional[int] = None isrc: Optional[str] = None release_date: Optional[str] = None genre: Optional[str] = None image_url: Optional[str] = None original_url: str = "" metadata: Dict[str, Any] = None def __post_init__(self): if self.metadata is None: self.metadata = {} @dataclass class DownloadItem: """Represents a download item in the queue""" id: str url: str metadata: UniversalMetadata quality: DownloadQuality status: DownloadStatus progress: float = 0.0 file_path: Optional[str] = None error_message: Optional[str] = None created_at: float = None def __post_init__(self): if self.created_at is None: self.created_at = time.time() class UniversalMusicDownloader: """Universal music downloader supporting multiple streaming services""" def __init__(self, download_dir: str = None, max_concurrent_downloads: int = 3): self.download_dir = download_dir or os.path.expanduser("~/Downloads/SwingMusic") self.max_concurrent_downloads = max_concurrent_downloads self.default_quality = DownloadQuality.HIGH self.download_queue: List[DownloadItem] = [] self.session = None # Ensure download directory exists os.makedirs(self.download_dir, exist_ok=True) async def _get_session(self) -> aiohttp.ClientSession: """Get or create aiohttp session""" if self.session is None: self.session = aiohttp.ClientSession() return self.session async def close(self): """Close aiohttp session""" if self.session: await self.session.close() def parse_url(self, url: str) -> Optional[ParsedURL]: """Parse and validate a music service URL""" return universal_url_parser.parse_url(url) async def get_metadata(self, url: str) -> Optional[UniversalMetadata]: """Get metadata from any supported music service URL""" try: # Parse URL parsed_url = universal_url_parser.parse_url(url) if not parsed_url: logger.warning(f"Could not parse URL: {url}") return None # Route to appropriate service if parsed_url.service == MusicService.SPOTIFY: return await self._get_spotify_metadata(parsed_url) elif parsed_url.service == MusicService.TIDAL: return await self._get_tidal_metadata(parsed_url) elif parsed_url.service == MusicService.APPLE_MUSIC: return await self._get_apple_music_metadata(parsed_url) elif parsed_url.service == MusicService.YOUTUBE: return await self._get_youtube_metadata(parsed_url) elif parsed_url.service == MusicService.YOUTUBE_MUSIC: return await self._get_youtube_music_metadata(parsed_url) elif parsed_url.service == MusicService.SOUNDCLOUD: return await self._get_soundcloud_metadata(parsed_url) elif parsed_url.service == MusicService.DEEZER: return await self._get_deezer_metadata(parsed_url) elif parsed_url.service == MusicService.MUSICBRAINZ: return await self._get_musicbrainz_metadata(parsed_url) elif parsed_url.service == MusicService.DISCOGS: return await self._get_discogs_metadata(parsed_url) else: logger.warning(f"Unsupported service: {parsed_url.service}") return None except Exception as e: logger.error(f"Error getting metadata for {url}: {e}") return None async def _get_spotify_metadata(self, parsed_url: ParsedURL) -> Optional[UniversalMetadata]: """Get metadata from Spotify""" try: return UniversalMetadata( service=MusicService.SPOTIFY, service_id=parsed_url.id, title=f"Spotify {parsed_url.item_type.title()}", artist="Unknown Artist", original_url=parsed_url.url ) except Exception as e: logger.error(f"Error getting Spotify metadata: {e}") return None async def _get_tidal_metadata(self, parsed_url: ParsedURL) -> Optional[UniversalMetadata]: """Get metadata from Tidal""" try: import aiohttp from bs4 import BeautifulSoup url = f"https://tidal.com/browse/{parsed_url.item_type}/{parsed_url.id}" session = await self._get_session() async with session.get(url, headers={'User-Agent': 'Mozilla/5.0'}) as response: if response.status == 200: html = await response.text() soup = BeautifulSoup(html, 'html.parser') title_elem = soup.find('meta', property='og:title') artist_elem = soup.find('meta', property='og:music:artist') image_elem = soup.find('meta', property='og:image') title = title_elem.get('content', '') if title_elem else '' artist = artist_elem.get('content', '') if artist_elem else 'Unknown Artist' image_url = image_elem.get('content', '') if image_elem else None return UniversalMetadata( service=MusicService.TIDAL, service_id=parsed_url.id, title=title or f"Tidal {parsed_url.item_type.title()}", artist=artist, image_url=image_url, original_url=parsed_url.url ) else: logger.warning(f"Tidal page not found: {response.status}") except Exception as e: logger.error(f"Error getting Tidal metadata: {e}") # Fallback metadata return UniversalMetadata( service=MusicService.TIDAL, service_id=parsed_url.id, title=f"Tidal {parsed_url.item_type.title()}", artist="Unknown Artist", original_url=parsed_url.url ) async def _get_apple_music_metadata(self, parsed_url: ParsedURL) -> Optional[UniversalMetadata]: """Get metadata from Apple Music""" try: return UniversalMetadata( service=MusicService.APPLE_MUSIC, service_id=parsed_url.id, title=f"Apple Music {parsed_url.item_type.title()}", artist="Unknown Artist", original_url=parsed_url.url ) except Exception as e: logger.error(f"Error getting Apple Music metadata: {e}") return None async def _get_youtube_metadata(self, parsed_url: ParsedURL) -> Optional[UniversalMetadata]: """Get metadata from YouTube""" try: return UniversalMetadata( service=MusicService.YOUTUBE, service_id=parsed_url.id, title=f"YouTube {parsed_url.item_type.title()}", artist="Unknown Artist", original_url=parsed_url.url ) except Exception as e: logger.error(f"Error getting YouTube metadata: {e}") return None async def _get_youtube_music_metadata(self, parsed_url: ParsedURL) -> Optional[UniversalMetadata]: """Get metadata from YouTube Music""" try: return UniversalMetadata( service=MusicService.YOUTUBE_MUSIC, service_id=parsed_url.id, title=f"YouTube Music {parsed_url.item_type.title()}", artist="Unknown Artist", original_url=parsed_url.url ) except Exception as e: logger.error(f"Error getting YouTube Music metadata: {e}") return None async def _get_soundcloud_metadata(self, parsed_url: ParsedURL) -> Optional[UniversalMetadata]: """Get metadata from SoundCloud""" try: return UniversalMetadata( service=MusicService.SOUNDCLOUD, service_id=parsed_url.id, title=f"SoundCloud {parsed_url.item_type.title()}", artist="Unknown Artist", original_url=parsed_url.url ) except Exception as e: logger.error(f"Error getting SoundCloud metadata: {e}") return None async def _get_deezer_metadata(self, parsed_url: ParsedURL) -> Optional[UniversalMetadata]: """Get metadata from Deezer""" try: return UniversalMetadata( service=MusicService.DEEZER, service_id=parsed_url.id, title=f"Deezer {parsed_url.item_type.title()}", artist="Unknown Artist", original_url=parsed_url.url ) except Exception as e: logger.error(f"Error getting Deezer metadata: {e}") return None async def _get_musicbrainz_metadata(self, parsed_url: ParsedURL) -> Optional[UniversalMetadata]: """Get metadata from MusicBrainz""" try: return UniversalMetadata( service=MusicService.MUSICBRAINZ, service_id=parsed_url.id, title=f"MusicBrainz {parsed_url.item_type.title()}", artist="Unknown Artist", original_url=parsed_url.url ) except Exception as e: logger.error(f"Error getting MusicBrainz metadata: {e}") return None async def _get_discogs_metadata(self, parsed_url: ParsedURL) -> Optional[UniversalMetadata]: """Get metadata from Discogs""" try: return UniversalMetadata( service=MusicService.DISCOGS, service_id=parsed_url.id, title=f"Discogs {parsed_url.item_type.title()}", artist="Unknown Artist", original_url=parsed_url.url ) except Exception as e: logger.error(f"Error getting Discogs metadata: {e}") return None def add_download(self, url: str, quality: DownloadQuality = None) -> Optional[str]: """Add a download to the queue""" try: if quality is None: quality = self.default_quality # Parse URL parsed_url = self.parse_url(url) if not parsed_url: logger.error(f"Invalid URL: {url}") return None # Create download item download_id = str(time.time()) item = DownloadItem( id=download_id, url=url, metadata=UniversalMetadata( service=parsed_url.service, service_id=parsed_url.id, title=f"{parsed_url.service.value.title()} {parsed_url.item_type.title()}", artist="Unknown Artist", original_url=url ), quality=quality, status=DownloadStatus.PENDING ) # Add to queue self.download_queue.append(item) logger.info(f"Added download: {url}") return download_id except Exception as e: logger.error(f"Error adding download: {e}") return None def get_download_status(self, download_id: str) -> Optional[DownloadItem]: """Get status of a download""" for item in self.download_queue: if item.id == download_id: return item return None def get_all_downloads(self) -> List[DownloadItem]: """Get all downloads""" return self.download_queue.copy() # Global instance universal_music_downloader = UniversalMusicDownloader()