import os import re import asyncio from typing import List, Optional import httpx import yt_dlp from app.config import settings from app.providers.base import BaseProvider, TrackMetadata from app.core.tagger import apply_id3_tags class UniversalProvider(BaseProvider): def can_handle(self, url: str) -> bool: lower = url.lower() return any(domain in lower for domain in [ "spotify.com", "music.apple.com", "youtube.com", "youtu.be", "music.youtube.com", "soundcloud.com", "snd.sc", "api.soundcloud.com" ]) async def resolve(self, url: str) -> List[TrackMetadata]: loop = asyncio.get_event_loop() return await loop.run_in_executor(None, self._resolve_sync, url) def _resolve_sync(self, url: str) -> List[TrackMetadata]: lower = url.lower() # Handle Spotify via spotifyscraper & open oEmbed fallback if "spotify.com" in lower: tracks = self._resolve_spotify(url) if tracks: return tracks # Handle Apple Music via iTunes search / oembed / yt-dlp if "music.apple.com" in lower: tracks = self._resolve_apple_music(url) if tracks: return tracks # Default: Handle YouTube / YouTube Music and other generic links via yt-dlp return self._resolve_ytdlp(url) def _resolve_spotify(self, url: str) -> List[TrackMetadata]: tracks: List[TrackMetadata] = [] try: from spotify_scraper import SpotifyClient m_track = re.search(r'track/([a-zA-Z0-9]+)', url) m_album = re.search(r'album/([a-zA-Z0-9]+)', url) m_playlist = re.search(r'playlist/([a-zA-Z0-9]+)', url) with SpotifyClient() as client: if m_track: t = client.get_track(m_track.group(1)) if t: artist = ", ".join(a.name for a in t.artists) if t.artists else "Unknown" album = t.album.name if (t.album and hasattr(t.album, "name")) else "Singles" cover = None if hasattr(t, "images") and t.images: cover = getattr(t.images[0], "url", None) elif hasattr(t, "album") and t.album and getattr(t.album, "images", None): cover = getattr(t.album.images[0], "url", None) tracks.append(TrackMetadata( service="spotify", service_id=str(t.id), title=t.name, artist=artist, album=album, duration_ms=getattr(t, "duration_ms", None), cover_url=cover, is_explicit=getattr(t, "explicit", False), source_url=t.url or url )) return tracks elif m_album: alb = client.get_album(m_album.group(1)) if alb and hasattr(alb, "tracks"): cover = getattr(alb.images[0], "url", None) if hasattr(alb, "images") and alb.images else None for idx, t in enumerate(alb.tracks, 1): artist = ", ".join(a.name for a in t.artists) if t.artists else "Unknown" tracks.append(TrackMetadata( service="spotify", service_id=str(t.id), title=t.name, artist=artist, album=alb.name, track_number=idx, duration_ms=getattr(t, "duration_ms", None), cover_url=cover, is_explicit=getattr(t, "explicit", False), source_url=t.url or url )) return tracks elif m_playlist: pl = client.get_playlist(m_playlist.group(1)) if pl and hasattr(pl, "tracks"): cover = getattr(pl.images[0], "url", None) if hasattr(pl, "images") and pl.images else None for idx, t in enumerate(pl.tracks, 1): artist = ", ".join(a.name for a in t.artists) if t.artists else "Unknown" album = getattr(t.album, "name", pl.name) if getattr(t, "album", None) else pl.name tracks.append(TrackMetadata( service="spotify", service_id=str(t.id), title=t.name, artist=artist, album=album, track_number=idx, duration_ms=getattr(t, "duration_ms", None), cover_url=cover, is_explicit=getattr(t, "explicit", False), source_url=t.url or url )) return tracks except Exception as e: print(f"Spotify scraper resolution failed: {e}") # Fallback to open oembed if scraper failed return self._resolve_spotify_oembed(url) def _resolve_spotify_oembed(self, url: str) -> List[TrackMetadata]: try: oembed_url = f"https://open.spotify.com/oembed?url={url}" with httpx.Client(timeout=10.0) as client: resp = client.get(oembed_url) if resp.status_code == 200: data = resp.json() full_title = data.get("title", "Unknown") artist = "Unknown" title = full_title if " - " in full_title: parts = full_title.split(" - ", 1) artist = parts[0].strip() title = parts[1].strip() return [TrackMetadata( service="spotify", service_id=url, title=title, artist=artist, album="Singles", cover_url=data.get("thumbnail_url"), source_url=url )] except Exception as e: print(f"Spotify oEmbed resolution failed: {e}") return [] def _resolve_apple_music(self, url: str) -> List[TrackMetadata]: # Extract song/album title from URL or yt-dlp m = re.search(r'/album/([^/]+)/(\d+)', url) if m: album_slug = m.group(1).replace("-", " ").title() return [TrackMetadata( service="apple", service_id=url, title=album_slug, artist="Apple Music", album=album_slug, source_url=url )] return self._resolve_ytdlp(url) def _resolve_ytdlp(self, url: str) -> List[TrackMetadata]: tracks: List[TrackMetadata] = [] ydl_opts = { "extract_flat": True, "quiet": True, "no_warnings": True, } try: with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=False) if not info: return [] entries = info.get("entries") if "entries" in info else [info] for entry in entries: if not entry: continue title = entry.get("title", "Unknown") uploader = entry.get("uploader") or entry.get("artist") or "Unknown" service_name = "soundcloud" if ("soundcloud" in (entry.get("webpage_url") or url).lower() or "snd.sc" in url.lower()) else "youtube" tracks.append(TrackMetadata( service=service_name, service_id=str(entry.get("id", "")), title=title, artist=uploader, album=entry.get("album"), duration_ms=int(entry.get("duration", 0) * 1000) if entry.get("duration") else None, cover_url=entry.get("thumbnail"), source_url=entry.get("webpage_url") or url )) except Exception as e: print(f"yt-dlp resolve error: {e}") return tracks async def download_track(self, track: TrackMetadata, dest_path: str) -> bool: loop = asyncio.get_event_loop() return await loop.run_in_executor(None, self._download_sync, track, dest_path) def _download_sync(self, track: TrackMetadata, dest_path: str) -> bool: tmp_output = dest_path + ".tmp.mp3" query = f"{track.artist} - {track.title}" if track.is_explicit: query += " explicit" ydl_opts = { "format": "bestaudio/best", "outtmpl": dest_path + ".tmp.%(ext)s", "postprocessors": [{ "key": "FFmpegExtractAudio", "preferredcodec": "mp3", "preferredquality": "320", }], "quiet": True, "no_warnings": True, "default_search": "ytsearch1:", } try: target_input = track.source_url if track.service in ("youtube", "soundcloud") and track.source_url else f"ytsearch1:{query}" with yt_dlp.YoutubeDL(ydl_opts) as ydl: ydl.download([target_input]) if os.path.exists(tmp_output): cover_bytes = None if track.cover_url: try: with httpx.Client(timeout=10.0) as http_client: resp = http_client.get(track.cover_url) if resp.status_code == 200: cover_bytes = resp.content except Exception as e: print(f"Warning: Failed to download cover: {e}") apply_id3_tags( tmp_output, title=track.title, artist=track.artist, album=track.album, track_number=track.track_number, year=track.year, genre=track.genre, cover_bytes=cover_bytes ) os.replace(tmp_output, dest_path) return True except Exception as e: print(f"Failed to download universal track {track.title}: {e}") return False