如何将 sys.stdout 复制到日志文件?

编辑: 由于看起来要么没有解决方案,要么我在做一些没有人知道的非标准的事情——我将修改我的问题,还要问: 当一个 python 应用程序进行大量的系统调用时,什么是完成日志记录的最佳方法?

我的应用程序有两种模式。在交互模式下,我希望所有输出都转到屏幕和日志文件,包括任何系统调用的输出。在守护进程模式下,所有输出都会转到日志。守护进程模式使用 os.dup2()工作得很好。如果不修改每个系统调用,我无法找到一种方法将所有输出“ T”到交互模式的日志中。


换句话说,我需要命令行‘ tee’对于 Python 应用程序 包括系统调用输出生成的任何输出的功能。

澄清:

为了重定向所有的输出,我做了这样的事情,它工作得很好:

# open our log file
so = se = open("%s.log" % self.name, 'w', 0)


# re-open stdout without buffering
sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)


# redirect stdout and stderr to the log file opened above
os.dup2(so.fileno(), sys.stdout.fileno())
os.dup2(se.fileno(), sys.stderr.fileno())

这样做的好处是不需要来自代码其余部分的特殊 print 调用。代码还运行一些 shell 命令,所以不必分别处理它们的每个输出。

简单地说,我想做同样的事情,除了 复制而不是重定向。

首先,我认为简单地逆转 dup2应该是可行的。为什么不行呢? 这是我的测试:

import os, sys


### my broken solution:
so = se = open("a.log", 'w', 0)
sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)


os.dup2(sys.stdout.fileno(), so.fileno())
os.dup2(sys.stderr.fileno(), se.fileno())
###


print("foo bar")


os.spawnve("P_WAIT", "/bin/ls", ["/bin/ls"], {})
os.execve("/bin/ls", ["/bin/ls"], os.environ)

文件“ a.log”应该与屏幕上显示的内容相同。

126553 次浏览

print语句将调用分配给 sys.stdout 的任何对象的 write()方法

我会组织一个小班同时给两个地方写信。

import sys


class Logger(object):
def __init__(self):
self.terminal = sys.stdout
self.log = open("log.dat", "a")


def write(self, message):
self.terminal.write(message)
self.log.write(message)


sys.stdout = Logger()

现在,print语句将同时回显到屏幕并附加到您的日志文件:

# prints "1 2" to <stdout> AND log.dat
print "%d %d" % (1,2)

这显然是快速和肮脏的。一些注意事项:

  • 您可能应该参数化日志文件名。
  • 如果需要将 sys.stdout 还原为 <stdout>,则可能需要 不会在程序执行期间记录日志。
  • 您可能希望能够一次写入多个日志文件,或者处理不同的日志级别,等等。

这些都足够简单明了,我很乐意把它们留给读者作为练习。这里的关键见解是,print只调用分配给 sys.stdout的“类似文件的对象”。

我以前也遇到过同样的问题,并且发现这个片段非常有用:

class Tee(object):
def __init__(self, name, mode):
self.file = open(name, mode)
self.stdout = sys.stdout
sys.stdout = self
def __del__(self):
sys.stdout = self.stdout
self.file.close()
def write(self, data):
self.file.write(data)
self.stdout.write(data)
def flush(self):
self.file.flush()

发信人: http://mail.python.org/pipermail/python-list/2007-May/438106.html

您真正需要的是来自标准库的 logging模块。创建一个日志记录器并附加两个处理程序,一个将写入文件,另一个将写入 stdout 或 stderr。

详情请参阅 日志记录到多个目的地

(啊,只要重读一遍你的问题,就会发现这并不完全适用。)

下面是一个使用 Python 日志记录模块的示例程序。这个日志模块自2.3以来已经有了所有版本。在此示例中,可以通过命令行选项配置日志记录。

在非常模式下,它将只记录到一个文件,在正常模式下,它将同时记录到一个文件和控制台。

import os
import sys
import logging
from optparse import OptionParser


def initialize_logging(options):
""" Log information based upon users options"""


logger = logging.getLogger('project')
formatter = logging.Formatter('%(asctime)s %(levelname)s\t%(message)s')
level = logging.__dict__.get(options.loglevel.upper(),logging.DEBUG)
logger.setLevel(level)


# Output logging information to screen
if not options.quiet:
hdlr = logging.StreamHandler(sys.stderr)
hdlr.setFormatter(formatter)
logger.addHandler(hdlr)


# Output logging information to file
logfile = os.path.join(options.logdir, "project.log")
if options.clean and os.path.isfile(logfile):
os.remove(logfile)
hdlr2 = logging.FileHandler(logfile)
hdlr2.setFormatter(formatter)
logger.addHandler(hdlr2)


return logger


def main(argv=None):
if argv is None:
argv = sys.argv[1:]


# Setup command line options
parser = OptionParser("usage: %prog [options]")
parser.add_option("-l", "--logdir", dest="logdir", default=".", help="log DIRECTORY (default ./)")
parser.add_option("-v", "--loglevel", dest="loglevel", default="debug", help="logging level (debug, info, error)")
parser.add_option("-q", "--quiet", action="store_true", dest="quiet", help="do not log to console")
parser.add_option("-c", "--clean", dest="clean", action="store_true", default=False, help="remove old log file")


# Process command line options
(options, args) = parser.parse_args(argv)


# Setup logger format and output locations
logger = initialize_logging(options)


# Examples
logger.error("This is an error message.")
logger.info("This is an info message.")
logger.debug("This is a debug message.")


if __name__ == "__main__":
sys.exit(main())

因为可以从代码中产生外部进程,所以可以使用 tee本身。我不知道有任何 Unix 系统调用可以完全做到 tee所做的事情。

# Note this version was written circa Python 2.6, see below for
# an updated 3.3+-compatible version.
import subprocess, os, sys


# Unbuffer output (this ensures the output is in the correct order)
sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)


tee = subprocess.Popen(["tee", "log.txt"], stdin=subprocess.PIPE)
os.dup2(tee.stdin.fileno(), sys.stdout.fileno())
os.dup2(tee.stdin.fileno(), sys.stderr.fileno())


print "\nstdout"
print >>sys.stderr, "stderr"
os.spawnve("P_WAIT", "/bin/ls", ["/bin/ls"], {})
os.execve("/bin/ls", ["/bin/ls"], os.environ)

您还可以使用 多重处理包模拟 tee(或者如果使用 Python 2.5或更早版本,则使用 正在处理)。

更新

下面是一个 Python 3.3 + 兼容的版本:

import subprocess, os, sys


tee = subprocess.Popen(["tee", "log.txt"], stdin=subprocess.PIPE)
# Cause tee's stdin to get a copy of our stdin/stdout (as well as that
# of any child processes we spawn)
os.dup2(tee.stdin.fileno(), sys.stdout.fileno())
os.dup2(tee.stdin.fileno(), sys.stderr.fileno())


# The flush flag is needed to guarantee these lines are written before
# the two spawned /bin/ls processes emit any output
print("\nstdout", flush=True)
print("stderr", file=sys.stderr, flush=True)


# These child processes' stdin/stdout are
os.spawnve("P_WAIT", "/bin/ls", ["/bin/ls"], {})
os.execve("/bin/ls", ["/bin/ls"], os.environ)

正如其他地方所描述的,也许最好的解决方案是直接使用日志记录模块:

import logging


logging.basicConfig(level=logging.DEBUG, filename='mylog.log')
logging.info('this should to write to the log file')

但是,在某些(罕见的)情况下,真的很想会重定向 stdout。我在扩展 django 的 runserver 命令时遇到了这种情况,该命令使用 print: 我不想破解 django 源代码,但是需要 print 语句来转到一个文件。

这是一种使用日志模块将 stdout 和 stderr 从 shell 重定向的方法:

import logging, sys


class LogFile(object):
"""File-like object to log text using the `logging` module."""


def __init__(self, name=None):
self.logger = logging.getLogger(name)


def write(self, msg, level=logging.INFO):
self.logger.log(level, msg)


def flush(self):
for handler in self.logger.handlers:
handler.flush()


logging.basicConfig(level=logging.DEBUG, filename='mylog.log')


# Redirect stdout and stderr
sys.stdout = LogFile('stdout')
sys.stderr = LogFile('stderr')


print 'this should to write to the log file'

只有在确实不能直接使用日志记录模块的情况下,才应该使用这个 LogFile 实现。

另一种使用日志模块的解决方案:

import logging
import sys


log = logging.getLogger('stdxxx')


class StreamLogger(object):


def __init__(self, stream, prefix=''):
self.stream = stream
self.prefix = prefix
self.data = ''


def write(self, data):
self.stream.write(data)
self.stream.flush()


self.data += data
tmp = str(self.data)
if '\x0a' in tmp or '\x0d' in tmp:
tmp = tmp.rstrip('\x0a\x0d')
log.info('%s%s' % (self.prefix, tmp))
self.data = ''




logging.basicConfig(level=logging.INFO,
filename='text.log',
filemode='a')


sys.stdout = StreamLogger(sys.stdout, '[stdout] ')


print 'test for stdout'

我用 Python 编写了一个 tee()实现,它应该可以在大多数情况下工作,而且它也可以在 Windows 上工作。

Https://github.com/pycontribs/tendo

另外,如果需要,还可以将它与 Python 中的 logging模块结合使用。

我正在编写一个运行 cmd-line 脚本的脚本。(因为在某些情况下,Linux 命令没有可行的替代品——例如 rsync。)

我真正想要的是在任何可能的情况下使用缺省的 python 日志记录机制,但是仍然可以在出现意料之外的错误时捕获任何错误。

这个代码似乎很管用。它可能不是特别优雅或高效(尽管它没有使用 string + = string,所以至少它没有特定的潜在瓶子- 我把它贴出来,以防它给别人任何有用的想法。

import logging
import os, sys
import datetime


# Get name of module, use as application name
try:
ME=os.path.split(__file__)[-1].split('.')[0]
except:
ME='pyExec_'


LOG_IDENTIFIER="uuu___( o O )___uuu "
LOG_IDR_LENGTH=len(LOG_IDENTIFIER)


class PyExec(object):


# Use this to capture all possible error / output to log
class SuperTee(object):
# Original reference: http://mail.python.org/pipermail/python-list/2007-May/442737.html
def __init__(self, name, mode):
self.fl = open(name, mode)
self.fl.write('\n')
self.stdout = sys.stdout
self.stdout.write('\n')
self.stderr = sys.stderr


sys.stdout = self
sys.stderr = self


def __del__(self):
self.fl.write('\n')
self.fl.flush()
sys.stderr = self.stderr
sys.stdout = self.stdout
self.fl.close()


def write(self, data):
# If the data to write includes the log identifier prefix, then it is already formatted
if data[0:LOG_IDR_LENGTH]==LOG_IDENTIFIER:
self.fl.write("%s\n" % data[LOG_IDR_LENGTH:])
self.stdout.write(data[LOG_IDR_LENGTH:])


# Otherwise, we can give it a timestamp
else:


timestamp=str(datetime.datetime.now())
if 'Traceback' == data[0:9]:
data='%s: %s' % (timestamp, data)
self.fl.write(data)
else:
self.fl.write(data)


self.stdout.write(data)




def __init__(self, aName, aCmd, logFileName='', outFileName=''):


# Using name for 'logger' (context?), which is separate from the module or the function
baseFormatter=logging.Formatter("%(asctime)s \t %(levelname)s \t %(name)s:%(module)s:%(lineno)d \t %(message)s")
errorFormatter=logging.Formatter(LOG_IDENTIFIER + "%(asctime)s \t %(levelname)s \t %(name)s:%(module)s:%(lineno)d \t %(message)s")


if logFileName:
# open passed filename as append
fl=logging.FileHandler("%s.log" % aName)
else:
# otherwise, use log filename as a one-time use file
fl=logging.FileHandler("%s.log" % aName, 'w')


fl.setLevel(logging.DEBUG)
fl.setFormatter(baseFormatter)


# This will capture stdout and CRITICAL and beyond errors


if outFileName:
teeFile=PyExec.SuperTee("%s_out.log" % aName)
else:
teeFile=PyExec.SuperTee("%s_out.log" % aName, 'w')


fl_out=logging.StreamHandler( teeFile )
fl_out.setLevel(logging.CRITICAL)
fl_out.setFormatter(errorFormatter)


# Set up logging
self.log=logging.getLogger('pyExec_main')
log=self.log


log.addHandler(fl)
log.addHandler(fl_out)


print "Test print statement."


log.setLevel(logging.DEBUG)


log.info("Starting %s", ME)
log.critical("Critical.")


# Caught exception
try:
raise Exception('Exception test.')
except Exception,e:
log.exception(str(e))


# Uncaught exception
a=2/0




PyExec('test_pyExec',None)

显然,如果您不像我一样喜欢奇思妙想,那么可以用另一个字符串替换 LOG _ IDENTIFIER,这个字符串是您不希望看到有人向日志中写入的。

以上的答案似乎都没有真正回答提出的问题。我知道这是一个老线索,但我认为这个问题比每个人都要简单得多:

class tee_err(object):


def __init__(self):
self.errout = sys.stderr


sys.stderr = self


self.log = 'logfile.log'
log = open(self.log,'w')
log.close()


def write(self, line):


log = open(self.log,'a')
log.write(line)
log.close()


self.errout.write(line)

现在,这将对普通的 sys.stderr 处理程序和文件重复所有操作。

下面是另一种解决方案,它比其他解决方案更通用——它支持将输出(写入 sys.stdout)分割到任意数量的类文件对象。没有包含 __stdout__本身的要求。

import sys


class multifile(object):
def __init__(self, files):
self._files = files
def __getattr__(self, attr, *args):
return self._wrap(attr, *args)
def _wrap(self, attr, *args):
def g(*a, **kw):
for f in self._files:
res = getattr(f, attr, *args)(*a, **kw)
return res
return g


# for a tee-like behavior, use like this:
sys.stdout = multifile([ sys.stdout, open('myfile.txt', 'w') ])


# all these forms work:
print 'abc'
print >>sys.stdout, 'line2'
sys.stdout.write('line3\n')

注意: 这是一个概念验证。这里的实现并不完整,因为它只包装类似文件的对象(例如 write)的 方法,而忽略了成员/properties/setattr 等。然而,对于大多数人来说,它可能已经足够好了。

我喜欢它的地方,除了它的通用性,就是它不会直接调用 writeflushos.dup2等等。

完成约翰 T 答案: https://stackoverflow.com/a/616686/395687

我添加了 __enter____exit__方法,使用它作为具有 with关键字的上下文管理器,它提供了这段代码

class Tee(object):
def __init__(self, name, mode):
self.file = open(name, mode)
self.stdout = sys.stdout
sys.stdout = self


def __del__(self):
sys.stdout = self.stdout
self.file.close()


def write(self, data):
self.file.write(data)
self.stdout.write(data)


def __enter__(self):
pass


def __exit__(self, _type, _value, _traceback):
pass

然后它可以被用作

with Tee('outfile.log', 'w'):
print('I am written to both stdout and outfile.log')

我知道这个问题已经得到了反复的回答,但是对于这个问题,我从 John T’s的答案中选取了主要的答案,并对其进行了修改,使其包含了建议的同花顺,并遵循了其链接的修订版本。我还添加了 Cladmi 的答案中提到的输入和输出,以便与 with 语句一起使用。此外,文件提到了使用 os.fsync()刷新文件,所以我也添加了这一点。我不知道你是否需要 真的,但它的存在。

import sys, os


class Logger(object):
"Lumberjack class - duplicates sys.stdout to a log file and it's okay"
#source: https://stackoverflow.com/q/616645
def __init__(self, filename="Red.Wood", mode="a", buff=0):
self.stdout = sys.stdout
self.file = open(filename, mode, buff)
sys.stdout = self


def __del__(self):
self.close()


def __enter__(self):
pass


def __exit__(self, *args):
self.close()


def write(self, message):
self.stdout.write(message)
self.file.write(message)


def flush(self):
self.stdout.flush()
self.file.flush()
os.fsync(self.file.fileno())


def close(self):
if self.stdout != None:
sys.stdout = self.stdout
self.stdout = None


if self.file != None:
self.file.close()
self.file = None

然后你就可以使用它了

with Logger('My_best_girlie_by_my.side'):
print("we'd sing sing sing")

或者

Log=Logger('Sleeps_all.night')
print('works all day')
Log.close()

根据@user5359531在@John T’s 回答下的评论中提出的请求,下面是这个答案中链接讨论的修订版本的参考文章:

Issue of redirecting the stdout to both file and screen
Gabriel Genellina gagsl-py2 at yahoo.com.ar
Mon May 28 12:45:51 CEST 2007


Previous message: Issue of redirecting the stdout to both file and screen
Next message: Formal interfaces with Python
Messages sorted by: [ date ] [ thread ] [ subject ] [ author ]


En Mon, 28 May 2007 06:17:39 -0300, 人言落日是天涯,望极天涯不见家
<kelvin.you at gmail.com> escribió:


> I wanna print the log to both the screen and file, so I simulatered a
> 'tee'
>
> class Tee(file):
>
>     def __init__(self, name, mode):
>         file.__init__(self, name, mode)
>         self.stdout = sys.stdout
>         sys.stdout = self
>
>     def __del__(self):
>         sys.stdout = self.stdout
>         self.close()
>
>     def write(self, data):
>         file.write(self, data)
>         self.stdout.write(data)
>
> Tee('logfile', 'w')
> print >>sys.stdout, 'abcdefg'
>
> I found that it only output to the file, nothing to screen. Why?
> It seems the 'write' function was not called when I *print* something.


You create a Tee instance and it is immediately garbage collected. I'd
restore sys.stdout on Tee.close, not __del__ (you forgot to call the
inherited __del__ method, btw).
Mmm, doesn't work. I think there is an optimization somewhere: if it looks
like a real file object, it uses the original file write method, not yours.
The trick would be to use an object that does NOT inherit from file:


import sys
class TeeNoFile(object):
def __init__(self, name, mode):
self.file = open(name, mode)
self.stdout = sys.stdout
sys.stdout = self
def close(self):
if self.stdout is not None:
sys.stdout = self.stdout
self.stdout = None
if self.file is not None:
self.file.close()
self.file = None
def write(self, data):
self.file.write(data)
self.stdout.write(data)
def flush(self):
self.file.flush()
self.stdout.flush()
def __del__(self):
self.close()


tee=TeeNoFile('logfile', 'w')
print 'abcdefg'
print 'another line'
tee.close()
print 'screen only'
del tee # should do nothing


--
Gabriel Genellina

我为 sys.stderr编写了一个完整的替代程序,并将重命名为 stderr的代码复制到 stdout,以使其也可用于替代 sys.stdout

为此,我创建了与当前 stderrstdout相同的对象类型,并将所有方法转发到原始系统 stderrstdout:

import os
import sys
import logging


class StdErrReplament(object):
"""
How to redirect stdout and stderr to logger in Python
https://stackoverflow.com/questions/19425736/how-to-redirect-stdout-and-stderr-to-logger-in-python


Set a Read-Only Attribute in Python?
https://stackoverflow.com/questions/24497316/set-a-read-only-attribute-in-python
"""
is_active = False


@classmethod
def lock(cls, logger):
"""
Attach this singleton logger to the `sys.stderr` permanently.
"""
global _stderr_singleton
global _stderr_default
global _stderr_default_class_type


# On Sublime Text, `sys.__stderr__` is set to None, because they already replaced `sys.stderr`
# by some `_LogWriter()` class, then just save the current one over there.
if not sys.__stderr__:
sys.__stderr__ = sys.stderr


try:
_stderr_default
_stderr_default_class_type


except NameError:
_stderr_default = sys.stderr
_stderr_default_class_type = type( _stderr_default )


# Recreate the sys.stderr logger when it was reset by `unlock()`
if not cls.is_active:
cls.is_active = True
_stderr_write = _stderr_default.write


logger_call = logger.debug
clean_formatter = logger.clean_formatter


global _sys_stderr_write
global _sys_stderr_write_hidden


if sys.version_info <= (3,2):
logger.file_handler.terminator = '\n'


# Always recreate/override the internal write function used by `_sys_stderr_write`
def _sys_stderr_write_hidden(*args, **kwargs):
"""
Suppress newline in Python logging module
https://stackoverflow.com/questions/7168790/suppress-newline-in-python-logging-module
"""


try:
_stderr_write( *args, **kwargs )
file_handler = logger.file_handler


formatter = file_handler.formatter
terminator = file_handler.terminator


file_handler.formatter = clean_formatter
file_handler.terminator = ""


kwargs['extra'] = {'_duplicated_from_file': True}
logger_call( *args, **kwargs )


file_handler.formatter = formatter
file_handler.terminator = terminator


except Exception:
logger.exception( "Could not write to the file_handler: %s(%s)", file_handler, logger )
cls.unlock()


# Only create one `_sys_stderr_write` function pointer ever
try:
_sys_stderr_write


except NameError:


def _sys_stderr_write(*args, **kwargs):
"""
Hides the actual function pointer. This allow the external function pointer to
be cached while the internal written can be exchanged between the standard
`sys.stderr.write` and our custom wrapper around it.
"""
_sys_stderr_write_hidden( *args, **kwargs )


try:
# Only create one singleton instance ever
_stderr_singleton


except NameError:


class StdErrReplamentHidden(_stderr_default_class_type):
"""
Which special methods bypasses __getattribute__ in Python?
https://stackoverflow.com/questions/12872695/which-special-methods-bypasses-getattribute-in-python
"""


if hasattr( _stderr_default, "__abstractmethods__" ):
__abstractmethods__ = _stderr_default.__abstractmethods__


if hasattr( _stderr_default, "__base__" ):
__base__ = _stderr_default.__base__


if hasattr( _stderr_default, "__bases__" ):
__bases__ = _stderr_default.__bases__


if hasattr( _stderr_default, "__basicsize__" ):
__basicsize__ = _stderr_default.__basicsize__


if hasattr( _stderr_default, "__call__" ):
__call__ = _stderr_default.__call__


if hasattr( _stderr_default, "__class__" ):
__class__ = _stderr_default.__class__


if hasattr( _stderr_default, "__delattr__" ):
__delattr__ = _stderr_default.__delattr__


if hasattr( _stderr_default, "__dict__" ):
__dict__ = _stderr_default.__dict__


if hasattr( _stderr_default, "__dictoffset__" ):
__dictoffset__ = _stderr_default.__dictoffset__


if hasattr( _stderr_default, "__dir__" ):
__dir__ = _stderr_default.__dir__


if hasattr( _stderr_default, "__doc__" ):
__doc__ = _stderr_default.__doc__


if hasattr( _stderr_default, "__eq__" ):
__eq__ = _stderr_default.__eq__


if hasattr( _stderr_default, "__flags__" ):
__flags__ = _stderr_default.__flags__


if hasattr( _stderr_default, "__format__" ):
__format__ = _stderr_default.__format__


if hasattr( _stderr_default, "__ge__" ):
__ge__ = _stderr_default.__ge__


if hasattr( _stderr_default, "__getattribute__" ):
__getattribute__ = _stderr_default.__getattribute__


if hasattr( _stderr_default, "__gt__" ):
__gt__ = _stderr_default.__gt__


if hasattr( _stderr_default, "__hash__" ):
__hash__ = _stderr_default.__hash__


if hasattr( _stderr_default, "__init__" ):
__init__ = _stderr_default.__init__


if hasattr( _stderr_default, "__init_subclass__" ):
__init_subclass__ = _stderr_default.__init_subclass__


if hasattr( _stderr_default, "__instancecheck__" ):
__instancecheck__ = _stderr_default.__instancecheck__


if hasattr( _stderr_default, "__itemsize__" ):
__itemsize__ = _stderr_default.__itemsize__


if hasattr( _stderr_default, "__le__" ):
__le__ = _stderr_default.__le__


if hasattr( _stderr_default, "__lt__" ):
__lt__ = _stderr_default.__lt__


if hasattr( _stderr_default, "__module__" ):
__module__ = _stderr_default.__module__


if hasattr( _stderr_default, "__mro__" ):
__mro__ = _stderr_default.__mro__


if hasattr( _stderr_default, "__name__" ):
__name__ = _stderr_default.__name__


if hasattr( _stderr_default, "__ne__" ):
__ne__ = _stderr_default.__ne__


if hasattr( _stderr_default, "__new__" ):
__new__ = _stderr_default.__new__


if hasattr( _stderr_default, "__prepare__" ):
__prepare__ = _stderr_default.__prepare__


if hasattr( _stderr_default, "__qualname__" ):
__qualname__ = _stderr_default.__qualname__


if hasattr( _stderr_default, "__reduce__" ):
__reduce__ = _stderr_default.__reduce__


if hasattr( _stderr_default, "__reduce_ex__" ):
__reduce_ex__ = _stderr_default.__reduce_ex__


if hasattr( _stderr_default, "__repr__" ):
__repr__ = _stderr_default.__repr__


if hasattr( _stderr_default, "__setattr__" ):
__setattr__ = _stderr_default.__setattr__


if hasattr( _stderr_default, "__sizeof__" ):
__sizeof__ = _stderr_default.__sizeof__


if hasattr( _stderr_default, "__str__" ):
__str__ = _stderr_default.__str__


if hasattr( _stderr_default, "__subclasscheck__" ):
__subclasscheck__ = _stderr_default.__subclasscheck__


if hasattr( _stderr_default, "__subclasses__" ):
__subclasses__ = _stderr_default.__subclasses__


if hasattr( _stderr_default, "__subclasshook__" ):
__subclasshook__ = _stderr_default.__subclasshook__


if hasattr( _stderr_default, "__text_signature__" ):
__text_signature__ = _stderr_default.__text_signature__


if hasattr( _stderr_default, "__weakrefoffset__" ):
__weakrefoffset__ = _stderr_default.__weakrefoffset__


if hasattr( _stderr_default, "mro" ):
mro = _stderr_default.mro


def __init__(self):
"""
Override any super class `type( _stderr_default )` constructor, so we can
instantiate any kind of `sys.stderr` replacement object, in case it was already
replaced by something else like on Sublime Text with `_LogWriter()`.


Assures all attributes were statically replaced just above. This should happen in case
some new attribute is added to the python language.


This also ignores the only two methods which are not equal, `__init__()` and `__getattribute__()`.
"""
different_methods = ("__init__", "__getattribute__")
attributes_to_check = set( dir( object ) + dir( type ) )


for attribute in attributes_to_check:


if attribute not in different_methods \
and hasattr( _stderr_default, attribute ):


base_class_attribute = super( _stderr_default_class_type, self ).__getattribute__( attribute )
target_class_attribute = _stderr_default.__getattribute__( attribute )


if base_class_attribute != target_class_attribute:
sys.stderr.write( "    The base class attribute `%s` is different from the target class:\n%s\n%s\n\n" % (
attribute, base_class_attribute, target_class_attribute ) )


def __getattribute__(self, item):


if item == 'write':
return _sys_stderr_write


try:
return _stderr_default.__getattribute__( item )


except AttributeError:
return super( _stderr_default_class_type, _stderr_default ).__getattribute__( item )


_stderr_singleton = StdErrReplamentHidden()
sys.stderr = _stderr_singleton


return cls


@classmethod
def unlock(cls):
"""
Detach this `stderr` writer from `sys.stderr` and allow the next call to `lock()` create
a new writer for the stderr.
"""


if cls.is_active:
global _sys_stderr_write_hidden


cls.is_active = False
_sys_stderr_write_hidden = _stderr_default.write






class StdOutReplament(object):
"""
How to redirect stdout and stderr to logger in Python
https://stackoverflow.com/questions/19425736/how-to-redirect-stdout-and-stderr-to-logger-in-python


Set a Read-Only Attribute in Python?
https://stackoverflow.com/questions/24497316/set-a-read-only-attribute-in-python
"""
is_active = False


@classmethod
def lock(cls, logger):
"""
Attach this singleton logger to the `sys.stdout` permanently.
"""
global _stdout_singleton
global _stdout_default
global _stdout_default_class_type


# On Sublime Text, `sys.__stdout__` is set to None, because they already replaced `sys.stdout`
# by some `_LogWriter()` class, then just save the current one over there.
if not sys.__stdout__:
sys.__stdout__ = sys.stdout


try:
_stdout_default
_stdout_default_class_type


except NameError:
_stdout_default = sys.stdout
_stdout_default_class_type = type( _stdout_default )


# Recreate the sys.stdout logger when it was reset by `unlock()`
if not cls.is_active:
cls.is_active = True
_stdout_write = _stdout_default.write


logger_call = logger.debug
clean_formatter = logger.clean_formatter


global _sys_stdout_write
global _sys_stdout_write_hidden


if sys.version_info <= (3,2):
logger.file_handler.terminator = '\n'


# Always recreate/override the internal write function used by `_sys_stdout_write`
def _sys_stdout_write_hidden(*args, **kwargs):
"""
Suppress newline in Python logging module
https://stackoverflow.com/questions/7168790/suppress-newline-in-python-logging-module
"""


try:
_stdout_write( *args, **kwargs )
file_handler = logger.file_handler


formatter = file_handler.formatter
terminator = file_handler.terminator


file_handler.formatter = clean_formatter
file_handler.terminator = ""


kwargs['extra'] = {'_duplicated_from_file': True}
logger_call( *args, **kwargs )


file_handler.formatter = formatter
file_handler.terminator = terminator


except Exception:
logger.exception( "Could not write to the file_handler: %s(%s)", file_handler, logger )
cls.unlock()


# Only create one `_sys_stdout_write` function pointer ever
try:
_sys_stdout_write


except NameError:


def _sys_stdout_write(*args, **kwargs):
"""
Hides the actual function pointer. This allow the external function pointer to
be cached while the internal written can be exchanged between the standard
`sys.stdout.write` and our custom wrapper around it.
"""
_sys_stdout_write_hidden( *args, **kwargs )


try:
# Only create one singleton instance ever
_stdout_singleton


except NameError:


class StdOutReplamentHidden(_stdout_default_class_type):
"""
Which special methods bypasses __getattribute__ in Python?
https://stackoverflow.com/questions/12872695/which-special-methods-bypasses-getattribute-in-python
"""


if hasattr( _stdout_default, "__abstractmethods__" ):
__abstractmethods__ = _stdout_default.__abstractmethods__


if hasattr( _stdout_default, "__base__" ):
__base__ = _stdout_default.__base__


if hasattr( _stdout_default, "__bases__" ):
__bases__ = _stdout_default.__bases__


if hasattr( _stdout_default, "__basicsize__" ):
__basicsize__ = _stdout_default.__basicsize__


if hasattr( _stdout_default, "__call__" ):
__call__ = _stdout_default.__call__


if hasattr( _stdout_default, "__class__" ):
__class__ = _stdout_default.__class__


if hasattr( _stdout_default, "__delattr__" ):
__delattr__ = _stdout_default.__delattr__


if hasattr( _stdout_default, "__dict__" ):
__dict__ = _stdout_default.__dict__


if hasattr( _stdout_default, "__dictoffset__" ):
__dictoffset__ = _stdout_default.__dictoffset__


if hasattr( _stdout_default, "__dir__" ):
__dir__ = _stdout_default.__dir__


if hasattr( _stdout_default, "__doc__" ):
__doc__ = _stdout_default.__doc__


if hasattr( _stdout_default, "__eq__" ):
__eq__ = _stdout_default.__eq__


if hasattr( _stdout_default, "__flags__" ):
__flags__ = _stdout_default.__flags__


if hasattr( _stdout_default, "__format__" ):
__format__ = _stdout_default.__format__


if hasattr( _stdout_default, "__ge__" ):
__ge__ = _stdout_default.__ge__


if hasattr( _stdout_default, "__getattribute__" ):
__getattribute__ = _stdout_default.__getattribute__


if hasattr( _stdout_default, "__gt__" ):
__gt__ = _stdout_default.__gt__


if hasattr( _stdout_default, "__hash__" ):
__hash__ = _stdout_default.__hash__


if hasattr( _stdout_default, "__init__" ):
__init__ = _stdout_default.__init__


if hasattr( _stdout_default, "__init_subclass__" ):
__init_subclass__ = _stdout_default.__init_subclass__


if hasattr( _stdout_default, "__instancecheck__" ):
__instancecheck__ = _stdout_default.__instancecheck__


if hasattr( _stdout_default, "__itemsize__" ):
__itemsize__ = _stdout_default.__itemsize__


if hasattr( _stdout_default, "__le__" ):
__le__ = _stdout_default.__le__


if hasattr( _stdout_default, "__lt__" ):
__lt__ = _stdout_default.__lt__


if hasattr( _stdout_default, "__module__" ):
__module__ = _stdout_default.__module__


if hasattr( _stdout_default, "__mro__" ):
__mro__ = _stdout_default.__mro__


if hasattr( _stdout_default, "__name__" ):
__name__ = _stdout_default.__name__


if hasattr( _stdout_default, "__ne__" ):
__ne__ = _stdout_default.__ne__


if hasattr( _stdout_default, "__new__" ):
__new__ = _stdout_default.__new__


if hasattr( _stdout_default, "__prepare__" ):
__prepare__ = _stdout_default.__prepare__


if hasattr( _stdout_default, "__qualname__" ):
__qualname__ = _stdout_default.__qualname__


if hasattr( _stdout_default, "__reduce__" ):
__reduce__ = _stdout_default.__reduce__


if hasattr( _stdout_default, "__reduce_ex__" ):
__reduce_ex__ = _stdout_default.__reduce_ex__


if hasattr( _stdout_default, "__repr__" ):
__repr__ = _stdout_default.__repr__


if hasattr( _stdout_default, "__setattr__" ):
__setattr__ = _stdout_default.__setattr__


if hasattr( _stdout_default, "__sizeof__" ):
__sizeof__ = _stdout_default.__sizeof__


if hasattr( _stdout_default, "__str__" ):
__str__ = _stdout_default.__str__


if hasattr( _stdout_default, "__subclasscheck__" ):
__subclasscheck__ = _stdout_default.__subclasscheck__


if hasattr( _stdout_default, "__subclasses__" ):
__subclasses__ = _stdout_default.__subclasses__


if hasattr( _stdout_default, "__subclasshook__" ):
__subclasshook__ = _stdout_default.__subclasshook__


if hasattr( _stdout_default, "__text_signature__" ):
__text_signature__ = _stdout_default.__text_signature__


if hasattr( _stdout_default, "__weakrefoffset__" ):
__weakrefoffset__ = _stdout_default.__weakrefoffset__


if hasattr( _stdout_default, "mro" ):
mro = _stdout_default.mro


def __init__(self):
"""
Override any super class `type( _stdout_default )` constructor, so we can
instantiate any kind of `sys.stdout` replacement object, in case it was already
replaced by something else like on Sublime Text with `_LogWriter()`.


Assures all attributes were statically replaced just above. This should happen in case
some new attribute is added to the python language.


This also ignores the only two methods which are not equal, `__init__()` and `__getattribute__()`.
"""
different_methods = ("__init__", "__getattribute__")
attributes_to_check = set( dir( object ) + dir( type ) )


for attribute in attributes_to_check:


if attribute not in different_methods \
and hasattr( _stdout_default, attribute ):


base_class_attribute = super( _stdout_default_class_type, self ).__getattribute__( attribute )
target_class_attribute = _stdout_default.__getattribute__( attribute )


if base_class_attribute != target_class_attribute:
sys.stdout.write( "    The base class attribute `%s` is different from the target class:\n%s\n%s\n\n" % (
attribute, base_class_attribute, target_class_attribute ) )


def __getattribute__(self, item):


if item == 'write':
return _sys_stdout_write


try:
return _stdout_default.__getattribute__( item )


except AttributeError:
return super( _stdout_default_class_type, _stdout_default ).__getattribute__( item )


_stdout_singleton = StdOutReplamentHidden()
sys.stdout = _stdout_singleton


return cls


@classmethod
def unlock(cls):
"""
Detach this `stdout` writer from `sys.stdout` and allow the next call to `lock()` create
a new writer for the stdout.
"""


if cls.is_active:
global _sys_stdout_write_hidden


cls.is_active = False
_sys_stdout_write_hidden = _stdout_default.write

要使用它,您可以只调用 StdErrReplament::lock(logger)StdOutReplament::lock(logger) 传递要用于发送输出文本的日志记录器。例如:

import os
import sys
import logging


current_folder = os.path.dirname( os.path.realpath( __file__ ) )
log_file_path = os.path.join( current_folder, "my_log_file.txt" )


file_handler = logging.FileHandler( log_file_path, 'a' )
file_handler.formatter = logging.Formatter( "%(asctime)s %(name)s %(levelname)s - %(message)s", "%Y-%m-%d %H:%M:%S" )


log = logging.getLogger( __name__ )
log.setLevel( "DEBUG" )
log.addHandler( file_handler )


log.file_handler = file_handler
log.clean_formatter = logging.Formatter( "", "" )


StdOutReplament.lock( log )
StdErrReplament.lock( log )


log.debug( "I am doing usual logging debug..." )
sys.stderr.write( "Tests 1...\n" )
sys.stdout.write( "Tests 2...\n" )

运行这段代码,您将在屏幕上看到:

enter image description here

文件内容:

enter image description here

如果您还希望在屏幕上看到 log.debug调用的内容,则需要向日志记录器添加一个流处理程序。在这种情况下是这样的:

import os
import sys
import logging


class ContextFilter(logging.Filter):
""" This filter avoids duplicated information to be displayed to the StreamHandler log. """
def filter(self, record):
return not "_duplicated_from_file" in record.__dict__


current_folder = os.path.dirname( os.path.realpath( __file__ ) )
log_file_path = os.path.join( current_folder, "my_log_file.txt" )


stream_handler = logging.StreamHandler()
file_handler = logging.FileHandler( log_file_path, 'a' )


formatter = logging.Formatter( "%(asctime)s %(name)s %(levelname)s - %(message)s", "%Y-%m-%d %H:%M:%S" )
file_handler.formatter = formatter
stream_handler.formatter = formatter
stream_handler.addFilter( ContextFilter() )


log = logging.getLogger( __name__ )
log.setLevel( "DEBUG" )
log.addHandler( file_handler )
log.addHandler( stream_handler )


log.file_handler = file_handler
log.stream_handler = stream_handler
log.clean_formatter = logging.Formatter( "", "" )


StdOutReplament.lock( log )
StdErrReplament.lock( log )


log.debug( "I am doing usual logging debug..." )
sys.stderr.write( "Tests 1...\n" )
sys.stdout.write( "Tests 2...\n" )

它在运行时的输出如下:

enter image description here

虽然它仍然保存到文件 my_log_file.txt:

enter image description here

当使用 StdErrReplament:unlock()禁用此功能时,它将只恢复 stderr流的标准行为,因为附加的日志记录器永远不能分离,因为其他人可以引用它的旧版本。这就是为什么它是一个永远不会消亡的全局单例。因此,在用 imp或其他东西重新加载这个模块的情况下,它将永远不会重新获取当前的 sys.stderr,因为它已经被注入到这个模块中并在内部保存了它。

如果您希望将所有输出记录到一个文件并将其输出到一个文本文件,那么可以执行以下操作。这有点俗气,但很有效:

import logging
debug = input("Debug or not")
if debug == "1":
logging.basicConfig(level=logging.DEBUG, filename='./OUT.txt')
old_print = print
def print(string):
old_print(string)
logging.info(string)
print("OMG it works!")

编辑: 注意,除非您将 sys.stderr 重定向到 sys.stdout,否则不会记录错误

编辑2: 第二个问题是,与内置函数不同,必须传递1个参数。

编辑3: 查看之前编写 stdin 和 stdout 到控制台和 file 的代码,stderr 只能进入 file

import logging, sys
debug = input("Debug or not")
if debug == "1":
old_input = input
sys.stderr.write = logging.info
def input(string=""):
string_in = old_input(string)
logging.info("STRING IN " + string_in)
return string_in
logging.basicConfig(level=logging.DEBUG, filename='./OUT.txt')
old_print = print
def print(string="", string2=""):
old_print(string, string2)
logging.info(string)
logging.info(string2)
print("OMG")
b = input()
print(a) ## Deliberate error for testing

你也可以根据上面的 Shx2的答案使用 class multifile添加 stderr:

class Log(object):


def __init__(self, path_log, mode="w", encoding="utf-8"):
h = open(path_log, mode, encoding=encoding)
sys.stdout = multifile([ sys.stdout, h ])
sys.stderr = multifile([ sys.stderr, h ])


def __enter__(self):
""" Necessary if called by with (or with... as) """
return self     # only necessary if "as"


def __exit__(self, type, value, tb):
""" Necessary if call by with """
pass


def __del__(self):
if sys is not None:
# restoring
sys.stdout = sys.__stdout__
sys.stderr = sys.__stderr__


log = Log("test.txt")
print("line 1")
print("line 2", file=sys.stderr)
del log
print("line 3 only on screen")

我已经使用雅各布 · 加布里尔森公认的解决方案大约一年了,但是现在不可避免的事情发生了,我的一个用户想在 Windows 上使用这个解决方案。看看其他提出的答案,我认为这些答案大多数在 捕获派生进程的输出失败(由原来的海报粗体) ,我认为唯一的办法是做 os.dup2()。我想我已经知道如何回答原始海报的 一模一样问题了,而且不需要使用 Unix 特定的工具 tee: 我现在可以捕获我的 Python 程序的所有输出,包括任何衍生的 shell 命令。这在 Windows、 Mac 和 Linux 上都适用。守则如下:

import os, sys, threading, platform


class StreamCapture:
def __init__(self,stream,writer,echo=True,monkeypatch=None):
self.active = True
self.writer = writer
self.stream = stream
self.fd = stream.fileno()
self.echo = echo
(r,w) = os.pipe()
self.pipe_read_fd = r
self.pipe_write_fd = w
self.dup_fd = os.dup(self.fd)
os.dup2(w,self.fd)
self.monkeypatch = monkeypatch if monkeypatch is not None else platform.system()=='Windows'
if self.monkeypatch:
self.oldwrite = stream.write
stream.write = lambda z: os.write(self.fd,z.encode() if type(z)==str else z)
t = threading.Thread(target=self.printer)
self.thread = t
t.start()
def printer(self):
while True:
data = os.read(self.pipe_read_fd,100000)
if(len(data)==0):
self.writer.close()
os.close(self.dup_fd)
os.close(self.pipe_read_fd)
return
self.writer.write(data)
if self.echo:
os.write(self.dup_fd,data)
def close(self):
if not self.active:
return
self.active = False
self.stream.flush()
if self.monkeypatch:
self.stream.write = self.oldwrite
os.dup2(self.dup_fd,self.fd)
os.close(self.pipe_write_fd)
def __enter__(self):
return self
def __exit__(self,a,b,c):
self.close()

你可以这样使用它(注意除了雅各布 · 加布里尔森的解决方案之外,其他解决方案都未能抓住的难点) :

print("This does not get saved to the log file")
with StreamCapture(sys.stdout,open('logfile.txt','wb')):
os.write(sys.stdout.fileno(),b"Hello, captured world!\n")
os.system('echo Hello from the shell')     # Hard case
print("More capturing")
print("This also does not get saved to the log file")

这不是一个简短而甜蜜的回答,但是我尽量保持简洁,而且尽可能的简单。这很复杂,原因如下:

  1. 因为我不能使用 tee,所以我必须在 Python 进程中以某种方式执行 tee的任务。我不清楚是否有一种可移植的 fork()ing 和与 os.pipe()通信的方式(这个表示在 Windows 中很难与分叉进程共享文件描述符) ,所以我决定使用 threading

  2. 在 Windows 中,当 sys.stdoutsys.stderr的底层 fileno()通过 os.dup2()通过 os.pipe()重新路由时,sys.stdoutsys.stderr真的不会欣赏。Python 解释器在执行第一个 print(...)命令后立即崩溃。

  3. 在 Windows 上,为了解决解释器崩溃的问题,我将 sys.stdout.write = ...修改为一个简单调用 os.write(...)的新函数。默认情况下,我只在检测到 Windows 时才这样做。因为我是 monkeypatch,所以我希望这将到达所有对 sys.stdout的缓存引用。我选择这种猴子补丁方法,而不是分配一个全新的流,例如 sys.stdout=...,因为我担心旧的 sys.stdout的副本会保留在解释器的各个部分缓存,但我猜测 sys.stdout.write不太可能被直接缓存。

  4. 如果对处理管道输出的线程进行守护,那么主线程一完成,该线程就会被杀死,但这并不能保证所有输出都已写入日志文件。实际上,有必要 没有将这些 helper 线程守护化,并允许它们在管道关闭时优雅地终止自己。

实际上,我并不完全确定自己是否完全正确——与精细操作系统特性交互的线程代码编写起来很可怕。尽管如此,它还是通过了我的测试。因为这有点复杂,所以我做了一个 PyPI 包:

pip install streamcapture

Github 是 给你

下面是一个上下文管理器,暂时的将标准输出复制到一个文件中。在我看来,这是一个改进,因为它重置 sys.stdout并关闭文件,即使发生异常,语法是在背景中不可见的变化的指示。扩展了约翰 · T 的解。

class DuplicateStdout:
def __init__(self, path):
self.stdout = sys.stdout
self.path = path
self.f = None
    

def write(self, s):
self.stdout.write(s)
self.f.write(s)


def __enter__(self):
self.f = open(self.path, "w")
sys.stdout = self
    

def __exit__(self, *args):
sys.stdout = self.stdout
self.f.close()

示例用法:

with DuplicateStdout("foo.log"):
print("Hey") # also in foo.log


print("There") # not in foo.log