zhaoyuan_pre_acceptance/modules/lib/log_manager.py

import logging
import os
import threading
from logging.handlers import QueueHandler, QueueListener
from queue import Queue

class LogManager:
    _instance = None
    _lock = threading.Lock()
    _configured = False  # 确保单例配置唯一性
    
    def __new__(cls, log_path="/home/kevin/kevin/zhaoyuan/zhaoyuan/log/app.log"):
        with cls._lock:
            if not cls._instance:
                cls._instance = super().__new__(cls)
                # 路径处理逻辑
                cls._instance._full_path = log_path
                cls._instance._init_logger()
            return cls._instance

    @classmethod
    def _validate_path(cls, path):
        """路径验证与创建"""
        default_path = os.path.join(os.getcwd(), "logs")
        target_path = path or default_path
        
        try:
            os.makedirs(target_path, exist_ok=True)
            # 测试写入权限
            test_file = os.path.join(target_path, "write_test.tmp")
            with open(test_file, "w") as f:
                f.write("permission_test")
            os.remove(test_file)
            return target_path
        except PermissionError:
            logging.error(f"Insufficient permissions for {target_path}, using default")
            os.makedirs(default_path, exist_ok=True)
            return default_path
        except Exception as e:
            logging.error(f"Path error: {str(e)}, using default")
            return default_path

    @staticmethod
    def _sanitize_filename(name):
        """文件名合法性过滤"""
        invalid_chars = {'/', '\\', ':', '*', '?', '"', '<', '>', '|'}
        cleaned = ''.join(c for c in name if c not in invalid_chars)
        return cleaned[:50]  # 限制文件名长度

    def _init_logger(self):
        """初始化日志系统"""
        self.log_queue = Queue(-1)
        self.logger = logging.getLogger("GlobalLogger")
        self.logger.setLevel(logging.DEBUG)

        if not self.logger.handlers:
            # 创建带线程标识的格式器
            formatter = logging.Formatter(
                "[%(asctime)s][%(levelname)s][%(threadName)s] %(message)s"
            )
            
            # 文件处理器（自动UTF-8编码）
            file_handler = logging.FileHandler(
                self._full_path, 
                encoding='utf-8',
                delay=True  # 延迟打开文件直到实际写入
            )
            file_handler.setFormatter(formatter)
            
            # 控制台处理器（仅ERROR级别）
            console_handler = logging.StreamHandler()
            console_handler.setLevel(logging.ERROR)
            console_handler.setFormatter(formatter)

            # 异步监听器
            self.listener = QueueListener(
                self.log_queue,
                file_handler,
                console_handler,
                respect_handler_level=True
            )
            self.listener.start()

            # 队列处理器配置
            queue_handler = QueueHandler(self.log_queue)
            queue_handler.setLevel(logging.DEBUG)
            self.logger.addHandler(queue_handler)
            self.logger.propagate = False

    def get_logger(self):
        """获取线程安全日志器"""
        return self.logger

    @classmethod
    def shutdown(cls):
        """安全关闭日志系统"""
        if cls._instance:
            cls._instance.listener.stop()
            cls._instance = None

# 使用示例
if __name__ == "__main__":
    # 自定义路径和文件名
    custom_logger = LogManager(
        log_path="/home/kevin/kevin/zhaoyuan/zhaoyuan/log/runtime.log"
    ).get_logger()
    
    custom_logger.info("Custom logger configured successfully")
    
    # 默认配置
    default_logger = LogManager().get_logger()
    default_logger.warning("Using default configuration")
    
    # 安全关闭
    LogManager.shutdown()