153 lines
5.9 KiB
Python
153 lines
5.9 KiB
Python
import os
|
|
import asyncio
|
|
import subprocess
|
|
from typing import List, Optional
|
|
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"
|
|
])
|
|
|
|
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 / Apple Music via SpotDL parser / metadata
|
|
if "spotify.com" in lower or "music.apple.com" in lower:
|
|
return self._resolve_spotdl(url)
|
|
|
|
# Handle YouTube / YouTube Music via yt-dlp
|
|
return self._resolve_ytdlp(url)
|
|
|
|
def _resolve_spotdl(self, url: str) -> List[TrackMetadata]:
|
|
# SpotDL resolves metadata seamlessly
|
|
tracks: List[TrackMetadata] = []
|
|
try:
|
|
from spotdl.types.song import Song
|
|
from spotdl.utils.search import parse_query
|
|
|
|
songs = parse_query([url])
|
|
for s in songs:
|
|
tracks.append(TrackMetadata(
|
|
service="spotify" if "spotify" in url else "apple",
|
|
service_id=s.url or s.song_id,
|
|
title=s.name,
|
|
artist=s.artist,
|
|
album=s.album_name,
|
|
track_number=s.track_number,
|
|
year=int(s.year) if s.year else None,
|
|
genre=s.genres[0] if s.genres else None,
|
|
duration_ms=int(s.duration * 1000) if s.duration else None,
|
|
is_explicit=bool(s.explicit),
|
|
cover_url=s.cover_url,
|
|
source_url=s.url
|
|
))
|
|
except Exception as e:
|
|
print(f"SpotDL resolve error, falling back to yt-dlp: {e}")
|
|
return self._resolve_ytdlp(url)
|
|
return tracks
|
|
|
|
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"
|
|
tracks.append(TrackMetadata(
|
|
service="youtube",
|
|
service_id=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 f"https://www.youtube.com/watch?v={entry.get('id')}"
|
|
))
|
|
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"
|
|
|
|
# Search and download best audio via yt-dlp
|
|
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 == "youtube" 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):
|
|
# Apply high quality tags from metadata
|
|
import httpx
|
|
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
|