TechOverflow Logo
TechOverflow

递归地将所有音频文件重新编码为 OPUS 的 Python 脚本

English Deutsch 简体中文

此 Python 脚本递归地将给定输入目录中的所有 FLAC、WAV 和 MP3 等文件转换为 OPUS。默认比特率为 96k，大致相当于 192k 到 256k MP3。你可以选择不同的

编码后，会自动检查源文件和转换文件的长度。如果长度差异超过 0.2 秒，则认为编码失败。

有一个 --delete 选项用于删除源文件（当然，仅当 OPUS 文件与源文件长度相同时）。谨慎使用，使用前做好备份，此脚本不提供任何明示或暗示的保证 - 它可能会丢失你的某些音乐！

脚本自动执行并行编码，因此即使转码许多文件也相当快。

除非命令行中给出 -o/--overwrite，否则不会覆盖任何文件。

如何使用

首先，安装 ffmpeg 和 ffprobe。现在使用以下命令安装 ffmpeg 的 python 绑定

install_ffmpeg_python.sh
pip install ffmpeg-python

以下是脚本提供的命令行选项：

ConvertToOPUS_usage.txt
usage: ConvertToOPUS.py [-h] [--bitrate BITRATE] [--delete] [-o] [-j THREADS] directory

Transcode MP3, FLAC & WAV files to OPUS.

positional arguments:
  directory             Directory containing the files to transcode

options:
  -h, --help            show this help message and exit
  --bitrate BITRATE     Bitrate for the OPUS files
  --delete              Delete original files after successful transcoding
  -o, --overwrite       Overwrite existing OPUS files
  -j THREADS, --threads THREADS
                        Number of parallel transcodes

源代码

ConvertToOPUS.py
#!/usr/bin/env python3
import os
import sys
import argparse
import ffmpeg
from concurrent.futures import ProcessPoolExecutor
import logging

logging.basicConfig(level=logging.INFO, format='%(levelname)s: %(message)s')

def get_file_duration(file_path):
    try:
        probe = ffmpeg.probe(file_path)
        audio_stream = next((stream for stream in probe['streams'] if stream['codec_type'] == 'audio'), None)
        return float(audio_stream['duration'])
    except Exception as e:
        logging.error(f"Failed to get duration for {file_path}: {e}")
        return None

def transcode_file(file_path, bitrate, delete_after_transcode, overwrite):
    try:
        output_file = file_path.rsplit('.', 1)[0] + '.opus'
        if not overwrite and os.path.exists(output_file):
            logging.info(f"{output_file} exists. Skipping due to no overwrite flag.")
            return

        stream = (
            ffmpeg
            .input(file_path)
            .output(output_file, ab=bitrate, loglevel="error")
        )

        if overwrite:
            stream = stream.overwrite_output()

        stream.run()

        original_duration = get_file_duration(file_path)
        transcoded_duration = get_file_duration(output_file)

        if original_duration is None or transcoded_duration is None:
            logging.error(f"Failed to transcode {file_path}. Could not retrieve file duration.")
            os.remove(output_file)
            return

        duration_diff = abs(original_duration - transcoded_duration)
        if duration_diff >= 0.2:
            logging.error(f"Transcoding failed for {file_path}. Duration mismatch.")
            os.remove(output_file)
        else:
            logging.info(f"Transcoded {file_path} to {output_file} successfully.")
            if delete_after_transcode:
                os.remove(file_path)
    except Exception as e:
        logging.error(f"Failed to transcode {file_path}: {e}")

def main():
    parser = argparse.ArgumentParser(description="Transcode MP3, FLAC & WAV files to OPUS.")
    parser.add_argument("directory", type=str, help="Directory containing the files to transcode")
    parser.add_argument("--bitrate", type=str, default="96k", help="Bitrate for the OPUS files")
    parser.add_argument("--delete", action="store_true", help="Delete original files after successful transcoding")
    parser.add_argument("-o", "--overwrite", action="store_true", help="Overwrite existing OPUS files")
    parser.add_argument("-j", "--threads", type=int, default=os.cpu_count(), help="Number of parallel transcodes")

    args = parser.parse_args()

    supported_extensions = ['mp3', 'flac', 'wav', 'm4a']

    # List all files with the supported extensions in the directory recursively
    files_to_transcode = []
    for root, dirs, files in os.walk(args.directory):
        for file in files:
            if file.split('.')[-1].lower() in supported_extensions:
                files_to_transcode.append(os.path.join(root, file))

    logging.info(f"Found {len(files_to_transcode)} files to transcode.")

    with ProcessPoolExecutor(max_workers=args.threads) as executor:
        for file_path in files_to_transcode:
            executor.submit(transcode_file, file_path, args.bitrate, args.delete, args.overwrite)

if __name__ == '__main__':
    main()
Check out similar posts by category: Audio, Audio/Video, Python