包装子流程'标准输出/标准错误

Question

我想捕获并显示通过 Python 子进程调用的进程的输出。

我以为我可以将我的类文件对象作为命名参数 stdout 和 stderr 传递，

我可以看到它访问 fileno 属性 - 所以它正在对对象执行某些操作。 但是，write() 方法永远不会被调用。我的方法是完全错误还是我只是错过了一些东西？

class Process(object):
    class StreamWrapper(object):
        def __init__(self, stream):
            self._stream = stream
            self._buffer = []
        def _print(self, msg):
            print repr(self), msg
        def __getattr__(self, name):
            if not name in ['fileno']:
                self._print("# Redirecting: %s" % name)
            return getattr(self._stream, name)
        def write(self, data):
            print "###########"
            self._buffer.append(data)
            self._stream.write(data)
            self._stream.flush()
        def getBuffer(self):
            return self._buffer[:]
    def __init__(self, *args, **kwargs):
        print ">> Running `%s`" % " ".join(args[0])
        self._stdout = self.StreamWrapper(sys.stdout)
        self._stderr = self.StreamWrapper(sys.stderr)
        kwargs.setdefault('stdout', self._stdout)
        kwargs.setdefault('stderr', self._stderr)
        self._process = subprocess.Popen(*args, **kwargs)
        self._process.communicate()

更新：

我还想使用 ANSI 控制字符来移动光标并覆盖以前的输出内容。我不知道这是否是正确的术语，但这里有一个例子来说明我的意思：我正在尝试自动化一些 GIT 的东西，并且它们的进度可以自我更新，而无需每次都写入新行。

更新 2

对我来说，立即显示子流程的输出很重要。我尝试使用 subprocess.PIPE 捕获输出并手动显示它，但我只能在进程完成后让它显示输出。但是，我想实时查看输出。

Answer 1

类似文件还不够接近。它必须是具有实际文件描述符的实际文件。使用 subprocess 对管道的支持并根据需要从中读取。

Answer 2

这是来自 Chromium CI 产品 Catapult 的函数（BSD-3-Clause执照）。

它是非阻塞的，并且会耗尽输出。也就是说，处理大多数答案中存在的问题。

我还会研究 os.set_blocking() 而不是这里的 fcntl 。

os.set_blocking() 是在 Python 3.5 中引入的，从 Python 3.12 开始支持 Windows 管道。因此，从 3.12 开始，代码也可以移植到 Windows。

def _IterProcessStdoutFcntl(process,
                            iter_timeout=None,
                            timeout=None,
                            buffer_size=4096,
                            poll_interval=1):
  """An fcntl-based implementation of _IterProcessStdout."""
  # pylint: disable=too-many-nested-blocks
  import fcntl
  try:
    # Enable non-blocking reads from the child's stdout.
    child_fd = process.stdout.fileno()
    fl = fcntl.fcntl(child_fd, fcntl.F_GETFL)
    fcntl.fcntl(child_fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)

    end_time = (time.time() + timeout) if timeout else None
    iter_end_time = (time.time() + iter_timeout) if iter_timeout else None

    while True:
      if end_time and time.time() > end_time:
        raise TimeoutError()
      if iter_end_time and time.time() > iter_end_time:
        yield None
        iter_end_time = time.time() + iter_timeout

      if iter_end_time:
        iter_aware_poll_interval = min(poll_interval,
                                       max(0, iter_end_time - time.time()))
      else:
        iter_aware_poll_interval = poll_interval

      read_fds, _, _ = select.select([child_fd], [], [],
                                     iter_aware_poll_interval)
      if child_fd in read_fds:
        data = _read_and_decode(child_fd, buffer_size)
        if not data:
          break
        yield data

      if process.poll() is not None:
        # If process is closed, keep checking for output data (because of timing
        # issues).
        while True:
          read_fds, _, _ = select.select([child_fd], [], [],
                                         iter_aware_poll_interval)
          if child_fd in read_fds:
            data = _read_and_decode(child_fd, buffer_size)
            if data:
              yield data
              continue
          break
        break
  finally:
    try:
      if process.returncode is None:
        # Make sure the process doesn't stick around if we fail with an
        # exception.
        process.kill()
    except OSError:
      pass
    process.wait()

Answer 3

你在课堂上上课有什么原因吗？ stdout 和 stderr 可以采用任何文件，例如 line 。因此，只需传递一个打开的文件类型或 stringIO 就足以修改流

import sys
sys.stdout = open('test.txt','w')
print "Testing!"
sys.stdout.write('\nhehehe')
sys.stdout = sys.__stdout__
sys.exit(0)

Answer 4

进程的 stdin、stdout 和 stderr 需要是真实的文件描述符。 （这实际上不是 Python 施加的限制，而是管道在操作系统级别上的工作方式。）因此您将需要不同的解决方案。

如果您想实时跟踪 stdout 和 stderr，您将需要异步 I/O 或线程。

• 异步 I/O：使用标准同步（=阻塞）I/O，对其中一个流的读取可能会阻塞，从而不允许实时访问另一个流。如果您使用的是 Unix，则可以使用非阻塞 I/O，如 这个答案。然而，在 Windows 上，您将无法采用这种方法。有关 Python 中异步 I/O 的更多信息以及一些替代方案，请参见 此视频。

• 线程：处理此问题的另一种常见方法是为要实时读取的每个文件描述符创建一个线程。线程只处理它们所分配的文件描述符，因此阻塞 I/O 不会造成损害。

Answer 5

请查看此处。

p = subprocess.Popen(cmd,
                 shell=True,
                 bufsize=64,
                 stdin=subprocess.PIPE,
                 stderr=subprocess.PIPE,
                 stdout=subprocess.PIPE)