生成AIショート動画自動生成チャレンジ: 動画クリップと音楽を統合してffmpegで動画生成

AIテキスト

前回までに、ACE-Stepを用いた楽曲生成と、Stable Diffusion WebUI ForgeおよびWan2.2による動画クリップ生成の手順を確認しました。

今回は、これらの素材を統合し、一本の動画として仕上げる処理を試します。

ffmpegで動画クリップと楽曲を統合

動画と音楽の統合には ffmpeg を使用します。ffmpeg は CLI 形式で提供されているツールですが、基本的な使い方であれば導入や操作はそれほど難しくありません。具体的な導入手順については下記も参考にしてください。

fffmpeg は非常に多機能なツールで、CLI で提供されていることから、最初はやや敷居が高く感じられるかもしれません。

ただし現在では、ChatGPTに自然言語で要件を伝えるだけで、適切なオプションを含んだコマンドラインを生成できます。生成されたコマンドをそのままターミナルに貼り付けて実行できるため、実用上のハードルは大きく下がっています。

コード例は長いので文末に置きました。

動画統合時の注意点

今回今回は、一定の長さを持つ音楽ファイルと、フォルダに配置した数秒程度の動画クリップをそのまま連結しています。
必要に応じて、ffmpegの機能を使ってフェード処理などを加えることも可能です。

すべての処理を一つのPythonスクリプトにまとめることもできますが、実際の運用では生成物の選別や微調整が発生します。そのため、処理をいくつかのツールに分離しておいたほうが、全体としての使い勝手は高くなります。

本記事では、以下のような役割分担を前提に構成しています。

  • Stable Diffusionによる静止画生成
  • ComfyUI + Wan2.2による動画クリップ生成
  • ComfyUI + ACE-Stepによる音楽生成
  • ffmpegによる動画クリップと音楽ファイルの統合

また、実際の用途によっては、動画に音声や字幕を追加するなど、さらなる加工が必要になるでしょう。

公開する場合はライセンスや諸条件もチェック

これまで紹介してきたStability Matrix、ComfyUI、Stable Diffusion、WebUI Forgeといった実行環境は、Wan2.2やACE-Stepを含め、いずれもローカル環境で動作します。ツール自体は基本的に無料で、商用利用が可能なものが中心です。

一方で、Stable Diffusionのチェックポイント(モデル)の選定やバージョンによっては、利用条件が異なる場合があります。商用利用が許可されたモデルを適切に選べば、生成物をYouTubeなどで公開することも可能ですが、公式の説明だけでは権利関係を十分に判断できないケースもあるため、配布元のライセンス表記や注意事項を個別に確認することが重要です。

項目補足・注意点
Stability Matrix環境管理ツール自体は無料。ライセンスは同梱ソフトに依存
ComfyUIオープンソース。商用利用可能
Stable Diffusionコアモデルはオープンライセンスだが、派生モデルは個別確認が必要
WebUI ForgeAUTOMATIC1111派生。商用利用可
Wan2.2ローカル実行可能。配布元ライセンスを要確認
ACE-Stepローカル実行可能。用途制限がないかモデルごとに確認が必要
商用利用チェックポイント(モデル)ごとに可否が異なる。特にStable Diffusion系は注意が必要。
YouTube公開商用利用可モデル+既存IPに類似しない生成物であれば可能

まとめ

本記事では、生成した動画クリップと楽曲を ffmpeg で統合し、一本のショート動画として仕上げる流れを確認しました。

素材生成から動画完成までを分離して扱うことで、調整や再生成を含めた運用がしやすくなります。

この構成をベースに、自動化や追加演出へと発展させていくことも可能です。

参考)コード例

ChatGPTに対し、フォルダに配置した動画ファイルと音楽ファイルを ffmpeg で統合するコードの生成を依頼しました。構成を単純にするため、動画の長さは30秒固定としています。

既存の処理との親和性を考慮してPythonを採用していますが、ffmpeg自体はCLIツールのため、シェルスクリプトなど別の形式で実装することも可能です。

"""Merge ./audios and ./movies into a single 30s MP4 using ffmpeg.

- Pick files sorted by filename.
- Combine multiple files until total >= 30s; if still short, pad (video last-frame hold, audio silence) to exactly 30s.
- If longer, cut to 30s.

Requirements:
- ffmpeg + ffprobe available in PATH.

Usage:
  python3 test.py
  python3 test.py --audios ./audios --movies ./movies --out output_30s.mp4 --seconds 30

Notes:
- Folder name is "movies" as requested.
"""

from __future__ import annotations

import argparse
import os
import shlex
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import List, Optional


TARGET_SECONDS_DEFAULT = 30.0


AUDIO_EXTS = {".mp3", ".wav", ".m4a", ".aac", ".flac", ".ogg", ".opus"}
VIDEO_EXTS = {".mp4", ".mov", ".mkv", ".webm", ".m4v", ".avi"}


@dataclass
class Paths:
    root: Path
    audios_dir: Path
    movies_dir: Path
    work_dir: Path


def run(cmd: List[str], *, cwd: Optional[Path] = None) -> None:
    printable = " ".join(shlex.quote(c) for c in cmd)
    print(f"\n$ {printable}")
    subprocess.run(cmd, cwd=str(cwd) if cwd else None, check=True)


def ffprobe_duration(path: Path) -> float:
    # Returns duration in seconds (float). If unknown, raises.
    cmd = [
        "ffprobe",
        "-v",
        "error",
        "-show_entries",
        "format=duration",
        "-of",
        "default=noprint_wrappers=1:nokey=1",
        str(path),
    ]
    out = subprocess.check_output(cmd, text=True).strip()
    if not out:
        raise RuntimeError(f"ffprobe returned empty duration for: {path}")
    try:
        return float(out)
    except ValueError as e:
        raise RuntimeError(f"failed to parse duration '{out}' for: {path}") from e


def list_media_files(folder: Path, exts: set[str]) -> List[Path]:
    if not folder.exists():
        raise FileNotFoundError(f"Folder not found: {folder}")
    files = [p for p in folder.iterdir() if p.is_file() and p.suffix.lower() in exts]
    return sorted(files, key=lambda p: p.name)


def pick_until_seconds(files: List[Path], target: float) -> List[Path]:
    picked: List[Path] = []
    total = 0.0
    for f in files:
        dur = ffprobe_duration(f)
        picked.append(f)
        total += dur
        if total >= target:
            break
    return picked


def write_concat_list(list_path: Path, files: List[Path]) -> None:
    # concat demuxer list file
    lines = []
    for f in files:
        # escape single quotes
        p = str(f.resolve()).replace("'", "\\'")
        lines.append(f"file '{p}'")
    list_path.write_text("\n".join(lines) + "\n", encoding="utf-8")


def ensure_dir(p: Path) -> None:
    p.mkdir(parents=True, exist_ok=True)


def build_concat_video(pths: Paths, video_files: List[Path]) -> Path:
    """Re-encode each clip to uniform format, then concat into a single video-only mp4."""
    if not video_files:
        raise RuntimeError("No video files found to process.")

    v_norm_dir = pths.work_dir / "v_norm"
    ensure_dir(v_norm_dir)

    norm_files: List[Path] = []
    for i, src in enumerate(video_files, start=1):
        dst = v_norm_dir / f"v_{i:03d}.mp4"
        # Normalize: 30fps, yuv420p, even dimensions. Remove audio.
        run(
            [
                "ffmpeg",
                "-y",
                "-i",
                str(src),
                "-an",
                "-vf",
                "fps=30,scale=trunc(iw/2)*2:trunc(ih/2)*2,format=yuv420p",
                "-c:v",
                "libx264",
                "-preset",
                "veryfast",
                "-crf",
                "18",
                "-movflags",
                "+faststart",
                str(dst),
            ]
        )
        norm_files.append(dst)

    concat_list = pths.work_dir / "video_concat.txt"
    write_concat_list(concat_list, norm_files)

    out = pths.work_dir / "video_concat.mp4"
    # Concat normalized files (re-encode once more for safety)
    run(
        [
            "ffmpeg",
            "-y",
            "-f",
            "concat",
            "-safe",
            "0",
            "-i",
            str(concat_list),
            "-c:v",
            "libx264",
            "-preset",
            "veryfast",
            "-crf",
            "18",
            "-pix_fmt",
            "yuv420p",
            "-r",
            "30",
            "-movflags",
            "+faststart",
            str(out),
        ]
    )
    return out


def build_concat_audio(pths: Paths, audio_files: List[Path]) -> Path:
    """Convert each audio to uniform WAV, concat, then encode to AAC."""
    if not audio_files:
        raise RuntimeError("No audio files found to process.")

    a_norm_dir = pths.work_dir / "a_norm"
    ensure_dir(a_norm_dir)

    norm_files: List[Path] = []
    for i, src in enumerate(audio_files, start=1):
        dst = a_norm_dir / f"a_{i:03d}.wav"
        # Normalize: 48kHz stereo PCM
        run(
            [
                "ffmpeg",
                "-y",
                "-i",
                str(src),
                "-vn",
                "-ac",
                "2",
                "-ar",
                "48000",
                "-c:a",
                "pcm_s16le",
                str(dst),
            ]
        )
        norm_files.append(dst)

    concat_list = pths.work_dir / "audio_concat.txt"
    write_concat_list(concat_list, norm_files)

    out = pths.work_dir / "audio_concat.m4a"
    run(
        [
            "ffmpeg",
            "-y",
            "-f",
            "concat",
            "-safe",
            "0",
            "-i",
            str(concat_list),
            "-c:a",
            "aac",
            "-b:a",
            "192k",
            "-fflags",
            "+genpts",
            "-reset_timestamps",
            "1",
            str(out),
        ]
    )
    return out


def mux_to_30s(video_path: Path, audio_path: Path, out_path: Path, seconds: float) -> None:
    """Mux video+audio and force exact duration.

    Fixes:
    - Audio stream exists but is silent due to non-zero start timestamps / PTS.
    - Output container reports longer duration when padding filters extend past the target.

    Strategy:
    - Reset PTS to start at 0.
    - Pad first, then TRIM LAST to exactly `seconds`.
    - Pass `-t seconds` as a final safety net.
    """

    # NOTE: trim/atrim must be last (after pad/apad), otherwise the container may keep the padded duration.
    fc = (
        f"[0:v]setpts=PTS-STARTPTS,"
        f"tpad=stop_mode=clone:stop_duration={seconds},"
        f"trim=0:{seconds},setpts=PTS-STARTPTS[v];"
        f"[1:a]asetpts=PTS-STARTPTS,"
        f"aresample=async=1:first_pts=0,"
        f"apad,atrim=0:{seconds},asetpts=PTS-STARTPTS[a]"
    )

    run(
        [
            "ffmpeg",
            "-y",
            "-i",
            str(video_path),
            "-i",
            str(audio_path),
            "-filter_complex",
            fc,
            "-map",
            "[v]",
            "-map",
            "[a]",
            "-t",
            f"{seconds}",
            "-c:v",
            "libx264",
            "-preset",
            "veryfast",
            "-crf",
            "18",
            "-pix_fmt",
            "yuv420p",
            "-c:a",
            "aac",
            "-b:a",
            "192k",
            "-movflags",
            "+faststart",
            str(out_path),
        ]
    )


def remux_for_vscode_compat(src_mp4: Path, dst_mp4: Path) -> None:
    """Re-mux to improve compatibility with VS Code's built-in player (Chromium).

    Some Chromium builds are picky about AAC details / sample rate / channel layout.
    Re-encode audio to AAC-LC, 44.1kHz, stereo, and copy video.
    """
    run(
        [
            "ffmpeg",
            "-y",
            "-i",
            str(src_mp4),
            "-map",
            "0:v:0",
            "-map",
            "0:a:0",
            "-c:v",
            "copy",
            "-c:a",
            "aac",
            "-profile:a",
            "aac_low",
            "-ar",
            "44100",
            "-ac",
            "2",
            "-b:a",
            "192k",
            "-movflags",
            "+faststart",
            str(dst_mp4),
        ]
    )


def parse_args() -> argparse.Namespace:
    ap = argparse.ArgumentParser()
    ap.add_argument("--audios", default="./audios", help="Audio folder")
    ap.add_argument("--movies", default="./movies", help="Video clips folder (movies)")
    ap.add_argument("--out", default="output_30s.mp4", help="Output mp4")
    ap.add_argument("--seconds", type=float, default=TARGET_SECONDS_DEFAULT, help="Target duration in seconds")
    ap.add_argument("--work", default="./.tmp_merge", help="Work directory")
    ap.add_argument(
        "--vscode-compat",
        action="store_true",
        help="Also write a VS Code/Chromium friendly MP4 (audio AAC-LC 44.1kHz).",
    )
    return ap.parse_args()


def main() -> int:
    args = parse_args()

    root = Path.cwd()
    pths = Paths(
        root=root,
        audios_dir=(root / args.audios).resolve(),
        movies_dir=(root / args.movies).resolve(),
        work_dir=(root / args.work).resolve(),
    )

    ensure_dir(pths.work_dir)

    audio_all = list_media_files(pths.audios_dir, AUDIO_EXTS)
    video_all = list_media_files(pths.movies_dir, VIDEO_EXTS)

    if not audio_all:
        raise RuntimeError(f"No audio files in: {pths.audios_dir}")
    if not video_all:
        raise RuntimeError(f"No video files in: {pths.movies_dir}")

    target = float(args.seconds)

    audio_pick = pick_until_seconds(audio_all, target)
    video_pick = pick_until_seconds(video_all, target)

    print("\n[Pick] audio:")
    for p in audio_pick:
        print(f"  - {p.name} ({ffprobe_duration(p):.2f}s)")
    print("[Pick] video:")
    for p in video_pick:
        print(f"  - {p.name} ({ffprobe_duration(p):.2f}s)")

    v_concat = build_concat_video(pths, video_pick)
    a_concat = build_concat_audio(pths, audio_pick)

    out_path = (root / args.out).resolve()
    mux_to_30s(v_concat, a_concat, out_path, target)

    if args.vscode_compat:
        compat_out = out_path.with_name(out_path.stem + "_vscode" + out_path.suffix)
        remux_for_vscode_compat(out_path, compat_out)
        print(f"[OK] wrote (vscode): {compat_out}")

    print(f"\n[OK] wrote: {out_path}")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except subprocess.CalledProcessError as e:
        print(f"\n[ERROR] command failed: {e}", file=sys.stderr)
        raise
タイトルとURLをコピーしました