opt/tools/log.py

import logging
import os
import queue
import threading
import time

from logging.handlers import RotatingFileHandler
from colorama import init



class LogColors:
    colors = [
        "\033[0m",    # RESET
        "\033[36m",   # CYAN
        "\033[92m",   # LIGHT_GREEN
        "\033[94m",   # LIGHT_BLUE
        "\033[95m",   # LIGHT_MAGENTA
        "\033[96m",   # LIGHT_CYAN
        "\033[90m",   # DARK_GRAY
        "\033[93m",   # LIGHT_YELLOW
        "\033[32m",   # GREEN
        "\033[33m",   # YELLOW
        "\033[34m",   # BLUE
        "\033[35m",   # MAGENTA
        "\033[91m",   # LIGHT_RED
        "\033[37m",   # WHITE
        "\033[31m",   # RED
    ]

    @classmethod
    def get_color(cls, index):
        return cls.colors[index]

class Logger:
    def __init__(self, log_file=r'Log\app.log', file_log_level=logging.DEBUG, console_log_level=logging.DEBUG):
        init(autoreset=True)  # 初始化colorama并自动重置颜色

        # 创建日志文件目录（如果不存在）
        log_dir = os.path.dirname(log_file)
        if not os.path.exists(log_dir):
            os.makedirs(log_dir)

        # 如果日志文件不存在，创建一个新的
        if not os.path.exists(log_file):
            with open(log_file, 'a'):
                pass

        # 创建Logger对象
        self.logger = logging.getLogger('Logger')
        self.logger.setLevel(logging.DEBUG)

        # 创建文件处理器，并设置日志级别和格式
        max_bytes = 5 * 1024 * 1024  # 最大文件大小 5MB
        backup_count = 5  # 保留的日志文件数量
        file_handler = RotatingFileHandler(log_file, maxBytes=max_bytes, backupCount=backup_count)
        file_handler.setLevel(file_log_level)
        formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
        file_handler.setFormatter(formatter)
        self.logger.addHandler(file_handler)

        # 创建控制台处理器，并设置日志级别和格式
        console_handler = logging.StreamHandler()
        console_handler.setLevel(console_log_level)
        console_handler.setFormatter(formatter)
        self.logger.addHandler(console_handler)

        self.log_queue = queue.Queue()
        self.thread = threading.Thread(target=self._log_worker)
        self.thread.start()

    def get_logger(self):
        return self.logger

    def _log_worker(self):
        while True:
            try:
                log_action = self.log_queue.get()
                if log_action is None:
                    break
                log_action()  # 执行存储的 lambda 函数
            except Exception as e:
                self.logger.error(f"Error in log worker: {e}")
            time.sleep(0.01)

    def debug(self, message):
        self.log_queue.put(lambda: self.logger.debug(f"{LogColors.get_color(8)}{message}{LogColors.get_color(0)}"))  # LIGHT_MAGENTA

    def info(self, message, color_index=13):
        self.log_queue.put(lambda: self.logger.info(f"{LogColors.get_color(color_index)}{message}{LogColors.get_color(0)}"))

    def warning(self, message):
        self.log_queue.put(lambda: self.logger.warning(f"{LogColors.get_color(9)}{message}{LogColors.get_color(0)}"))  # YELLOW

    def error(self, message):
        self.log_queue.put(lambda: self.logger.error(f"{LogColors.get_color(14)}{message}{LogColors.get_color(0)}"))  # RED

    def exception(self, message):
        self.log_queue.put(lambda: self.logger.exception(f"{LogColors.get_color(14)}{message}{LogColors.get_color(0)}"))  # RED

    def critical(self, message):
        self.log_queue.put(lambda: self.logger.critical(f"{LogColors.get_color(12)}{message}{LogColors.get_color(0)}"))  # MAGENTA


# 使用示例
logger = Logger(log_file=r'Log\app.log', file_log_level=logging.DEBUG, console_log_level=logging.DEBUG)