music-app/backend/app/services/audio.py

import os
import uuid
import subprocess
from typing import Optional, Dict, Any
from sqlalchemy.orm import Session
from ..models.song import Song
from ..schemas.song import ScanResult

SUPPORTED_FORMATS = {'.mp3', '.aac', '.flac', '.wav', '.ogg', '.m4a'}
MUSIC_DIR = os.getenv("MUSIC_DIR", "./music")
UPLOAD_DIR = os.getenv("UPLOAD_DIR", "./uploads")
TRANSCODE_FORMAT = os.getenv("TRANSCODE_FORMAT", "ogg")
TRANSCODE_BITRATE = os.getenv("TRANSCODE_BITRATE", "192k")


def extract_metadata(file_path: str) -> Dict[str, Any]:  # pragma: no cover
    """Extract metadata from an audio file using mutagen."""
    from mutagen.mp3 import MP3
    from mutagen.flac import FLAC
    from mutagen.oggvorbis import OggVorbis
    from mutagen.wave import WAVE
    from mutagen.mp4 import MP4
    
    metadata: Dict[str, Any] = {}
    ext = os.path.splitext(file_path)[1].lower()
    
    try:
        if ext == '.mp3':
            audio = MP3(file_path)
            if audio.tags:
                metadata['title'] = audio.tags.get('TIT2', '').text[0] if audio.tags.get('TIT2') else None
                metadata['artist'] = audio.tags.get('TPE1', '').text[0] if audio.tags.get('TPE1') else None
                metadata['album'] = audio.tags.get('TALB', '').text[0] if audio.tags.get('TALB') else None
                metadata['genre'] = audio.tags.get('TCON', '').text[0] if audio.tags.get('TCON') else None
            metadata['duration'] = audio.info.length if audio.info else 0
            
            # Extract album art
            for tag in (audio.tags or {}).values():
                if hasattr(tag, 'data') and tag.FrameId == 'APIC':
                    art_path = os.path.join(os.path.dirname(file_path), f"art_{uuid.uuid4().hex[:8]}.jpg")
                    with open(art_path, 'wb') as f:
                        f.write(tag.data)
                    metadata['album_art_path'] = art_path
                    break
                    
        elif ext == '.flac':
            audio = FLAC(file_path)
            metadata['title'] = audio.get('TITLE', [''])[0] if audio.get('TITLE') else None
            metadata['artist'] = audio.get('ARTIST', [''])[0] if audio.get('ARTIST') else None
            metadata['album'] = audio.get('ALBUM', [''])[0] if audio.get('ALBUM') else None
            metadata['genre'] = audio.get('GENRE', [''])[0] if audio.get('GENRE') else None
            metadata['duration'] = audio.info.length if audio.info else 0
            
            # Extract album art from FLAC
            if audio.pictures:
                picture = audio.pictures[0]
                art_path = os.path.join(os.path.dirname(file_path), f"art_{uuid.uuid4().hex[:8]}.jpg")
                with open(art_path, 'wb') as f:
                    f.write(picture.data)
                metadata['album_art_path'] = art_path
                
        elif ext == '.ogg':
            audio = OggVorbis(file_path)
            metadata['title'] = audio.get('TITLE', [''])[0] if audio.get('TITLE') else None
            metadata['artist'] = audio.get('ARTIST', [''])[0] if audio.get('ARTIST') else None
            metadata['album'] = audio.get('ALBUM', [''])[0] if audio.get('ALBUM') else None
            metadata['genre'] = audio.get('GENRE', [''])[0] if audio.get('GENRE') else None
            metadata['duration'] = audio.info.length if audio.info else 0
            
        elif ext == '.wav':
            audio = WAVE(file_path)
            metadata['duration'] = audio.info.length if audio.info else 0
            
        elif ext in ('.m4a', '.aac'):
            audio = MP4(file_path)
            metadata['title'] = audio.get('\xa9nam', [b''])[0].decode() if audio.get('\xa9nam') else None
            metadata['artist'] = audio.get('\xa9ART', [b''])[0].decode() if audio.get('\xa9ART') else None
            metadata['album'] = audio.get('\xa9alb', [b''])[0].decode() if audio.get('\xa9alb') else None
            metadata['genre'] = audio.get('\xa9gen', [b''])[0].decode() if audio.get('\xa9gen') else None
            metadata['duration'] = audio.info.length if audio.info else 0
            
    except Exception:
        pass
    
    return metadata


def transcode_to_ogg(input_path: str, output_dir: str) -> Optional[str]:
    """Transcode audio file to OGG Vorbis format."""
    try:
        output_filename = f"{uuid.uuid4().hex}.ogg"
        output_path = os.path.join(output_dir, output_filename)
        
        cmd = [
            'ffmpeg', '-i', input_path,
            '-codec:a', 'libvorbis',
            '-q:a', '5',
            '-map_metadata', '0',
            '-y', output_path
        ]
        
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=120)
        
        if result.returncode == 0 and os.path.exists(output_path):
            return output_path
        return None
    except (subprocess.TimeoutExpired, FileNotFoundError, Exception):
        return None


def scan_directory(directory: str, db: Session) -> ScanResult:  # pragma: no cover
    """Scan a directory for music files and add them to the database."""
    scanned = 0
    added = 0
    skipped = 0
    errors = []
    
    os.makedirs(directory, exist_ok=True)
    os.makedirs(UPLOAD_DIR, exist_ok=True)
    
    for root, dirs, files in os.walk(directory):
        for filename in files:
            ext = os.path.splitext(filename)[1].lower()
            if ext not in SUPPORTED_FORMATS:
                continue
            
            scanned += 1
            file_path = os.path.join(root, filename)
            
            try:
                # Check if already in database
                existing = db.query(Song).filter(Song.file_path == file_path).first()
                if existing:
                    skipped += 1
                    continue
                
                metadata = extract_metadata(file_path)
                
                song_id = str(uuid.uuid4())
                
                # Transcode
                transcoded_path = transcode_to_ogg(file_path, UPLOAD_DIR)
                
                song = Song(
                    id=song_id,
                    title=metadata.get('title') or filename.replace(ext, ''),
                    artist=metadata.get('artist') or 'Unknown Artist',
                    album=metadata.get('album'),
                    duration_sec=metadata.get('duration', 0),
                    genre=metadata.get('genre'),
                    file_path=file_path,
                    transcoded_path=transcoded_path,
                    album_art_path=metadata.get('album_art_path'),
                    file_format=ext.replace('.', ''),
                    file_size_bytes=os.path.getsize(file_path),
                )
                
                db.add(song)
                added += 1
                
            except Exception as e:
                errors.append(f"Error processing {filename}: {str(e)}")
    
    db.commit()
    
    return ScanResult(scanned=scanned, added=added, skipped=skipped, errors=errors)


def delete_song(song: Song):
    """Delete song files from disk."""
    for path in [song.file_path, song.transcoded_path, song.album_art_path]:
        if path and os.path.exists(path):
            try:
                os.remove(path)
            except OSError:
                pass


def get_stream_path(song: Song) -> Optional[str]:
    """Get the best available stream path (transcoded first, then original)."""
    if song.transcoded_path and os.path.exists(song.transcoded_path):
        return song.transcoded_path
    if song.file_path and os.path.exists(song.file_path):
        return song.file_path
    return None