update:日志添加版本显示

Author: openrz

requirements.txt

@@ -6,4 +6,5 @@ pydantic==2.5.0
 python-dotenv==1.0.0
 configparser==6.0.0
 asyncio-mqtt==0.16.1
-pycryptodome==3.23.0
+pycryptodome==3.23.0
+loguru==0.7.2

src/server.py

@@ -15,6 +15,7 @@
     JSONRPCProtocol,
 )
 from src.utils.util import get_local_ip
+from .utils import __version__
 from contextlib import asynccontextmanager
 from fastapi.responses import RedirectResponse
 from fastapi.middleware.cors import CORSMiddleware
@@ -90,7 +91,7 @@ async def lifespan(app: FastAPI):
 app = FastAPI(
     title="MCP Endpoint Server",
     description="高效的WebSocket中转服务器",
-    version="1.0.0",
+    version=__version__,
     lifespan=lifespan,
 )
 
@@ -117,7 +118,7 @@ async def root():
     response = JSONRPCProtocol.create_success_response(
         result={
             "message": "MCP Endpoint Server",
-            "version": "1.0.0",
+            "version": __version__,
             "status": "running",
         }
     )
@@ -222,6 +223,11 @@ def signal_handler(signum, frame):
 
 def main():
     """主函数"""
+    # 设置uvicorn日志拦截
+    from .utils.logger import logger_manager
+
+    logger_manager.setup_uvicorn_logging()
+
     # 注册信号处理器
     signal.signal(signal.SIGINT, signal_handler)
     signal.signal(signal.SIGTERM, signal_handler)
@@ -241,6 +247,8 @@ def main():
         reload=debug,
         log_level=config.get("server", "log_level", "INFO").lower(),
         access_log=False,
+        log_config=None,  # 禁用uvicorn默认日志配置
+        use_colors=True,  # 启用颜色支持
     )
 
 

src/utils/__init__.py

@@ -0,0 +1,5 @@
+"""
+工具模块
+"""
+
+__version__ = "0.0.6"

src/utils/logger.py

@@ -2,61 +2,97 @@
 日志管理工具
 """
 
-import logging
-import logging.handlers
 import os
+import sys
+import logging
 from typing import Optional
+from loguru import logger
 from .config import config
 
+# 版本号
+from . import __version__ as VERSION
+
+
+class InterceptHandler(logging.Handler):
+    """拦截标准库日志并转发到loguru"""
+
+    def emit(self, record):
+        # 获取对应的loguru级别
+        try:
+            level = logger.level(record.levelname).name
+        except ValueError:
+            level = record.levelno
+
+        # 查找调用者
+        frame, depth = sys._getframe(6), 6
+        while frame and frame.f_code.co_filename == __file__:
+            frame = frame.f_back
+            depth += 1
+
+        logger.opt(depth=depth, exception=record.exc_info).log(
+            level, record.getMessage()
+        )
+
 
 class LoggerManager:
     """日志管理器"""
 
     def __init__(self):
-        self.logger = None
         self._setup_logger()
 
     def _setup_logger(self):
         """设置日志器"""
-        self.logger = logging.getLogger("mcp_server")
-
-        # 设置日志级别
-        log_level = config.get("server", "log_level", "INFO")
-        self.logger.setLevel(getattr(logging, log_level.upper(), logging.INFO))
-
-        # 清除现有的处理器
-        self.logger.handlers.clear()
-
-        # 创建格式化器
-        formatter = logging.Formatter(
-            "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
-            datefmt="%Y-%m-%d %H:%M:%S",
+        # 移除默认的处理器
+        logger.remove()
+
+        # 自定义格式：时间[版本号][模块路径]-级别-消息
+        # 不同部分使用不同颜色
+        custom_format = (
+            f"<green>{{time:YYMMDD HH:mm:ss}}</green>"
+            f"<blue>[{VERSION}][{{name}}]</blue>"
+            "<level>-{level}-</level>"
+            "<green>{message}</green>"
         )
 
         # 控制台处理器
-        console_handler = logging.StreamHandler()
-        console_handler.setFormatter(formatter)
-        self.logger.addHandler(console_handler)
+        logger.add(
+            sys.stdout,
+            format=custom_format,
+            level=config.get("server", "log_level", "INFO"),
+            colorize=True,
+            backtrace=True,
+            diagnose=True,
+            enqueue=True,
+            catch=True,
+        )
 
-        # 文件处理器
+        # 文件处理器（不带颜色）
         log_file = config.get("logging", "log_file", "logs/mcp_server.log")
         if log_file:
             # 确保日志目录存在
             os.makedirs(os.path.dirname(log_file), exist_ok=True)
 
+            # 文件格式（不带颜色）
+            file_format = f"{{time:YYYY-MM-DD HH:mm:ss}} [{VERSION}][{{name}}] {{level}} - {{message}}"
+
             # 获取文件大小限制
             max_file_size = config.get("logging", "max_file_size", "10MB")
             max_bytes = self._parse_size(max_file_size)
 
             # 获取备份数量
             backup_count = config.getint("logging", "backup_count", 5)
 
-            # 创建轮转文件处理器
-            file_handler = logging.handlers.RotatingFileHandler(
-                log_file, maxBytes=max_bytes, backupCount=backup_count, encoding="utf-8"
+            logger.add(
+                log_file,
+                format=file_format,
+                level=config.get("server", "log_level", "INFO"),
+                rotation=max_bytes,
+                retention=backup_count,
+                compression="zip",
+                encoding="utf-8",
+                enqueue=True,
+                catch=True,
             )
-            file_handler.setFormatter(formatter)
-            self.logger.addHandler(file_handler)
 
     def _parse_size(self, size_str: str) -> int:
         """解析文件大小字符串"""
@@ -70,19 +106,36 @@ def _parse_size(self, size_str: str) -> int:
         else:
             return int(size_str)
 
-    def get_logger(self) -> logging.Logger:
+    def get_logger(self):
         """获取日志器"""
-        return self.logger
+        return logger
 
     def reload(self):
         """重新加载日志配置"""
         self._setup_logger()
 
+    def setup_uvicorn_logging(self):
+        """设置uvicorn日志拦截"""
+        import logging
+
+        # 拦截标准库日志
+        logging.basicConfig(handlers=[InterceptHandler()], level=0, force=True)
+
+        # 直接禁用uvicorn.access日志
+        logging.getLogger("uvicorn.access").disabled = True
+        logging.getLogger("uvicorn.access").propagate = False
+
+        # 拦截其他uvicorn日志
+        for name in logging.root.manager.loggerDict.keys():
+            if not name.startswith("uvicorn.access"):
+                logging.getLogger(name).handlers = []
+                logging.getLogger(name).propagate = True
+
 
 # 全局日志管理器实例
 logger_manager = LoggerManager()
 
 
-def get_logger() -> logging.Logger:
+def get_logger(name: str = "mcp_server"):
     """获取日志器"""
-    return logger_manager.get_logger()
+    return logger.bind(name=name)