python执行ffmpeg

python执行ffmpeg命令

1. 能拿到ffmpeg正常输出
2. ffmpeg抛出异常时可以拿到异常信息
3. 返回ffmpeg处理进度

以下代码依赖的pexpect,progressbar需要安装下

import pexpect
import subprocess
import progressbar
import logging

def exec_progress(command, video_duration_seconds):
    """
    执行ffmpeg命令，并根据ffmpeg输出中的"time=xxx"匹配进度, ffmpeg执行失败时抛出FfmpegException
    :param command: ffmpeg命令
    :param video_duration_seconds: 视频总时长
    """
    thread = pexpect.spawn(command)
    cpl = thread.compile_pattern_list([pexpect.EOF,
                                       "frame=.*time=([\d:\.]*)(.*)",
                                       '(.+)'])

    progress = progressbar.ProgressBar(max_value=video_duration_seconds).start()

    output_list = []
    while True:
        i = thread.expect_list(cpl, timeout=None)
        if i == 0:
            progress.finish()
            break
        elif i == 1:
            seconds = duration_to_seconds(thread.match.group(1))
            progress.update(seconds)
        elif i == 2:
            logging.debug(thread.match.group(0))
            output_list.append(thread.match.group(0))
            pass

    thread.close()
    if thread.exitstatus:
        raise FfmpegException(thread.exitstatus, command, "\n".join(output_list))


def exec_output(command):
    """
    执行ffmpeg命令并返回所有输出，如果执行失败，抛出FfmpegException
    :param command: ffmpeg命令
    :return: ffmpeg标准输出
    """
    try:
        process = subprocess.check_output(command, shell=True, stderr=subprocess.STDOUT)
        return process

    except subprocess.CalledProcessError as err:
        raise FfmpegException(err.returncode, err.cmd, err.output)

# 这个方法应该单独抽到别的模块        
def duration_to_seconds(duration):
    time_arr = duration[0:duration.find(".")].split(":")
    if len(time_arr) == 3:
        return int(time_arr[0]) * 3600 + int(time_arr[1]) * 60 + int(time_arr[2])
    logging.error("unrecognized duration %s", duration)
    return 0

class FfmpegException(Exception):
    """
    使用ffmpeg报错时抛出
    """

    def __init__(self, returncode, cmd, output=None):
        self.returncode = returncode
        self.cmd = cmd
        self.output = output

    def __str__(self):
        return "Command '%s' returned non-zero exit status %d" % (self.cmd, self.returncode)

简单说明下

1. exec_progress执行耗时较长需要实时拿到处理进度的场景，如压缩，裁剪等
2. exec_output执行耗时较短直接拿到输出结果的场景，如获取视频信息等