用 Python 包一层 ffmpeg 跑批量任务：进度、重试、日志都别省

前面十几篇都是单条 ffmpeg 命令，真到生产就是"一文件夹 500 个视频要统一转码/加水印/提取"。直接 shell 循环能跑，但没法看进度、出错就断、日志一团。这篇讲我怎么用 Python 把 ffmpeg 包一层，让批量任务可控。

核心：subprocess 跑命令，别用 os.system

import subprocess

def run_ffmpeg(args: list):
    cmd = ["ffmpeg", "-hide_banner", "-y", *args]
    p = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    if p.returncode != 0:
        raise RuntimeError(p.stderr)
    return p

-hide_banner 少输出噪音；-y 覆盖不询问（批处理必备）。

进度条

ffmpeg 的进度在 stderr，格式像 frame=123 fps=... time=00:01:23 ...。抓 time= 算百分比：

import re
m = re.search(r"time=(\d+):(\d+):(\d+\.\d+)", line)

配合 tqdm 直接出进度条。我一般只在 stderr 里逐行扫，匹配到就 pbar.update。

失败重试

网络源（m3u8）容易中途断。加重试：

from tenacity import retry, stop_after_attempt, wait_fixed

@retry(stop=stop_after_attempt(3), wait=wait_fixed(5))
def transcode(src, dst):
    run_ffmpeg([...])

或者手写 for attempt in range(3)。注意重试前删掉半截的坏文件，否则 -y 覆盖还好，没覆盖就留了个损坏文件后面 concat 崩。

日志别只打印

批量跑一晚上，出了问题得能复盘。每条任务写一行结构化日志（文件名、起止时间、退出码、大小）：

logging.info({"file": src, "ok": True, "size": os.path.getsize(dst)})

出错的单独归档到 failed/ 目录，别让一个坏文件卡住整批。

并发还是串行

ffmpeg 本身能跑满一颗核。多进程的收益 = 核数，但 IO（磁盘/网络）会成瓶颈。我一般 concurrent.futures.ProcessPoolExecutor(max_workers=os.cpu_count()//2)，留点余量给系统。前面并发下载那篇的线程池思路在这也能用，但 ffmpeg 是 CPU 密集，用进程不是线程。

一个容易翻车的点

Windows 上路径带空格、中文，subprocess 传 list 就不会被 shell 拆错；千万别拼成字符串再 shell=True，既不安全又容易编码翻车。我见过有人中文路径直接 ffmpeg input.mp4 报"找不到文件"，就是字符串拼接的锅。

小结

批量处理真正的难点不是 ffmpeg 命令，是可观测、可恢复、可控并发。把上面四块（进度、重试、日志、进程池）拼起来，500 个文件挂机跑通宵也不用盯着。