zsglpt/app_logger.py

#!/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/ 目录")