如何为 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，唉...

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 的 stdin，连接任意其他文件对象。 （编辑：现有答案解释了为什么这是不可能的）

Python 2.6 是否有解决此需求/想要的机制，或者可能另一个 Popen 到 cat 在 python 圈子里被认为是完美的吗？

谢谢。

Answer 1

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

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()，然后再将它们传递给创建的子进程。

因此，如果您不想将系统 shell 管道与 Popen 一起使用：

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

...

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

话虽如此，我认为你更好的选择是使用 unix PIPE 方式，如 subprocess文档。

Answer 4

这应该很容易。首先，使用 os.pipe 创建一个管道，然后使用 Popen将管道的读取端作为标准输入的过滤器。然后，对于目录中名称与模式匹配的每个文件，只需将其内容传递到管道的写入端。这应该与 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