# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述日志 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/31 19:58 # User : geovindu # Product : PyCharm # Project : PyAlgorithms # File : LogHelper.py import os import sys import time import threading import queue import datetime import traceback import json import zipfile import random import re from enum import Enum from typing import Tuple, Optional, Dict, List, Any, Callable import atexit import platform import zoneinfo class LogLevel(Enum): 日志等级枚举与 C# 版本对齐 OFF 0 # 关闭输出 ERROR 1 INFO 2 DEBUG 3 # 类型别名 DesensitizeFunc Callable[[str, str, str], str] AlertCallbackFunc Callable[[str, str, Dict[str, Any]], None] class LogHelper: Python 异步日志工具类 新增特性业务模块分目录隔离存储 完整能力清单 1. 【新增】按业务模块分目录存储日志防止多业务日志混杂 2. 异步队列批量写入多线程安全、优雅退出 3. 文本 / JSON结构化日志动态切换异常堆栈独立字段 stack_trace 4. 控制台彩色输出开关自动按大小切割日志、备份自动zip压缩 5. 过期日志自动清理磁盘超限策略优先删zip再删log 6. 分级独立磁盘上限 全局总磁盘上限兜底 7. 全局/分级日志采样策略减轻高频日志IO压力 8. 日志脱敏回调函数支持携带模块名 9. 告警回调磁盘超限、队列拥堵事件通知 10.自定义时区、自定义时间格式化模板 日志文件生成样式 Logs/ ├─ default/ # 旧代码默认模块 │ ├─ info_20260731.log │ └─ debug_20260731.log ├─ user_service/ # 用户业务模块 │ └─ info_20260731.log ├─ order_service/ # 订单业务模块 │ └─ debug_20260731.log ├─ pay_service/ # 支付业务模块 │ └─ error_20260731.log └─ goods_service/ # 商品业务模块 └─ info_20260731.log # 【全局配置区 按需修改】 LOG_ROOT os.path.join(os.getcwd(), Logs) DEFAULT_MODULE default # 默认业务模块名称 KEEP_DAYS 7 # 日志保留天数 MAX_FILE_SIZE_MB 10 # 单个日志文件最大MB MAX_FILE_SIZE_BYTES MAX_FILE_SIZE_MB * 1024 * 1024 QUEUE_MAXSIZE 10000 QUEUE_ALERT_THRESHOLD 7000 BATCH_WRITE_COUNT 10 ENABLE_CONSOLE_OUTPUT True ENABLE_JSON_FORMAT False COMPRESS_BAK_LOG True # 磁盘容量限制 MB MAX_DISK_TOTAL_MB 500 LEVEL_DISK_LIMIT { debug: 150, info: 250, error: 100 } # 采样配置 0~1.0 GLOBAL_SAMPLING_RATE 1.0 LEVEL_SAMPLING_RATE { debug: 1.0, info: 1.0, error: 1.0 } DISK_CHECK_INTERVAL_SEC 60 TIME_ZONE_NAME Asia/Shanghai TIME_FORMAT_TEMPLATE %Y-%m-%d %H:%M:%S _log_level LogLevel.DEBUG # 外部回调注册 _desensitize_callback: Optional[DesensitizeFunc] None _alert_callback: Optional[AlertCallbackFunc] None # 内部变量 _log_queue: queue.Queue queue.Queue(QUEUE_MAXSIZE) _stop_event threading.Event() _writer_thread: Optional[threading.Thread] None _last_clean_time: datetime.datetime _last_disk_check_time 0.0 _init_finished False _file_operate_lock threading.Lock() _pid os.getpid() _tz_info: Optional[zoneinfo.ZoneInfo] None # 控制台颜色 _COLOR_MAP { DEBUG: \033[34m, INFO: \033[32m, ERROR: \033[31m, RESET: \033[0m } if platform.system() Windows: try: import ctypes kernel32 ctypes.windll.kernel32 kernel32.SetConsoleMode(kernel32.GetStdHandle(-11), 7) except Exception: for k in _COLOR_MAP: _COLOR_MAP[k] classmethod def _safe_module_name(cls, module: str) - str: 安全过滤模块名称防止路径穿越、非法字符 :param module: :return: if not isinstance(module, str) or module.strip() : return cls.DEFAULT_MODULE # 只允许字母数字下划线横杠 safe_str re.sub(r[^\w\-], _, module.strip()) return safe_str classmethod def get_module_log_dir(cls, module: str) - str: 获取指定业务模块日志根目录 :param module: :return: mod_name cls._safe_module_name(module) return os.path.join(cls.LOG_ROOT, mod_name) classmethod def _init_timezone(cls): :return: try: cls._tz_info zoneinfo.ZoneInfo(cls.TIME_ZONE_NAME) except Exception: cls._tz_info None classmethod def get_now_datetime(cls) - datetime.datetime: :return: if cls._tz_info: return datetime.datetime.now(tzcls._tz_info) return datetime.datetime.now() # 回调注册API classmethod def register_desensitize(cls, func: DesensitizeFunc): 脱敏回调签名func(level:str, module:str, msg:str) - str cls._desensitize_callback func classmethod def register_alert_callback(cls, func: AlertCallbackFunc): :param func: :return: cls._alert_callback func # 动态配置API classmethod def set_log_level(cls, level: LogLevel): :param level: :return: cls._log_level level classmethod def switch_json_format(cls, enable: bool): :param enable: :return: cls.ENABLE_JSON_FORMAT enable classmethod def switch_console(cls, enable: bool): cls.ENABLE_CONSOLE_OUTPUT enable classmethod def set_timezone(cls, tz_name: str): :param tz_name: :return: cls.TIME_ZONE_NAME tz_name cls._init_timezone() classmethod def set_time_format(cls, fmt_str: str): :param fmt_str: :return: cls.TIME_FORMAT_TEMPLATE fmt_str classmethod def set_sampling_rate(cls, level: str, rate: float): :param level: :param rate: :return: rate max(0.0, min(1.0, rate)) cls.LEVEL_SAMPLING_RATE[level.lower()] rate # 采样判断 classmethod def _need_sample(cls, level: str) - bool: :param level: :return: lv level.lower() lv_rate cls.LEVEL_SAMPLING_RATE.get(lv, 1.0) final_rate cls.GLOBAL_SAMPLING_RATE * lv_rate if final_rate 1.0: return True return random.random() final_rate # 磁盘空间管理适配模块目录 classmethod def _get_dir_total_size_mb(cls, folder: str, filter_prefix: str None) - float: :param folder: :param filter_prefix: :return: total_bytes 0 if not os.path.exists(folder): return 0.0 for entry in os.scandir(folder): if entry.is_dir(): # 递归遍历模块子目录 total_bytes cls._get_dir_total_size_mb(entry.path, filter_prefix) continue if not entry.is_file(): continue if not (entry.name.endswith(.log) or entry.name.endswith(.zip)): continue if filter_prefix is not None and not entry.name.startswith(f{filter_prefix}_): continue total_bytes entry.stat().st_size return total_bytes / 1024 / 1024 classmethod def _get_sorted_log_files(cls, folder: str, filter_prefix: str None): :param folder: :param filter_prefix: :return: file_list [] if not os.path.exists(folder): return file_list for entry in os.scandir(folder): if entry.is_dir(): file_list.extend(cls._get_sorted_log_files(entry.path, filter_prefix)) continue if not entry.is_file(): continue if not (entry.name.endswith(.log) or entry.name.endswith(.zip)): continue if filter_prefix is not None and not entry.name.startswith(f{filter_prefix}_): continue file_list.append((entry.stat().st_mtime, entry.path)) file_list.sort(keylambda x: x[0]) return [item[1] for item in file_list] classmethod def _trigger_alert(cls, event_type: str, message: str, ext: Dict[str, Any] None): :param event_type: :param message: :param ext: :return: if cls._alert_callback is None: return try: ext_info ext if ext is not None else {} cls._alert_callback(event_type, message, ext_info) except Exception: pass classmethod def _try_free_disk_space(cls, level: str None) - bool: :param level: :return: while True: if level is not None: usage_mb cls._get_dir_total_size_mb(cls.LOG_ROOT, filter_prefixlevel) limit_mb cls.LEVEL_DISK_LIMIT.get(level, cls.MAX_DISK_TOTAL_MB) if usage_mb limit_mb: return True else: usage_mb cls._get_dir_total_size_mb(cls.LOG_ROOT) if usage_mb cls.MAX_DISK_TOTAL_MB: return True all_files cls._get_sorted_log_files(cls.LOG_ROOT, filter_prefixlevel) if not all_files: return False deleted False # 优先删除zip for fp in all_files: if fp.endswith(.zip): try: os.remove(fp) print(f[LogHelper CLEAN] 磁盘超限删除老旧压缩包{os.path.basename(fp)}, filesys.stderr) deleted True break except Exception: continue if deleted: continue # 删除log for fp in all_files: if fp.endswith(.log): try: os.remove(fp) print(f[LogHelper CLEAN] 磁盘超限删除老旧日志{os.path.basename(fp)}, filesys.stderr) deleted True break except Exception: continue if not deleted: return False classmethod def _check_disk_limit(cls, level: str) - bool: :param level: :return: now_ts time.time() if now_ts - cls._last_disk_check_time cls.DISK_CHECK_INTERVAL_SEC: return False cls._last_disk_check_time now_ts lv level.lower() lv_usage cls._get_dir_total_size_mb(cls.LOG_ROOT, lv) lv_limit cls.LEVEL_DISK_LIMIT.get(lv, cls.MAX_DISK_TOTAL_MB) total_usage cls._get_dir_total_size_mb(cls.LOG_ROOT) over_flag False if lv_usage lv_limit: cls._trigger_alert( DISK_OVER_LIMIT, f日志【{level}】分级磁盘占用超限({lv_usage:.2f}MB/{lv_limit}MB)开始清理, {level: level, usage_mb: lv_usage, limit_mb: lv_limit, total_usage: total_usage} ) success cls._try_free_disk_space(lv) if not success: over_flag True elif total_usage cls.MAX_DISK_TOTAL_MB: cls._trigger_alert( DISK_OVER_LIMIT, f日志总磁盘占用超限({total_usage:.2f}MB/{cls.MAX_DISK_TOTAL_MB}MB)开始清理, {usage_mb: total_usage, limit_mb: cls.MAX_DISK_TOTAL_MB} ) success cls._try_free_disk_space() if not success: over_flag True if over_flag: print(f[LogHelper FATAL] 清理后磁盘依旧超限暂停持久化写入【{level}】日志, filesys.stderr) return over_flag classmethod def _init(cls) - None: :return: if cls._init_finished: return os.makedirs(cls.LOG_ROOT, exist_okTrue) cls._init_timezone() cls._last_clean_time cls.get_now_datetime() cls._stop_event.clear() cls._writer_thread threading.Thread( targetcls._write_log_loop, nameLogHelper-Writer, daemonTrue ) cls._writer_thread.start() atexit.register(cls._shutdown) cls._init_finished True classmethod def _get_caller_info(cls) - Tuple[str, str, str, int]: :return: try: frame sys._getframe(4) while frame: if frame.f_code.co_filename ! __file__: class_name 未知类 if self in frame.f_locals: class_name frame.f_locals[self].__class__.__name__ elif cls in frame.f_locals: class_name frame.f_locals[cls].__name__ return ( os.path.basename(frame.f_code.co_filename), class_name, frame.f_code.co_name, frame.f_lineno ) frame frame.f_back except Exception: pass return (未知文件, 未知类, 未知方法, 0) classmethod def _build_log_data(cls, level: str, module: str, msg: str, stack_trace: str ) - str | Dict[str, Any]: :param level: :param module: :param msg: :param stack_trace: :return: mod_name cls._safe_module_name(module) # 脱敏回调携带模块名 if cls._desensitize_callback is not None: try: msg cls._desensitize_callback(level, mod_name, msg) except Exception: pass filename, class_name, method_name, line_no cls._get_caller_info() now cls.get_now_datetime() log_time_str now.strftime(cls.TIME_FORMAT_TEMPLATE) timestamp now.timestamp() thread_name threading.current_thread().name if cls.ENABLE_JSON_FORMAT: data { time: log_time_str, timestamp: timestamp, level: level, module: mod_name, pid: cls._pid, thread: thread_name, file: filename, class: class_name, function: method_name, line: line_no, message: msg, stack_trace: stack_trace if stack_trace else None } return data else: text ( f[{log_time_str}] [{level}] [MODULE:{mod_name}] [PID:{cls._pid}] [THREAD:{thread_name}] f[{class_name}.{method_name}·{line_no}行·{filename}] {msg} ) if stack_trace: text f\n{stack_trace} text \n return text classmethod def _enqueue_log(cls, level: str, module: str, msg: str, stack_trace: str ) - None: :param level: :param module: :param msg: :param stack_trace: :return: if not cls._need_sample(level): return try: log_item cls._build_log_data(level, module, msg, stack_trace) # 控制台输出 if cls.ENABLE_CONSOLE_OUTPUT: if isinstance(log_item, dict): console_str json.dumps(log_item, ensure_asciiFalse) else: console_str log_item.rstrip(\n) color cls._COLOR_MAP.get(level, ) reset cls._COLOR_MAP[RESET] print(f{color}{console_str}{reset}) queue_size cls._log_queue.qsize() if queue_size cls.QUEUE_ALERT_THRESHOLD: cls._trigger_alert( QUEUE_BLOCKED, f日志队列拥堵当前队列长度:{queue_size}, {queue_size: queue_size, max_queue: cls.QUEUE_MAXSIZE} ) mod_safe cls._safe_module_name(module) cls._log_queue.put_nowait((level.lower(), mod_safe, log_item)) except queue.Full: print([LogHelper WARN] 日志队列已满丢弃日志, filesys.stderr) except Exception as e: print(f[LogHelper ERROR] 组装日志异常:{e}, filesys.stderr) classmethod def _get_log_path(cls, level: str, module: str) - str: :param level: :param module: :return: mod_dir cls.get_module_log_dir(module) os.makedirs(mod_dir, exist_okTrue) date_str cls.get_now_datetime().strftime(%Y%m%d) return os.path.join(mod_dir, f{level}_{date_str}.log) classmethod def _compress_file(cls, src_file: str): :param src_file: :return: if not os.path.exists(src_file) or not cls.COMPRESS_BAK_LOG: return try: zip_path src_file.replace(.log, .zip) with zipfile.ZipFile(zip_path, w, compressionzipfile.ZIP_DEFLATED) as zf: zf.write(src_file, arcnameos.path.basename(src_file)) os.remove(src_file) except Exception as e: print(f[LogHelper] 日志压缩失败:{src_file}, err:{e}, filesys.stderr) classmethod def _check_and_split(cls, log_path: str) - None: :param log_path: :return: with cls._file_operate_lock: if not os.path.exists(log_path): return try: if os.path.getsize(log_path) cls.MAX_FILE_SIZE_BYTES: base os.path.splitext(log_path)[0] bak_path f{base}_{time.strftime(%H%M%S)}.log os.rename(log_path, bak_path) threading.Thread(targetcls._compress_file, args(bak_path,), daemonTrue).start() except Exception as e: print(f[LogHelper] 文件切割异常:{e}, filesys.stderr) classmethod def _clean_old_logs(cls) - None: :return: now cls.get_now_datetime() expire_time now - datetime.timedelta(dayscls.KEEP_DAYS) with cls._file_operate_lock: if not os.path.exists(cls.LOG_ROOT): return for root, _, files in os.walk(cls.LOG_ROOT): for filename in files: if not (filename.endswith(.log) or filename.endswith(.zip)): continue file_path os.path.join(root, filename) try: mtime datetime.datetime.fromtimestamp(os.path.getmtime(file_path), tzcls._tz_info) if mtime expire_time: os.remove(file_path) except Exception as e: print(f[LogHelper] 清理过期文件失败 {filename}:{e}, filesys.stderr) classmethod def _write_log_loop(cls) - None: :return: # buffer key: (level, module) buffer: Dict[Tuple[str, str], List[Any]] {} while not cls._stop_event.is_set(): try: level, module, log_item cls._log_queue.get(timeout0.8) key (level, module) buffer.setdefault(key, []).append(log_item) if len(buffer[key]) cls.BATCH_WRITE_COUNT: cls._flush_buffer(buffer, key) except queue.Empty: if buffer: cls._flush_buffer(buffer) continue except Exception as e: print(f[LogHelper] 写入循环异常: {e}, filesys.stderr) if buffer: cls._flush_buffer(buffer) classmethod def _flush_buffer(cls, buffer: Dict[Tuple[str, str], List[Any]], target_key: Tuple[str, str] None): :param buffer: :param target_key: :return: target_keys [target_key] if target_key else list(buffer.keys()) for key in target_keys: level, module key msg_list buffer.get(key) if not msg_list: continue try: disk_over cls._check_disk_limit(level) if disk_over: buffer[key].clear() continue log_path cls._get_log_path(level, module) cls._check_and_split(log_path) lines [] for item in msg_list: if isinstance(item, dict): lines.append(json.dumps(item, ensure_asciiFalse) \n) else: lines.append(item) with open(log_path, a, encodingutf-8) as f: f.writelines(lines) f.flush() buffer[key].clear() except Exception as e: print(f[LogHelper] 文件写入失败 level:{level}, module:{module}, err:{e}, filesys.stderr) now cls.get_now_datetime() if now.date() ! cls._last_clean_time.date(): cls._clean_old_logs() cls._last_clean_time now # 对外API【兼容旧代码 新增模块版本】 # 原有接口默认模块 default历史代码无需改动 classmethod def debug(cls, msg: str): :param msg: :return: if cls._log_level.value LogLevel.DEBUG.value: cls._enqueue_log(DEBUG, cls.DEFAULT_MODULE, msg) classmethod def info(cls, msg: str): :param msg: :return: if cls._log_level.value LogLevel.INFO.value: cls._enqueue_log(INFO, cls.DEFAULT_MODULE, msg) classmethod def error(cls, msg: str, exc: Optional[Exception] None): :param msg: :param exc: :return: if cls._log_level.value LogLevel.ERROR.value: return stack_str if exc: stack_str .join(traceback.format_exception(type(exc), exc, exc.__traceback__)) cls._enqueue_log(ERROR, cls.DEFAULT_MODULE, msg, stack_str) # 新增支持指定业务模块的日志接口 classmethod def debug_mod(cls, module: str, msg: str): :param module: :param msg: :return: if cls._log_level.value LogLevel.DEBUG.value: cls._enqueue_log(DEBUG, module, msg) classmethod def info_mod(cls, module: str, msg: str): :param module: :param msg: :return: if cls._log_level.value LogLevel.INFO.value: cls._enqueue_log(INFO, module, msg) classmethod def error_mod(cls, module: str, msg: str, exc: Optional[Exception] None): :param module: :param msg: :param exc: :return: if cls._log_level.value LogLevel.ERROR.value: return stack_str if exc: stack_str .join(traceback.format_exception(type(exc), exc, exc.__traceback__)) cls._enqueue_log(ERROR, module, msg, stack_str) classmethod def _shutdown(cls): :return: cls._stop_event.set() try: cls._log_queue.join() except Exception: pass if cls._writer_thread: cls._writer_thread.join(timeout3.0) # 模块导入自动初始化 LogHelper._init()调用# encoding: utf-8 # 版权所有 2026 ©涂聚文有限公司™ ® # 许可信息查看言語成了邀功盡責的功臣還需要行爲每日來值班嗎 # 描述日志 # Author : geovindu,Geovin Du 涂聚文. # IDE : PyCharm 2024.3.6 python 3.11 # os : windows 10 # database : mysql 9.0 sql server 2019, postgreSQL 17.0 Oracle 21c Neo4j # Datetime : 2026/7/31 14:16 # User : geovindu # Product : PyCharm # Project : ictsimple # File : logtest.py from loghelper import LogHelper,LogLevel import time import os if __name__ __main__: # 时区、时间格式动态修改示例 # LogHelper.set_timezone(UTC) # LogHelper.set_time_format(%Y-%m-%d %H:%M:%S.%f) # #TIME_ZONE_NAME Asia/Shanghai #TIME_FORMAT_TEMPLATE %Y-%m-%d %H:%M:%S # 1. 脱敏回调新增参数 module def demo_desensitize(level: str, module: str, msg: str) - str: import re msg re.sub(r1[3-9]\d{9}, lambda m: m.group()[:3] **** m.group()[7:], msg) return msg LogHelper.register_desensitize(demo_desensitize) # 2. 告警回调 def demo_alert(event_type, msg, ext): print(f\n【ALERT】{event_type} | {msg} | {ext}\n) LogHelper.register_alert_callback(demo_alert) # 3. 原有调用方式默认模块 default LogHelper.info(【默认模块】系统启动手机号13800138000) # 4. 【核心新增】按业务模块输出日志 LogHelper.info_mod(user_service, 用户模块注册成功 13900139000) LogHelper.debug_mod(order_service, 订单模块创建订单请求) try: 1 / 0 except Exception as e: LogHelper.error_mod(pay_service, 支付模块异常, e) # 切换JSON格式 LogHelper.switch_json_format(True) LogHelper.info_mod(goods_service, 商品模块查询库存) input(回车退出...)