python-3.x - 如何传递或包装打印命令（stdout），以便 print 每次调用也调用一个函数？

Question

我正在尝试自动化一项长期运行的作业，并且我希望能够将所有控制台输出上传到另一个日志，例如 CloudWatch Logs。在大多数情况下，这可以通过制作和使用自定义函数而不是打印来完成。但是机器学习中有一些功能，例如 Model.summary() 或进度条，同时训练它们自己输出到标准输出。

我可以通过内部控制台日志在最后获得所有控制台输出。但我需要的是实时上传标准输出，因为它由任何人调用。这样人们就可以通过查看 Cloudwatch 上的日志来检查进度，而不必登录机器并检查内部控制台日志。

基本上我需要的是：

From: call_to_stdout -> Console(and probably other stuff)
To:   call_to_stdout -> uploadLog() -> Console(and probably other stuff)

我需要的伪代码

class stdout_PassThru:
   def __init__(self, in_old_stdout):
      self.old_stdout = in_old_stdout

   def write(self, msg):
      self.old_stdout.write(msg)
      uploadLogToCloudwatch(msg)

def uploadLogToCloudwatch(msg):
   # Botocore stuff to upload to Cloudwatch

myPassThru = stdout_PassThru(sys.stdout)
sys.stdout = myPassThru

我试过用谷歌搜索这个，但我得到的最好的是 stringIO 东西，我可以在其中捕获标准输出，但在我调用的函数结束并且我可以再次插入代码之前，我无法对它做任何事情。我想在每次使用标准输出时运行我的上传日志代码。

这甚至可能吗？谢谢，麻烦您了。

编辑：有人建议重定向/输出到文件。问题在于，在输出内容时，它只是流式传输/写入文件。我需要调用一个函数，该函数在每次调用标准输出时都有效，这不是流。如果 stdout 每次刷新时都会输出，那么调用函数也会很好。

标签： python-3.xstdout

Answer 1

我解决了我的问题。有点隐藏在其他一些答案中。我对这个解决方案的最初问题是，当它在 Jupyter Notebook 中进行测试时， sys.stdout = myClass(sys.stdout) 导致 Jupyter ......等待？不确定，但它永远不会完成对段落的处理。

但是当我将它放入 python 文件并使用 python test.py 运行时，它运行得非常完美，并且符合预期。

从某种意义上说，这使我可以通过调用 print，同时在每次调用 print 时执行我自己的函数。

def addLog(message):
    # my boto function to upload Cloudwatch logs

class sendToLog:
    def __init__(self, stream):
        self.stream = stream
    def write(self, o):
        self.stream.write(o)
        addLog(o)
        self.stream.flush()
    def writelines(self, o):
        self.stream.writelines(o)
        addLog(o)
        self.stream.flush()
    def __getattr__(self, attr):
        return getattr(self.stream, attr)

sys.stdout = sendToLog(sys.stdout)