32blogby Studio Mitsu

Automatiza el procesamiento de video con FFmpeg y Python

Aprende a llamar FFmpeg desde Python para automatizar el procesamiento de video por lotes. Cubre subprocess, ffmpeg-python, barras de progreso, codificación paralela y manejo de errores.

by omitsu13 min read

This article contains affiliate links.

Contenido

La forma más sencilla de llamar a FFmpeg desde Python es subprocess.run(["ffmpeg", "-i", "input.mov", "-c:v", "libx264", "output.mp4"]). Para procesamiento por lotes, combina esto con pathlib.glob() para iterar archivos, tqdm para barras de progreso y concurrent.futures para codificar en paralelo usando múltiples núcleos.

Si estás convirtiendo videos a mano, automatizar el proceso con Python bien vale el esfuerzo. Ya sea que necesites convertir por lotes 100 archivos MP4 a WebM o redimensionar una carpeta de grabaciones de una vez, un solo script puede encargarse de todo. Una vez que lo configuras, no hay vuelta atrás.

Este artículo recorre cómo llamar a FFmpeg desde Python — empezando con lo básico usando subprocess, luego pasando a la biblioteca ffmpeg-python, un convertidor por lotes de carpetas completo, barras de progreso, codificación paralela, y un patrón robusto de manejo de errores con reintentos.

Por qué usar Python con FFmpeg

FFmpeg es una herramienta CLI completa y potente por sí sola. Pero cuando estás lidiando con grandes cantidades de archivos o lógica condicional compleja, los scripts de shell empiezan a mostrar sus límites.

Combinar Python te da varias ventajas.

  • Enumeración y filtrado de archivos: apunta solo a archivos que coincidan con ciertas extensiones, tamaños o patrones de nombres
  • Manejo de errores y reintentos: registra los fallos y reprocésalos automáticamente
  • Visualización de progreso: saber en tiempo real cuántos archivos están hechos y cuántos quedan
  • Registro estructurado: guardar resultados de conversión en un formato que puedas revisar después

Los scripts de shell pueden manejar algo de esto, pero Python mantiene la lógica legible a medida que crece la complejidad. La documentación oficial de FFmpeg cubre las opciones del CLI, mientras que Python se encarga de la capa de orquestación.

Input Folderglob filterIteratePythonsubprocessCallFFmpegEncodeWriteOutput FolderDone / Failed

Lo básico con subprocess

La forma más simple de llamar a FFmpeg desde Python es subprocess.run(). Como FFmpeg es una herramienta CLI, pasas los argumentos de línea de comandos como una lista y simplemente funciona.

python
import subprocess
from pathlib import Path


def convert_to_mp4(input_path: Path, output_path: Path) -> bool:
    """Convert a file to MP4. Returns True on success, False on failure."""
    cmd = [
        "ffmpeg",
        "-i", str(input_path),
        "-c:v", "libx264",
        "-crf", "23",
        "-preset", "medium",
        "-c:a", "aac",
        "-b:a", "128k",
        "-y",               # overwrite without asking
        str(output_path),
    ]

    result = subprocess.run(
        cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    return result.returncode == 0


if __name__ == "__main__":
    src = Path("input.mov")
    dst = Path("output.mp4")

    if convert_to_mp4(src, dst):
        print(f"Success: {dst}")
    else:
        print("Conversion failed")

Algunas cosas que vale la pena notar.

  • La bandera "-y" sobrescribe el archivo de salida si ya existe — útil para procesamiento por lotes
  • stdout=subprocess.PIPE y stderr=subprocess.PIPE capturan la salida de FFmpeg en lugar de dejar que inunde la terminal
  • FFmpeg escribe sus logs en stderr incluso en éxito. Accede a la salida de error con result.stderr.decode()

Usando la biblioteca ffmpeg-python

Llamar a subprocess directamente es sencillo, pero la lista de comandos se vuelve difícil de manejar cuando añades muchas opciones. La biblioteca ffmpeg-python ofrece un enfoque más limpio.

Instálala con pip.

python
# pip install ffmpeg-python

Construyes el pipeline de FFmpeg usando encadenamiento de métodos.

python
import ffmpeg


def transcode(input_path: str, output_path: str) -> None:
    """Transcode using ffmpeg-python."""
    (
        ffmpeg
        .input(input_path)
        .output(
            output_path,
            vcodec="libx264",
            crf=23,
            preset="medium",
            acodec="aac",
            audio_bitrate="128k",
        )
        .overwrite_output()
        .run(capture_stdout=True, capture_stderr=True)
    )


if __name__ == "__main__":
    transcode("input.mov", "output.mp4")

Cuando ocurre un error, se lanza ffmpeg.Error. El atributo e.stderr contiene la salida de error estándar de FFmpeg, que es útil para depuración.

python
import ffmpeg


def transcode_safe(input_path: str, output_path: str) -> None:
    try:
        (
            ffmpeg
            .input(input_path)
            .output(output_path, vcodec="libx264", crf=23)
            .overwrite_output()
            .run(capture_stdout=True, capture_stderr=True)
        )
    except ffmpeg.Error as e:
        print("FFmpeg error:")
        print(e.stderr.decode())
        raise

Conversión por lotes de una carpeta de videos

Ahora el evento principal. El script a continuación convierte todos los archivos MOV de una carpeta a MP4. Usa pathlib.glob() para enumerar archivos y los procesa uno por uno.

python
import subprocess
from pathlib import Path
from typing import List


INPUT_DIR = Path("./originals")
OUTPUT_DIR = Path("./converted")
INPUT_EXT = ".mov"
OUTPUT_EXT = ".mp4"


def convert_file(src: Path, dst: Path) -> bool:
    """Convert a single file to MP4."""
    dst.parent.mkdir(parents=True, exist_ok=True)

    cmd = [
        "ffmpeg",
        "-i", str(src),
        "-c:v", "libx264",
        "-crf", "23",
        "-preset", "fast",
        "-c:a", "aac",
        "-b:a", "128k",
        "-movflags", "+faststart",  # optimize for web playback
        "-y",
        str(dst),
    ]

    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    return result.returncode == 0


def batch_convert(input_dir: Path, output_dir: Path) -> None:
    """Convert all INPUT_EXT files in input_dir to OUTPUT_EXT."""
    files = list(input_dir.glob(f"**/*{INPUT_EXT}"))

    if not files:
        print(f"No {INPUT_EXT} files found in {input_dir}")
        return

    print(f"Found {len(files)} file(s) to convert")
    success_count = 0
    fail_list: List[Path] = []

    for i, src in enumerate(files, start=1):
        # preserve the relative folder structure in the output
        relative = src.relative_to(input_dir)
        dst = output_dir / relative.with_suffix(OUTPUT_EXT)

        print(f"[{i}/{len(files)}] {src.name} → {dst.name}", end=" ... ")

        if convert_file(src, dst):
            print("OK")
            success_count += 1
        else:
            print("FALLO")
            fail_list.append(src)

    print(f"\nHecho: {success_count}/{len(files)} exitosos")

    if fail_list:
        print("Archivos fallidos:")
        for f in fail_list:
            print(f"  {f}")


if __name__ == "__main__":
    batch_convert(INPUT_DIR, OUTPUT_DIR)

El patrón glob recursivo procesa subcarpetas automáticamente. La salida refleja la estructura de directorios original, así que incluso los archivos profundamente anidados se mantienen organizados.

Añadir una barra de progreso

Cuando procesas una gran cantidad de archivos, una barra de progreso hace la espera mucho más manejable. tqdm añade una en una sola línea.

python
import subprocess
from pathlib import Path
from typing import List
from tqdm import tqdm


# pip install tqdm


def convert_file(src: Path, dst: Path) -> bool:
    dst.parent.mkdir(parents=True, exist_ok=True)
    cmd = [
        "ffmpeg", "-i", str(src),
        "-c:v", "libx264", "-crf", "23", "-preset", "fast",
        "-c:a", "aac", "-b:a", "128k",
        "-y", str(dst),
    ]
    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    return result.returncode == 0


def batch_convert_with_progress(input_dir: Path, output_dir: Path) -> None:
    files = list(input_dir.glob("**/*.mov"))
    fail_list: List[Path] = []

    with tqdm(total=len(files), unit="file", ncols=80) as pbar:
        for src in files:
            relative = src.relative_to(input_dir)
            dst = output_dir / relative.with_suffix(".mp4")

            pbar.set_description(src.name[:30])  # show filename on the left

            if not convert_file(src, dst):
                fail_list.append(src)

            pbar.update(1)

    if fail_list:
        print(f"\nFallidos: {len(fail_list)} archivo(s)")
        for f in fail_list:
            print(f"  {f}")


if __name__ == "__main__":
    batch_convert_with_progress(Path("./originals"), Path("./converted"))

set_description() muestra el nombre del archivo actual a la izquierda de la barra. Truncar con [:30] evita que nombres de archivo largos rompan la disposición.

Manejo de errores y lógica de reintentos

Un aspecto complicado del manejo de errores de FFmpeg es que un código de salida exitoso no garantiza que el archivo de salida sea válido. Si la entrada tiene corrupción de datos a mitad del archivo, FFmpeg puede terminar con código de salida 0 mientras produce un archivo no reproducible.

El script a continuación combina la lógica de reintentos con un paso básico de validación de salida usando ffprobe.

python
import subprocess
import time
from pathlib import Path


MAX_RETRIES = 3
RETRY_DELAY = 2  # seconds


def verify_output(path: Path) -> bool:
    """Validate the output file using ffprobe."""
    result = subprocess.run(
        ["ffprobe", "-v", "error", "-show_entries", "format=duration",
         "-of", "default=noprint_wrappers=1:nokey=1", str(path)],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    if result.returncode != 0:
        return False
    output = result.stdout.decode().strip()
    return bool(output) and float(output) > 0


def convert_with_retry(src: Path, dst: Path, retries: int = MAX_RETRIES) -> bool:
    """Convert with up to `retries` attempts."""
    for attempt in range(1, retries + 1):
        dst.parent.mkdir(parents=True, exist_ok=True)

        cmd = [
            "ffmpeg", "-i", str(src),
            "-c:v", "libx264", "-crf", "23", "-preset", "fast",
            "-c:a", "aac", "-b:a", "128k",
            "-y", str(dst),
        ]

        result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)

        if result.returncode != 0:
            print(f"  Intento {attempt}/{retries}: error de FFmpeg (salida {result.returncode})")
            if attempt < retries:
                time.sleep(RETRY_DELAY)
            continue

        if not verify_output(dst):
            print(f"  Intento {attempt}/{retries}: validación de salida fallida")
            dst.unlink(missing_ok=True)  # remove the broken file
            if attempt < retries:
                time.sleep(RETRY_DELAY)
            continue

        return True

    return False


def batch_convert_robust(input_dir: Path, output_dir: Path) -> None:
    files = list(input_dir.glob("**/*.mov"))
    success, failed = 0, []

    for i, src in enumerate(files, start=1):
        relative = src.relative_to(input_dir)
        dst = output_dir / relative.with_suffix(".mp4")
        print(f"[{i}/{len(files)}] {src.name}")

        if convert_with_retry(src, dst):
            print(f"  Hecho: {dst}")
            success += 1
        else:
            print(f"  Omitido tras {MAX_RETRIES} intentos fallidos: {src}")
            failed.append(src)

    print(f"\nResultado: {success} exitosos / {len(failed)} fallidos / {len(files)} total")

    if failed:
        log_path = output_dir / "failed.txt"
        log_path.write_text("\n".join(str(f) for f in failed))
        print(f"Log de fallos guardado en: {log_path}")


if __name__ == "__main__":
    batch_convert_robust(Path("./originals"), Path("./converted"))

Los archivos fallidos se escriben en failed.txt en el directorio de salida para que puedas revisarlos y reprocesarlos después.

Para una base sólida en FFmpeg en sí, consulta el tutorial de uso de FFmpeg.

Codificación paralela con concurrent.futures

Todos los scripts anteriores procesan archivos de uno en uno. Si tu máquina tiene múltiples núcleos de CPU — y la mayoría los tiene — estás desperdiciando rendimiento. El módulo concurrent.futures de Python facilita la codificación simultánea de múltiples archivos.

python
import subprocess
from concurrent.futures import ProcessPoolExecutor, as_completed
from pathlib import Path


MAX_WORKERS = 4  # ajusta según tus núcleos de CPU


def convert_file(src: Path, dst: Path) -> tuple[Path, bool]:
    """Convierte un archivo. Retorna (ruta_origen, éxito)."""
    dst.parent.mkdir(parents=True, exist_ok=True)
    cmd = [
        "ffmpeg", "-i", str(src),
        "-c:v", "libx264", "-crf", "23", "-preset", "fast",
        "-c:a", "aac", "-b:a", "128k",
        "-y", str(dst),
    ]
    result = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    return src, result.returncode == 0


def batch_convert_parallel(input_dir: Path, output_dir: Path) -> None:
    files = list(input_dir.glob("**/*.mov"))
    tasks: dict = {}
    success, failed = 0, []

    with ProcessPoolExecutor(max_workers=MAX_WORKERS) as executor:
        for src in files:
            relative = src.relative_to(input_dir)
            dst = output_dir / relative.with_suffix(".mp4")
            future = executor.submit(convert_file, src, dst)
            tasks[future] = src

        for future in as_completed(tasks):
            src_path, ok = future.result()
            if ok:
                success += 1
                print(f"  Hecho: {src_path.name}")
            else:
                failed.append(src_path)
                print(f"  FALLO: {src_path.name}")

    print(f"\nResultado: {success} exitosos / {len(failed)} fallidos / {len(files)} total")


if __name__ == "__main__":
    batch_convert_parallel(Path("./originals"), Path("./converted"))

Algunas consideraciones importantes sobre la codificación paralela.

  • MAX_WORKERS debe coincidir con tus núcleos disponibles. El codificador x264 de FFmpeg usa múltiples hilos por proceso, así que configurar demasiados workers causa contención de hilos y en realidad ralentiza las cosas. Para una máquina de 8 núcleos, 2-4 workers es un buen punto de partida
  • El uso de memoria escala linealmente con el número de codificaciones paralelas. Si trabajas con material 4K, monitorea el uso de RAM
  • La E/S de disco puede convertirse en el cuello de botella — especialmente con discos mecánicos. Los SSD manejan lecturas/escrituras paralelas mucho mejor

Con un lote de 50 grabaciones de pantalla (1080p, 2-5 minutos cada una), 4 workers pueden ofrecer una aceleración de 3-4x respecto al procesamiento de un solo hilo. No es una mejora perfecta de N veces debido a la sobrecarga de E/S, pero la diferencia es significativa.

Para lotes aún más grandes donde tu máquina local no es suficiente, considera trasladar el trabajo a un servidor VPS de codificación. Una instancia en la nube con 16+ núcleos puede procesar cientos de archivos mientras tú sigues trabajando localmente.

Kamatera

VPS en la nube de grado empresarial con centros de datos globales

  • 13 centros de datos (EE.UU., Europa, Asia, Oriente Medio)
  • Desde $4/mes por 1GB RAM — pago por uso
  • Prueba gratuita de 30 días

Preguntas frecuentes

¿Cuál es la mejor forma de llamar a FFmpeg desde Python?

Usa subprocess.run() con el comando como lista de strings. Es el método más confiable y te da control total sobre las opciones de FFmpeg. La biblioteca ffmpeg-python es un wrapper conveniente pero añade una dependencia.

¿Sigue funcionando ffmpeg-python? No se ha actualizado desde 2019.

Sí. La última versión es la v0.2.0 de 2019, pero la biblioteca es estable porque simplemente envuelve el CLI de FFmpeg. Mientras FFmpeg esté instalado y la interfaz CLI no cambie (raramente lo hace), ffmpeg-python sigue funcionando. Si prefieres una alternativa más activamente mantenida, mira python-ffmpeg o typed-ffmpeg.

¿Cómo muestro una barra de progreso para la codificación de cada archivo individual?

Los ejemplos de este artículo muestran progreso a nivel de archivo (archivo 1 de N). Para progreso a nivel de fotograma dentro de una codificación individual, parsea la salida stderr de FFmpeg línea por línea buscando los campos frame= y time=. El paquete better-ffmpeg-progress hace esto automáticamente.

¿Puede FFmpeg salir con código 0 y producir un archivo roto?

Sí. Esto ocurre cuando la entrada tiene corrupción parcial o el espacio en disco se agota durante la codificación. Siempre valida la salida con ffprobe — verifica que la duración sea mayor a cero y coincida con el valor esperado.

¿Cuántos procesos paralelos de FFmpeg debería ejecutar?

Empieza con el número de núcleos físicos de CPU dividido entre 2. El codificador x264 de FFmpeg usa múltiples hilos por proceso, así que ejecutar demasiados en paralelo causa contención de hilos y ralentiza las cosas. Monitorea el uso de CPU y ajusta.

¿Debería usar subprocess.run() o subprocess.Popen()?

Usa subprocess.run() para la mayoría del procesamiento por lotes — bloquea hasta que FFmpeg termina, lo que simplifica el manejo de errores. Usa Popen() solo cuando necesites leer stderr en tiempo real (por ejemplo, para parsear progreso a nivel de fotograma).

¿Qué versión de Python necesito?

Python 3.8 o posterior. Los scripts usan pathlib, argumentos con nombre para subprocess.run() y anotaciones de tipo. El ejemplo de procesamiento paralelo usa la sintaxis tuple[Path, bool] que requiere Python 3.9+.

¿Cómo ejecuto la codificación FFmpeg en un servidor remoto?

Sube tus archivos vía rsync o scp, ejecuta el script Python por SSH (usa nohup o tmux para que sobreviva a la desconexión), y luego descarga los resultados. Un VPS con núcleos de CPU dedicados es ideal para esto.

Conclusión

Aquí tienes un resumen de lo que cubrió este artículo.

  • subprocess.run() es la forma más simple de llamar a FFmpeg desde Python, usando las mismas opciones que pasarías en la línea de comandos
  • ffmpeg-python te permite construir pipelines con encadenamiento de métodos, lo cual es especialmente útil para grafos de filtros complejos
  • La conversión por lotes usa pathlib.glob() para enumerar archivos y preserva la estructura de carpetas original en la salida
  • tqdm añade una barra de progreso en una línea — esencial para trabajos por lotes de larga duración
  • El manejo robusto de errores significa no confiar solo en el código de salida. Valida la salida con ffprobe y reintenta en caso de fallo
  • La codificación paralela con concurrent.futures reduce el tiempo de procesamiento en proporción a tus núcleos de CPU

Para llevar estos scripts más lejos, considera añadir argparse para argumentos CLI y logging para escribir logs estructurados a un archivo.


Artículos relacionados: