你如何在 Python 中处理管道和输出？答案

【问题标题】：How do you handle pipes and output in Python?你如何在 Python 中处理管道和输出？
【发布时间】：2013-10-19 03:45:09
【问题描述】：

我正在尝试在我的 python 脚本中执行此命令：

avprobeCommand = "avprobe -of json -show_streams {0} | grep '\"duration\"' | sed -n 1p | sed 's/ //g'".format(hiOutput)
output = subprocess.check_output([avprobeCommand])

我不断得到：

    Traceback (most recent call last):
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.py", line 552, in __bootstrap_inner
    self.run()
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/threading.py", line 505, in run
    self.__target(*self.__args, **self.__kwargs)
  File "/Users/jmakaila/Documents/Development/Present/Web/video_dev/present-live-transcoder/Transcoder.py", line 60, in transcode
    output = subprocess.check_output([avprobeCommand])
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/subprocess.py", line 537, in check_output
    process = Popen(stdout=PIPE, *popenargs, **kwargs)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/subprocess.py", line 679, in __init__
    errread, errwrite)
  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/subprocess.py", line 1228, in _execute_child
    raise child_exception
OSError: [Errno 2] No such file or directory

我已经尝试将 args 拆分，但我一直收到 -of json -show_streams 部分的错误，据记录，它看起来像这样：

subprocess.check_output(["avprobe", "-of json", "-show_streams", "{0}".format(hiOutput)

【问题讨论】：

这里列出了替代方法：stackoverflow.com/q/295459/1858225 我建议使用pipes 模块：docs.python.org/2/library/pipes.html

标签： python subprocess avprobe

【解决方案1】：

将命令作为字符串传递，并传递shell=True:

import pipes
import subprocess

avprobeCommand = """avprobe -of json -show_streams {0} | grep '"duration"' | sed -n 1p | sed 's/ //g'""".format(pipes.quote(hiOutput))
output = subprocess.check_output(avprobeCommand, shell=True)

更新：应使用pipes.quote 转义参数。（如果您使用 Python 3.3+，请使用 shlex.quote）。

【讨论】：

@J.F.Sebastian，感谢您的评论。我根据您的评论更新了答案。

【解决方案2】：

在您的情况下，您可以将后处理移至 Python：

import json
from subprocess import check_output as qx

data = json.loads(qx(["avprobe", "-of", "json", "-show_streams", hiOutput]))
result = data["duration"]         # grep '"duration"'
             .partition("\n")[0]  # sed -n 1p
             .replace(" ", "")    # sed 's/ //g'

有关更一般的情况，请参阅How do I use subprocess.Popen to connect multiple processes by pipes?

【讨论】：