import os import uuid import subprocess import shutil 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 from mutagen.id3 import ID3 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