```python import logging import os import sys import uuid import socket LOCAL_IP = socket.gethostbyname(socket.gethostname()) TASK_TYPE = os.environ.get("AIPAAS_TASK_TYPE", "").lower() INSTANCE_ID = os.environ.get("AIPAAS_INSTANCE_ID", uuid.uuid4()) LOGS_PATH = os.environ.get("AIPAAS_LOGS_PATH", "/logs").rstrip("/") DEFAULT_LOGS_PATH = "{}/{}_{}_{}.log".format(LOGS_PATH, TASK_TYPE, INSTANCE_ID, LOCAL_IP) try: if sys.version_info[:2] < (3, 6): raise ImportError from concurrent_log_handler import \ ConcurrentRotatingFileHandler as RotatingFileHandler from concurrent_log_handler.queue import setup_logging_queues setup_logging_queues() except ImportError: from logging.handlers import RotatingFileHandler # 每个日志文件的最大字节数 MAX_BYTES = 1024 * 1024 * 100 # 最大日志文件备份数 BACKUP_COUNT = 10 # 默认格式化输出 DEFAULT_FORMATTER = \ "%(asctime)s %(levelname)s %(name)s " \ "[p-%(process)d t-%(thread)d] " \ "[%(filename)s-%(funcName)s-line:%(lineno)d] %(message)s" #自定义的日志格式，可以自定义字段 AIPAAS_FORMATTER = \ "%(asctime)s %(levelname)s %(name)s " \ "[instance_id: %(instance_id)s ip: %(ip)s " \ "[p-%(process)d t-%(thread)d] " \ "[%(filename)s-%(funcName)s-line:%(lineno)d] %(message)s" LEVEL_DEBUG = logging.DEBUG # 10 LEVEL_INFO = logging.INFO # 20 LEVEL_WARNING = logging.WARNING # 30 LEVEL_ERROR = logging.ERROR # 40 LEVEL_CRITICAL = logging.CRITICAL # 50 LEVEL_FATAL = logging.FATAL # 50 LEVEL_RELATIONS = { "debug": LEVEL_DEBUG, "info": LEVEL_INFO, "warn": LEVEL_WARNING, "warning": LEVEL_WARNING, "error": LEVEL_ERROR, "crit": LEVEL_CRITICAL, "critical": LEVEL_CRITICAL, "fatal": LEVEL_FATAL } def get_formatter(fmt=DEFAULT_FORMATTER): """ 定义日志输出格式 """ formatter = logging.Formatter(fmt=fmt) return formatter class LoggerFactory: """ 日志工具类实现 """ __instance = None logger = None def __init__(self, log_path=None, **kwargs): # 获取日志格式 self.formatter = get_formatter() # 指定文件输出 log_dir = os.path.dirname(log_path) try: os.makedirs(log_dir, exist_ok=True) self.file_handler = self._get_file_handler(log_path=log_path) except Exception as e: print("create log dir error: {}".format(e)) self.file_handler = None # 标准输出 self.stream_handler = self._stream_handler() def _get_file_handler(self, log_path): """ :param log_path: 日志路径 :return: """ file_handler = RotatingFileHandler( filename=log_path, maxBytes=MAX_BYTES, backupCount=BACKUP_COUNT, encoding="utf-8", delay=False) file_handler.setFormatter(self.formatter) file_handler.suffix = '%Y-%m-%d_' return file_handler def _stream_handler(self): stream_handler = logging.StreamHandler() stream_handler.setFormatter(self.formatter) return stream_handler def get_logger(self, name=None, level=LEVEL_INFO): if isinstance(level, str): level = LEVEL_RELATIONS.get(level.lower(), "info") if self.logger is None: self.logger = logging.getLogger(name) if self.file_handler is not None: self.logger.addHandler(self.file_handler) if self.stream_handler is not None: self.logger.addHandler(self.stream_handler) self.logger.setLevel(level) self.logger.name = name return self.logger def __new__(cls, *args, **kwargs): renew_flag = kwargs.get("renew_flag", False) if cls.__instance is None or renew_flag: cls.__instance = object.__new__(cls) return cls.__instance get_logger = LoggerFactory(log_path=DEFAULT_LOGS_PATH).get_logger logger = get_logger("aipaas") ```