Python脚本如何记录进程运行异常日志

wen python案例 2026-06-27 296

本文目录导读：

使用logging模块记录异常
使用装饰器自动记录异常
全局异常捕获
完整的多进程异常日志示例
带邮件通知的异常日志
关键建议

我来为你介绍几种Python记录进程运行异常日志的方法：

使用logging模块记录异常

import logging
import traceback
import sys
# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log', encoding='utf-8'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)
def risky_function():
    try:
        # 可能出错的代码
        result = 1 / 0
    except Exception as e:
        # 记录完整的异常信息
        logger.error(f"发生异常: {str(e)}")
        logger.error(f"异常类型: {type(e).__name__}")
        logger.error(f"详细追踪:\n{traceback.format_exc()}")
        raise
# 使用示例
if __name__ == "__main__":
    try:
        risky_function()
    except Exception as e:
        logger.critical(f"进程终止: {str(e)}")
        sys.exit(1)

使用装饰器自动记录异常

import logging
import functools
import traceback
logging.basicConfig(level=logging.INFO, filename='app.log')
logger = logging.getLogger(__name__)
def log_exceptions(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f"函数 {func.__name__} 异常:")
            logger.error(f"参数: args={args}, kwargs={kwargs}")
            logger.error(f"错误信息: {str(e)}")
            logger.error(f"追踪信息:\n{traceback.format_exc()}")
            raise
    return wrapper
@log_exceptions
def process_data(data):
    return data / 0  # 这里会触发异常
# 使用
try:
    process_data(10)
except:
    print("异常已记录到日志")

全局异常捕获

import logging
import sys
import traceback
logging.basicConfig(level=logging.INFO, filename='app.log')
logger = logging.getLogger(__name__)
def global_exception_handler(exc_type, exc_value, exc_traceback):
    """全局异常处理器"""
    if issubclass(exc_type, KeyboardInterrupt):
        # 不记录键盘中断
        sys.__excepthook__(exc_type, exc_value, exc_traceback)
        return
    logger.critical(f"未捕获的异常:")
    logger.critical(f"类型: {exc_type.__name__}")
    logger.critical(f"值: {exc_value}")
    logger.critical(f"追踪信息:\n{''.join(traceback.format_tb(exc_traceback))}")
# 设置全局异常处理器
sys.excepthook = global_exception_handler
# 测试
def cause_error():
    raise ValueError("这是一个测试异常")
cause_error()

完整的多进程异常日志示例

import logging
import multiprocessing
import traceback
import time
import random
class ProcessLogger:
    def __init__(self, log_file='process.log'):
        self.log_file = log_file
        self.setup_logger()
    def setup_logger(self):
        self.logger = logging.getLogger('ProcessLogger')
        self.logger.setLevel(logging.DEBUG)
        # 文件处理器
        fh = logging.FileHandler(self.log_file, encoding='utf-8')
        fh.setLevel(logging.DEBUG)
        # 控制台处理器
        ch = logging.StreamHandler()
        ch.setLevel(logging.INFO)
        # 格式器
        formatter = logging.Formatter(
            '%(asctime)s - %(processName)s - %(levelname)s - %(message)s'
        )
        fh.setFormatter(formatter)
        ch.setFormatter(formatter)
        self.logger.addHandler(fh)
        self.logger.addHandler(ch)
    def worker_function(self, worker_id):
        """工作进程函数"""
        process_name = multiprocessing.current_process().name
        try:
            self.logger.info(f"进程 {process_name} (ID: {worker_id}) 开始工作")
            # 模拟随机异常
            if random.random() < 0.3:
                raise ValueError(f"进程 {worker_id} 遇到随机错误")
            # 模拟正常工作
            time.sleep(random.uniform(0.5, 2))
            result = random.randint(1, 100)
            self.logger.info(f"进程 {process_name} 完成工作，结果: {result}")
            return result
        except Exception as e:
            self.logger.error(f"进程 {process_name} 异常:")
            self.logger.error(f"错误信息: {str(e)}")
            self.logger.error(f"追踪信息:\n{traceback.format_exc()}")
            return None
    def run_workers(self, num_workers=5):
        """运行多个工作进程"""
        with multiprocessing.Pool(processes=num_workers) as pool:
            results = pool.map(self.worker_function, range(num_workers))
        # 统计结果
        successful = [r for r in results if r is not None]
        failed = [r for r in results if r is None]
        self.logger.info(f"执行完成: 成功 {len(successful)}, 失败 {len(failed)}")
        return results
# 使用示例
if __name__ == "__main__":
    process_logger = ProcessLogger()
    results = process_logger.run_workers(5)

带邮件通知的异常日志

import logging
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
import traceback
class EmailLogHandler(logging.Handler):
    def __init__(self, smtp_server, port, from_addr, to_addrs, password):
        super().__init__()
        self.smtp_server = smtp_server
        self.port = port
        self.from_addr = from_addr
        self.to_addrs = to_addrs
        self.password = password
    def emit(self, record):
        if record.levelno >= logging.ERROR:  # 只发送ERROR及以上级别
            try:
                msg = MIMEMultipart()
                msg['From'] = self.from_addr
                msg['To'] = ', '.join(self.to_addrs)
                msg['Subject'] = f'异常告警: {record.name}'
                body = self.format(record)
                msg.attach(MIMEText(body, 'plain'))
                with smtplib.SMTP(self.smtp_server, self.port) as server:
                    server.starttls()
                    server.login(self.from_addr, self.password)
                    server.send_message(msg)
            except:
                pass
# 配置日志
logger = logging.getLogger('AppLogger')
logger.setLevel(logging.DEBUG)
# 文件处理器
file_handler = logging.FileHandler('app.log', encoding='utf-8')
file_handler.setLevel(logging.DEBUG)
# 邮件处理器 (需要配置实际的邮件服务器信息)
email_handler = EmailLogHandler(
    smtp_server='smtp.gmail.com',
    port=587,
    from_addr='your_email@gmail.com',
    to_addrs=['admin@example.com'],
    password='your_password'
)
email_handler.setLevel(logging.ERROR)
# 格式器
formatter = logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s\n%(exc_info)s'
)
file_handler.setFormatter(formatter)
email_handler.setFormatter(formatter)
logger.addHandler(file_handler)
logger.addHandler(email_handler)
# 使用
try:
    # 你的代码
    risky_operation()
except Exception as e:
    logger.error(f"进程运行异常: {str(e)}", exc_info=True)

关键建议

选择合适的日志级别：
- DEBUG: 详细调试信息
- INFO: 一般运行信息
- WARNING: 警告信息
- ERROR: 错误信息
- CRITICAL: 严重错误
包含关键信息：
- 时间戳
- 进程ID/名称
- 错误类型
- 完整追踪信息
- 相关上下文数据

日志轮转：

from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler('app.log', maxBytes=1024*1024, backupCount=5)


选择最适合你项目需求的方案即可。

本文地址： https://www.bmjup.com/post/9774.html