Python 如何捕获日志中的异常并发送通知

logging

ERROR

CRITICAL

logging

logging.error()

logging.exception()

exc_info=True

logging.exception()

logging.handlers.SMTPHandler

logging.Handler

import logging
import traceback
import requests # 假设用于发送HTTP请求到通知服务
import json
import os # 用于获取环境变量，避免硬编码敏感信息

# 配置基础日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)

# 定义一个简单的自定义通知Handler
class CustomNotificationHandler(logging.Handler):
    def __init__(self, webhook_url, level=logging.ERROR):
        super().__init__(level)
        self.webhook_url = webhook_url
        self.session = requests.Session() # 使用session保持连接，提高效率

    def emit(self, record):
        # 过滤掉非异常的日志，或者根据需求只处理特定级别的日志
        if not record.exc_info and record.levelno < logging.ERROR:
            return

        # 格式化日志信息，包括异常堆栈
        message = self.format(record)

        # 准备发送到通知服务的payload
        # 这里以一个简单的Webhook为例，实际可能需要根据服务调整
        payload = {
            "text": f"? **生产环境异常告警** ?\n\n**应用:** MyAwesomeApp\n**级别:** {record.levelname}\n**消息:** {record.message}\n**时间:** {self.formatTime(record, '%Y-%m-%d %H:%M:%S')}\n\n**详细信息:**\n```\n{message}\n```"
        }

        try:
            # 异步发送通知是更好的实践，这里为简化直接发送
            # 生产环境建议使用线程池、Celery等异步任务队列
            response = self.session.post(self.webhook_url, json=payload, timeout=5)
            response.raise_for_status() # 检查HTTP请求是否成功
        except requests.exceptions.RequestException as e:
            # 如果通知发送失败，我们应该记录下来，但不能再次触发通知循环
            print(f"Failed to send notification: {e}")
            # 这里可以考虑将失败的通知信息记录到另一个更稳定的地方，如数据库或专门的告警系统
        except Exception as e:
            print(f"Unexpected error in CustomNotificationHandler: {e}")

# 获取Webhook URL，通常从环境变量或配置文件读取
# SLACK_WEBHOOK_URL = os.getenv("SLACK_WEBHOOK_URL") 
# 示例用一个虚拟的URL
SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/T00000000/B00000000/XXXXXXXXXXXXXXXXXXXXXXXX" 

if SLACK_WEBHOOK_URL:
    notification_handler = CustomNotificationHandler(SLACK_WEBHOOK_URL, level=logging.ERROR)
    # 给通知处理器设置一个更简洁的Formatter，因为我们已经在payload中格式化了大部分信息
    notification_handler.setFormatter(logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S'))

    # 将自定义的Handler添加到root logger
    logging.getLogger().addHandler(notification_handler)
else:
    logging.warning("SLACK_WEBHOOK_URL not set, notification handler will not be active.")

# 模拟一个会抛出异常的函数
def problematic_function(a, b):
    return a / b

# 捕获并记录异常
try:
    result = problematic_function(10, 0)
    print(result)
except ZeroDivisionError as e:
    logging.error("An error occurred during division.", exc_info=True) # exc_info=True是关键
    # 或者直接使用 logging.exception()
    # logging.exception("Another error occurred during division.")

print("\n--- 模拟其他日志 ---")
logging.info("This is an informational message.")
logging.warning("This is a warning, but won't trigger notification by default.")

CustomNotificationHandler

emit

emit

CustomNotificationHandler

emit

# ... (CustomNotificationHandler的init方法) ...

    def emit(self, record):
        # ... (过滤逻辑) ...

        # 格式化堆栈信息
        exc_text = ""
        if record.exc_info:
            exc_text = "".join(traceback.format_exception(*record.exc_info))

        # 针对Slack的Payload
        slack_message_blocks = [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"? 异常告警: {record.levelname} ?"
                }
            },
            {
                "type": "section",
                "fields": [
                    { "type": "mrkdwn", "text": f"*应用:* MyAwesomeApp" },
                    { "type": "mrkdwn", "text": f"*时间:* {self.formatTime(record, '%Y-%m-%d %H:%M:%S')}" },
                    { "type": "mrkdwn", "text": f"*级别:* {record.levelname}" },
                    { "type": "mrkdwn", "text": f"*消息:* {record.message}" }
                ]
            }
        ]

        if exc_text:
            slack_message_blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": "*堆栈信息:*\n```\n" + exc_text + "\n```"
                }
            })

        payload = {
            "blocks": slack_message_blocks
        }

        try:
            response = self.session.post(self.webhook_url, json=payload, timeout=5)
            response.raise_for_status()
        except requests.exceptions.RequestException as e:
            print(f"Failed to send Slack notification: {e}")