140 lines
4.1 KiB
Python
140 lines
4.1 KiB
Python
# !/usr/bin/env python
|
|
# -*- coding:utf-8 -*-
|
|
# @Filename: logger.py
|
|
# @Author: lychang
|
|
# @Time: 2021/7/15 16:41
|
|
import json
|
|
import logging
|
|
import sys
|
|
import time
|
|
from typing import Any, Dict
|
|
|
|
|
|
def _ensure_str(msg: Any) -> str:
|
|
"""确保消息为字符串类型"""
|
|
return msg if isinstance(msg, str) else str(msg)
|
|
|
|
|
|
class JsonFormatter(logging.Formatter):
|
|
"""自定义JSON格式化器"""
|
|
default_msec_format = '%s.%03d'
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__()
|
|
|
|
def format(self, record: logging.LogRecord) -> str:
|
|
"""格式化日志记录为JSON字符串"""
|
|
formatted_record = self._build_record_dict(record)
|
|
return json.dumps(formatted_record, ensure_ascii=False)
|
|
|
|
def _build_record_dict(self, record: logging.LogRecord) -> Dict[str, Any]:
|
|
"""构建日志字典结构"""
|
|
return {
|
|
'deal_time': self.formatTime(record),
|
|
'level': record.levelname,
|
|
'app_type': record.name,
|
|
'business_type': getattr(record, 'business_type', ''),
|
|
'deal_stage': getattr(record, 'deal_stage', ''),
|
|
'object_uuid': getattr(record, 'object_uuid', ''),
|
|
'trace_id': getattr(record, 'trace_id', ''),
|
|
'source': getattr(record, 'source', ''),
|
|
'message': record.getMessage(),
|
|
}
|
|
|
|
def formatTime(self, record: logging.LogRecord, datefmt: str = None) -> str:
|
|
"""格式化时间（包含毫秒）"""
|
|
ct = self.converter(record.created)
|
|
if datefmt:
|
|
s = time.strftime(datefmt, ct)
|
|
else:
|
|
s = time.strftime(self.default_time_format, ct)
|
|
return self.default_msec_format % (s, record.msecs)
|
|
|
|
|
|
# 初始化日志系统
|
|
INSTLOG = logging.getLogger("private_assistant")
|
|
INSTLOG.setLevel(logging.INFO)
|
|
handler = logging.StreamHandler(sys.stdout)
|
|
handler.setFormatter(JsonFormatter())
|
|
INSTLOG.addHandler(handler)
|
|
INSTLOG.propagate = False # 防止日志向上传播
|
|
|
|
DEFAULT_EXTRA = {
|
|
"business_type": "",
|
|
"deal_stage": "",
|
|
"object_uuid": "",
|
|
"trace_id": "",
|
|
"source": ""
|
|
}
|
|
|
|
|
|
def set_extra(
|
|
deal_stage: str = "",
|
|
business_type: str = "",
|
|
object_uuid: str = "",
|
|
trace_id: str = "",
|
|
source: str = ""
|
|
) -> None:
|
|
"""设置全局日志附加信息"""
|
|
global DEFAULT_EXTRA
|
|
DEFAULT_EXTRA.update({
|
|
'deal_stage': deal_stage,
|
|
'business_type': business_type,
|
|
'object_uuid': object_uuid,
|
|
'trace_id': trace_id,
|
|
'source': source,
|
|
})
|
|
|
|
|
|
def _merge_extra(**kwargs: Dict) -> Dict:
|
|
"""合并用户extra与默认配置"""
|
|
extra = kwargs.get('extra', {})
|
|
return {
|
|
'business_type': extra.get('business_type', DEFAULT_EXTRA['business_type']),
|
|
'deal_stage': extra.get('deal_stage', DEFAULT_EXTRA['deal_stage']),
|
|
'object_uuid': extra.get('object_uuid', DEFAULT_EXTRA['object_uuid']),
|
|
'trace_id': extra.get('trace_id', DEFAULT_EXTRA['trace_id']),
|
|
'source': extra.get('source', DEFAULT_EXTRA['source']),
|
|
}
|
|
|
|
|
|
def _log_wrapper(logger_method: Any) -> Any:
|
|
"""日志方法装饰器"""
|
|
|
|
def wrapper(msg: Any, *args: Any, **kwargs: Any) -> None:
|
|
msg = _ensure_str(msg)
|
|
kwargs['extra'] = _merge_extra(**kwargs)
|
|
logger_method(msg, *args, **kwargs)
|
|
|
|
return wrapper
|
|
|
|
|
|
# 使用装饰器统一处理日志方法
|
|
info = _log_wrapper(INSTLOG.info)
|
|
debug = _log_wrapper(INSTLOG.debug)
|
|
warning = _log_wrapper(INSTLOG.warning)
|
|
error = _log_wrapper(INSTLOG.error)
|
|
|
|
|
|
def exception(msg: Any, *args: Any, **kwargs: Any) -> None:
|
|
"""异常日志特殊处理"""
|
|
msg = _ensure_str(msg)
|
|
kwargs['extra'] = _merge_extra(**kwargs)
|
|
INSTLOG.error(msg, *args, exc_info=True, **kwargs)
|
|
|
|
|
|
def set_log_level(level: str) -> None:
|
|
"""设置日志级别"""
|
|
level_map = {
|
|
'DEBUG': logging.DEBUG,
|
|
'INFO': logging.INFO,
|
|
'WARNING': logging.WARNING,
|
|
'ERROR': logging.ERROR,
|
|
}
|
|
INSTLOG.setLevel(level_map.get(level.upper(), logging.INFO))
|
|
|
|
|
|
def set_log_root_name(name: str) -> None:
|
|
"""设置日志根名称"""
|
|
INSTLOG.name = name
|