sqllineage/utils/log.py

import logging
import os
from datetime import datetime

# 定义全局变量 log_path
cur_path = os.path.dirname(os.path.realpath(__file__))
log_path = os.path.join(os.path.dirname(cur_path), 'logs')


class Log():
    def __init__(self, logger_name='my_logger'):
        self.logger = logging.getLogger(logger_name)
        if self.logger.hasHandlers():
            self.logger.handlers.clear()
        self.logger.setLevel(logging.INFO)

        if not os.path.exists(log_path):
            os.makedirs(log_path)

        self.update_log_file()

    def update_log_file(self):
        current_date = datetime.now().strftime("%Y_%m_%d")
        self.log_name = os.path.join(log_path, f'{current_date}.log')

        for handler in self.logger.handlers[:]:
            self.logger.removeHandler(handler)

        fh = logging.FileHandler(self.log_name, 'a', encoding='utf-8')
        fh.setLevel(logging.INFO)

        ch = logging.StreamHandler()
        ch.setLevel(logging.INFO)

        formatter = logging.Formatter(
            '[%(asctime)s] %(filename)s line:%(lineno)d [%(levelname)s]%(message)s',
            datefmt="%Y-%m-%d %H:%M:%S"
        )
        fh.setFormatter(formatter)
        ch.setFormatter(formatter)

        self.logger.addHandler(fh)
        self.logger.addHandler(ch)

    def getlog(self):
        current_date = datetime.now().strftime("%Y_%m_%d")
        log_date = os.path.basename(self.log_name).split('.')[0]
        if current_date != log_date:
            self.update_log_file()
        return self.logger

    def info(self, msg, *args, **kwargs):
        logger = self.getlog()
        logger.info(msg, *args, **kwargs)

    def error(self, msg, *args, **kwargs):
        logger = self.getlog()
        logger.error(msg, *args, **kwargs)

    def warning(self, msg, *args, **kwargs):
        logger = self.getlog()
        logger.warning(msg, *args, **kwargs)


if __name__ == "__main__":
    log = Log()
    log.info("---测试开始----")
    log.error("操作步骤1,2,3")
    log.warning("----测试结束----")