32blogby StudioMitsu
ffmpeg12 min read

FFmpeg × Pythonで動画バッチ処理を自動化する

PythonからFFmpegを呼び出して動画のバッチ処理を自動化する方法。subprocess、ffmpeg-python、進捗表示、エラーハンドリングまで実践的に解説。

FFmpegPython自動化バッチ処理動画処理
目次

動画の変換作業を手作業でやっているなら、Pythonで自動化する価値は十分ある。100本のMP4をまとめてWebM変換したり、解像度を一括で変更したりするのも、スクリプト1本で片付けられる。

この記事では、PythonからFFmpegを呼び出す方法を基礎から説明する。subprocess で直接叩く方法から、ffmpeg-python ライブラリを使った書き方、フォルダ内を一括変換するスクリプト、進捗表示、エラーハンドリングまで僕が順番に解説する。

なぜPythonでFFmpegを使うのか

FFmpegはCLIツールとして完成度が高く、単体でも強力だ。でも、ファイルが大量にあったり、処理の組み合わせが複雑になってくると、シェルスクリプトだけでは辛くなってくる。

Pythonを組み合わせることで、次のことが楽になる。

  • ファイルの列挙と条件フィルタリング: 特定の拡張子、特定サイズ以上のファイルだけを対象にする
  • エラーハンドリングとリトライ: 失敗したファイルを記録して再実行する
  • 進捗表示: 何本中何本が終わったかをリアルタイムで把握する
  • ログ出力: 変換結果を後からレビューできる形で保存する

シェルスクリプトでもできなくはないが、Pythonのほうがロジックが複雑になったときに読みやすく保ちやすい。

subprocessで基本を押さえる

PythonからFFmpegを呼び出す一番シンプルな方法は subprocess.run() を使うことだ。FFmpegはCLIツールなので、コマンドライン引数をリストで渡すだけで動く。

python
import subprocess
from pathlib import Path


def convert_to_mp4(input_path: Path, output_path: Path) -> bool:
    """MP4に変換する。成功したらTrue、失敗したらFalseを返す。"""
    cmd = [
        "ffmpeg",
        "-i", str(input_path),
        "-c:v", "libx264",
        "-crf", "23",
        "-preset", "medium",
        "-c:a", "aac",
        "-b:a", "128k",
        "-y",               # 上書き確認なし
        str(output_path),
    ]

    result = subprocess.run(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    return result.returncode == 0


if __name__ == "__main__":
    src = Path("input.mov")
    dst = Path("output.mp4")

    if convert_to_mp4(src, dst):
        print(f"変換成功: {dst}")
    else:
        print("変換失敗")

いくつかポイントを説明する。

  • "-y" フラグで出力ファイルが既存でも上書きする。バッチ処理では基本的に付けておく
  • stdout=subprocess.PIPEstderr=subprocess.PIPE で出力をキャプチャする。これがないとターミナルにFFmpegの大量ログが流れる
  • FFmpegはログを stderr に出力する(正常時も)。エラー情報を取り出したい場合は result.stderr.decode() を参照する

ffmpeg-pythonライブラリを使う

subprocess で直接叩く方法はシンプルだが、オプションが増えるとコマンドリストが長くなって管理しにくくなる。そこで ffmpeg-python ライブラリを使う方法がある。

インストールはこれだけだ。

python
# pip install ffmpeg-python

使い方はメソッドチェーンでFFmpegのパイプラインを組み立てる形になる。

python
import ffmpeg


def transcode(input_path: str, output_path: str) -> None:
    """ffmpeg-pythonを使ったトランスコード。"""
    (
        ffmpeg
        .input(input_path)
        .output(
            output_path,
            vcodec="libx264",
            crf=23,
            preset="medium",
            acodec="aac",
            audio_bitrate="128k",
        )
        .overwrite_output()
        .run(capture_stdout=True, capture_stderr=True)
    )


if __name__ == "__main__":
    transcode("input.mov", "output.mp4")

エラーが発生した場合は ffmpeg.Error が投げられる。e.stderr にFFmpegの標準エラー出力が入っているので、デバッグに使える。

python
import ffmpeg


def transcode_safe(input_path: str, output_path: str) -> None:
    try:
        (
            ffmpeg
            .input(input_path)
            .output(output_path, vcodec="libx264", crf=23)
            .overwrite_output()
            .run(capture_stdout=True, capture_stderr=True)
        )
    except ffmpeg.Error as e:
        print("FFmpegエラー:")
        print(e.stderr.decode())
        raise

フォルダ内の動画を一括変換する

ここからが本番だ。フォルダ内にある全MOVファイルをMP4に変換するスクリプトを組む。pathlibglob() でファイルを列挙して、順番に変換する。

python
import subprocess
from pathlib import Path


INPUT_DIR = Path("./originals")
OUTPUT_DIR = Path("./converted")
INPUT_EXT = ".mov"
OUTPUT_EXT = ".mp4"


def convert_file(src: Path, dst: Path) -> bool:
    """1ファイルをMP4に変換する。"""
    dst.parent.mkdir(parents=True, exist_ok=True)

    cmd = [
        "ffmpeg",
        "-i", str(src),
        "-c:v", "libx264",
        "-crf", "23",
        "-preset", "fast",
        "-c:a", "aac",
        "-b:a", "128k",
        "-movflags", "+faststart",  # Web再生向け最適化
        "-y",
        str(dst),
    ]

    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    return result.returncode == 0


def batch_convert(input_dir: Path, output_dir: Path) -> None:
    """input_dir内の全INPUT_EXTファイルをOUTPUT_EXTに変換する。"""
    files = list(input_dir.glob(f"**/*{INPUT_EXT}"))

    if not files:
        print(f"{input_dir} に {INPUT_EXT} ファイルが見つかりません")
        return

    print(f"{len(files)} 件のファイルを変換します")
    success_count = 0
    fail_list: list[Path] = []

    for i, src in enumerate(files, start=1):
        # 入力ディレクトリからの相対パスを保ちながら出力先を決める
        relative = src.relative_to(input_dir)
        dst = output_dir / relative.with_suffix(OUTPUT_EXT)

        print(f"[{i}/{len(files)}] {src.name} → {dst.name}", end=" ... ")

        if convert_file(src, dst):
            print("OK")
            success_count += 1
        else:
            print("FAILED")
            fail_list.append(src)

    print(f"\n完了: {success_count}/{len(files)} 件成功")

    if fail_list:
        print("失敗したファイル:")
        for f in fail_list:
            print(f"  {f}")


if __name__ == "__main__":
    batch_convert(INPUT_DIR, OUTPUT_DIR)

glob の再帰パターンでサブフォルダも含めて処理する。出力側のフォルダ構造も元の構造を保ったまま作成されるので、ファイルが散在していても整理しながら変換できる。

進捗表示をつける

大量のファイルを変換するときは、あと何本残っているかがひと目でわかる進捗バーがほしくなる。tqdm を使えば1行で追加できる。

python
import subprocess
from pathlib import Path
from tqdm import tqdm


# pip install tqdm


def convert_file(src: Path, dst: Path) -> bool:
    dst.parent.mkdir(parents=True, exist_ok=True)
    cmd = [
        "ffmpeg", "-i", str(src),
        "-c:v", "libx264", "-crf", "23", "-preset", "fast",
        "-c:a", "aac", "-b:a", "128k",
        "-y", str(dst),
    ]
    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    return result.returncode == 0


def batch_convert_with_progress(input_dir: Path, output_dir: Path) -> None:
    files = list(input_dir.glob("**/*.mov"))
    fail_list: list[Path] = []

    with tqdm(total=len(files), unit="file", ncols=80) as pbar:
        for src in files:
            relative = src.relative_to(input_dir)
            dst = output_dir / relative.with_suffix(".mp4")

            pbar.set_description(src.name[:30])  # ファイル名を左に表示

            if not convert_file(src, dst):
                fail_list.append(src)

            pbar.update(1)

    if fail_list:
        print(f"\n失敗: {len(fail_list)} 件")
        for f in fail_list:
            print(f"  {f}")


if __name__ == "__main__":
    batch_convert_with_progress(Path("./originals"), Path("./converted"))

tqdmset_description() でファイル名をバーの左に表示できる。長いファイル名は [:30] でカットしておくと表示が崩れない。

エラーハンドリングとリトライ

FFmpegのエラーハンドリングで厄介なのは、exit code 0でも出力ファイルが壊れている場合がある ことだ。例えば、入力ファイルの途中にデータ欠損があると、FFmpegは処理を完了して終了コード0を返すが、出力動画は再生できない。

リトライロジックと出力ファイルの簡易検証を組み合わせたスクリプトを示す。

python
import subprocess
import time
from pathlib import Path


MAX_RETRIES = 3
RETRY_DELAY = 2  # 秒


def verify_output(path: Path) -> bool:
    """ffprobeで出力ファイルを検証する。"""
    result = subprocess.run(
        ["ffprobe", "-v", "error", "-show_entries", "format=duration",
         "-of", "default=noprint_wrappers=1:nokey=1", str(path)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    if result.returncode != 0:
        return False
    # duration が取れれば最低限の整合性はある
    output = result.stdout.decode().strip()
    return bool(output) and float(output) > 0


def convert_with_retry(src: Path, dst: Path, retries: int = MAX_RETRIES) -> bool:
    """最大retries回リトライしながら変換する。"""
    for attempt in range(1, retries + 1):
        dst.parent.mkdir(parents=True, exist_ok=True)

        cmd = [
            "ffmpeg", "-i", str(src),
            "-c:v", "libx264", "-crf", "23", "-preset", "fast",
            "-c:a", "aac", "-b:a", "128k",
            "-y", str(dst),
        ]

        result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        if result.returncode != 0:
            print(f"  試行 {attempt}/{retries}: FFmpegエラー (exit {result.returncode})")
            if attempt < retries:
                time.sleep(RETRY_DELAY)
            continue

        if not verify_output(dst):
            print(f"  試行 {attempt}/{retries}: 出力ファイル検証失敗")
            dst.unlink(missing_ok=True)  # 壊れたファイルを削除
            if attempt < retries:
                time.sleep(RETRY_DELAY)
            continue

        return True

    return False


def batch_convert_robust(input_dir: Path, output_dir: Path) -> None:
    files = list(input_dir.glob("**/*.mov"))
    success, failed = 0, []

    for i, src in enumerate(files, start=1):
        relative = src.relative_to(input_dir)
        dst = output_dir / relative.with_suffix(".mp4")
        print(f"[{i}/{len(files)}] {src.name}")

        if convert_with_retry(src, dst):
            print(f"  完了: {dst}")
            success += 1
        else:
            print(f"  スキップ({MAX_RETRIES}回失敗): {src}")
            failed.append(src)

    print(f"\n結果: 成功 {success} / 失敗 {len(failed)} / 合計 {len(files)}")

    if failed:
        log_path = output_dir / "failed.txt"
        log_path.write_text("\n".join(str(f) for f in failed))
        print(f"失敗リスト: {log_path}")


if __name__ == "__main__":
    batch_convert_robust(Path("./originals"), Path("./converted"))

失敗したファイルは failed.txt として出力ディレクトリに保存する。後からまとめて確認・再実行できる。

FFmpegについての基本的な使い方は FFmpegの使い方チュートリアル もあわせて参照してほしい。

まとめ

この記事で解説した内容をまとめる。

  • subprocess.run() を使えば、PythonからFFmpegをシンプルに呼び出せる。コマンドラインと同じオプションがそのまま使える
  • ffmpeg-python はメソッドチェーンでパイプラインを構築できる。複雑なフィルターグラフを扱う場合に特に有効だ
  • バッチ変換pathlib.glob() でファイルを列挙して、フォルダ構造を保ちながら変換できる
  • tqdm で進捗バーを1行で追加できる。長時間のバッチ処理では必須だ
  • エラーハンドリング では exit code だけを信頼しない。ffprobe で出力ファイルを検証してリトライする設計が堅牢だ

スクリプトを育てていくなら、argparse でCLI引数を受け付けるようにしたり、logging モジュールでファイルにログを残したりするとさらに使いやすくなる。僕自身もこのパターンで何度もバッチ処理を組んでいる。