Python已配置带队列的轮转文件日志，如何新增邮件处理器告警warn及更高级别日志

嘿，我来帮你搞定这个需求！你已经搭好了带队列和轮转文件的日志系统，要新增邮件告警其实很简单——只需要给你的QueueListener再加一个邮件处理器，并让它只关注WARNING及以上级别的日志就好。我直接给你修改后的完整代码和关键说明：

完整修改后的代码 #

import logging
import logging.handlers
import logging.config
import os
import platform
from queue import Queue
import json  # 别忘了导入json，你的JSONLogFormatter要用

class JSONLogFormatter(logging.Formatter):
    def format(self, record):
        # 这里保留你原来的格式化逻辑，示例如下：
        log_record = {
            "level": record.levelname,
            "message": record.getMessage(),
            "timestamp": self.formatTime(record),
            "module": record.module
        }
        return json.dumps(log_record)

def get_log_path(app_name):
    # 保留你原来的路径生成逻辑，示例如下：
    if platform.system() == "Windows":
        log_path = os.path.join(os.environ["APPDATA"], app_name, "app.log")
    else:
        log_path = os.path.join("/var/log", app_name, "app.log")
    # 确保目录存在
    os.makedirs(os.path.dirname(log_path), exist_ok=True)
    return log_path

def initialize(app_name="app_name", log_level=logging.DEBUG, log_size=10*1024*1024, num_files=10):
    
    # Create the main logger
    logger = logging.getLogger(__name__)
    logger.setLevel(log_level)
    # 避免重复添加处理器（如果多次调用initialize的话）
    if logger.handlers:
        return logger, None

    # Create a queue for log records
    log_queue = Queue()

    # Create a QueueHandler to send logs to the queue
    queue_handler = logging.handlers.QueueHandler(log_queue)
    logger.addHandler(queue_handler)

    # Get the log path
    log_path = get_log_path(app_name)

    # 1. 原来的轮转文件处理器
    rotating_file_handler = logging.handlers.RotatingFileHandler(
        log_path,
        maxBytes=log_size,  # 修正你原来的参数名错误，应该是maxBytes
        backupCount=num_files
    )
    rotating_file_handler.setFormatter(JSONLogFormatter())
    rotating_file_handler.setLevel(log_level)  # 记录所有指定级别的日志

    # 2. 新增的邮件处理器
    # 替换成你自己的邮件配置
    SMTP_CONFIG = {
        "smtp_server": "smtp.example.com",
        "smtp_port": 587,  # 587是TLS常用端口，465是SSL常用端口
        "sender_email": "your-alert@example.com",
        "recipient_emails": ["your-email@example.com"],
        "username": "your-alert@example.com",
        "password": "your-email-password-or-app-token"
    }

    smtp_handler = logging.handlers.SMTPHandler(
        mailhost=(SMTP_CONFIG["smtp_server"], SMTP_CONFIG["smtp_port"]),
        fromaddr=SMTP_CONFIG["sender_email"],
        toaddrs=SMTP_CONFIG["recipient_emails"],
        subject="⚠️ 应用告警日志",
        credentials=(SMTP_CONFIG["username"], SMTP_CONFIG["password"]),
        secure=()  # 用TLS传空元组，用SSL可改为None或用SMTP_SSLHandler
    )
    # 设置只处理WARNING及以上级别的日志
    smtp_handler.setLevel(logging.WARNING)
    # 给邮件设置更易读的格式（不用JSON）
    smtp_handler.setFormatter(logging.Formatter(
        """
        时间: %(asctime)s
        级别: %(levelname)s
        模块: %(module)s
        消息: %(message)s
        """
    ))

    # 把两个处理器都传给QueueListener，后台线程异步处理
    queue_listener = logging.handlers.QueueListener(
        log_queue,
        rotating_file_handler,
        smtp_handler  # 新增这一行，添加邮件处理器
    )

    return logger, queue_listener

def _test():
    logger, listener = initialize()
    if listener:
        listener.start()

    try:
        logger.info("这条日志只会写入文件，不会发邮件")
        logger.warning("这条是警告日志，会写入文件+发邮件告警")
        logger.error("这条错误日志同样会触发邮件")
    finally:
        if listener:
            listener.stop()

if __name__ == "__main__":
    _test()

关键说明 #

• 邮件配置替换：把SMTP_CONFIG里的参数换成你实际的邮件服务商信息，个人邮箱可能需要开启「应用专用密码」或允许第三方应用登录。如果用SSL加密（比如Gmail的465端口），可以改用logging.handlers.SMTP_SSLHandler。
• 级别控制：通过smtp_handler.setLevel(logging.WARNING)，确保只有警告、错误、严重错误级别的日志才会触发邮件，避免信息轰炸。
• 异步处理保障：邮件和文件日志都通过QueueListener在后台线程处理，完全不会影响主程序的运行效率，和你原来的设计逻辑保持一致。
• 避免重复日志：新增了if logger.handlers的判断，防止多次调用initialize时重复添加处理器导致日志重复。

测试验证 #

运行_test()函数，你会收到包含警告和错误日志的邮件，同时所有日志都会写入到指定的轮转文件里，完美实现你的需求~

备注：内容来源于stack exchange，提问作者Peter Paquette