捕获和打印标准输出和标准错误

Question

我正在尝试捕获并继续打印 subprocess.run 的 stdout/stderr。基本上是在找一个内部的tee函数

目标（交互式python模式）：

>>> import subprocess
>>> p = subprocess.run(['echo', 'random text'], stdout=(subprocess.PIPE, subprocess.STDOUT), stderr=(subprocess.PIPE, subprocess.STDERR))
random text
>>> out, err = p.communicate()
>>> print(out)
'random text'
>>> print(err)
''

上面的代码不起作用，但它显示了我想要完成的事情。关于如何确保 stdout 和 stderr 正常打印但也被捕获的任何想法？

Answer 1

我无法使用 subprocess.run 来捕获和打印输出，但我最终使用 popen 来完成此操作。 Popen 不允许超时，因此我必须为我们提供一个信号处理程序来促进这一点。

def tee(command: Sequence[str], *args, **kwargs):
    import subprocess
    import select
    from signal import alarm, signal, SIGALRM, SIGKILL
    stdout2 = ''
    stderr2 = ''

    # timeout configuration
    class TimeoutAlarm(Exception):
        pass

    def timeout_alarm_handler(signum, frame):
        raise TimeoutAlarm

    # capture timeout if specified in kwargs
    timeout = kwargs.pop('timeout') if kwargs.get('timeout') else None  # type: # Optional[int]
    if timeout:
        assert isinstance(timeout, int), "Signal handler only support integers"
        signal(SIGALRM, timeout_alarm_handler)
        alarm(int(timeout))

    # Configure stdout and stderr if specified
    _ = kwargs.pop('stdout') if kwargs.get('stdout') else None
    _ = kwargs.pop('stderr') if kwargs.get('stderr') else None

    # remove universal_newlines as we always use universal_newlines
    _ = kwargs.pop('universal_newlines') if kwargs.get('universal_newlines') is not None else None

    # Execute a child program in a new process
    with subprocess.Popen(command, *args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True, **kwargs) as process:
        try:  # TimeoutAlarm try block
            stdout_fn = process.stdout.fileno()
            stderr_fn = process.stderr.fileno()

            # Continue reading stdout and stderr as long as the process has not terminated
            while process.poll() is None:
                # blocks until one file descriptor is ready
                fds, _, _ = select.select([stdout_fn, stderr_fn], [], [])

                # Capture the stdout
                if stdout_fn in fds:
                    new_so = process.stdout.readline()
                    sys.stdout.write(new_so)
                    stdout2 += new_so

                # Capture the stderr
                if stderr_fn in fds:
                    new_se = process.stderr.readline()
                    sys.stderr.write(new_se)
                    stderr2 += new_se
            return_code = process.returncode
        except TimeoutAlarm:
            os.kill(process.pid, SIGKILL)
            return_code = process.returncode
            raise subprocess.TimeoutExpired(cmd=command, timeout=timeout)

    if return_code == 127:
        raise OSError('Command not found')
    elif return_code == 126:
        raise OSError('Command invoked cannot execute')

    return subprocess.CompletedProcess(args, return_code, stdout2, stderr2)

感谢 Andrew Hoos：https://gist.github.com/AndrewHoos/9f03c74988469b517a7a