如何判断Subprocess调用的FFmpeg是运行中还是等待用户输入？

背景 #

我们可以通过returncode检查subprocess子进程的状态，示例代码：

p = subprocess.Popen(command, shell=True)
return_code = p.wait()
print(return_code)

returncode的文档说明：

子进程的返回码，由poll()和wait()设置。若值为None，表示进程尚未终止。

当通过subprocess运行FFmpeg命令时，若目标文件已存在，FFmpeg会弹出确认提示：

File 'output.mp4' already exists. Overwrite? [y/N]

需要判断进程是正常运行，还是卡在等待用户输入的状态。

解决方案 #

1. 提前规避提示（推荐） #

最直接的方式是给FFmpeg命令加参数跳过确认环节：

• 加-y：自动覆盖已存在文件，不弹出提示
• 加-n：若文件已存在直接终止进程，不提示

示例命令：

ffmpeg -i input.mp4 -y output.mp4

从根源避免等待输入的情况，无需后续检测。

2. 实时监控输出判断提示 #

如果必须保留提示并检测，可捕获FFmpeg的输出，实时检查是否出现特定提示文本：

import subprocess
import select

command = "ffmpeg -i input.mp4 output.mp4"
p = subprocess.Popen(
    command,
    shell=True,
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,  # 将stderr合并到stdout统一捕获
    text=True
)

# 实时监控进程输出
while True:
    # 用select检查是否有输出可读，避免阻塞
    ready, _, _ = select.select([p.stdout], [], [], 1)
    if ready:
        line = p.stdout.readline()
        if not line:
            break  # 进程结束，输出为空
        # 检查是否出现覆盖提示
        if "Overwrite? [y/N]" in line:
            print("进程正在等待用户输入覆盖确认")
            # 可选择自动输入确认，比如：
            # p.stdin.write("y\n")
            # p.stdin.flush()
    # 检查进程是否已终止
    return_code = p.poll()
    if return_code is not None:
        print(f"进程已终止，返回码: {return_code}")
        break

原理：FFmpeg的覆盖提示会输出到标准错误流（stderr），将其合并到stdout后，即可通过字符串匹配判断进程是否在等待输入。

3. 通过进程状态判断（仅类Unix系统适用） #

不想解析输出的话，可通过/proc文件系统查看进程状态（仅Linux等类Unix系统支持）：

import subprocess
import os
import time

command = "ffmpeg -i input.mp4 output.mp4"
p = subprocess.Popen(command, shell=True, stdin=subprocess.PIPE)

while True:
    return_code = p.poll()
    if return_code is not None:
        print(f"进程已终止，返回码: {return_code}")
        break
    # 读取进程状态文件
    try:
        with open(f"/proc/{p.pid}/status", "r") as f:
            for line in f.read().splitlines():
                if line.startswith("State:"):
                    state = line.split()[1]
                    # 状态为T时，表示进程等待终端输入（S_TTYIN）
                    if state == "T":
                        print("进程正在等待用户输入")
                    break
    except FileNotFoundError:
        pass  # 进程已结束，跳过
    time.sleep(0.5)

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