navidrome-music-sync/app/providers/universal.py

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"
        ])

    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"
                    tracks.append(TrackMetadata(
                        service="youtube",
                        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 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"

        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):
                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