VberAdminPy/VapAdmin/BACKEND/utils/log_util.py

import os
import sys
import time

from loguru import logger

from core.settings import log_settings


class LoggerUtil:
		_instance = None
		_initialized = False
		
		def __new__(cls, *args, **kwargs):
				if not cls._instance:
						cls._instance = super().__new__(cls)
				return cls._instance
		
		def __init__(self):
				if not self._initialized:
						self._rotation = log_settings.rotation or "10 MB"
						self._retention = log_settings.retention or "30 days"
						self._encoding = "utf-8"
						self._backtrace = True
						self._diagnose = True
						self._enqueue = True
						self._compression = 'zip'
						self._format = '<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> <level>{level: <5}</level>' \
						               '[<magenta>{process}</magenta>:<yellow>{thread}</yellow>]' \
						               '[<cyan>{name}</cyan><<cyan>{function}</cyan>>:<yellow>{line}</yellow>] <level>{message}</level>'
						
						self.log_path = os.path.join(os.getcwd(), log_settings.path)
						os.makedirs(self.log_path, exist_ok=True)
						if log_settings.is_detail:
								self._detail_logger()
						else:
								if logger.level(log_settings.level).no == logger.level("DEBUG").no:
										logger.add(
												os.path.join(self.log_path, f'{time.strftime("%Y-%m-%d")}.debug.log'),
												level="DEBUG",
												filter=lambda record: record['level'].no == logger.level("DEBUG").no,
												format=self._format,
												rotation=self._rotation, retention=self._retention, backtrace=self._backtrace,
												diagnose=self._diagnose,
										)
								logger.add(
										os.path.join(self.log_path, f'{time.strftime("%Y-%m-%d")}.log'),
										level="INFO",
										filter=lambda record: record['level'].no < logger.level("ERROR").no,
										format=self._format,
										rotation=self._rotation, retention=self._retention, backtrace=self._backtrace,
										diagnose=self._diagnose,
										encoding=self._encoding, enqueue=self._enqueue, compression=self._compression
								)
								logger.add(
										os.path.join(self.log_path, f'{time.strftime("%Y-%m-%d")}.error.log'),
										level="ERROR",
										filter=lambda record: record['level'].no >= logger.level("ERROR").no,
										format=self._error_format,
										rotation=self._rotation, retention=self._retention, backtrace=self._backtrace,
										diagnose=self._diagnose,
										encoding=self._encoding, enqueue=self._enqueue, compression=self._compression
								)
						
						self._initialized = True
						self.logger = logger
		
		def _detail_logger(self):
				# 配置DEBUG日志
				logger.add(
						os.path.join(self.log_path, f'{time.strftime("%Y-%m-%d")}/debug.log'),
						level='DEBUG', filter=lambda record: record['level'].no == logger.level("DEBUG").no, format=self._format,
						rotation=self._rotation, retention=self._retention, backtrace=self._backtrace, diagnose=self._diagnose,
						encoding=self._encoding, enqueue=self._enqueue, compression=self._compression
				)
				# 配置INFO日志
				logger.add(
						os.path.join(self.log_path, f'{time.strftime("%Y-%m-%d")}/info.log'),
						level='INFO', filter=lambda record: record['level'].no == logger.level("INFO").no, format=self._format,
						rotation=self._rotation, retention=self._retention, backtrace=self._backtrace, diagnose=self._diagnose,
						encoding=self._encoding, enqueue=self._enqueue, compression=self._compression
				)
				
				# 配置WARNING日志
				logger.add(
						os.path.join(self.log_path, f'{time.strftime("%Y-%m-%d")}/warning.log'),
						level='WARNING', filter=lambda record: record['level'].no == logger.level("WARNING").no,
						format=self._format,
						rotation=self._rotation, retention=self._retention, backtrace=self._backtrace, diagnose=self._diagnose,
						encoding=self._encoding, enqueue=self._enqueue, compression=self._compression
				)
				
				# 配置ERROR日志
				logger.add(
						os.path.join(self.log_path, f'{time.strftime("%Y-%m-%d")}/error.log'),
						level='ERROR', filter=lambda record: record['level'].no == logger.level("ERROR").no,
						format=self._error_format,
						rotation=self._rotation, retention=self._retention, backtrace=self._backtrace, diagnose=self._diagnose,
						encoding=self._encoding, enqueue=self._enqueue, compression=self._compression
				)
				
				# 配置CRITICAL日志
				logger.add(
						os.path.join(self.log_path, f'{time.strftime("%Y-%m-%d")}/critical.log'),
						level='CRITICAL', filter=lambda record: record['level'].no == logger.level("CRITICAL").no,
						format=self._error_format,
						rotation=self._rotation, retention=self._retention, backtrace=self._backtrace, diagnose=self._diagnose,
						encoding=self._encoding, enqueue=self._enqueue, compression=self._compression
				)
				
				logger.add(sys.stderr, level="CRITICAL", backtrace=self._backtrace, diagnose=self._diagnose,
				           format=self._format, colorize=True,
				           filter=lambda record: record["level"].no == logger.level("CRITICAL").no)
		
		def _error_format(self, record):
				print(record)
				log = self._format
				if record["exception"]:
						log += "\n[<cyan>{file.path}</cyan>] "
						log += " <red>exception:{exception}</red>\n"
				return log
		
		@staticmethod
		def get_instance():
				if LoggerUtil._instance is None:
						LoggerUtil._instance = LoggerUtil()
				return LoggerUtil._instance
		
		def __getattr__(self, name):
				return getattr(self.logger, name)