自定義structlog

# -*- coding: utf-8 -*-

import os
import logging
import logging.handlers
from copy import deepcopy

import structlog

DEFAULT_LOG_PATH = "./nlp-server.log"
DEFAULT_LOG_LEVEL = "debug"

# 日誌脫敏處理，將匹配到的信息進行替換
def dict_get(dictionaries):
    variable_name = os.environ.get("aicc.log.sensitive.fields", "Phone").split("|")
    variable_list = [i.strip() for i in variable_name]

    for k, v in dictionaries.items():
        if k in variable_list:
            any_to_str = str(dictionaries[k])
            dictionaries[k] = any_to_str.replace(any_to_str[:-4], "***")
        else:
            if type(v) is dict:
                dict_get(v)


# 自定義過濾器
def desensitization_log(_, __, event_dict):
    handel_list = deepcopy(event_dict.get("event"))
    if type(handel_list) is dict:
        dict_get(handel_list)
    return handel_list


def init_logger():
    timestamper = structlog.processors.TimeStamper(fmt="%Y-%m-%d %H:%M:%S")
    shared_processors = [
        structlog.stdlib.add_log_level,
        timestamper,
    ]
    structlog.configure(
        processors=shared_processors + [
            structlog.stdlib.add_logger_name,
            structlog.stdlib.add_log_level,
            structlog.stdlib.PositionalArgumentsFormatter(),
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.StackInfoRenderer(),
            structlog.processors.format_exc_info,
            structlog.processors.UnicodeDecoder(),
            desensitization_log,
            structlog.processors.JSONRenderer(indent=1, sort_keys=True),
        ],
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )
    root_logger = logging.getLogger()
    try:
        log_path = os.environ.get("aicc.event.logpath", DEFAULT_LOG_PATH) or DEFAULT_LOG_PATH
        log_level = os.environ.get("aicc.event.loglevel", DEFAULT_LOG_LEVEL) or DEFAULT_LOG_LEVEL
        if log_level.lower() == "error":
            set_debug_level = logging.ERROR
        else:
            set_debug_level = logging.DEBUG
            root_logger.setLevel(set_debug_level)
        if not root_logger.handlers:
            # Add log rotate and format handler for exporter log
            formatter = logging.Formatter(
                '%(asctime)s [%(process)d:%(thread)d] %(filename)s[line:%(lineno)d] : %(message)s')
            handler = logging.handlers.RotatingFileHandler(log_path,
                                                            maxBytes=30 * 1024 * 1024,
                                                            backupCount=10)
            handler.setFormatter(formatter)
            root_logger.addHandler(handler)
            # Add log to stdout
            std_handler = logging.StreamHandler()
            std_handler.setFormatter(formatter)
            root_logger.addHandler(std_handler)
            root_logger.debug("NLP-Server init log succeed, log level is %s.", set_debug_level)
        return root_logger
    except Exception as err:
        root_logger.error('NLP-Server init log failed %s' % (err))
        raise err




複製代碼