ffmpeg_utils.py

import asyncio
import math
import os
import subprocess
from pathlib import Path


class FfmpegError(RuntimeError):
    pass


async def run_ffmpeg(args: list[str]) -> None:
    process = await asyncio.create_subprocess_exec(
        *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await process.communicate()
    if process.returncode != 0:
        detail = stderr.decode("utf-8", errors="replace")[-4000:] or stdout.decode("utf-8", errors="replace")[-4000:]
        raise FfmpegError(detail)


def probe_duration_seconds(path: Path) -> float | None:
    result = subprocess.run(
        [
            "ffprobe",
            "-v",
            "error",
            "-show_entries",
            "format=duration",
            "-of",
            "default=noprint_wrappers=1:nokey=1",
            str(path),
        ],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode != 0:
        return None
    try:
        return float(result.stdout.strip())
    except ValueError:
        return None


async def compress_audio(input_path: Path, output_path: Path) -> None:
    await run_ffmpeg(
        [
            "ffmpeg",
            "-y",
            "-i",
            str(input_path),
            "-vn",
            "-ar",
            "16000",
            "-ac",
            "1",
            "-c:a",
            "libmp3lame",
            "-b:a",
            "32k",
            str(output_path),
        ]
    )


async def chunk_audio(input_path: Path, output_dir: Path, chunk_duration_seconds: int) -> list[Path]:
    output_dir.mkdir(parents=True, exist_ok=True)
    output_pattern = output_dir / "chunk_%03d.mp3"
    await run_ffmpeg(
        [
            "ffmpeg",
            "-y",
            "-i",
            str(input_path),
            "-f",
            "segment",
            "-segment_time",
            str(chunk_duration_seconds),
            "-c",
            "copy",
            str(output_pattern),
        ]
    )
    return sorted(output_dir.glob("chunk_*.mp3"))


def size_mb(path: Path) -> float:
    return os.path.getsize(path) / (1024 * 1024)


def needs_chunking(path: Path, target_size_mb: int) -> bool:
    return size_mb(path) > target_size_mb


def estimated_chunk_duration(duration_seconds: float | None, final_size_mb: float, target_size_mb: int, fallback_seconds: int) -> int:
    if not duration_seconds or final_size_mb <= 0 or final_size_mb <= target_size_mb:
        return fallback_seconds
    chunks = max(2, math.ceil(final_size_mb / target_size_mb))
    return max(60, math.floor(duration_seconds / chunks))