如何从ffmpeg-python异步进程中获取实时进度信息？

核心问题拆解 #

你的当前代码通过communicate()等待进程结束后才获取输出，无法实现实时追踪。要拿到FFmpeg的实时进度或日志，需要逐行读取进程的输出流，同时FFmpeg默认将进度日志输出到stderr而非stdout，因此需要调整管道配置。

解决方案1：解析默认stderr输出 #

直接捕获FFmpeg的标准错误输出，通过正则提取进度信息：

import ffmpeg
import threading
import re

def parse_progress(line: str, total_duration: float) -> float | None:
    """从FFmpeg日志行提取进度百分比"""
    time_match = re.search(r'time=(\d+):(\d+):(\d+\.\d+)', line)
    if time_match:
        h, m, s = map(float, time_match.groups())
        current_time = h * 3600 + m * 60 + s
        return round((current_time / total_duration) * 100, 2)
    return None

def get_total_duration(input_path: str) -> float:
    """获取输入视频总时长（秒）"""
    probe_data = ffmpeg.probe(input_path)
    return float(probe_data['streams'][0]['duration'])

def ffmpeg_task(vid_path: str, audio_path: str, output_path: str):
    # 提前获取视频总时长，用于计算进度
    total_duration = get_total_duration(vid_path)
    
    # 启动异步FFmpeg进程，捕获stderr
    video_stream = ffmpeg.input(vid_path)
    audio_stream = ffmpeg.input(audio_path)
    process = ffmpeg.output(audio_stream, video_stream, output_path) \
        .overwrite_output() \
        .run_async(pipe_stderr=True)
    
    # 逐行读取实时输出
    with process.stderr:
        for line in iter(process.stderr.readline, b''):
            line_str = line.decode('utf-8').strip()
            # 存储实时日志（可写入文件/队列，这里仅打印示例）
            print(f"[{output_path}] 实时日志: {line_str}")
            
            # 解析并输出进度
            progress = parse_progress(line_str, total_duration)
            if progress:
                print(f"[{output_path}] 当前进度: {progress}%")
    
    # 等待进程结束，获取返回码
    ret_code = process.wait()
    print(f"[{output_path}] 任务完成，返回码: {ret_code}")

# 启动多线程任务
threading.Thread(target=ffmpeg_task, args=("vid_part1.mp4", "audio_part1.mp3", "output1.mp4")).start()
threading.Thread(target=ffmpeg_task, args=("vid_part2.mp4", "audio_part2.mp3", "output2.mp4")).start()

解决方案2：使用FFmpeg -progress参数（更稳定） #

FFmpeg提供-progress参数输出机器可读的键值对格式进度数据，解析更可靠：

import ffmpeg
import threading

def parse_progress_kv(line: str) -> dict | None:
    """解析-progress输出的键值对"""
    line = line.strip()
    if '=' in line:
        key, val = line.split('=', 1)
        return {key.strip(): val.strip()}
    return None

def ffmpeg_task(vid_path: str, audio_path: str, output_path: str):
    # 配置FFmpeg将进度输出到stdout
    video_stream = ffmpeg.input(vid_path)
    audio_stream = ffmpeg.input(audio_path)
    process = ffmpeg.output(audio_stream, video_stream, output_path) \
        .overwrite_output() \
        .global_args('-progress', 'pipe:1')  # 将进度定向到stdout
        .run_async(pipe_stdout=True)
    
    # 逐行解析进度数据
    with process.stdout:
        for line in iter(process.stdout.readline, b''):
            line_str = line.decode('utf-8')
            progress_data = parse_progress_kv(line_str)
            if progress_data:
                # 解析进度百分比
                if 'out_time_ms' in progress_data and 'duration_ms' in progress_data:
                    current_ms = int(progress_data['out_time_ms'])
                    total_ms = int(progress_data['duration_ms'])
                    progress = round((current_ms / total_ms) * 100, 2)
                    print(f"[{output_path}] 进度: {progress}%")
                # 可提取其他字段：frame、fps、bitrate、speed等
                # print(f"[{output_path}] 进度详情: {progress_data}")
    
    ret_code = process.wait()
    print(f"[{output_path}] 任务完成，返回码: {ret_code}")

# 启动多线程任务
threading.Thread(target=ffmpeg_task, args=("vid_part1.mp4", "audio_part1.mp3", "output1.mp4")).start()

多线程注意事项 #

• 避免用全局变量传递进度数据，建议使用queue.Queue将进度信息发送到主线程统一处理
• 每个线程独立管理自身的FFmpeg进程，无资源冲突风险
• 可给任务添加唯一标识（如任务ID），方便区分不同线程的输出

内容的提问来源于stack exchange，提问作者duruburak