如何为 Popen 的标准输入连接多个文件

Question

我正在将一个 bash 脚本移植到 python 2.6，并且想要替换一些代码：

cat $( ls -tr xyz_`date +%F`_*.log ) | filter args > bzip2

我想我想要类似于http://docs.python.org/release/2.6/library/subprocess.html 的“替换 shell 管道”示例，唉...

p1 = Popen(["filter", "args"], stdin=*?WHAT?*, stdout=PIPE)
p2 = Popen(["bzip2"], stdin=p1.stdout, stdout=PIPE)
output = p2.communicate()[0]

但是，我不确定如何最好地提供p1 的stdin 值，以便连接输入文件。看来我可以补充...

p0 = Popen(["cat", "file1", "file2"...], stdout=PIPE)
p1 = ... stdin=p0.stdout ...

...但这似乎超出了使用（缓慢、低效）管道来调用具有重要功能的外部程序的范围。 （任何体面的 shell 都会在内部执行 cat。）

所以，我可以想象一个满足文件对象 API 要求的自定义类，因此可以用于 p1 的标准输入，连接任意其他文件对象。 （编辑：现有答案解释了为什么这是不可能的）

python 2.6 是否有一种机制可以满足这种需求/想要，或者在 python 圈子中是否可以认为另一个 Popen 到 cat 非常好？

谢谢。

Answer 1

您可以用 Python 代码替换您正在执行的所有操作，但您的外部实用程序除外。这样，只要您的外部工具是可移植的，您的程序就会保持可移植性。您还可以考虑将 C++ 程序转换为库并使用 Cython 与其交互。正如 Messa 所示，date 被替换为 time.strftime，通配符由 glob.glob 完成，cat 可以替换为读取列表中的所有文件并将它们写入程序的输入。对bzip2 的调用可以替换为bz2 模块，但这会使您的程序复杂化，因为您必须同时读取和写入。为此，您需要使用 p.communicate 或在数据量很大时使用线程（select.select 会是更好的选择，但它不适用于 Windows）。

import sys
import bz2
import glob
import time
import threading
import subprocess

output_filename = '../whatever.bz2'
input_filenames = glob.glob(time.strftime("xyz_%F_*.log"))
p = subprocess.Popen(['filter', 'args'], stdin=subprocess.PIPE, stdout=subprocess.PIPE)
output = open(output_filename, 'wb')
output_compressor = bz2.BZ2Compressor()

def data_reader():
    for filename in input_filenames:
        f = open(filename, 'rb')
        p.stdin.writelines(iter(lambda: f.read(8192), ''))
    p.stdin.close()

input_thread = threading.Thread(target=data_reader)
input_thread.start()

with output:
    for chunk in iter(lambda: p.stdout.read(8192), ''):
        output.write(output_compressor.compress(chunk))

    output.write(output_compressor.flush())

input_thread.join()
p.wait()

补充：如何检测文件输入类型

您可以使用 libmagic 的文件扩展名或 Python 绑定来检测文件的压缩方式。这是一个代码示例，它同时执行这两种操作，并自动选择magic（如果可用）。您可以选择适合您需要的部分，并根据您的需要进行调整。 open_autodecompress 应该检测 mime 编码并使用适当的解压缩器打开文件（如果可用）。

import os
import gzip
import bz2
try:
    import magic
except ImportError:
    has_magic = False
else:
    has_magic = True


mime_openers = {
    'application/x-bzip2': bz2.BZ2File,
    'application/x-gzip': gzip.GzipFile,
}

ext_openers = {
    '.bz2': bz2.BZ2File,
    '.gz': gzip.GzipFile,
}


def open_autodecompress(filename, mode='r'):
    if has_magic:
        ms = magic.open(magic.MAGIC_MIME_TYPE)
        ms.load()
        mimetype = ms.file(filename)
        opener = mime_openers.get(mimetype, open)
    else:
        basepart, ext = os.path.splitext(filename)
        opener = ext_openers.get(ext, open)
    return opener(filename, mode)

Answer 2

如果您查看 subprocess 模块实现的内部，您将看到 std{in,out,err} 应该是支持 fileno() 方法的文件对象，因此是一个简单的带有 python 接口的类文件对象（或即使是 StringIO 对象）也不适合这里。

如果是迭代器，而不是文件对象，你可以使用itertools.chain。

当然，牺牲内存消耗你可以这样做：

import itertools, os

# ...

files = [f for f in os.listdir(".") if os.path.isfile(f)]
input = ''.join(itertools.chain(open(file) for file in files))
p2.communicate(input)

Answer 3

使用子进程时，您必须考虑这样一个事实，即 Popen 内部将使用文件描述符（处理程序）并在将它们传递给创建的子进程之前为 stdin、stdout 和 stderr 调用 os.dup2()。

所以如果你不想在 Popen 中使用系统 shell 管道：

p0 = Popen(["cat", "file1", "file2"...], stdout=PIPE)
p1 = Popen(["filter", "args"], stdin=p0.stdout, stdout=PIPE)

...

我认为您的另一个选择是在 python 中编写一个 cat 函数并以类似 cat 的方式生成一个 file 并将这个 file 传递给 p1 stdin，不要考虑一个实现 io API 的类，因为它不会像我说的那样工作，因为在内部子进程只会获取文件描述符。

话虽如此，我认为您更好的选择是使用 unix PIPE 方式，例如 subprocess doc。

Answer 4

这应该很容易。首先，使用os.pipe 创建一个管道，然后以管道的读取端作为标准输入打开filter。然后对于目录中名称与模式匹配的每个文件，只需将其内容传递到管道的写入端。这应该与 shell 命令cat ..._*.log | filter args 所做的完全相同。

更新：对不起，来自os.pipe 的管道不需要，我忘记了subprocess.Popen(..., stdin=subprocess.PIPE) 实际上为你创建了一个。另外一个管道不能塞满太多数据，只有在读取之前的数据后才能将更多数据写入管道。

所以解决方案（例如wc -l）将是：

import glob
import subprocess

p = subprocess.Popen(["wc", "-l"], stdin=subprocess.PIPE)

processDate = "2011-05-18"  # or time.strftime("%F")
for name in glob.glob("xyz_%s_*.log" % processDate):
    f = open(name, "rb")
    # copy all data from f to p.stdin
    while True:
        data = f.read(8192)
        if not data:
            break  # reached end of file
        p.stdin.write(data)
    f.close()

p.stdin.close()
p.wait()

使用示例：

$ hexdump /dev/urandom | head -n 10000 >xyz_2011-05-18_a.log 
$ hexdump /dev/urandom | head -n 10000 >xyz_2011-05-18_b.log 
$ hexdump /dev/urandom | head -n 10000 >xyz_2011-05-18_c.log 
$ ./example.py 
   30000