Blog

import subprocess
import os
import json
import re
import concurrent.futures

def process_file(file_path, input_directory, verbose):
    """Processes a single .aax file."""
    filename = os.path.basename(file_path)
    filename_no_ext = os.path.splitext(filename)[0]
    output_dir = os.path.join(input_directory, filename_no_ext)
    output_mp3 = os.path.join(output_dir, f"{filename_no_ext}.mp3")

    if verbose:
        print(f"Processing file: {file_path}")
        print(f"Output directory: {output_dir}")
        print(f"Output MP3: {output_mp3}")

    os.makedirs(output_dir, exist_ok=True)

    if os.path.exists(output_mp3):
        if verbose:
            print(f"Output file {output_mp3} already exists. Skipping conversion.")
        return

    ffmpeg_command = ["ffmpeg", "-y", "-activation_bytes", "XXXXXXXXX", "-i", file_path, "-codec:a", "libmp3lame", output_mp3]
    try:
        subprocess.run(ffmpeg_command, check=True, capture_output=not verbose)
    except subprocess.CalledProcessError as e:
        print(f"Error: ffmpeg conversion failed for {file_path}")
        if verbose:
            print(e.stderr.decode())
        return

    if verbose:
        print(f"Conversion complete for {file_path}")

    try:
        ffprobe_output = subprocess.run(["ffprobe", "-i", output_mp3, "-show_chapters", "-print_format", "json"], capture_output=True, text=True, check=True).stdout
        chapters_data = json.loads(ffprobe_output)
        chapter_count = len(chapters_data.get("chapters", []))
    except (subprocess.CalledProcessError, json.JSONDecodeError):
        chapter_count = 0

    if chapter_count > 0:
        if verbose:
            print(f"Found {chapter_count} chapters. Splitting...")

        for i, chapter in enumerate(chapters_data["chapters"]):
            start_time = float(chapter["start_time"])
            end_time = float(chapter["end_time"])
            chapter_title = chapter.get("tags", {}).get("title", f"Chapter {i+1}")
            chapter_title = re.sub(r'[\\/*?:"<>|]', "", chapter_title)
            chapter_output = os.path.join(output_dir, f"{filename_no_ext}_{chapter_title}.mp3")
            duration = end_time - start_time

            if verbose:
                print(f"Extracting chapter: {chapter_title}, Start: {start_time}, Duration: {duration}")
            
            ffmpeg_chapter_command = ["ffmpeg", "-ss", str(start_time), "-t", str(duration), "-i", output_mp3, "-c", "copy", chapter_output]
            subprocess.run(ffmpeg_chapter_command, capture_output=not verbose)

    elif verbose:
        print(f"No chapters found in {output_mp3}.")


def convert_aax_to_mp3(input_directory=".", verbose=True, max_workers=None):
    """Converts .aax files to MP3 with multicore support."""

    try:
        subprocess.run(["ffmpeg", "-version"], check=True, capture_output=True)
        subprocess.run(["ffprobe", "-version"], check=True, capture_output=True)
        subprocess.run(["jq", "--version"], check=True, capture_output=True)
    except FileNotFoundError:
        print("Error: ffmpeg, ffprobe, and/or jq are not installed. Please install them.")
        return

    aax_files = [os.path.join(input_directory, filename) for filename in os.listdir(input_directory) if filename.endswith(".aax")]

    with concurrent.futures.ProcessPoolExecutor(max_workers=max_workers) as executor:
        futures = [executor.submit(process_file, file_path, input_directory, verbose) for file_path in aax_files]
        concurrent.futures.wait(futures)  # Wait for all processes to finish

    print("Script finished.")


if __name__ == "__main__":
    import argparse

    parser = argparse.ArgumentParser(description="Convert AAX files to MP3 with multicore support.")
    parser.add_argument("-i", "--input", dest="input_directory", default=".", help="Input directory containing .aax files (default: current directory).")
    parser.add_argument("-v", "--verbose", action="store_true", help="Enable verbose output.")
    parser.add_argument("-w", "--workers", dest="max_workers", type=int, default=None, help="Maximum number of worker processes (default: number of CPUs).")

    args = parser.parse_args()

    convert_aax_to_mp3(input_directory=args.input_directory, verbose=args.verbose, max_workers=args.max_workers)