Запустите подпроцесс и распечатайте вывод в лог

Question

Я ищу способ вызывать сценарии оболочки из python и записывать их stdout и stderr в файл, используя ведение журнала. Вот мой код:

import logging
import tempfile
import shlex
import os

def run_shell_command(command_line):
    command_line_args = shlex.split(command_line)

    logging.info('Subprocess: \"' + command_line + '\"')

    process_succeeded = True
    try:
        process_output_filename = tempfile.mktemp(suffix = 'subprocess_tmp_file_')
        process_output = open(process_output_filename, 'w')

        command_line_process = subprocess.Popen(command_line_args,\
                                                stdout = process_output,\
                                                stderr = process_output)
        command_line_process.wait()
        process_output.close()

        process_output = open(process_output_filename, 'r')
        log_subprocess_output(process_output)
        process_output.close()

        os.remove(process_output_filename)
    except:
        exception = sys.exc_info()[1]
        logging.info('Exception occured: ' + str(exception))
        process_succeeded = False

    if process_succeeded:
        logging.info('Subprocess finished')
    else:
        logging.info('Subprocess failed')

    return process_succeeded

И я уверен, что есть способ сделать это без создания временного файла для хранения результатов процесса. Есть идеи?

Answer 1

Я уверен, что есть способ сделать это без создания временного файла для хранения результатов процесса

Вы просто должны проверить документацию Popen в частности о stdout а также stderr:

stdin, stdout а также stderr укажите обработчики стандартного ввода, стандартного вывода и стандартных файлов ошибок исполняемой программы соответственно. Допустимые значения PIPE существующий файловый дескриптор (положительное целое число), существующий файловый объект и None, PIPE указывает, что новый канал для дочернего объекта должен быть создан. С настройками по умолчанию None перенаправление не произойдет; дескрипторы файла ребенка будут унаследованы от родителя. Дополнительно, stderr может быть STDOUT что указывает на то, что stderr данные из дочернего процесса должны быть записаны в тот же дескриптор файла, что и для stdout,

Таким образом, вы можете увидеть, что вы можете использовать либо файловый объект, либо PIPE значение. Это позволяет вам использовать communicate() метод для получения вывода:

from StringIO import StringIO
process = subprocess.Popen(arguments, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
output, error = process.communicate()
log_subprocess_output(StringIO(output))

---

Я бы переписал ваш код как:

import shlex
import logging
import subprocess
from StringIO import StringIO

def run_shell_command(command_line):
    command_line_args = shlex.split(command_line)

    logging.info('Subprocess: "' + command_line + '"')

    try:
        command_line_process = subprocess.Popen(
            command_line_args,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
        )

        process_output, _ =  command_line_process.communicate()

        # process_output is now a string, not a file,
        # you may want to do:
        # process_output = StringIO(process_output)
        log_subprocess_output(process_output)
    except (OSError, CalledProcessError) as exception:
        logging.info('Exception occured: ' + str(exception))
        logging.info('Subprocess failed')
        return False
    else:
        # no exception was raised
        logging.info('Subprocess finished')

    return True

Answer 2

Вы можете попытаться передать канал напрямую без буферизации всего вывода подпроцесса в памяти:

from subprocess import Popen, PIPE, STDOUT

process = Popen(command_line_args, stdout=PIPE, stderr=STDOUT)
with process.stdout:
    log_subprocess_output(process.stdout)
exitcode = process.wait() # 0 means success

где log_subprocess_output() может выглядеть так:

def log_subprocess_output(pipe):
    for line in iter(pipe.readline, b''): # b'\n'-separated lines
        logging.info('got line from subprocess: %r', line)

Answer 3

Я пытался добиться того же на check_call а также check_ouput. Я обнаружил, что это решение работает.

import logging
import threading
import os
import subprocess

logging.basicConfig(format='%(levelname)s:%(message)s', level=logging.INFO)

class LogPipe(threading.Thread):

    def __init__(self, level):
        """Setup the object with a logger and a loglevel
        and start the thread
        """
        threading.Thread.__init__(self)
        self.daemon = False
        self.level = level
        self.fdRead, self.fdWrite = os.pipe()
        self.pipeReader = os.fdopen(self.fdRead)
        self.start()

    def fileno(self):
        """Return the write file descriptor of the pipe"""
        return self.fdWrite

    def run(self):
        """Run the thread, logging everything."""
        for line in iter(self.pipeReader.readline, ''):
            logging.log(self.level, line.strip('\n'))

        self.pipeReader.close()

    def close(self):
        """Close the write end of the pipe."""
        os.close(self.fdWrite)

   def write(self):
       """If your code has something like sys.stdout.write"""
       logging.log(self.level, message)

   def flush(self):
       """If you code has something like this sys.stdout.flush"""
       pass

После его реализации я выполнил следующие шаги:

try:
    # It works on multiple handlers as well
    logging.basicConfig(handlers=[logging.FileHandler(log_file), logging.StreamHandler()])
    sys.stdout = LogPipe(logging.INFO)
    sys.stderr = LogPipe(logging.ERROR)
...
    subprocess.check_call(subprocess_cmd, stdout=sys.stdout, stderr=sys.stderr)
    export_output = subprocess.check_output(subprocess_cmd, stderr=sys.stderr)
...
finally:
    sys.stdout.close()
    sys.stderr.close()
    # It is neccessary to close the file handlers properly.
    sys.stdout = sys.__stdout__
    sys.stderr = sys.__stderr__
    logging.shutdown()
    os.remove(log_file)

Answer 4

Это сработало для меня:

      from subprocess import Popen, PIPE, STDOUT

command = f"shell command with arguments"
process = Popen(command, shell=True, stdout=PIPE, stderr=STDOUT)

with process.stdout:
    for line in iter(process.stdout.readline, b''):
        print(line.decode("utf-8").strip())

С обработкой исключений:

      from subprocess import Popen, PIPE, STDOUT, CalledProcessError

command = f"shell command with arguments"
process = Popen(command, shell=True, stdout=PIPE, stderr=STDOUT)

with process.stdout:
    try:
        for line in iter(process.stdout.readline, b''):
            print(line.decode("utf-8").strip())
            
    except CalledProcessError as e:
        print(f"{str(e)}")

Answer 5

Обработка потока вывода команды в реальном времени может быть достигнута путем итерацииstdoutкак subprocess.Popenбежит.

Эта реализация:

• использует оператор with, так что стандартные файловые дескрипторы закрываются и процесс ожидает
• передает аргументы ключевого слова в конструктор подпроцесса
• по умолчаниюtext=Trueдля автоматического декодирования байтовых строк в строки
• поднимает вопросCalledProcessErrorв случае неудачи, еслиcheck=Trueнравитьсяделает
• возвращаетCompletedProcessв случае успеха, как subprocess.runделает
• перенаправляет stderr на stdout (чтобы обрабатывать их отдельно, см. мой многопоточный ответ )

      import logging
from subprocess import PIPE, STDOUT, CalledProcessError, CompletedProcess, Popen


def stream_command(
    args,
    *,
    stdout_handler=logging.info,
    check=True,
    text=True,
    stdout=PIPE,
    stderr=STDOUT,
    **kwargs,
):
    """Mimic subprocess.run, while processing the command output in real time."""
    with Popen(args, text=text, stdout=stdout, stderr=stderr, **kwargs) as process:
        for line in process.stdout:
            stdout_handler(line[:-1])
    retcode = process.poll()
    if check and retcode:
        raise CalledProcessError(retcode, process.args)
    return CompletedProcess(process.args, retcode)

Регистрация в файле становится такой же простой, как и настройка. logging:

      logging.basicConfig(
    level=logging.INFO,
    filename="./capture.log",
    filemode="w",
    encoding="utf-8",
)
logging.info("test from python")
stream_command(["echo", "test from subprocess"])

С полученным файлом:

      $ cat ./capture.log
INFO:root:test from python
INFO:root:test from subprocess

Поведение можно настроить по своему усмотрению (printвместоlogging.infoили и то, и другое и т. д.):

      stream_command(["echo", "test"])
# INFO:root:test
stream_command("cat ./nonexist", shell=True, check=False)
# ERROR:root:cat: ./nonexist: No such file or directory
stream_command(["echo", "test"], stdout_handler=print)
# test
stdout_lines = []
def handler(line):
    print(line)
    logging.info(line)
    stdout_lines.append(line)
stream_command(["echo", "test"], stdout_handler=handler)
print(stdout_lines)
# test
# INFO:root:test
# ['test']