"""分级日志(info.log / error.log + console),自动携带 trace_id,金融敏感字段脱敏。""" import logging import re import sys from logging.handlers import TimedRotatingFileHandler from pathlib import Path from utils.request_id import get_request_id _TRACE_FMT = "%(asctime)s %(levelname)-7s [%(trace_id)s] %(name)s: %(message)s" # 手机号(11位)/ 身份证(18位,末位可 X)→ 日志脱敏 _PHONE = re.compile(r"(? bool: record.trace_id = get_request_id() or "-" return True class SensitiveMaskFilter(logging.Filter): """身份证保留前4后4、手机号保留前3后4,其余用 * 掩码。""" def filter(self, record: logging.LogRecord) -> bool: msg = record.getMessage() msg = _IDCARD.sub(lambda m: m.group()[:4] + "**********" + m.group()[-4:], msg) msg = _PHONE.sub(lambda m: m.group()[:3] + "****" + m.group()[-4:], msg) record.msg = msg record.args = () return True def get_logger(name: str = "app") -> logging.Logger: return logging.getLogger(name) def setup_logging(log_dir: str = "./logs") -> None: """幂等初始化:console + 按天滚动的 info.log / error.log。""" global _logging_configured if _logging_configured: return _logging_configured = True root = logging.getLogger() root.setLevel(logging.INFO) formatter = logging.Formatter(_TRACE_FMT) Path(log_dir).mkdir(parents=True, exist_ok=True) handlers = [ logging.StreamHandler(sys.stderr), TimedRotatingFileHandler(str(Path(log_dir) / "info.log"), when="midnight", backupCount=30, encoding="utf-8"), TimedRotatingFileHandler(str(Path(log_dir) / "error.log"), when="midnight", backupCount=30, encoding="utf-8"), ] handlers[0].setLevel(logging.INFO) handlers[1].setLevel(logging.INFO) handlers[2].setLevel(logging.ERROR) for h in handlers: h.setFormatter(formatter) h.addFilter(TraceIdFilter()) h.addFilter(SensitiveMaskFilter()) root.addHandler(h)