energy-data-trans/utils/log/trans_log.py

# -*- coding: utf-8 -*-
# @Time    : 2024/5/16
# @Author  : 魏志亮

import datetime
import logging
import sys
from os import *

from utils.conf.read_conf import read_conf, yaml_conf


def set_trance_id(trace_id):
    """设置当前线程的链路ID"""
    environ['trace_id'] = trace_id


class ContextFilter(logging.Filter):
    """一个自定义的日志过滤器，用于在日志记录中添加链路ID"""

    def filter(self, record):
        record.trace_id = ''
        if 'trace_id' in environ.keys():
            record.trace_id = environ['trace_id']

        return True


logger = logging.getLogger("etl_tools")
logger.setLevel(logging.INFO)
stout_handle = logging.StreamHandler(sys.stdout)
stout_handle.setFormatter(
    logging.Formatter("%(asctime)s-%(trace_id)s: %(message)s"))
stout_handle.setLevel(logging.INFO)
stout_handle.addFilter(ContextFilter())
logger.addHandler(stout_handle)

config_path = path.abspath(__file__).split("utils")[0] + 'conf' + sep + 'etl_config_dev.yaml'
config_path = environ.get('ETL_CONF', config_path)
config = yaml_conf(environ.get('ETL_CONF', config_path))
log_path_dir = read_conf(config, 'log_path_dir', "/data/logs")

log_path = log_path_dir + sep + r'etl_tools_' + (environ['env'] if 'env' in environ else 'dev')
file_path = path.join(log_path)

if not path.exists(file_path):
    makedirs(file_path, exist_ok=True)
file_name = file_path + sep + str(datetime.date.today()) + '.log'

file_handler = logging.FileHandler(file_name, encoding='utf-8')
file_handler.setFormatter(
    logging.Formatter("%(asctime)s-%(trace_id)s: %(message)s"))
file_handler.setLevel(logging.INFO)
file_handler.addFilter(ContextFilter())
logger.addHandler(file_handler)


def trans_print(*args):
    logger.info("  ".join([str(a) for a in args]))