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 open oEmbed API & yt-dlp fallback if "spotify.com" in lower: tracks = self._resolve_spotify_oembed(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_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() # data has title ("Track Name" or "Track Name - Single"), thumbnail_url full_title = data.get("title", "Unknown") # Spotfiy oEmbed titles are often "Title" or "Artist - Title" 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