tkrzCompressor - Audio compressor

compress.py

from pathlib import Path
from pydub import AudioSegment
from concurrent.futures import ThreadPoolExecutor, as_completed

OUTPUT_FOLDER_NAME = 'outputFiles'
OUTPUT_FOLDER = Path(OUTPUT_FOLDER_NAME)
OUTPUT_EXTENSION = 'mp3'
OUTPUT_BITRATE = '320k'


def make_output_directories(files):
    OUTPUT_FOLDER.mkdir(exist_ok=True)

    total_files = 0
    converted_files = 0

    for file in files:
        total_files += 1

        folder = OUTPUT_FOLDER.joinpath(file.parent)
        output_file = folder.joinpath(f'{file.stem}.{OUTPUT_EXTENSION}')

        if output_file.exists():
            converted_files += 1
        else:
            folder.mkdir(exist_ok=True)

    if converted_files == total_files:
        exit('tkrzCompressor - All files are already converted!')


def process_file(file):
    output_path = OUTPUT_FOLDER.joinpath(file.parent).joinpath(file.stem).with_suffix(f'.{OUTPUT_EXTENSION}')
    if not output_path.exists():
        audio = AudioSegment.from_wav(f'{file.parent}/{file.stem}.wav')
        audio.export(output_path, format=OUTPUT_EXTENSION, bitrate=OUTPUT_BITRATE)


with ThreadPoolExecutor() as executor:
    files = [file for file in Path('.').glob('**/*.wav') if file.parents[1] != OUTPUT_FOLDER]

    if len(files) == 0:
        print("tkrzCompressor - No tracks to process! Exiting...", flush=True)
        exit(0)

    make_output_directories(files)
    futures = {executor.submit(process_file, file): file for file in files}

    for future in as_completed(futures):
        folder, file_name = futures[future]
        try:
            future.result()
        except Exception:
            print(
                f"tkrzCompressor - Failed to process {folder}/{file_name}!", flush=True)
        else:
            print(
                f'tkrzCompressor - Successfully compressed {folder}/{file_name}.', flush=True)

print(f'tkrzCompressor - Done!', flush=True)