运行子进程并将输出打印到日志记录

Question

我正在寻找从 python 调用 shell 脚本并使用日志记录将它们的 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

您可以尝试直接传递管道而不将整个子进程输出缓冲到内存中：

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 2

我确信有办法做到这一点而无需创建临时 存储过程输出的文件

您只需检查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 3

我试图在check_call 和check_ouput 上实现相同的目标。我发现this solution 正在工作。

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)}")