converters/audio_converter.py

import os
import shutil
from pathlib import Path
from typing import Callable, Dict, Any

# Setup FFmpeg
ffmpeg_path = shutil.which("ffmpeg") or shutil.which("ffmpeg.exe")

if ffmpeg_path:
    os.environ["FFMPEG_BINARY"] = ffmpeg_path

from pydub import AudioSegment


class AudioConverter:

    SUPPORTED_FORMATS = {
        ".mp3",
        ".wav",
        ".flac",
        ".ogg",
        ".aac",
        ".m4a",
        ".wma"
    }

    def convert(
        self,
        input_path: str,
        output_path: str,
        options: Dict[str, Any] | None = None,
        progress_callback: Callable | None = None
    ) -> bool:

        options = options or {}

        try:
            self._update(progress_callback, 10)

            # Check input file
            if not os.path.exists(input_path):
                raise FileNotFoundError(f"Input file not found: {input_path}")

            input_ext = Path(input_path).suffix.lower()
            output_ext = Path(output_path).suffix.lower()

            # Validate formats
            if input_ext not in self.SUPPORTED_FORMATS:
                raise ValueError(f"Unsupported input format: {input_ext}")

            if output_ext not in self.SUPPORTED_FORMATS:
                raise ValueError(f"Unsupported output format: {output_ext}")

            # Create output folder
            Path(output_path).parent.mkdir(
                parents=True,
                exist_ok=True
            )

            # Load audio
            audio = AudioSegment.from_file(input_path)

            self._update(progress_callback, 35)

            # Trim audio
            start = int(options.get("start_ms", 0))
            end = int(options.get("end_ms", len(audio)))

            if start < 0:
                start = 0

            if end > len(audio):
                end = len(audio)

            audio = audio[start:end]

            self._update(progress_callback, 55)

            # Normalize safely
            if options.get("normalize", True):

                # Prevent silent audio crash
                if audio.dBFS != float("-inf"):

                    target_dbfs = options.get("target_dbfs", -20.0)

                    change = target_dbfs - audio.dBFS

                    audio = audio.apply_gain(change)

            self._update(progress_callback, 75)

            # Bitrate
            bitrate = options.get("bitrate", "192k")

            # Export format
            export_format = output_ext.replace(".", "")

            export_args = {}

            # WAV does not use bitrate
            if export_format != "wav":
                export_args["bitrate"] = bitrate

            # AAC handling
            if export_format == "aac":
                export_format = "adts"

            # Export audio
            audio.export(
                output_path,
                format=export_format,
                **export_args
            )

            self._update(progress_callback, 100)

            return True

        except Exception as e:
            print(f"Audio conversion error: {e}")
            return False

    def _update(self, callback, value):

        try:
            if callback:
                callback(value)
        except Exception as e:
            print(f"Progress callback error: {e}")