#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 日志管理模块 提供标准化的日志记录功能 """ import logging import os from logging.handlers import RotatingFileHandler from datetime import datetime import threading # 全局日志配置 _loggers = {} _logger_lock = threading.Lock() class ColoredFormatter(logging.Formatter): """带颜色的日志格式化器（用于控制台）""" # ANSI颜色代码 COLORS = { 'DEBUG': '\033[36m', # 青色 'INFO': '\033[32m', # 绿色 'WARNING': '\033[33m', # 黄色 'ERROR': '\033[31m', # 红色 'CRITICAL': '\033[35m', # 紫色 } RESET = '\033[0m' def format(self, record): """格式化日志记录""" # 添加颜色 levelname = record.levelname if levelname in self.COLORS: record.levelname = f"{self.COLORS[levelname]}{levelname}{self.RESET}" # 格式化 result = super().format(record) # 恢复原始levelname（避免影响其他handler） record.levelname = levelname return result def setup_logger(name='app', level=None, log_file=None, max_bytes=10*1024*1024, backup_count=5): """ 设置日志记录器 Args: name: 日志器名称 level: 日志级别（DEBUG, INFO, WARNING, ERROR, CRITICAL） log_file: 日志文件路径 max_bytes: 日志文件最大大小（字节） backup_count: 保留的备份文件数量 Returns: logging.Logger: 配置好的日志器 """ with _logger_lock: # 如果已经存在，直接返回 if name in _loggers: return _loggers[name] # 创建日志器 logger = logging.getLogger(name) # 设置日志级别 if level is None: level = os.environ.get('LOG_LEVEL', 'INFO') logger.setLevel(getattr(logging, level.upper())) # 清除已有的处理器（避免重复） logger.handlers.clear() # 日志格式 detailed_formatter = logging.Formatter( '[%(asctime)s] [%(name)s] [%(levelname)s] [%(filename)s:%(lineno)d] - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) simple_formatter = logging.Formatter( '[%(asctime)s] [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) colored_formatter = ColoredFormatter( '[%(asctime)s] [%(levelname)s] %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) # 控制台处理器（带颜色） console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) console_handler.setFormatter(colored_formatter) logger.addHandler(console_handler) # 文件处理器（如果指定了文件路径） if log_file: # 确保日志目录存在 log_dir = os.path.dirname(log_file) if log_dir and not os.path.exists(log_dir): os.makedirs(log_dir, exist_ok=True) # 主日志文件（详细格式） file_handler = RotatingFileHandler( log_file, maxBytes=max_bytes, backupCount=backup_count, encoding='utf-8' ) file_handler.setLevel(logging.DEBUG) file_handler.setFormatter(detailed_formatter) logger.addHandler(file_handler) # 错误日志文件（仅记录WARNING及以上） error_file = log_file.replace('.log', '_error.log') error_handler = RotatingFileHandler( error_file, maxBytes=max_bytes, backupCount=backup_count, encoding='utf-8' ) error_handler.setLevel(logging.WARNING) error_handler.setFormatter(detailed_formatter) logger.addHandler(error_handler) # 防止日志向上传播（避免重复） logger.propagate = False # 缓存日志器 _loggers[name] = logger return logger def get_logger(name='app'): """ 获取日志记录器 Args: name: 日志器名称 Returns: logging.Logger: 日志器实例 """ if name in _loggers: return _loggers[name] else: # 如果不存在，创建一个默认的 return setup_logger(name) class LoggerAdapter: """日志适配器，提供便捷的日志记录方法""" def __init__(self, logger_name='app', context=None): """ 初始化日志适配器 Args: logger_name: 日志器名称 context: 上下文信息（如用户ID、账号ID等） """ self.logger = get_logger(logger_name) self.context = context or {} def _format_message(self, message): """格式化消息，添加上下文信息""" if self.context: context_str = ' '.join([f"[{k}={v}]" for k, v in self.context.items()]) return f"{context_str} {message}" return message def debug(self, message): """记录调试信息""" self.logger.debug(self._format_message(message)) def info(self, message): """记录普通信息""" self.logger.info(self._format_message(message)) def warning(self, message): """记录警告信息""" self.logger.warning(self._format_message(message)) def error(self, message, exc_info=False): """记录错误信息""" self.logger.error(self._format_message(message), exc_info=exc_info) def critical(self, message, exc_info=False): """记录严重错误信息""" self.logger.critical(self._format_message(message), exc_info=exc_info) def exception(self, message): """记录异常信息（自动包含堆栈跟踪）""" self.logger.exception(self._format_message(message)) class AuditLogger: """审计日志记录器（用于记录关键操作）""" def __init__(self, log_file='logs/audit.log'): """初始化审计日志""" self.logger = setup_logger('audit', level='INFO', log_file=log_file) def log_user_login(self, user_id, username, ip_address, success=True): """记录用户登录""" status = "成功" if success else "失败" self.logger.info(f"用户登录{status}: user_id={user_id}, username={username}, ip={ip_address}") def log_admin_login(self, username, ip_address, success=True): """记录管理员登录""" status = "成功" if success else "失败" self.logger.info(f"管理员登录{status}: username={username}, ip={ip_address}") def log_user_created(self, user_id, username, created_by=None): """记录用户创建""" self.logger.info(f"用户创建: user_id={user_id}, username={username}, created_by={created_by}") def log_user_deleted(self, user_id, username, deleted_by): """记录用户删除""" self.logger.warning(f"用户删除: user_id={user_id}, username={username}, deleted_by={deleted_by}") def log_password_reset(self, user_id, username, reset_by): """记录密码重置""" self.logger.warning(f"密码重置: user_id={user_id}, username={username}, reset_by={reset_by}") def log_config_change(self, config_name, old_value, new_value, changed_by): """记录配置修改""" self.logger.warning(f"配置修改: {config_name} 从 {old_value} 改为 {new_value}, changed_by={changed_by}") def log_security_event(self, event_type, description, ip_address=None): """记录安全事件""" self.logger.warning(f"安全事件 [{event_type}]: {description}, ip={ip_address}") # 全局审计日志实例 audit_logger = AuditLogger() # 辅助函数 def log_exception(logger, message="发生异常"): """记录异常（包含堆栈跟踪）""" if isinstance(logger, str): logger = get_logger(logger) logger.exception(message) def log_performance(logger, operation, duration_ms, threshold_ms=1000): """记录性能信息""" if isinstance(logger, str): logger = get_logger(logger) if duration_ms > threshold_ms: logger.warning(f"性能警告: {operation} 耗时 {duration_ms}ms (阈值: {threshold_ms}ms)") else: logger.debug(f"性能: {operation} 耗时 {duration_ms}ms") # 初始化默认日志器 def init_logging(log_level='INFO', log_file='logs/app.log'): """ 初始化日志系统 Args: log_level: 日志级别 log_file: 日志文件路径 """ # 创建主应用日志器 setup_logger('app', level=log_level, log_file=log_file) # 创建数据库日志器 setup_logger('database', level=log_level, log_file='logs/database.log') # 创建自动化日志器 setup_logger('automation', level=log_level, log_file='logs/automation.log') # 创建审计日志器（已在AuditLogger中创建） print("✓ 日志系统初始化完成") if __name__ == '__main__': # 测试日志系统 init_logging(log_level='DEBUG') logger = get_logger('app') logger.debug("这是调试信息") logger.info("这是普通信息") logger.warning("这是警告信息") logger.error("这是错误信息") logger.critical("这是严重错误信息") # 测试上下文日志 adapter = LoggerAdapter('app', {'user_id': 123, 'username': 'test'}) adapter.info("用户操作日志") # 测试审计日志 audit_logger.log_user_login(123, 'test_user', '127.0.0.1', success=True) audit_logger.log_security_event('LOGIN_ATTEMPT', '多次登录失败', '192.168.1.1') print("\n日志测试完成，请检查 logs/ 目录")