音声ファイルを無音区間で自動分割するPythonスクリプトをClaude Codeと作った話

#!/usr/bin/env python3
"""音声ファイルを無音区間でセグメントに分割する CLI ツール。

依存: ffmpeg / ffprobe (PATH に通っていること)
"""

import argparse
import csv
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path

SILENCE_RE = re.compile(r"silence_end:\s*([\d.]+)\s*\|\s*silence_duration:\s*([\d.]+)")

@dataclass(frozen=True)
class Silence:
    """検出された無音区間。"""
    start: float
    end: float
    duration: float

@dataclass(frozen=True)
class Content:
    """Silence と Silence の間のコンテンツ区間。"""
    start: float
    end: float

@dataclass(frozen=True)
class Segment:
    """切り出し対象のセグメント（padding 適用済み）。"""
    index: int
    start: float
    end: float

def probe_duration(path: Path) -> float:
    """ffprobe で音声ファイルの長さ（秒）を返す。"""
    result = subprocess.run(
        [
            "ffprobe",
            "-v", "error",
            "-show_entries", "format=duration",
            "-of", "default=noprint_wrappers=1:nokey=1",
            str(path),
        ],
        text=True,
        capture_output=True,
        check=True,
    )
    return float(result.stdout.strip())

def detect_silences(path: Path, noise_db: float, min_silence: float) -> list[Silence]:
    """ffmpeg silencedetect フィルタで無音区間を検出して返す。"""
    result = subprocess.run(
        [
            "ffmpeg",
            "-hide_banner",
            "-nostats",
            "-i", str(path),
            "-af", f"silencedetect=noise={noise_db}dB:d={min_silence}",
            "-f", "null", "-",
        ],
        text=True,
        capture_output=True,
        check=True,
    )
    # ffmpeg はフィルタのログを stdout ではなく stderr に出力する
    silences: list[Silence] = []
    for line in result.stderr.splitlines():
        m = SILENCE_RE.search(line)
        if m:
            end = float(m.group(1))
            duration = float(m.group(2))
            # silence_start 行は使わず、end - duration で start を算出する
            silences.append(Silence(start=end - duration, end=end, duration=duration))
    return silences

def merge_nearby(silences: list[Silence], merge_gap: float) -> list[Silence]:
    """merge_gap 秒以内に連続する無音区間を1つに統合する。"""
    merged: list[Silence] = []
    for si in silences:
        if merged and si.start - merged[-1].end <= merge_gap:
            prev = merged[-1]
            new_end = max(prev.end, si.end)
            merged[-1] = Silence(start=prev.start, end=new_end, duration=new_end - prev.start)
        else:
            merged.append(si)
    return merged

def build_timeline(silences: list[Silence], total: float) -> list[Silence | Content]:
    """Silence リストから Silence と Content を交互に並べたタイムラインを構築する。"""
    timeline: list[Silence | Content] = []
    pos = 0.0
    for si in silences:
        if si.start > pos:
            timeline.append(Content(start=pos, end=si.start))
        timeline.append(si)
        pos = si.end
    # 末尾の無音の後の微小区間（ffmpeg の silence_end が total より手前になる検出誤差）は無視する
    if total - pos > 0.1:
        timeline.append(Content(start=pos, end=total))
    return timeline

def select_boundaries(timeline: list[Silence | Content], target_count: int | None, total: float) -> list[Silence]:
    """タイムラインから境界 Silence を選ぶ。先頭・末尾の Silence は除外する。"""
    # build_timeline が末尾の微小 Content を 0.1s 閾値で省略するため同じ閾値で判定する
    eps = 0.1
    inner = [b for b in timeline if isinstance(b, Silence) and b.start > eps and total - b.end > eps]
    if target_count is None:
        # target_count 未指定のときはすべての内部 Silence を境界にする
        return inner
    needed = target_count - 1
    if needed < 0:
        raise ValueError(f"target-count must be >= 1, got {target_count}")
    if len(inner) < needed:
        raise ValueError(
            f"need {needed} boundaries for {target_count} segments, "
            f"but only {len(inner)} silences detected"
        )
    # 無音が長いほど明確な区切り目である可能性が高いため、長さ降順で上位を選ぶ
    selected = sorted(inner, key=lambda si: si.duration, reverse=True)[:needed]
    return sorted(selected, key=lambda si: si.start)

def build_segments(timeline: list[Silence | Content], boundaries: list[Silence], total: float, padding: float) -> list[Segment]:
    """タイムラインと境界 Silence から Segment を生成し、前後に padding 秒を付与する。"""
    first, last = timeline[0], timeline[-1]
    content_start = first.end if isinstance(first, Silence) else first.start
    content_end = last.start if isinstance(last, Silence) else last.end
    pairs = list(zip(
        [content_start] + [b.end for b in boundaries],
        [b.start for b in boundaries] + [content_end],
        strict=True,
    ))
    segments: list[Segment] = []
    for i, (start, end) in enumerate(pairs, 1):
        padded_start = max(0.0, start - padding)
        padded_end = min(total, end + padding)
        if padded_end <= padded_start:
            raise ValueError(f"invalid segment {i}: {start:.3f}..{end:.3f}")
        segments.append(Segment(index=i, start=padded_start, end=padded_end))
    return segments

def write_segment(
    segment: Segment,
    source: Path,
    output_dir: Path,
    prefix: str,
    extension: str,
) -> None:
    """ffmpeg でセグメントを切り出してファイルに保存する。"""
    output = output_dir / f"{prefix}{segment.index:03d}.{extension}"
    command = [
        "ffmpeg",
        "-hide_banner",
        "-loglevel", "error",
        "-y",
        # -ss を -i より前に置くと高速な keyframe seek になる
        # 後ろに置くと精度が上がる代わりに先頭から全フレームをデコードするため遅い
        "-ss", f"{segment.start:.6f}",
        "-i", str(source),
        "-t", f"{segment.end - segment.start:.6f}",
        "-vn",  # 映像ストリームを除外
    ]
    if extension == "m4a":
        command += ["-c:a", "aac", "-b:a", "128k"]
    elif extension == "mp3":
        command += ["-af", "aresample=44100", "-c:a", "libmp3lame", "-b:a", "96k"]
    else:
        raise ValueError(f"unsupported extension: {extension}")
    command.append(str(output))
    subprocess.run(command, check=True)

def write_manifest(segments: list[Segment], output_dir: Path, prefix: str) -> None:
    """セグメントの境界情報を TSV ファイルに出力する。"""
    tsv_path = output_dir / f"{prefix.rstrip('-_')}-manifest.tsv"
    with tsv_path.open("w", encoding="utf-8", newline="") as f:
        writer = csv.writer(f, delimiter="\t")
        writer.writerow(["file", "start", "end", "duration"])
        for s in segments:
            writer.writerow(
                [
                    f"{prefix}{s.index:03d}",
                    f"{s.start:.3f}",
                    f"{s.end:.3f}",
                    f"{s.end - s.start:.3f}",
                ]
            )

def main() -> None:
    parser = argparse.ArgumentParser(description="音声ファイルを無音区間でセグメントに分割する")
    parser.add_argument("input", help="入力音声ファイル")
    parser.add_argument("--output-dir", default="out")
    parser.add_argument("--target-count", type=int, default=None)
    parser.add_argument("--prefix", default=None, help="出力ファイルのプレフィックス（省略時は元ファイル名）")
    parser.add_argument("--extension", choices=["mp3", "m4a"], default="mp3")
    parser.add_argument("--noise-db", type=float, default=-30.0)
    parser.add_argument("--min-silence", type=float, default=1.0)
    parser.add_argument("--merge-gap", type=float, default=0.1)
    parser.add_argument("--padding", type=float, default=0.04)
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    source = Path(args.input)
    output_dir = Path(args.output_dir)
    prefix = args.prefix if args.prefix is not None else f"{source.stem}-"

    try:
        total = probe_duration(source)
        raw_silences = detect_silences(source, args.noise_db, args.min_silence)
        silences = merge_nearby(raw_silences, args.merge_gap)
        timeline = build_timeline(silences, total)
        boundaries = select_boundaries(timeline, args.target_count, total)
        segments = build_segments(timeline, boundaries, total, args.padding)
    except (ValueError, subprocess.CalledProcessError) as e:
        parser.exit(1, f"error: {e}\n")

    print(f"detected {len(raw_silences)} silences → merged to {len(silences)} → selected {len(boundaries)} boundaries → {len(segments)} segments")

    if args.dry_run:
        if isinstance(timeline[0], Silence):
            print(f"  silence: 0.000s .. {segments[0].start:.3f}s  ({segments[0].start:.3f}s)")
        for i, s in enumerate(segments):
            print(f"  {s.index:03d}: {s.start:.3f}s .. {s.end:.3f}s  ({s.end - s.start:.3f}s)")
            if i < len(segments) - 1:
                silence_start = s.end
                silence_end = segments[i + 1].start
                print(f"  silence: {silence_start:.3f}s .. {silence_end:.3f}s  ({silence_end - silence_start:.3f}s)")
        if isinstance(timeline[-1], Silence):
            print(f"  silence: {segments[-1].end:.3f}s .. {total:.3f}s  ({total - segments[-1].end:.3f}s)")
        return

    output_dir.mkdir(parents=True, exist_ok=True)
    write_manifest(segments, output_dir, prefix)
    for s in segments:
        write_segment(s, source, output_dir, prefix, args.extension)
    print(f"wrote {len(segments)} files to {output_dir}")

if __name__ == "__main__":
    main()