如何在 Python 日志系统中为警告及以上级别消息自动发送邮件通知-Python教程-PHP中文网

如何在 Python 日志系统中为警告及以上级别消息自动发送邮件通知

本文介绍如何在已使用 QueueHandler + RotatingFileHandler 实现异步轮转日志的基础上，无缝集成 SMTPHandler，使 logger.warning()、logger.error() 或 logger.critical() 触发时自动发送邮件告警。

本文介绍如何在已使用 queuehandler + rotatingfilehandler 实现异步轮转日志的基础上，无缝集成 smtphandler，使 `logger.warning()`、`logger.error()` 或 `logger.critical()` 触发时自动发送邮件告警。

在生产环境中，仅将日志写入文件往往不足以保障快速响应——尤其当关键异常发生时，开发者需要即时感知。Python 标准库的 logging 模块支持多处理器（handler）并行工作，而 QueueListener 正是协调异步日志分发的理想中枢。本文将基于您已有的轮转日志架构，扩展一个高可靠、低侵入的邮件告警能力。

✅ 核心实现要点

复用现有队列机制：无需额外线程或修改主日志流，直接将 SMTPHandler 注册到 QueueListener 的 handler 列表中；
精准控制触发级别：通过 mailHandler.setLevel(logging.WARNING) 限定仅 WARNING 及以上（即 WARNING / ERROR / CRITICAL）触发邮件；
启用级别过滤开关：必须设置 respect_handler_level=True（默认为 False），否则所有日志（包括 INFO）都会被转发给 SMTPHandler；
避免弃用方法：logger.warn() 已被弃用，请统一使用 logger.warning()（Python 3.3+ 强制要求）。

? 配置 SMTPHandler 示例（含安全建议）

import logging
from logging.handlers import SMTPHandler, RotatingFileHandler, QueueListener
from queue import Queue

def initialize(
    app_name="app_name",
    log_level=logging.DEBUG,
    log_size=10 * 1024 * 1024,
    num_files=10,
    # 邮件配置（请按实际替换）
    mailhost=("smtp.gmail.com", 587),
    fromaddr="your_app@example.com",
    toaddrs=["admin@example.com"],
    subject="[ALERT] Your App Log Warning",
    credentials=("your_email@gmail.com", "your_app_password_or_app_token"),
    secure=()  # 使用 TLS；若需 SSL，设为 None 或传入空元组（取决于 SMTP 服务）
):
    logger = logging.getLogger(__name__)
    logger.setLevel(log_level)
    logger.propagate = False  # 防止日志向上冒泡至 root logger 导致重复输出

    # 1. 创建队列与 QueueHandler
    log_queue = Queue()
    queue_handler = logging.handlers.QueueHandler(log_queue)
    logger.addHandler(queue_handler)

    # 2. 创建 RotatingFileHandler（JSON 格式化）
    log_path = get_log_path(app_name)  # 您已实现的路径生成函数
    rotating_file_handler = RotatingFileHandler(
        log_path, maxBytes=log_size, backupCount=num_files
    )
    rotating_file_handler.setFormatter(JSONLogFormatter())  # 您自定义的 JSON 格式器

    # 3. 创建 SMTPHandler（仅处理 WARNING+）
    mail_handler = SMTPHandler(
        mailhost=mailhost,
        fromaddr=fromaddr,
        toaddrs=toaddrs,
        subject=subject,
        credentials=credentials,
        secure=secure
    )
    mail_handler.setLevel(logging.WARNING)  # 关键：设定触发级别
    mail_handler.setFormatter(logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    ))

    # 4. 启动 QueueListener，同时监听多个 handler，并启用级别过滤
    queue_listener = QueueListener(
        log_queue,
        rotating_file_handler,
        mail_handler,
        respect_handler_level=True  # ⚠️ 必须设为 True！
    )

    return logger, queue_listener

⚠️ 注意事项与最佳实践

凭证安全：切勿硬编码邮箱密码。推荐使用环境变量（如 os.getenv("SMTP_PASSWORD")）或密钥管理服务；
Gmail 用户注意：需开启「App Password」（两步验证后生成），而非账户密码；或使用 OAuth2（更复杂但更安全）；
网络可靠性：SMTPHandler 默认同步发送邮件，若 SMTP 服务不可达，可能阻塞 QueueListener 线程。如需更高可用性，可封装为异步任务（如结合 threading.Thread 或 concurrent.futures），但会增加复杂度；
邮件频控：生产环境建议添加限流逻辑（例如每分钟最多发 5 封），避免因高频错误导致邮箱被封或告警疲劳；
测试验证：务必在 _test() 中显式调用 logger.warning() 和 logger.error() 进行端到端测试，确认邮件真实到达。

✅ 最终验证示例

def _test():
    logger, listener = initialize(
        toaddrs=["your_real_email@example.com"],
        credentials=("your_email@gmail.com", "your_app_token")
    )
    listener.start()

    try:
        logger.info("This goes only to file (no email).")
        logger.warning("This triggers an email alert.")
        logger.error("Critical issue detected — email sent immediately.")
    finally:
        listener.stop()
        listener.join()  # 确保队列中剩余日志被处理完毕

通过以上配置，您在不干扰原有高性能日志写入的前提下，获得了及时、可控、可审计的异常告警能力——真正实现“日志归档”与“事件响应”的双轨协同。