LAS/common/logger.py

import os
import threading
from datetime import datetime, timezone, timedelta

class Logger:
    def __init__(self, module_name, log_path="./logs", max_log_size=5*1024*1024, max_log_files=10, log_lvl=1):
        """
        :param module_name: 模块名，日志文件前缀
        :param log_path: 日志目录
        :param max_log_size: 单个日志文件最大字节数
        :param max_log_files: 保留的最大日志文件数
        :param log_lvl: 日志等级（0=DEBUG, 1=INFO, 2=WARN, 3=ERROR）
        """
        self.module_name = module_name
        self.log_path = log_path
        self.max_log_size = max_log_size
        self.max_log_files = max_log_files
        self.log_lvl = log_lvl

        os.makedirs(log_path, exist_ok=True)

        self.log_file = os.path.join(log_path, f"{module_name}.log")
        self.lock = threading.Lock()

        # 维护文件大小计数器
        if os.path.exists(self.log_file):
            self.cur_size = os.path.getsize(self.log_file)
        else:
            self.cur_size = 0

    def _get_bj_time_ms(self):
        tz = timezone(timedelta(hours=8))  # 北京时间
        now = datetime.now(tz)
        return now.strftime("%Y-%m-%d %H:%M:%S.%f")[:-3]

    def _rotate_log(self):
        """日志切分"""
        if os.path.exists(self.log_file) and self.cur_size >= self.max_log_size:
            timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
            new_log_file = os.path.join(self.log_path, f"{self.module_name}_{timestamp}.log")
            os.rename(self.log_file, new_log_file)
            self.cur_size = 0
            self._manage_log_files()

    def _manage_log_files(self):
        """控制日志文件数量"""
        log_files = sorted(
            (f for f in os.listdir(self.log_path) if f.endswith(".log")),
            key=lambda x: os.path.getctime(os.path.join(self.log_path, x))
        )
        while len(log_files) > self.max_log_files:
            oldest_file = log_files.pop(0)
            os.remove(os.path.join(self.log_path, oldest_file))

    def _log(self, lvl_name, text):
        with self.lock:
            self._rotate_log()
            formatted_now = self._get_bj_time_ms()
            line = f"[{formatted_now}][{lvl_name}]: {text}\n"
            with open(self.log_file, "a", encoding="utf-8") as f:
                f.write(line)
            self.cur_size += len(line.encode("utf-8"))

    # ========== 对外接口 ==========
    def LOGDBG(self, text):
        if self.log_lvl <= 0:
            self._log("DEBUG", text)

    def LOGINFO(self, text):
        if self.log_lvl <= 1:
            self._log("INFO", text)

    def LOGWARN(self, text):
        if self.log_lvl <= 2:
            self._log("WARN", text)

    def LOGERR(self, text):
        if self.log_lvl <= 3:
            self._log("ERROR", text)


# ================== 全局实例 ==================
g_logger = Logger(
    module_name="MyModule",
    log_path="./logs",
    max_log_size=50*1024*1024,  # 2 MB
    max_log_files=5,
    log_lvl=1
)

# ========== 使用示例 ==========
if __name__ == "__main__":
    g_logger.LOGINFO("启动服务成功")
    g_logger.LOGDBG("这是调试信息")
    g_logger.LOGWARN("这是警告")
    g_logger.LOGERR("这是错误信息")