""" 日志模块 - 使用环境变量配置 """ import os import sys import threading from datetime import datetime from collections import deque import atexit # 日志级别定义 LOG_LEVELS = {"debug": 0, "info": 1, "warning": 2, "error": 3, "critical": 4} # 文件写入状态标志（仅由 writer 线程修改，无需锁保护） _file_writing_disabled = False _disable_reason = None # 全局文件句柄（仅由 writer 线程访问，无需文件锁） _log_file_handle = None # ----------------------------------------------------------------- # 高性能无锁队列：用 deque + Condition 替代 Queue # deque.append / deque.popleft 在 CPython 中受 GIL 保护，是原子操作， # 不需要额外的 Lock 做入队保护，只用 Condition 做"有数据"通知。 # ----------------------------------------------------------------- _log_deque: deque = deque() _deque_condition = threading.Condition(threading.Lock()) _writer_thread = None _writer_running = False # ----------------------------------------------------------------- # 缓存日志级别，避免每次都读 os.getenv（高并发热路径） # ----------------------------------------------------------------- _cached_log_level: int = LOG_LEVELS["info"] _cached_log_file: str = "log.txt" # ENABLE_LOG=0/false/no/off 时彻底关闭日志 _log_enabled: bool = True def _refresh_config(): """从环境变量刷新缓存配置（模块加载时及需要时调用）""" global _cached_log_level, _cached_log_file, _log_enabled level = os.getenv("LOG_LEVEL", "info").lower() _cached_log_level = LOG_LEVELS.get(level, LOG_LEVELS["info"]) _cached_log_file = os.getenv("LOG_FILE", "log.txt") _log_enabled = os.getenv("ENABLE_LOG", "1").strip().lower() not in ("0", "false", "no", "off") def _get_current_log_level() -> int: return _cached_log_level def _get_log_file_path() -> str: return _cached_log_file # ----------------------------------------------------------------- # 文件句柄管理（仅在 writer 线程内调用，不需要 _file_lock） # ----------------------------------------------------------------- def _close_log_file(): global _log_file_handle if _log_file_handle is not None: try: _log_file_handle.flush() _log_file_handle.close() except Exception: pass finally: _log_file_handle = None def _open_log_file(mode: str = "a") -> bool: global _log_file_handle, _file_writing_disabled, _disable_reason _close_log_file() try: # 使用较大缓冲区（64 KB），由 writer 线程定期 flush，减少系统调用 _log_file_handle = open(_cached_log_file, mode, encoding="utf-8", buffering=65536) return True except (PermissionError, OSError, IOError) as e: _file_writing_disabled = True _disable_reason = str(e) print(f"Warning: Cannot open log file, disabling file writing: {e}", file=sys.stderr) print("Log messages will continue to display in console only.", file=sys.stderr) return False except Exception as e: print(f"Warning: Failed to open log file: {e}", file=sys.stderr) return False def _clear_log_file(): """清空日志文件（启动时调用，此时 writer 线程尚未启动，直接操作安全）""" global _file_writing_disabled, _disable_reason try: with open(_cached_log_file, "w", encoding="utf-8") as f: pass # 覆盖清空 _open_log_file("a") except (PermissionError, OSError, IOError) as e: _file_writing_disabled = True _disable_reason = str(e) print( f"Warning: File system appears to be read-only or permission denied. " f"Disabling log file writing: {e}", file=sys.stderr, ) print("Log messages will continue to display in console only.", file=sys.stderr) except Exception as e: print(f"Warning: Failed to clear log file: {e}", file=sys.stderr) # ----------------------------------------------------------------- # Writer 线程：批量从 deque 取出并写入，减少系统调用次数 # ----------------------------------------------------------------- _BATCH_SIZE = 1000 # 单次最多批量写入条数 _FLUSH_INTERVAL = 2 # 秒：无新消息时强制 flush 周期 def _log_writer_worker(): global _writer_running last_flush_time = 0.0 while True: # 等待数据或超时 with _deque_condition: if not _log_deque and _writer_running: _deque_condition.wait(timeout=_FLUSH_INTERVAL) # 批量取出 batch = [] for _ in range(_BATCH_SIZE): if _log_deque: batch.append(_log_deque.popleft()) else: break if batch and not _file_writing_disabled: # 一次 write 调用搞定整批，最大化减少系统调用 chunk = "\n".join(batch) + "\n" try: if _log_file_handle is None: _open_log_file("a") if _log_file_handle is not None: _log_file_handle.write(chunk) except Exception as e: print(f"Warning: Failed to write log batch: {e}", file=sys.stderr) _close_log_file() try: _open_log_file("a") except Exception: pass # 定时 flush now = _now_ts() if now - last_flush_time >= _FLUSH_INTERVAL: if _log_file_handle is not None: try: _log_file_handle.flush() except Exception: pass last_flush_time = now # 退出条件：已停止 + deque 已清空 if not _writer_running and not _log_deque: break # 最终 flush & close if _log_file_handle is not None: try: _log_file_handle.flush() except Exception: pass _close_log_file() def _now_ts() -> float: import time return time.monotonic() def _start_writer_thread(): global _writer_thread, _writer_running if _writer_thread is None or not _writer_thread.is_alive(): _writer_running = True _writer_thread = threading.Thread(target=_log_writer_worker, daemon=True, name="LogWriter") _writer_thread.start() def _stop_writer_thread(): global _writer_running _writer_running = False # 唤醒 writer 线程让它能感知退出信号 with _deque_condition: _deque_condition.notify_all() if _writer_thread and _writer_thread.is_alive(): _writer_thread.join(timeout=3.0) # ----------------------------------------------------------------- # 入队（热路径，极轻量） # ----------------------------------------------------------------- _MAX_QUEUE_SIZE = 5000 # 防止极端情况内存无限增长 def _write_to_file(message: str): if _file_writing_disabled: return # deque.append 在 CPython 受 GIL 保护，无需额外锁 if len(_log_deque) >= _MAX_QUEUE_SIZE: return # 过载保护：丢弃而非阻塞 _log_deque.append(message) # 非阻塞通知 writer（acquire 失败直接跳过，不影响主线程） if _deque_condition.acquire(blocking=False): try: _deque_condition.notify() finally: _deque_condition.release() # ----------------------------------------------------------------- # 核心日志函数（热路径） # ----------------------------------------------------------------- def _log(level: str, message: str): # 最快短路：日志整体已禁用时直接返回，零开销 if not _log_enabled: return level = level.lower() level_val = LOG_LEVELS.get(level) if level_val is None: print(f"Warning: Unknown log level '{level}'", file=sys.stderr) return # 热路径：直接与缓存值比较，无函数调用开销 if level_val < _cached_log_level: return timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") entry = f"[{timestamp}] [{level.upper()}] {message}" if level in ("error", "critical"): print(entry, file=sys.stderr) else: print(entry) _write_to_file(entry) def set_log_level(level: str): """动态设置日志级别（同时更新缓存）""" global _cached_log_level level = level.lower() if level not in LOG_LEVELS: print(f"Warning: Unknown log level '{level}'. Valid levels: {', '.join(LOG_LEVELS.keys())}") return False _cached_log_level = LOG_LEVELS[level] return True class Logger: """支持 log('info', 'msg') 和 log.info('msg') 两种调用方式""" def __call__(self, level: str, message: str): _log(level, message) def debug(self, message: str): _log("debug", message) def info(self, message: str): _log("info", message) def warning(self, message: str): _log("warning", message) def error(self, message: str): _log("error", message) def critical(self, message: str): _log("critical", message) def get_current_level(self) -> str: current_level = _get_current_log_level() for name, value in LOG_LEVELS.items(): if value == current_level: return name return "info" def get_log_file(self) -> str: return _get_log_file_path() def close(self): """手动关闭（优雅退出用）""" _stop_writer_thread() def get_queue_size(self) -> int: return len(_log_deque) # 导出全局日志实例 log = Logger() # 导出的公共接口 __all__ = ["log", "set_log_level", "LOG_LEVELS"] # 模块加载时：读取配置缓存 → 清空日志文件 → 启动 writer 线程 _refresh_config() if _log_enabled: _clear_log_file() _start_writer_thread() # 注册退出清理 atexit.register(_stop_writer_thread) # 使用说明: # 1. 设置日志级别: export LOG_LEVEL=debug (或在 .env 中设置) # 2. 设置日志文件: export LOG_FILE=log.txt (或在 .env 中设置) # 3. 日志级别已缓存，热路径零 os.getenv 调用 # 4. 写入线程批量处理（最多 200 条/次），64 KB 缓冲区，每 0.5 s flush 一次 # 5. 队列上限 5000 条，超出时丢弃新日志（过载保护，不阻塞主线程） # 6. 动态调整级别：set_log_level('debug') 立即生效 # 7. 彻底关闭日志（最高性能）：export ENABLE_LOG=0 (或 false/no/off) # 关闭后不会启动 writer 线程、不写文件、不打印控制台，_log 直接 return