ab-av1を使用して、cwd内の動画ファイルをav1にエンコードするpythonスクリプト（9割9分AI製）

auto-ab-av1.py

import subprocess
import json
import re
import sys
from pathlib import Path
from typing import Optional, List

# 定数
AB_AV1_EXECUTABLE = "ab-av1.exe"
FFPROBE_EXECUTABLE = "ffprobe"
OUTPUT_DIR_NAME = "encoded"
VIDEO_STREAM_INDEX = "v:0"
PROCESS_TIMEOUT_SECONDS = 5

# サポートする動画拡張子
VIDEO_EXTENSIONS = [
    ".mp4",
    ".mkv",
    ".avi",
    ".mov",
    ".wmv",
    ".flv",
    ".webm",
    ".m4v",
    ".ts",
    ".m2ts",
]

# インターレース判定用の進行モード
PROGRESSIVE_FIELD_ORDERS = ["progressive", "unknown", ""]


def run_ffprobe_command(video_file: str, show_entries: str) -> dict:
    """ffprobeコマンドを実行してJSON結果を返す
    
    Args:
        video_file: 動画ファイルパス
        show_entries: ffprobeの--show-entriesパラメータ
        
    Returns:
        ffprobeの実行結果をパースしたdict
        
    Raises:
        subprocess.CalledProcessError: ffprobe実行に失敗した場合
        json.JSONDecodeError: JSON解析に失敗した場合
    """
    cmd = [
        FFPROBE_EXECUTABLE,
        "-v",
        "error",
        "-select_streams",
        VIDEO_STREAM_INDEX,
        "-show_entries",
        show_entries,
        "-of",
        "json",
        video_file,
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
    return json.loads(result.stdout)


def get_video_codec(video_file: str) -> str:
    """動画のコーデックを取得
    
    Args:
        video_file: 動画ファイルパス
        
    Returns:
        コーデック名（取得できない場合は空文字列）
    """
    try:
        data = run_ffprobe_command(video_file, "stream=codec_name")
        if "streams" in data and len(data["streams"]) > 0:
            return data["streams"][0].get("codec_name", "")
    except (subprocess.CalledProcessError, json.JSONDecodeError) as e:
        print(f"警告: コーデック情報の取得に失敗しました: {e}")
    return ""


def is_interlaced(video_file: str) -> bool:
    """動画がインターレースかどうかを判定
    
    Args:
        video_file: 動画ファイルパス
        
    Returns:
        インターレースの場合True、プログレッシブの場合False
    """
    try:
        data = run_ffprobe_command(video_file, "stream=field_order")
        if "streams" in data and len(data["streams"]) > 0:
            field_order = data["streams"][0].get("field_order", "progressive")
            # field_orderがtt、bb、tb、btのいずれかであればインターレース
            return field_order not in PROGRESSIVE_FIELD_ORDERS
    except (subprocess.CalledProcessError, json.JSONDecodeError) as e:
        print(f"警告: フィールドオーダー情報の取得に失敗しました: {e}")
    return False


def build_ab_av1_command(
    action: str, video_file: str, interlaced: bool = False, **kwargs
) -> List[str]:
    """ab-av1コマンドを構築
    
    Args:
        action: ab-av1のアクション（"crf-search" または "encode"）
        video_file: 動画ファイルパス
        interlaced: インターレース動画の場合True
        **kwargs: 追加のオプション（crf, outputなど）
        
    Returns:
        コマンドライン引数のリスト
    """
    cmd = [AB_AV1_EXECUTABLE, action, "--input", video_file]
    
    # CRF値の指定（encodeアクション用）
    if "crf" in kwargs:
        cmd.extend(["--crf", str(kwargs["crf"])])
    
    # 出力ファイルの指定（encodeアクション用）
    if "output" in kwargs:
        cmd.extend(["--output", str(kwargs["output"])])
    
    # インターレース対応フィルター
    if interlaced:
        # vmafが適切に比較できないため、crf-searchのみ、モード0（フレームベース）を使用
        vfilter = "bwdif=mode=0:parity=auto:deint=all" if action == "crf-search" else "bwdif=mode=1:parity=auto:deint=all"
        cmd.extend(["--vfilter", vfilter])
    
    return cmd


def extract_crf_from_output(output: str) -> Optional[float]:
    """ab-av1の出力からCRF値を抽出
    
    Args:
        output: ab-av1コマンドの標準出力
        
    Returns:
        抽出されたCRF値、見つからない場合はNone
    """
    for line in output.split("\n"):
        match = re.search(r"(?:crf\s+)?(\d+(?:\.\d+)?)", line)
        if match:
            return float(match.group(1))
    return None


def run_crf_search(video_file: str, interlaced: bool = False) -> Optional[float]:
    """ab-av1でCRF値を検索
    
    Args:
        video_file: 動画ファイルパス
        interlaced: インターレース動画の場合True
        
    Returns:
        検出されたCRF値、失敗した場合はNone
    """
    cmd = build_ab_av1_command("crf-search", video_file, interlaced)
    
    try:
        process = subprocess.Popen(cmd, stdout=subprocess.PIPE, text=True)
        stdout, _ = process.communicate()
        return extract_crf_from_output(stdout)
    except (subprocess.SubprocessError, OSError) as e:
        print(f"エラー: CRF検索の実行に失敗しました: {e}")
        return None


def get_output_path(video_file: str) -> Path:
    """出力ファイルパスを取得
    
    Args:
        video_file: 入力動画ファイルパス
        
    Returns:
        出力ファイルの完全パス
    """
    video_path = Path(video_file)
    output_dir = Path.cwd() / OUTPUT_DIR_NAME
    output_dir.mkdir(exist_ok=True)
    return output_dir / video_path.name


def encode_video(video_file: str, crf_value: float, interlaced: bool = False) -> bool:
    """ab-av1でエンコード
    
    Args:
        video_file: 動画ファイルパス
        crf_value: CRF値
        interlaced: インターレース動画の場合True
        
    Returns:
        エンコード成功時True、失敗時False
    """
    output_file = get_output_path(video_file)
    cmd = build_ab_av1_command(
        "encode", video_file, interlaced, crf=crf_value, output=output_file
    )

    print(f"実行中: {' '.join(cmd)}\n")
    print(f"出力先: {output_file}\n")
    print(
        "エンコード処理を開始します（進捗情報が表示されるまで数秒かかる場合があります）...\n"
    )

    # プロセスを起動（進捗が表示されるまでバックグラウンドで実行）
    process = None
    try:
        process = subprocess.Popen(cmd)
        # プロセスの完了を待つ
        process.wait()

        # 完了メッセージを表示
        if process.returncode == 0:
            print("\n✓ エンコード完了")
            return True
        else:
            print(f"\n✗ エンコードエラー (exit code: {process.returncode})")
            return False

    except KeyboardInterrupt:
        print("\n\n中断されました。プロセスを終了しています...")
        if process is not None:
            try:
                process.terminate()
                process.wait(timeout=PROCESS_TIMEOUT_SECONDS)
            except subprocess.TimeoutExpired:
                process.kill()
        raise
    except (subprocess.SubprocessError, OSError) as e:
        print(f"\n✗ エンコード実行エラー: {e}")
        return False


def process_video(video_file: str) -> bool:
    """動画ファイルを処理
    
    Args:
        video_file: 動画ファイルパス
        
    Returns:
        処理成功時True、失敗時またはスキップ時False
    """
    print(f"\n{'='*60}")
    print(f"処理中: {video_file}")
    print(f"{'='*60}\n")

    # 1. コーデックがAV1であれば終了
    codec = get_video_codec(video_file)
    print(f"コーデック: {codec}")
    if codec == "av1":
        print("既にAV1でエンコードされています。スキップします。")
        return False

    # 2. インターレースかどうかを確認
    interlaced = is_interlaced(video_file)
    print(f"インターレース: {'はい' if interlaced else 'いいえ'}")

    # 3. CRF値を検索
    print("\nCRF値を検索中...")
    crf_value = run_crf_search(video_file, interlaced)

    if crf_value is None:
        print("CRF値の取得に失敗しました。")
        return False

    print(f"\n推奨CRF値: {crf_value}")

    # 4. エンコード
    print("\nエンコード中...")
    success = encode_video(video_file, crf_value, interlaced)

    if success:
        print(f"\n{video_file} の処理が完了しました。")
    
    return success


def find_video_files(directory: Path) -> List[Path]:
    """指定ディレクトリ内の動画ファイルを検索
    
    Args:
        directory: 検索対象ディレクトリ
        
    Returns:
        見つかった動画ファイルのリスト
    """
    video_files = []
    for ext in VIDEO_EXTENSIONS:
        video_files.extend(directory.glob(f"*{ext}"))
    return video_files


def main() -> None:
    """メイン処理"""
    current_dir = Path.cwd()
    video_files = find_video_files(current_dir)

    if not video_files:
        print("動画ファイルが見つかりませんでした。")
        return

    print(f"{len(video_files)}個の動画ファイルが見つかりました。\n")

    success_count = 0
    skip_count = 0
    error_count = 0

    for idx, video_file in enumerate(video_files, 1):
        print(f"\n[{idx}/{len(video_files)}] ", end="")
        try:
            success = process_video(str(video_file))
            if success:
                success_count += 1
            else:
                skip_count += 1
        except KeyboardInterrupt:
            print("\n\n処理が中断されました。")
            sys.exit(1)
        except Exception as e:
            print(f"\n✗ エラーが発生しました: {e}")
            error_count += 1
            continue

    # 処理結果のサマリーを表示
    print(f"\n{'='*60}")
    print("すべての処理が完了しました！")
    print(f"成功: {success_count}件、スキップ: {skip_count}件、エラー: {error_count}件")
    print(f"{'='*60}")


if __name__ == "__main__":
    main()

仕様

• cwd内の動画ファイルを走査し、av1でエンコードされていない動画をエンコードする。
• エンコードにはab-av1を使用し、vmaf 95になるようなcrfでエンコードする。
• 入力動画がインターレース方式であれば、bwdifを用いてデインタレースしてエンコードする（mode=1）
• 出力先は、(cwd)/encoded/(input file name)

プロンプト

cwd内の動画ファイルのそれぞれに対して以下を実行して。ab-av1、ffmpeg、ffprobeが使えます。
動画のコーデックがav1であれば終了。
動画がインターレースかどうか取得。
"ab-av1.exe crf-search --input (inputfile)"を実行。動画が、インターレースであれば、--vfilter "bwdif=mode=0:parity=auto:deint=all" オプションを付加。
ab-av1の実行結果から適切なcrf値を取得して以下のコマンドを叩いて。"ab-av1.exe encode --crf (crf) --input (inputfile)" なお、動画がインターレースであれば --vfilter bwdif を付加。