aio-ffmpeg

Строго типизированная асинхронная обёртка над ffmpeg и ffprobe для Python 3.11+.

Процессы запускаются через asyncio.create_subprocess_exec(). Нет зависимостей времени выполнения — только стандартная библиотека Python.

Сравнение с аналогами

Возможности

• Асинхронный запуск процессов: asyncio.create_subprocess_exec(), контроль конкурентности через asyncio.Semaphore.
• Типизация: frozen dataclass-модели со slots=True, PEP 561 py.typed, mypy --strict.
• Прогресс: потоковый разбор протокола -progress pipe:1 (кадры, время, битрейт, скорость, процент).
• Graceful shutdown: отправка q в stdin перед системными сигналами завершения. Предотвращает повреждение заголовков MP4.
• Три уровня API:
  • FFmpegClient — готовые методы для типовых задач.
  • MediaPipeline — декларативный конвейер цепочек обработки.
  • FFmpegCommand — построитель аргументов CLI с валидацией.
• FFprobe: типизированный разбор контейнеров, видео/аудио/субтитр-потоков, глав и метаданных.
• Аппаратное ускорение: автоопределение NVENC, AMF, QSV, D3D11VA, VideoToolbox.
• Интеграция с async-yt-dlp: конвейер загрузки и постобработки медиафайлов.

Методы FFmpegClient

Установка

Требуется Python 3.11+, установленные в системе ffmpeg и ffprobe (в PATH или через аргументы клиента / переменную FFMPEG_PATH).

# Базовая установка (без runtime-зависимостей):
pip install aio-ffmpeg

# С интеграцией с async-yt-dlp:
pip install "aio-ffmpeg[ytdlp]"

Или через uv:

uv add aio-ffmpeg

Быстрый старт

1. Анализ медиафайла

import asyncio
from aio_ffmpeg import FFmpegClient


async def main() -> None:
    client = FFmpegClient()
    info = await client.probe("video.mp4")

    print(f"Формат: {info.format.format_long_name}")
    print(f"Длительность: {info.duration} сек")

    if info.primary_video:
        v = info.primary_video
        print(f"Видео: {v.codec_name}, {v.width}x{v.height} @ {v.frame_rate:.2f} fps")

    if info.primary_audio:
        a = info.primary_audio
        print(f"Аудио: {a.codec_name}, {a.sample_rate} Hz, каналов: {a.channels}")


asyncio.run(main())

2. Транскодирование с прогрессом

import asyncio
from aio_ffmpeg import FFmpegClient, ProgressInfo


async def main() -> None:
    client = FFmpegClient()

    async def on_progress(p: ProgressInfo) -> None:
        print(f"Прогресс: {p.percent:.1f}% | Скорость: {p.speed} | FPS: {p.fps:.1f}")

    result = await client.transcode(
        input="input.mp4",
        output="output_720p.mp4",
        video_codec="libx264",
        crf=23,
        resolution=(1280, 720),
        audio_codec="aac",
        audio_bitrate="128k",
        on_progress=on_progress,
    )
    print(f"Готово за {result.duration_seconds:.2f} сек")


asyncio.run(main())

3. Декларативный конвейер (MediaPipeline)

import asyncio
from aio_ffmpeg import FFmpegClient


async def main() -> None:
    client = FFmpegClient()

    # обрезка → масштабирование → нормализация звука → вывод
    await (
        client.pipeline("input.mp4")
        .trim(start=10, duration=60)
        .scale(1280, 720)
        .normalize_audio(target_lufs=-16.0)
        .output("highlight_720p.mp4")
        .run()
    )


asyncio.run(main())

4. Интеграция с async-yt-dlp

import asyncio
from aio_ffmpeg import FFmpegClient
from aio_ffmpeg.integration import process_download_result


async def main() -> None:
    client = FFmpegClient()

    # download_result — объект с протоколом DownloadResultProtocol
    post_result = await process_download_result(
        download_result,
        action="extract_audio",
        client=client,
        audio_format="mp3",
        audio_bitrate="320k",
    )
    print(f"Результат: {post_result.output_path}")


asyncio.run(main())

Примеры

В каталоге examples/ — готовые примеры:

• simple_transcode.py — перекодирование с прогрессом
• extract_audio.py — извлечение звуковых дорожек
• video_thumbnails.py — скриншоты, миниатюры, раскадровка
• watermark_and_filters.py — водяные знаки и фильтры через MediaPipeline
• stream_concat.py — склейка медиафайлов
• hardware_acceleration.py — кодирование с аппаратным ускорением
• ytdlp_pipeline.py — конвейер с async-yt-dlp

Разработка

# Dev-окружение
uv sync --extra dev

# Линтинг
uv run ruff check src/ tests/
uv run ruff format --check src/ tests/

# Типизация
uv run mypy src/

# Тесты
uv run pytest tests/ -v

Лицензия

Распространяется под лицензией MIT.