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