как объединить несколько файлов для стандартного ввода Popen - PullRequest
Новая
       7

как объединить несколько файлов для стандартного ввода Popen

9 голосов
/ 18 мая 2011

Я портирую bash-скрипт на python 2.6 и хочу заменить код: 

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

Полагаю, мне нужно что-то похожее на пример "Замена оболочки трубопровода" на http://docs.python.org/release/2.6/library/subprocess.html, ala ... 

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 ...

... но это, кажется, выходит за рамки использования (медленных, неэффективных) каналов для вызова внешних программ со значительной функциональностью. (Любая приличная оболочка выполняет cat внутри.)

Итак, я могу представить собственный класс, который удовлетворяет требованиям API файлового объекта и, следовательно, может использоваться для стандартного ввода p1, объединяя произвольные другие файловые объекты. ( РЕДАКТИРОВАТЬ: существующие ответы объясняют, почему это невозможно )

Есть ли в python 2.6 механизм для удовлетворения этой потребности / желания, или может ли другой Popen до cat считаться идеально подходящим для кругов питона?

Спасибо.

Ответы [ 4 ]

5 голосов
/ 18 мая 2011

Вы можете заменить все, что вы делаете, кодом Python, кроме вашей внешней утилиты. Таким образом, ваша программа останется переносимой, пока ваша внешняя утилита переносима. Вы также можете подумать о том, чтобы превратить программу C ++ в библиотеку и использовать Cython для взаимодействия с ней. Как показал Месса, 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()

Дополнение: Как определить тип ввода файла

Вы можете использовать расширение файла или привязки Python для libmagic, чтобы определить, как файл сжимается. Вот пример кода, который выполняет оба действия и автоматически выбирает 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)
2 голосов
/ 18 мая 2011

Если вы загляните внутрь реализации модуля 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)
1 голос
/ 18 мая 2011

При использовании подпроцесса вы должны учитывать тот факт, что внутренне Popen будет использовать дескриптор файла (обработчик) и вызывать os.dup2 () для stdin, stdout и stderr, прежде чем передавать их в созданный дочерний процесс.

Так что, если вы не хотите использовать системную оболочку с Popen: 

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

...

Я думаю, что другой вариант - написать функцию cat на python и сгенерировать файл в catПодобно этому способу и передав этот файл в p1 stdin, не думайте о классе, который реализует API io, потому что он не будет работать, как я сказал, потому что внутри дочерний процесс просто получит дескрипторы файлов.

С учетом сказанного, я думаю, что вам лучше использовать unix PIPE, как в subprocess doc .

1 голос
/ 18 мая 2011

Это должно быть легко.Сначала создайте pipe , используя os.pipe , затем вставьте filter со считанным концом канала в качестве стандартного ввода.Затем для каждого файла в каталоге с именем, совпадающим с шаблоном, просто передайте его содержимое в конец записи канала.Это должно быть точно так же, как и команда оболочки 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
...