mirror of
https://github.com/LifeArchiveProject/WeChatDataAnalysis.git
synced 2026-06-18 15:54:08 +08:00
fix(logging): 日志写入改用 output_dir 并避免重复初始化
- 日志目录改为 get_output_dir()/logs/...(兼容 WECHAT_TOOL_DATA_DIR) - 清理/关闭旧 handlers,避免重复输出和 Windows 文件锁 - get_logger/get_log_file_path 未初始化时自动 setup_logging - 新增 tests/test_logging_config_data_dir.py
This commit is contained in:
@@ -10,8 +10,13 @@ from starlette.exceptions import HTTPException as StarletteHTTPException
|
||||
from starlette.responses import FileResponse
|
||||
from starlette.staticfiles import StaticFiles
|
||||
|
||||
from . import __version__ as APP_VERSION
|
||||
from .logging_config import setup_logging, get_logger
|
||||
|
||||
# 初始化日志系统
|
||||
setup_logging()
|
||||
logger = get_logger(__name__)
|
||||
|
||||
from . import __version__ as APP_VERSION
|
||||
from .path_fix import PathFixRoute
|
||||
from .chat_realtime_autosync import CHAT_REALTIME_AUTOSYNC
|
||||
from .routers.chat import router as _chat_router
|
||||
@@ -30,10 +35,6 @@ from .routers.wrapped import router as _wrapped_router
|
||||
from .sns_stage_timing import add_sns_stage_timing_headers
|
||||
from .wcdb_realtime import WCDB_REALTIME, shutdown as _wcdb_shutdown
|
||||
|
||||
# 初始化日志系统
|
||||
setup_logging()
|
||||
logger = get_logger(__name__)
|
||||
|
||||
app = FastAPI(
|
||||
title="微信数据库解密工具",
|
||||
description="现代化的微信数据库解密工具,支持微信信息检测和数据库解密功能",
|
||||
|
||||
@@ -53,9 +53,9 @@ class WeChatLogger:
|
||||
return cls._instance
|
||||
|
||||
def __init__(self):
|
||||
if not self._initialized:
|
||||
self.setup_logging()
|
||||
WeChatLogger._initialized = True
|
||||
# Lazy-init in `setup_logging()` / accessors to avoid double-initialization when
|
||||
# callers instantiate the manager and then call `setup_logging()` again.
|
||||
pass
|
||||
|
||||
def setup_logging(self, log_level: str = "INFO"):
|
||||
"""设置日志配置"""
|
||||
@@ -66,7 +66,9 @@ class WeChatLogger:
|
||||
|
||||
# 创建日志目录
|
||||
now = datetime.now()
|
||||
log_dir = Path("output/logs") / str(now.year) / f"{now.month:02d}" / f"{now.day:02d}"
|
||||
from .app_paths import get_output_dir
|
||||
|
||||
log_dir = get_output_dir() / "logs" / str(now.year) / f"{now.month:02d}" / f"{now.day:02d}"
|
||||
log_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# 设置日志文件名
|
||||
@@ -77,6 +79,10 @@ class WeChatLogger:
|
||||
root_logger = logging.getLogger()
|
||||
for handler in root_logger.handlers[:]:
|
||||
root_logger.removeHandler(handler)
|
||||
try:
|
||||
handler.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 配置日志格式
|
||||
# 文件格式(无颜色)
|
||||
@@ -109,22 +115,48 @@ class WeChatLogger:
|
||||
|
||||
# 只为uvicorn日志器添加文件处理器,保持其原有的控制台处理器(带颜色)
|
||||
uvicorn_logger = logging.getLogger("uvicorn")
|
||||
for handler in uvicorn_logger.handlers[:]:
|
||||
if isinstance(handler, logging.FileHandler):
|
||||
uvicorn_logger.removeHandler(handler)
|
||||
try:
|
||||
handler.close()
|
||||
except Exception:
|
||||
pass
|
||||
uvicorn_logger.addHandler(file_handler)
|
||||
uvicorn_logger.setLevel(level)
|
||||
|
||||
# 只为uvicorn.access日志器添加文件处理器
|
||||
uvicorn_access_logger = logging.getLogger("uvicorn.access")
|
||||
for handler in uvicorn_access_logger.handlers[:]:
|
||||
if isinstance(handler, logging.FileHandler):
|
||||
uvicorn_access_logger.removeHandler(handler)
|
||||
try:
|
||||
handler.close()
|
||||
except Exception:
|
||||
pass
|
||||
uvicorn_access_logger.addHandler(file_handler)
|
||||
uvicorn_access_logger.setLevel(level)
|
||||
|
||||
# 只为uvicorn.error日志器添加文件处理器
|
||||
uvicorn_error_logger = logging.getLogger("uvicorn.error")
|
||||
for handler in uvicorn_error_logger.handlers[:]:
|
||||
if isinstance(handler, logging.FileHandler):
|
||||
uvicorn_error_logger.removeHandler(handler)
|
||||
try:
|
||||
handler.close()
|
||||
except Exception:
|
||||
pass
|
||||
uvicorn_error_logger.addHandler(file_handler)
|
||||
uvicorn_error_logger.setLevel(level)
|
||||
|
||||
# 配置FastAPI日志器
|
||||
fastapi_logger = logging.getLogger("fastapi")
|
||||
fastapi_logger.handlers = []
|
||||
for handler in fastapi_logger.handlers[:]:
|
||||
fastapi_logger.removeHandler(handler)
|
||||
try:
|
||||
handler.close()
|
||||
except Exception:
|
||||
pass
|
||||
fastapi_logger.addHandler(file_handler)
|
||||
fastapi_logger.addHandler(console_handler)
|
||||
fastapi_logger.setLevel(level)
|
||||
@@ -136,6 +168,8 @@ class WeChatLogger:
|
||||
logger.info(f"日志文件: {self.log_file}")
|
||||
logger.info(f"日志级别: {logging.getLevelName(level)}")
|
||||
logger.info("=" * 60)
|
||||
|
||||
WeChatLogger._initialized = True
|
||||
|
||||
return self.log_file
|
||||
|
||||
@@ -145,6 +179,8 @@ class WeChatLogger:
|
||||
|
||||
def get_log_file_path(self) -> Path:
|
||||
"""获取当前日志文件路径"""
|
||||
if not hasattr(self, "log_file"):
|
||||
self.setup_logging()
|
||||
return self.log_file
|
||||
|
||||
|
||||
@@ -157,10 +193,14 @@ def setup_logging(log_level: str = "INFO") -> Path:
|
||||
def get_logger(name: str) -> logging.Logger:
|
||||
"""获取日志器的便捷函数"""
|
||||
logger_manager = WeChatLogger()
|
||||
if not WeChatLogger._initialized:
|
||||
logger_manager.setup_logging()
|
||||
return logger_manager.get_logger(name)
|
||||
|
||||
|
||||
def get_log_file_path() -> Path:
|
||||
"""获取当前日志文件路径的便捷函数"""
|
||||
logger_manager = WeChatLogger()
|
||||
if not WeChatLogger._initialized:
|
||||
logger_manager.setup_logging()
|
||||
return logger_manager.get_log_file_path()
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
import importlib
|
||||
from pathlib import Path
|
||||
from tempfile import TemporaryDirectory
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(ROOT / "src"))
|
||||
|
||||
|
||||
def _close_logging_handlers() -> None:
|
||||
# Close handlers to avoid Windows temp dir cleanup failures (FileHandler holds a lock).
|
||||
import logging
|
||||
|
||||
for logger_name in ("", "uvicorn", "uvicorn.access", "uvicorn.error", "fastapi"):
|
||||
lg = logging.getLogger(logger_name)
|
||||
for h in lg.handlers[:]:
|
||||
try:
|
||||
h.close()
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
lg.removeHandler(h)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
class TestLoggingConfigDataDir(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self._prev_data_dir = os.environ.get("WECHAT_TOOL_DATA_DIR")
|
||||
self._td = TemporaryDirectory()
|
||||
os.environ["WECHAT_TOOL_DATA_DIR"] = self._td.name
|
||||
|
||||
import wechat_decrypt_tool.app_paths as app_paths
|
||||
import wechat_decrypt_tool.logging_config as logging_config
|
||||
|
||||
importlib.reload(app_paths)
|
||||
importlib.reload(logging_config)
|
||||
|
||||
self.logging_config = logging_config
|
||||
|
||||
def tearDown(self):
|
||||
_close_logging_handlers()
|
||||
|
||||
if self._prev_data_dir is None:
|
||||
os.environ.pop("WECHAT_TOOL_DATA_DIR", None)
|
||||
else:
|
||||
os.environ["WECHAT_TOOL_DATA_DIR"] = self._prev_data_dir
|
||||
self._td.cleanup()
|
||||
|
||||
def test_setup_logging_uses_wechat_tool_data_dir(self):
|
||||
log_file = self.logging_config.setup_logging()
|
||||
|
||||
base = Path(self._td.name) / "output" / "logs"
|
||||
self.assertTrue(log_file.is_relative_to(base))
|
||||
self.assertTrue(log_file.exists())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
Reference in New Issue
Block a user