python - 使用Python发送带有输出日志和成功状态的电子邮件

Question

我想在每个脚本运行后自动发送电子邮件，无论是否成功。
电子邮件应包括日志，如果有错误，则应包括堆栈跟踪。
主题应包含一个后缀，以告知运行是否存在错误(“ERROR”)或没有错误(“SUCCESS”)。
当前，我正在使用下面的代码，该代码很难做到:

if __name__ == '__main__':
    # setup logging and e-mail messaging
    logger = logging.getLogger(__name__)  # Creating an instance of logger
    logger.setLevel(logging.DEBUG)
    mail_handler = BufferingSMTPHandler(fromaddr='xyz@test.de',
                                        toaddr=['xyz@tzest.de'],
                                        subject='JOB: Testjob')
    mail_handler.setLevel(logging.INFO)  # Set minimum level to receive messages
    logger.addHandler(mail_handler)
    # rest of __main__ is at the end of the file!


def handle_exception(exc_type, exc_value, exc_traceback):
    if issubclass(exc_type, KeyboardInterrupt):
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return

    logger.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))
    mail_handler.set_failed(True)
    mail_handler.flush()

if __name__ == '__main__':
    .... do work here
    logger.info('TEST')
    ....
    mail_handler.set_failed(False)  # reached end of the script, so success is assumed
    mail_handler.flush()

BufferingSMTPHandler如下所示:

import logging
import logging.handlers
import smtplib


class BufferingSMTPHandler(logging.handlers.BufferingHandler):
    def __init__(self, subject, fromaddr, toaddr, mailhost='mysmtpserver', capacity=500):
        logging.handlers.BufferingHandler.__init__(self, capacity)
        self.mailhost = mailhost
        self.fromaddr = fromaddr
        self.toaddr = toaddr
        self.subject = subject
        self.failed = False

        self.setFormatter(logging.Formatter(
            '{levelname} - {asctime} - {lineno} - {name} -  In {funcName}: {message}',
            style='{'))  # Formatter to prettify logs

    def set_failed(self, failed):
        self.failed = failed

    def flush(self):  # Method to send emails
        if len(self.buffer) <= 0:
            return

        with smtplib.SMTP(self.mailhost) as smtp:
            if self.failed:
                subject = self.subject + ' - ERROR'
            else:
                subject = self.subject + ' - SUCCESS'
            body = ''

            for record in self.buffer:
                body += self.format(record) + '\n'  # Populating body of the message with formatted logs

            msg = f'Subject: {subject}\n\n{body}'

            smtp.sendmail(self.fromaddr, self.toaddr, msg)
            self.buffer.clear()  # Clearing buffer to allude sending same logs

有没有一种方法可以在不向主脚本中添加太多代码的情况下执行此操作？
例如。以这种方式(伪代码):

from status_mailer import AutoStatusMailer

if __name__ == '__main__':
  mailer = AutoStatusMailer(fromaddr='xyz@test.de', toaddr=['xyz@tzest.de'], subject='JOB: Testjob')
  logger = mailer.get_logger()
  .... do work here
  logger.info('TEST')
  ...
  ... status mail is automatically sent afterwards

Answer 1

我根据在StackOverflow上找到的信息将一个名为AutoStatusMailer的类放在一起，以解决该问题:

import atexit
import collections
import logging
import smtplib
import sys
from email.message import EmailMessage


class TailLogHandler(logging.Handler):
    def __init__(self, log_queue, formatter=None):
        logging.Handler.__init__(self)
        self.log_queue = log_queue
        if not formatter:
            formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
        self.setFormatter(formatter)

    def emit(self, record):
        self.log_queue.append(self.format(record))


class TailLogger(object):
    def __init__(self, maxlen, formatter=None):
        self._log_queue = collections.deque(maxlen=maxlen)
        self._log_handler = TailLogHandler(self._log_queue, formatter)

    def contents(self):
        return '\n'.join(self._log_queue)

    @property
    def log_handler(self):
        return self._log_handler


class AutoStatusMailer:
    def __init__(self, fromaddr, toaddr, subject, mailhost, log_length=500, log_level=logging.DEBUG, log_console=True,
                 call_original_excepthook_on_error=False):
        self.mailhost = mailhost
        self.fromaddr = fromaddr
        self.toaddr = toaddr
        self.subject = subject
        self.failed = False
        self.call_original_excepthook_on_error = call_original_excepthook_on_error

        self.logger = logging.getLogger('auto_status_mail_logger')  # Creating an instance of logger
        self.logger.setLevel(log_level)

        if log_console:
            console_handler = logging.StreamHandler(sys.stdout)
            console_handler.setLevel(log_level)
            self.logger.addHandler(console_handler)

        self.tail = TailLogger(log_length)
        tail_handler = self.tail.log_handler
        tail_handler.setLevel(log_level)  # Set minimum level to receive messages
        self.logger.addHandler(tail_handler)

        self.original_excepthook = sys.excepthook
        sys.excepthook = self.handle_exception
        atexit.register(self.send_mail)

    def get_logger(self):
        return self.logger

    def handle_exception(self, exc_type, exc_value, exc_traceback):
        if issubclass(exc_type, KeyboardInterrupt):
            sys.__excepthook__(exc_type, exc_value, exc_traceback)
            return

        self.logger.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_traceback))
        self.failed = True

        if self.call_original_excepthook_on_error:
            self.original_excepthook(exc_type, exc_value, exc_traceback)

    def send_mail(self):
        with smtplib.SMTP(self.mailhost) as smtp:
            if self.failed:
                subject = self.subject + ' - ERROR'
            else:
                subject = self.subject + ' - SUCCESS'
            msg = EmailMessage()
            msg.set_content(self.tail.contents())
            msg['Subject'] = subject
            msg['From'] = self.fromaddr
            msg['To'] = ', '.join(self.toaddr)
            smtp.send_message(msg)

它的用法如下:

from auto_status_mailer import AutoStatusMailer

auto_status_mailer = AutoStatusMailer(fromaddr='test@test.de',
                                      toaddr=['test@test.de', 'test2@test.de'],
                                      subject='JOB: My Job',
                                      mailhost='mri.server.lan')

logger = auto_status_mailer.get_logger()

logger.info("Test")
# raise ValueError("TEST")