187 lines
7.5 KiB
Python
187 lines
7.5 KiB
Python
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]:
|
|
"""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:
|
|
"""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 |