NhanNguyen1309/audio-separation-api

1from __future__ import annotations2 3import subprocess4from pathlib import Path5 6from app.config import settings7from app.errors import ApiError8from app.models import StemMode9 10 11SUPPORTED_EXTENSIONS = {".mp3", ".wav", ".m4a"}12SUPPORTED_MIME_PREFIXES = {"audio/"}13SUPPORTED_MIME_TYPES = {"application/octet-stream"}14SUPPORTED_STEM_MODES: set[str] = {"2-stem", "4-stem"}15 16 17def ensure_supported_extension(filename: str) -> str:18    extension = Path(filename).suffix.lower()19    if extension not in SUPPORTED_EXTENSIONS:20        raise ApiError(21            code="unsupported_file_type",22            message="Only MP3, WAV, and M4A files are supported.",23            status_code=400,24        )25    return extension26 27 28def ensure_supported_stem_mode(stem_mode: str) -> StemMode:29    if stem_mode not in SUPPORTED_STEM_MODES:30        raise ApiError(31            code="unsupported_file_type",32            message="Only 2-stem and 4-stem separation are supported in this version.",33            status_code=400,34        )35    return stem_mode  # type: ignore[return-value]36 37 38def ensure_supported_mime_type(content_type: str | None) -> None:39    if not content_type:40        return41 42    if content_type in SUPPORTED_MIME_TYPES:43        return44 45    if any(content_type.startswith(prefix) for prefix in SUPPORTED_MIME_PREFIXES):46        return47 48    raise ApiError(49        code="unsupported_file_type",50        message="The uploaded file type is not supported.",51        status_code=400,52    )53 54 55def ensure_file_size(size_bytes: int) -> None:56    if size_bytes > settings.max_file_size_bytes:57        raise ApiError(58            code="file_too_large",59            message="This version supports files up to 60MB.",60            status_code=400,61        )62 63 64def probe_duration_seconds(path: Path) -> float:65    command = [66        "ffprobe",67        "-v",68        "error",69        "-show_entries",70        "format=duration",71        "-of",72        "default=noprint_wrappers=1:nokey=1",73        str(path),74    ]75    try:76        result = subprocess.run(command, capture_output=True, text=True, timeout=30, check=False)77    except (subprocess.SubprocessError, OSError) as exc:78        raise ApiError(code="ffprobe_failed", message="Could not inspect the audio file.", status_code=400) from exc79 80    if result.returncode != 0:81        raise ApiError(code="ffprobe_failed", message="Could not inspect the audio file.", status_code=400)82 83    try:84        return float(result.stdout.strip())85    except ValueError as exc:86        raise ApiError(code="ffprobe_failed", message="Could not read the audio duration.", status_code=400) from exc87 88 89def ensure_duration(path: Path) -> float:90    duration = probe_duration_seconds(path)91    if duration > settings.max_duration_seconds:92        raise ApiError(93            code="duration_too_long",94            message="This version supports songs up to 10 minutes.",95            status_code=400,96        )97    return duration98 99 100def ensure_decodable(path: Path) -> None:101    command = ["ffmpeg", "-v", "error", "-i", str(path), "-f", "null", "-"]102    try:103        result = subprocess.run(command, capture_output=True, text=True, timeout=60, check=False)104    except (subprocess.SubprocessError, OSError) as exc:105        raise ApiError(code="invalid_audio", message="The audio file could not be decoded.", status_code=400) from exc106 107    if result.returncode != 0:108        raise ApiError(code="invalid_audio", message="The audio file could not be decoded.", status_code=400)109 110 111def validate_saved_audio(path: Path) -> float:112    duration = ensure_duration(path)113    ensure_decodable(path)114    return duration115