gpt4 book ai didi

python - 如何在 Python 中无缝包装 Bash shell IO

转载 作者:太空宇宙 更新时间:2023-11-03 19:15:57 25 4
gpt4 key购买 nike

如何将 bash shell session 包装在 Python 脚本中,以便 Python 可以将 stdout 和 stderr 存储到数据库,并偶尔写入 stdin?

我尝试使用 subprocess 和类似 tee 的 Python 类来重定向 IO,但它似乎使用 fileno 完全绕过 Python。

shell.py:

import os
import sys
from StringIO import StringIO
from subprocess import Popen, PIPE

class TeeFile(StringIO):
def __init__(self, file, auto_flush=False):
#super(TeeFile, self).__init__()
StringIO.__init__(self)
self.file = file
self.auto_flush = auto_flush
self.length = 0

def write(self, s):
print 'writing' # This is never called!!!
self.length += len(s)
self.file.write(s)
#super(TeeFile, self).write(s)
StringIO.write(self, s)
if self.auto_flush:
self.file.flush()

def flush(self):
self.file.flush()
StringIO.flush(self)

def fileno(self):
return self.file.fileno()

cmd = ' '.join(sys.argv[1:])
stderr = TeeFile(sys.stderr, True)
stdout = TeeFile(sys.stdout, True)

p = Popen(cmd, shell=True, stdin=PIPE, stdout=stdout, stderr=stderr, close_fds=True)

例如运行 python shell.py ping google.com 会运行正确的命令并显示输出,但 Python 永远看不到标准输出。

最佳答案

#!/usr/bin/env python

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

from twisted.internet import protocol
from twisted.internet import reactor
import re

class MyPP(protocol.ProcessProtocol):
def __init__(self, verses):
self.verses = verses
self.data = ""
def connectionMade(self):
print "connectionMade!"
for i in range(self.verses):
self.transport.write("Aleph-null bottles of beer on the wall,\n" +
"Aleph-null bottles of beer,\n" +
"Take one down and pass it around,\n" +
"Aleph-null bottles of beer on the wall.\n")
self.transport.closeStdin() # tell them we're done
def outReceived(self, data):
print "outReceived! with %d bytes!" % len(data)
self.data = self.data + data
def errReceived(self, data):
print "errReceived! with %d bytes!" % len(data)
def inConnectionLost(self):
print "inConnectionLost! stdin is closed! (we probably did it)"
def outConnectionLost(self):
print "outConnectionLost! The child closed their stdout!"
# now is the time to examine what they wrote
#print "I saw them write:", self.data
(dummy, lines, words, chars, file) = re.split(r'\s+', self.data)
print "I saw %s lines" % lines
def errConnectionLost(self):
print "errConnectionLost! The child closed their stderr."
def processExited(self, reason):
print "processExited, status %d" % (reason.value.exitCode,)
def processEnded(self, reason):
print "processEnded, status %d" % (reason.value.exitCode,)
print "quitting"
reactor.stop()

pp = MyPP(10)
reactor.spawnProcess(pp, "wc", ["wc"], {})
reactor.run()

这就是将命令 IO 作为协议(protocol)处理的 Twisted 方式。顺便说一句,你的 StringIO 脚本太复杂了。而是检查 Popen.communicate() 方法。请注意,stdin/stdout 是文件描述符,需要并行读取,因为如果输出较长,它们的缓冲区将溢出。如果您想通过这种方式传输大量数据,请使用 Twisted 方式,或者万一 Popen 方式触发单独的线程来读取 stdout 可能会按行并立即将它们放入数据库。 Twisted howto on process protocol.

关于python - 如何在 Python 中无缝包装 Bash shell IO,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/11297579/

25 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com