zhaoyuan/modules/lib/log_manager.py

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

def resource_path():
    """ 获取资源绝对路径，兼容开发环境和单文件模式 """
    if hasattr(sys, '_MEIPASS'):
        base_path = sys._MEIPASS
    else:
        base_path = os.path.abspath(".")
    return base_path

class LogManager:
    _instance = None
    _lock = threading.Lock()
    _configured = False  # 确保单例配置唯一性

    def __new__(cls, log_path=None):
        base_path = resource_path()
        with cls._lock:
            if not cls._instance:
                cls._instance = super().__new__(cls)
                # 路径处理逻辑
                if log_path is None:
                    log_path = Path(base_path) / "app.log"
                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][%(filename)s:%(lineno)d] %(message)s"
            )

            # 文件处理器（自动UTF-8编码）
            file_handler = logging.FileHandler(
                self._full_path,
                encoding='utf-8',
                delay=True  # 延迟打开文件直到实际写入
            )
            file_handler.setFormatter(formatter)

            # 异步监听器 - 仅包含文件处理器
            self.listener = QueueListener(
                self.log_queue,
                file_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("这条信息只会写入文件，不会显示在终端")  # 终端不再显示

    # 默认配置
    default_logger = LogManager().get_logger()
    default_logger.warning("警告信息也只会写入文件")

    # 安全关闭
    LogManager.shutdown()