259 lines
11 KiB
Python
259 lines
11 KiB
Python
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
|