you-music/backend/app/services/downloader.py

import asyncio
import os
import re
import subprocess
from typing import Optional, Tuple
from pathlib import Path
import aiohttp
import mutagen
from mutagen.mp3 import MP3
from mutagen.id3 import ID3, TIT2, TPE1, TALB, APIC
from app.core.config import settings
import logging

logger = logging.getLogger(__name__)


class MusicDownloader:
    """Download music from various sources using yt-dlp (similar to xiaomusic)"""
    
    def __init__(self):
        self.download_path = settings.MUSIC_DIR
        self.temp_path = settings.TEMP_DIR
        self.proxy = settings.PROXY
        self.ffmpeg_location = settings.FFMPEG_LOCATION
        
    async def download_music(
        self, 
        url: str, 
        output_name: Optional[str] = None
    ) -> Tuple[bool, str, Optional[str]]:
        """
        Download music from URL using yt-dlp
        Returns: (success, file_path, error_message)
        """
        try:
            # Prepare output template
            if output_name:
                title = f"{output_name}.%(ext)s"
            else:
                title = "%(title)s.%(ext)s"
            
            # Build yt-dlp command arguments (similar to xiaomusic)
            cmd_args = [
                "yt-dlp",
                "--no-playlist",
                "-x",  # Extract audio
                "--audio-format", settings.YT_DLP_AUDIO_FORMAT,
                "--audio-quality", settings.YT_DLP_AUDIO_QUALITY,
                "--paths", self.download_path,
                "-o", title,
                "--ffmpeg-location", self.ffmpeg_location,
            ]
            
            if self.proxy:
                cmd_args.extend(["--proxy", self.proxy])
            
            cmd_args.append(url)
            
            logger.info(f"Downloading: {' '.join(cmd_args)}")
            
            # Execute download
            process = await asyncio.create_subprocess_exec(
                *cmd_args,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            
            stdout, stderr = await process.communicate()
            
            if process.returncode == 0:
                # Find the downloaded file
                output_file = await self._find_downloaded_file(output_name)
                if output_file:
                    logger.info(f"Download successful: {output_file}")
                    return True, output_file, None
                else:
                    return False, "", "Downloaded file not found"
            else:
                error_msg = stderr.decode() if stderr else "Unknown error"
                logger.error(f"Download failed: {error_msg}")
                return False, "", error_msg
                
        except Exception as e:
            logger.exception(f"Download exception: {e}")
            return False, "", str(e)
    
    async def _find_downloaded_file(self, output_name: Optional[str]) -> Optional[str]:
        """Find the most recently downloaded file"""
        try:
            files = []
            for ext in ['.mp3', '.m4a', '.opus', '.webm']:
                if output_name:
                    pattern = f"{output_name}{ext}"
                    file_path = os.path.join(self.download_path, pattern)
                    if os.path.exists(file_path):
                        return file_path
                else:
                    # Find most recent file
                    for file in Path(self.download_path).glob(f"*{ext}"):
                        files.append(file)
            
            if files:
                # Return most recent file
                latest_file = max(files, key=lambda x: x.stat().st_mtime)
                return str(latest_file)
            
            return None
        except Exception as e:
            logger.error(f"Error finding downloaded file: {e}")
            return None
    
    async def download_playlist(
        self, 
        url: str, 
        playlist_name: str
    ) -> Tuple[bool, list[str], Optional[str]]:
        """Download entire playlist"""
        try:
            output_dir = os.path.join(self.download_path, playlist_name)
            os.makedirs(output_dir, exist_ok=True)
            
            title = f"{playlist_name}/%(title)s.%(ext)s"
            
            cmd_args = [
                "yt-dlp",
                "--yes-playlist",
                "-x",
                "--audio-format", settings.YT_DLP_AUDIO_FORMAT,
                "--audio-quality", settings.YT_DLP_AUDIO_QUALITY,
                "--paths", self.download_path,
                "-o", title,
                "--ffmpeg-location", self.ffmpeg_location,
            ]
            
            if self.proxy:
                cmd_args.extend(["--proxy", self.proxy])
            
            cmd_args.append(url)
            
            logger.info(f"Downloading playlist: {' '.join(cmd_args)}")
            
            process = await asyncio.create_subprocess_exec(
                *cmd_args,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            
            await process.wait()
            
            # Get downloaded files
            downloaded_files = []
            if os.path.exists(output_dir):
                for file in Path(output_dir).glob("*.mp3"):
                    downloaded_files.append(str(file))
            
            return True, downloaded_files, None
            
        except Exception as e:
            logger.exception(f"Playlist download exception: {e}")
            return False, [], str(e)
    
    async def get_music_metadata(self, file_path: str) -> dict:
        """Extract metadata from audio file using mutagen"""
        try:
            audio = mutagen.File(file_path, easy=True)
            if audio is None:
                return {}
            
            metadata = {
                "title": audio.get("title", [os.path.basename(file_path)])[0] if audio.get("title") else os.path.basename(file_path),
                "artist": audio.get("artist", ["Unknown"])[0] if audio.get("artist") else "Unknown",
                "album": audio.get("album", [""])[0] if audio.get("album") else "",
                "duration": audio.info.length if hasattr(audio, 'info') else 0,
            }
            
            return metadata
        except Exception as e:
            logger.error(f"Error extracting metadata: {e}")
            return {
                "title": os.path.basename(file_path),
                "artist": "Unknown",
                "album": "",
                "duration": 0,
            }
    
    async def get_duration(self, file_path: str) -> float:
        """Get audio duration"""
        try:
            audio = mutagen.File(file_path)
            if audio and hasattr(audio, 'info'):
                return audio.info.length
            return 0.0
        except Exception:
            return 0.0
    
    def extract_youtube_id(self, url: str) -> Optional[str]:
        """Extract YouTube video ID from URL"""
        patterns = [
            r'(?:youtube\.com\/watch\?v=|youtu\.be\/)([a-zA-Z0-9_-]{11})',
            r'youtube\.com\/embed\/([a-zA-Z0-9_-]{11})',
        ]
        
        for pattern in patterns:
            match = re.search(pattern, url)
            if match:
                return match.group(1)
        
        return None
    
    def is_youtube_url(self, url: str) -> bool:
        """Check if URL is from YouTube"""
        return 'youtube.com' in url or 'youtu.be' in url
    
    def is_bilibili_url(self, url: str) -> bool:
        """Check if URL is from Bilibili"""
        return 'bilibili.com' in url


# Global instance
music_downloader = MusicDownloader()