Python audio compressor using ffmpeg with Opus support for highly efficient speech compression. Supports batch processing, dry runs, and customizable bitrates.

convert_audio_ffmpeg.py

#!/usr/bin/env python
"""
This module provides a command-line tool for compressing audio files using ffmpeg.

It supports various audio formats (Opus, AAC, OGG, MP3), customizable bitrates,
mono/stereo conversion, parallel processing, and dry-run mode. It also handles
input and output directories and provides sensible default settings.

Key features:

*   **Format support:** Compresses to Opus, AAC, OGG, and MP3. Opus is recommended
    for speech compression due to its high efficiency at low bitrates.
*   **Bitrate control:** Allows specifying custom bitrates or uses format-specific
    "sweet spot" defaults for optimal compression.
*   **Mono/Stereo conversion:** Supports converting audio to mono to further reduce
    file size (recommended for speech).
*   **Parallel processing:** Uses multiprocessing to significantly speed up
    compression of multiple files.
*   **Dry-run mode:** Allows previewing the compression process without modifying
    any files.
*   **Input/Output directory handling:** Accepts input and output directories as
    arguments, with defaults for convenience.
*   **Error Handling:** Includes error checking for invalid inputs and ffmpeg
    failures.

Usage:
    python compress_audio.py [-f FORMAT] [-b BITRATE] [-d] [-j JOBS] [-m] [-i INPUT] [-o OUTPUT]

    -f, --format    Output format (opus, aac, ogg, mp3). Default: opus.
    -b, --bitrate   Bitrate in kbps. If not specified, a format-specific
                    'sweet spot' is used.
    -d, --dry-run   Perform a dry run (no compression).
    -j, --jobs      Number of parallel jobs. Default: 10.
    -m, --mono      Convert to mono. Default: stereo.
    -i, --input     Input directory. Default: current directory.
    -o, --output    Output directory. Default: 'compressed'.

Example:
    python compress_audio.py -f opus -b 16 -m -i input_audio -o output_audio

"""

import os
import subprocess
import multiprocessing
import argparse

def compress_audio(input_file, bitrate, dry_run, format, mono, output_dir):
    """Compresses an audio file using ffmpeg.

    This function takes an input audio file (currently supporting only MP3 as input),
    compresses it using ffmpeg with the specified parameters, and saves the
    compressed file to the specified output directory.

    Args:
        input_file (str): Path to the input audio file.
        bitrate (int): Bitrate in kbps for the output audio.
        dry_run (bool): If True, performs a dry run (no compression) and prints
            what would be done.
        format (str): Output audio format ("opus", "aac", "ogg", or "mp3").
        mono (bool): If True, converts the audio to mono; otherwise, keeps it stereo.
        output_dir (str): Path to the output directory where the compressed
            file will be saved.

    Returns:
        None. Prints messages to the console indicating the compression progress
        or any errors encountered.

    Raises:
        subprocess.CalledProcessError: If the ffmpeg command fails.

    Example:
        compress_audio("input.mp3", 16, False, "opus", True, "compressed")
    """

    output_ext = ".opus" if format == "opus" else ".m4a" if format == "aac" else ".ogg" if format == "ogg" else ".mp3"
    output_file = os.path.join(output_dir, os.path.basename(input_file).replace(".mp3", f"-{bitrate}k_{'mono' if mono else 'stereo'}{output_ext}"))

    if not os.path.exists(output_file):
        if dry_run:
            print(f"[Dry Run] Would compress: \"{input_file}\" -> \"{output_file}\"")
        else:
            try:
                ffmpeg_command = ["ffmpeg", "-i", input_file]
                if mono:
                    ffmpeg_command.extend(["-ac", "1"])
                if format == "opus":
                    ffmpeg_command.extend(["-c:a", "libopus", "-b:a", f"{bitrate}k"])
                elif format == "aac":
                    ffmpeg_command.extend(["-c:a", "aac", "-b:a", f"{bitrate}k"])
                elif format == "ogg":
                    ffmpeg_command.extend(["-c:a", "libvorbis", "-b:a", f"{bitrate}k"])
                else:  # mp3
                    ffmpeg_command.extend(["-ab", f"{bitrate}k"])
                ffmpeg_command.append(output_file)
                subprocess.run(ffmpeg_command, check=True)
                print(f"Compressed: \"{input_file}\" -> \"{output_file}\"")
            except subprocess.CalledProcessError as e:
                print(f"Error compressing: \"{input_file}\": {e}")
    else:
        print(f"Skipping (already exists): \"{input_file}\" -> \"{output_file}\"")

def main():
    """Main function to parse arguments and run the compression process.

    This function parses command-line arguments, sets default values, validates
    input, and manages the parallel compression process.

    Args:
        None. Reads command-line arguments.

    Returns:
        None.

    Raises:
        SystemExit: If there are errors in the command-line arguments or input.
    """

    max_cpu = multiprocessing.cpu_count()
    parser = argparse.ArgumentParser(description="Compress MP3 files to various audio formats.")

    parser.add_argument("-f", "--format", choices=["opus", "aac", "ogg", "mp3"], default="opus",
                        help="Output format (default: opus).")
    parser.add_argument("-b", "--bitrate", type=int, metavar="B",
                        help="Bitrate in kbps. If not specified, a format-specific 'sweet spot' is used.")
    parser.add_argument("-d", "--dry-run", action="store_true", help="Perform a dry run (no compression)")
    parser.add_argument("-j", "--jobs", type=int, default=10, metavar="N",
                        help=f"Number of parallel jobs (default: 10, max: {max_cpu})")
    parser.add_argument("-m", "--mono", action="store_true", help="Convert to mono (default: stereo)")
    parser.add_argument("-i", "--input", type=str, default=".", help="Input directory (default: current directory)")
    parser.add_argument("-o", "--output", type=str, default="compressed", help="Output directory (default: compressed)")
    parser.add_argument("-c", "--create-output-dir", action="store_true", default=False,
                        help="Create the output directory if it does not exist (default: False)")

    if len(os.sys.argv) == 1:
        parser.print_help()
        os.sys.exit(1)

    args = parser.parse_args()

    format = args.format
    bitrate = args.bitrate
    dry_run = args.dry_run
    num_jobs = args.jobs
    mono = args.mono
    input_dir = args.input
    output_dir = args.output
    create_output_dir = args.create_output_dir

    sweet_spots = {
        "opus": 16,  # Excellent for speech
        "aac": 96,   # Good compromise for music and speech
        "ogg": 96,   # Good quality
        "mp3": 64,   # Good for speech
    }

    if bitrate is None:
        bitrate = sweet_spots[format]
        print(f"Using default bitrate for {format}: {bitrate} kbps")

    error_occurred = False

    if num_jobs < 1:
        print("Error: Number of jobs must be at least 1.")
        error_occurred = True
    if num_jobs > max_cpu:
        print(f"Error: Number of jobs cannot exceed {max_cpu} (number of CPUs on this machine).")
        error_occurred = True

    if not os.path.isdir(input_dir):
        print(f"Error: Input directory \"{input_dir}\" does not exist.")
        error_occurred = True

    if not os.path.exists(output_dir):
        if create_output_dir:
            try:
                os.makedirs(output_dir)
                print(f"Created output directory \"{output_dir}\".")
            except OSError as e:
                print(f"Error creating the output directory \"{output_dir}\": {e}")
                error_occurred = True
        else:
            print(f"Error: Output directory \"{output_dir}\" does not exist.")
            error_occurred = True

    os.makedirs(output_dir, exist_ok=True)
    if not os.path.exists(output_dir):
        print(f"Error creating the output directory \"{output_dir}\".")
        error_occurred = True
    else:
        print(f"Output directory \"{output_dir}\" is ready.")

    if error_occurred:
        parser.print_help()
        os.sys.exit(1)

    mp3_files = [os.path.join(input_dir, f) for f in os.listdir(input_dir) if f.endswith(".mp3")]

    if not mp3_files:
        print(f"No mp3 files found in \"{input_dir}\"")
        os.sys.exit(1)

    if num_jobs == 1:
        for input_file in mp3_files:
            compress_audio(input_file, bitrate, dry_run, format, mono, output_dir)
    else:
        with multiprocessing.Pool(processes=num_jobs) as pool:
            pool.starmap(compress_audio, [(input_file, bitrate, dry_run, format, mono, output_dir) for input_file in mp3_files])

if __name__ == "__main__":
    main()