Python 异常处理与日志记录

异常处理是任何编程语言中的重要组成部分，Python 也不例外。Python 提供了丰富的异常处理机制，让开发者可以更好地管理程序中出现的错误。除了捕获和处理异常外，记录异常信息也是至关重要的，以便日后排查问题和改进程序。本文将介绍如何在 Python 中捕获异常，并将异常信息记录到日志文件中。

异常处理

在 Python 中，使用 try-except 语句来捕获异常。try 代码块中放置可能引发异常的代码，except 代码块中处理异常情况。

 
try:
    # 可能引发异常的代码
    result = 10 / 0
except ZeroDivisionError as e:
    # 处理异常情况
    print("除零错误发生:", e)

在上面的例子中，当除以零时会触发 ZeroDivisionError 异常，except 代码块会捕获这个异常并进行处理。

异常日志记录

除了简单地在控制台打印异常信息，我们还可以将异常信息记录到日志文件中，以便后续分析。Python 提供了内置的 logging 模块，可以轻松实现这一功能。

 
import logging
# 配置日志记录器
logging.basicConfig(filename='error.log', level=logging.ERROR)
try:
    result = 10 / 0
except ZeroDivisionError as e:
    # 记录异常信息到日志文件
    logging.error("除零错误发生: %s", e)

在上述代码中，我们首先通过 basicConfig 方法配置了日志记录器，指定了日志文件名为 error.log，并设置记录级别为 ERROR，这意味着只有 ERROR 级别及以上的日志才会被记录。然后，在 except 代码块中，我们使用 logging.error 方法将异常信息记录到日志文件中。

日志文件分析

通过记录异常信息到日志文件，我们可以随时查看程序运行中出现的异常情况，以便及时定位和解决问题。可以使用文本编辑器或日志分析工具来查看日志文件内容，分析异常发生的原因和频率。

 
error.log:
ERROR:root:除零错误发生: division by zero

上述日志文件记录了一条除零错误的异常信息，其中包含了异常类型和具体错误信息。通过分析日志文件，我们可以清晰地了解到程序中存在的问题，并采取相应的措施进行修复。

高级日志记录配置

除了基本的日志记录配置外，logging 模块还提供了更多高级的配置选项，以满足不同场景下的需求。

1. 添加时间戳

可以通过配置 format 参数来添加时间戳到日志记录中，以便更好地跟踪异常发生的时间。

 
import logging
logging.basicConfig(filename='error.log', level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s')

2. 指定日志格式

可以通过配置 format 参数来指定日志的格式，以便更清晰地展示异常信息。

 
import logging
logging.basicConfig(filename='error.log', level=logging.ERROR, format='%(asctime)s - %(levelname)s - %(message)s - %(filename)s:%(lineno)d')

3. 添加异常堆栈信息

可以通过设置 exc_info 参数为 True 来添加异常的堆栈信息到日志记录中。

 
import logging
try:
    result = 10 / 0
except ZeroDivisionError as e:
    logging.error("除零错误发生: %s", e, exc_info=True)

4. 日志分级

除了 ERROR 级别外，logging 模块还支持其他几种日志级别，包括 DEBUG、INFO、WARNING、CRITICAL 等。可以根据实际需求选择合适的日志级别进行记录。

 
import logging
logging.basicConfig(filename='app.log', level=logging.DEBUG)

日志轮换与归档

对于大型项目或长期运行的程序，日志文件可能会变得非常庞大。为了解决这个问题，logging 模块支持日志轮换和归档功能，可以按照一定的策略将日志文件分割、压缩或移动到其他位置。

 
import logging
from logging.handlers import RotatingFileHandler
handler = RotatingFileHandler('app.log', maxBytes=10000, backupCount=5)
handler.setLevel(logging.DEBUG)
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
handler.setFormatter(formatter)
logger = logging.getLogger(__name__)
logger.addHandler(handler)
logger.debug('This is a debug message')

在上述代码中，我们使用 RotatingFileHandler 类来实现日志轮换功能，设置了日志文件的最大字节数为 10000 字节，备份文件的数量为 5。当日志文件大小超过设定的最大字节数时，会自动创建新的日志文件，并将旧的日志文件备份。

日志配置文件

在较大的项目中，可能需要更加灵活和复杂的日志配置。为了避免在代码中硬编码日志配置，可以将日志配置信息单独存放在一个配置文件中，例如 logging.conf。

 
[loggers]
keys=root
 
[handlers]
keys=fileHandler
 
[formatters]
keys=myFormatter
 
[logger_root]
level=DEBUG
handlers=fileHandler
 
[handler_fileHandler]
class=FileHandler
level=DEBUG
formatter=myFormatter
args=('app.log',)
 
[formatter_myFormatter]
format=%(asctime)s - %(levelname)s - %(message)s

然后在 Python 代码中使用 logging.config 模块加载配置文件。

 
import logging
import logging.config
 
logging.config.fileConfig('logging.conf')
 
logger = logging.getLogger(__name__)
logger.debug('This is a debug message')

异常链追踪

有时候，程序中的异常不仅仅是由当前位置的代码引发的，可能还涉及到调用栈上的其他函数或模块。为了更好地追踪异常的来源，可以使用 logging.exception 方法来记录异常链。

 
import logging
 
try:
    result = 10 / 0
except ZeroDivisionError as e:
    logging.exception("除零错误发生")

上述代码会将当前异常以及调用栈上的所有异常信息记录到日志中，方便后续排查问题。

异常处理装饰器

为了简化异常处理的代码，可以定义一个异常处理的装饰器，统一处理函数中的异常情况。

 
import logging
from functools import wraps
 
def handle_exceptions(logger):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                logger.exception("异常发生: %s", e)
        return wrapper
    return decorator
 
logger = logging.getLogger(__name__)
 
@handle_exceptions(logger)
def divide(x, y):
    return x / y
 
result = divide(10, 0)

上述代码定义了一个名为 handle_exceptions 的装饰器，可以将函数中的异常信息记录到日志中。然后，我们将 divide 函数应用了该装饰器，即使在函数内部发生异常，也会被捕获并记录到日志中。

日志记录的性能考虑

在实际项目中，日志记录的性能也是需要考虑的重要因素。过于频繁或过度详细的日志记录可能会对程序的性能产生负面影响，因此需要在记录足够信息的同时，尽量减少对性能的影响。

1. 日志级别控制

在配置日志记录器时，可以根据不同的场景和需求设置不同的日志级别。在开发和测试阶段，可以设置为 DEBUG 级别以记录更详细的信息，而在生产环境中，可以设置为 WARNING 或更高级别以减少日志记录的数量，从而降低对性能的影响。

 
import logging
 
logging.basicConfig(level=logging.DEBUG)  # 开发/测试环境
# logging.basicConfig(level=logging.WARNING)  # 生产环境

2. 异步日志记录

为了减少日志记录对主程序的阻塞，可以使用异步日志记录器，将日志记录操作放到独立的线程或进程中执行，从而提高程序的响应速度。

 
import logging
import logging.handlers
 
logger = logging.getLogger(__name__)
handler = logging.handlers.QueueHandler()
logger.addHandler(handler)
 
queue = logging.handlers.QueueListener(handler, *logger.handlers)
queue.start()

3. 批量写入

在高并发环境中，频繁地向日志文件写入可能会导致文件 I/O 压力过大，影响系统性能。为了解决这个问题，可以考虑将日志记录操作改为批量写入，例如使用队列缓存日志消息，然后定期将队列中的消息批量写入日志文件。

 
import logging
import queue
import threading
 
log_queue = queue.Queue()
 
def log_worker():
    while True:
        record = log_queue.get()
        logger = logging.getLogger(record.name)
        logger.handle(record)
        log_queue.task_done()
 
thread = threading.Thread(target=log_worker, daemon=True)
thread.start()
 
logging.basicConfig(level=logging.INFO)
logging.handlers.QueueHandler(log_queue)

日志记录的安全性考虑

除了性能和功能需求外，日志记录的安全性也是需要考虑的重要因素。在实际项目中，可能存在一些敏感信息需要记录，如用户密码、API 密钥等，如果不加以保护，可能会造成信息泄露和安全漏洞。

1. 敏感信息过滤

在记录日志时，应当避免将敏感信息直接记录到日志中，尤其是在生产环境中。可以通过过滤器来检查日志消息，将其中的敏感信息进行替换或删除。

 
import logging
 
class SensitiveFilter(logging.Filter):
    def filter(self, record):
        record.msg = record.msg.replace('password', '***')
        return True
 
logger = logging.getLogger(__name__)
logger.addFilter(SensitiveFilter())

2. 安全存储

日志文件本身可能包含敏感信息，因此需要采取相应的安全措施来保护日志文件的安全性。可以将日志文件存储在受限制的访问路径下，设置合适的文件权限，以及定期清理和归档日志文件。

 
import logging
 
logging.basicConfig(filename='/var/log/app.log', level=logging.INFO)

3. 加密传输

如果日志信息需要通过网络传输，应当采取加密的方式来保护数据的安全性，防止被中间人窃取或篡改。可以使用 TLS/SSL 协议来加密日志传输通道。

 
import logging
import logging.handlers
 
handler = logging.handlers.SysLogHandler(address=('localhost', 514))
handler.socktype = socket.SOCK_STREAM
handler.ssl_context = ssl.create_default_context()
 
logger = logging.getLogger(__name__)
logger.addHandler(handler)

日志记录审计

除了安全性考虑外，日志记录还可以用于审计目的，帮助监控系统的运行情况、用户行为和操作记录，以及追踪安全事件和异常情况。通过分析日志记录，可以及时发现潜在的安全威胁和漏洞，并采取相应的措施进行预防和应对。

 
import logging
 
logging.basicConfig(filename='audit.log', level=logging.INFO)
 
def audit_log(func):
    def wrapper(*args, **kwargs):
        result = func(*args, **kwargs)
        logging.info(f'{func.__name__} called with arguments: {args}, {kwargs}. Result: {result}')
        return result
    return wrapper
 
@audit_log
def add(x, y):
    return x + y
 
add(3, 4)

在上述代码中，我们定义了一个名为 audit_log 的装饰器，用于记录函数调用的参数和结果到审计日志文件中。通过对关键操作和敏感函数应用该装饰器，可以实现对系统行为的全面监控和审计。

日志监控和警报

除了记录日志和审计功能外，日志监控和警报也是关键的安全实践。通过监控日志文件的变化、异常和异常频率，可以及时发现系统中的异常行为和安全威胁，从而采取相应的措施进行响应和修复。

1. 实时监控

使用监控工具或脚本定期扫描日志文件，监控日志文件的变化和异常情况，及时发现和处理异常事件。

 
import os
import time
 
logfile = '/var/log/app.log'
 
def monitor_logfile(logfile):
    while True:
        with open(logfile, 'r') as f:
            f.seek(0, os.SEEK_END)
            new_data = f.read()
            if new_data:
                print("发现新日志:", new_data)
        time.sleep(10)
 
monitor_logfile(logfile)

2. 异常警报

当发现异常日志时，可以通过发送警报通知管理员或安全团队，以便及时响应和处理。

 
import smtplib
from email.mime.text import MIMEText
 
def send_alert(subject, message):
    sender_email = "your_email@example.com"
    receiver_email = "admin@example.com"
    password = "your_email_password"
    
    msg = MIMEText(message)
    msg['Subject'] = subject
    msg['From'] = sender_email
    msg['To'] = receiver_email
    
    server = smtplib.SMTP('smtp.example.com', 587)
    server.starttls()
    server.login(sender_email, password)
    server.sendmail(sender_email, receiver_email, msg.as_string())
    server.quit()
 
# 在发现异常时调用 send_alert 函数发送邮件警报

3. 日志分析

使用日志分析工具或脚本对日志文件进行分析，识别异常行为、异常模式和安全事件，以及监控系统运行状态和性能表现。

 
import re
 
def analyze_logfile(logfile):
    pattern = r'ERROR|WARNING'
    with open(logfile, 'r') as f:
        for line in f:
            if re.search(pattern, line):
                print("发现异常日志:", line)
 
analyze_logfile(logfile)

自动化响应

为了进一步提高安全性和响应速度，可以将日志监控和警报的流程自动化，通过脚本或自动化工具来自动检测、警报和响应异常事件。

 
import subprocess
 
def automate_response(logfile):
    pattern = r'ERROR|WARNING'
    result = subprocess.run(['grep', '-E', pattern, logfile], capture_output=True)
    if result.stdout:
        send_alert("发现异常日志", result.stdout.decode())
 
automate_response(logfile)

日志备份和恢复

除了监控和警报外，日志备份和恢复也是关键的安全实践。定期备份日志文件可以防止日志丢失或被篡改，同时在需要时可以快速恢复到历史状态，帮助排查问题和恢复系统功能。

1. 定期备份

使用定时任务或脚本定期备份日志文件，将当前日志文件复制到备份目录或存储介质中，以确保日志数据的安全性和可靠性。

 
import shutil
import time
 
def backup_logfile(logfile, backup_dir):
    timestamp = time.strftime('%Y%m%d%H%M%S')
    backup_file = f'{backup_dir}/app_{timestamp}.log'
    shutil.copy(logfile, backup_file)

2. 日志归档

对于长期运行的系统，日志文件可能会变得非常庞大，为了节省存储空间和提高检索效率，可以定期对日志文件进行归档，将旧的日志文件压缩或移动到归档目录中。

 
import os
import shutil
import zipfile
 
def archive_logfile(logfile, archive_dir):
    if os.path.exists(logfile):
        shutil.move(logfile, archive_dir)
        with zipfile.ZipFile(f'{archive_dir}/app_logs.zip', 'w') as zipf:
            zipf.write(f'{archive_dir}/app.log', arcname='app.log')

3. 日志恢复

在需要恢复日志文件时，可以从备份或归档中找到对应的日志文件，并将其复制到原始位置进行恢复。

 
def restore_logfile(backup_file, original_file):
    shutil.copy(backup_file, original_file)

日志清理和管理

为了避免日志文件过大或占用过多存储空间，可以定期清理和管理日志文件，删除过期和不必要的日志数据，以保持系统的性能和稳定性。

 
import os
import glob
 
def cleanup_logs(log_dir, max_age):
    current_time = time.time()
    for logfile in glob.glob(f'{log_dir}/*.log'):
        if os.path.getmtime(logfile) < current_time - max_age:
            os.remove(logfile)

日志文件加密和权限控制

除了备份和管理外，保护日志文件的安全性也是至关重要的。通过加密日志文件和限制文件权限，可以防止未授权访问和篡改，保障日志数据的保密性和完整性。

1. 文件加密

使用加密算法对日志文件进行加密，以确保日志数据在存储和传输过程中的安全性。可以使用加密库如 cryptography 来实现对日志文件的加密和解密操作。

 
from cryptography.fernet import Fernet
 
def encrypt_logfile(logfile, key):
    with open(logfile, 'rb') as f:
        data = f.read()
    fernet = Fernet(key)
    encrypted_data = fernet.encrypt(data)
    with open(logfile + '.enc', 'wb') as f:
        f.write(encrypted_data)
 
def decrypt_logfile(encrypted_file, key):
    with open(encrypted_file, 'rb') as f:
        encrypted_data = f.read()
    fernet = Fernet(key)
    decrypted_data = fernet.decrypt(encrypted_data)
    with open(encrypted_file[:-4], 'wb') as f:
        f.write(decrypted_data)

2. 文件权限控制

通过设置文件权限来限制对日志文件的访问权限，只允许授权用户或组访问和修改日志文件，防止未授权用户篡改或查看日志内容。

 
import os
 
def set_file_permissions(logfile, mode):
    os.chmod(logfile, mode)

日志访问审计

对于敏感信息和重要日志数据，应当实施严格的访问控制和审计机制，记录和监控对日志文件的访问和操作记录，及时发现和处理未授权的访问和异常行为。

 
import subprocess
 
def audit_log_access(logfile):
    result = subprocess.run(['ls', '-l', logfile], capture_output=True)
    print("日志文件权限信息:", result.stdout.decode())

结论

通过加密日志文件、限制文件权限和实施访问审计，可以有效保护日志数据的安全性和完整性，防止未授权访问和篡改，保障系统的安全运行和数据保护。

在实际项目中，建议结合加密库、操作系统权限控制和审计工具，建立完善的日志安全管理机制，以确保日志系统的安全性和可靠性。

希望本文的介绍能够帮助您更好地理解和应用日志文件加密和权限控制在实际项目中的重要性和作用。

	try:
	# 可能引发异常的代码
	result = 10 / 0
	except ZeroDivisionError as e:
	# 处理异常情况
	print("除零错误发生:", e)

	import logging

	# 配置日志记录器
	logging.basicConfig(filename='error.log', level=logging.ERROR)

	try:
	result = 10 / 0
	except ZeroDivisionError as e:
	# 记录异常信息到日志文件
	logging.error("除零错误发生: %s", e)

	import logging

	logging.basicConfig(filename='app.log', level=logging.DEBUG)

	import logging
	from logging.handlers import RotatingFileHandler

	handler = RotatingFileHandler('app.log', maxBytes=10000, backupCount=5)
	handler.setLevel(logging.DEBUG)
	formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
	handler.setFormatter(formatter)

	logger = logging.getLogger(__name__)
	logger.addHandler(handler)

	logger.debug('This is a debug message')

	[loggers]
	keys=root

	[handlers]
	keys=fileHandler

	[formatters]
	keys=myFormatter

	[logger_root]
	level=DEBUG
	handlers=fileHandler

	[handler_fileHandler]
	class=FileHandler
	level=DEBUG
	formatter=myFormatter
	args=('app.log',)

	[formatter_myFormatter]
	format=%(asctime)s - %(levelname)s - %(message)s

	import logging
	import logging.config

	logging.config.fileConfig('logging.conf')

	logger = logging.getLogger(__name__)
	logger.debug('This is a debug message')

	import logging
	from functools import wraps

	def handle_exceptions(logger):
	def decorator(func):
	@wraps(func)
	def wrapper(args, *kwargs):
	try:
	return func(args, *kwargs)
	except Exception as e:
	logger.exception("异常发生: %s", e)
	return wrapper
	return decorator

	logger = logging.getLogger(__name__)

	@handle_exceptions(logger)
	def divide(x, y):
	return x / y

	result = divide(10, 0)

	import logging

	logging.basicConfig(level=logging.DEBUG) # 开发/测试环境
	# logging.basicConfig(level=logging.WARNING) # 生产环境

	import logging
	import logging.handlers

	logger = logging.getLogger(__name__)
	handler = logging.handlers.QueueHandler()
	logger.addHandler(handler)

	queue = logging.handlers.QueueListener(handler, *logger.handlers)
	queue.start()

	import logging
	import queue
	import threading

	log_queue = queue.Queue()

	def log_worker():
	while True:
	record = log_queue.get()
	logger = logging.getLogger(record.name)
	logger.handle(record)
	log_queue.task_done()

	thread = threading.Thread(target=log_worker, daemon=True)
	thread.start()

	logging.basicConfig(level=logging.INFO)
	logging.handlers.QueueHandler(log_queue)

	import logging

	class SensitiveFilter(logging.Filter):
	def filter(self, record):
	record.msg = record.msg.replace('password', '***')
	return True

	logger = logging.getLogger(__name__)
	logger.addFilter(SensitiveFilter())

	import logging

	logging.basicConfig(filename='/var/log/app.log', level=logging.INFO)

	import logging
	import logging.handlers

	handler = logging.handlers.SysLogHandler(address=('localhost', 514))
	handler.socktype = socket.SOCK_STREAM
	handler.ssl_context = ssl.create_default_context()

	logger = logging.getLogger(__name__)
	logger.addHandler(handler)

	import logging

	logging.basicConfig(filename='audit.log', level=logging.INFO)

	def audit_log(func):
	def wrapper(args, *kwargs):
	result = func(args, *kwargs)
	logging.info(f'{func.__name__} called with arguments: {args}, {kwargs}. Result: {result}')
	return result
	return wrapper

	@audit_log
	def add(x, y):
	return x + y

	add(3, 4)

	import os
	import time

	logfile = '/var/log/app.log'

	def monitor_logfile(logfile):
	while True:
	with open(logfile, 'r') as f:
	f.seek(0, os.SEEK_END)
	new_data = f.read()
	if new_data:
	print("发现新日志:", new_data)
	time.sleep(10)

	monitor_logfile(logfile)

	import smtplib
	from email.mime.text import MIMEText

	def send_alert(subject, message):
	sender_email = "your_email@example.com"
	receiver_email = "admin@example.com"
	password = "your_email_password"

	msg = MIMEText(message)
	msg['Subject'] = subject
	msg['From'] = sender_email
	msg['To'] = receiver_email

	server = smtplib.SMTP('smtp.example.com', 587)
	server.starttls()
	server.login(sender_email, password)
	server.sendmail(sender_email, receiver_email, msg.as_string())
	server.quit()

	# 在发现异常时调用 send_alert 函数发送邮件警报

	import re

	def analyze_logfile(logfile):
	pattern = r'ERROR\|WARNING'
	with open(logfile, 'r') as f:
	for line in f:
	if re.search(pattern, line):
	print("发现异常日志:", line)

	analyze_logfile(logfile)

	import subprocess

	def automate_response(logfile):
	pattern = r'ERROR\|WARNING'
	result = subprocess.run(['grep', '-E', pattern, logfile], capture_output=True)
	if result.stdout:
	send_alert("发现异常日志", result.stdout.decode())

	automate_response(logfile)

	import shutil
	import time

	def backup_logfile(logfile, backup_dir):
	timestamp = time.strftime('%Y%m%d%H%M%S')
	backup_file = f'{backup_dir}/app_{timestamp}.log'
	shutil.copy(logfile, backup_file)

	import os
	import shutil
	import zipfile

	def archive_logfile(logfile, archive_dir):
	if os.path.exists(logfile):
	shutil.move(logfile, archive_dir)
	with zipfile.ZipFile(f'{archive_dir}/app_logs.zip', 'w') as zipf:
	zipf.write(f'{archive_dir}/app.log', arcname='app.log')

	def restore_logfile(backup_file, original_file):
	shutil.copy(backup_file, original_file)

	import os
	import glob

	def cleanup_logs(log_dir, max_age):
	current_time = time.time()
	for logfile in glob.glob(f'{log_dir}/*.log'):
	if os.path.getmtime(logfile) < current_time - max_age:
	os.remove(logfile)

	from cryptography.fernet import Fernet

	def encrypt_logfile(logfile, key):
	with open(logfile, 'rb') as f:
	data = f.read()
	fernet = Fernet(key)
	encrypted_data = fernet.encrypt(data)
	with open(logfile + '.enc', 'wb') as f:
	f.write(encrypted_data)

	def decrypt_logfile(encrypted_file, key):
	with open(encrypted_file, 'rb') as f:
	encrypted_data = f.read()
	fernet = Fernet(key)
	decrypted_data = fernet.decrypt(encrypted_data)
	with open(encrypted_file[:-4], 'wb') as f:
	f.write(decrypted_data)

	import os

	def set_file_permissions(logfile, mode):
	os.chmod(logfile, mode)

	import subprocess

	def audit_log_access(logfile):
	result = subprocess.run(['ls', '-l', logfile], capture_output=True)
	print("日志文件权限信息:", result.stdout.decode())