优化后端日志记录，增强接口访问记录的终端显示效果

ruoyi-fastapi-backend/app.py

@@ -9,4 +9,5 @@
         port=AppConfig.app_port,
         root_path=AppConfig.app_root_path,
         reload=AppConfig.app_reload,
+        access_log=False
     )

ruoyi-fastapi-backend/middlewares/trace_middleware/span.py

@@ -3,11 +3,14 @@
 @author: peng
 @file: span.py
 @time: 2025/1/17  16:57
+@modifier: left666
+@modify_time: 2025/9/25  9:58
 """
-
+import time
 from contextlib import asynccontextmanager
 from starlette.types import Scope, Message
 from .ctx import TraceCtx
+from utils.log_util import logger
 
 
 class Span:
@@ -18,12 +21,18 @@ class Span:
 
     def __init__(self, scope: Scope):
         self.scope = scope
+        self.client_ip = scope.get('client')[0]
+        self.start_time = time.perf_counter()
+        self.status_code = None
 
     async def request_before(self):
         """
         request_before: 处理header信息等, 如记录请求体信息
         """
         TraceCtx.set_id()
+        # 安全获取客户端真实IP
+        if x_forwarded_for := [v.decode() for k, v in self.scope.get('headers', []) if k.lower() == b'x-forwarded-for']:
+            self.client_ip = x_forwarded_for[0].split(',')[0].strip()
 
     async def request_after(self, message: Message):
         """
@@ -44,6 +53,13 @@ async def response(self, message: Message):
         """
         if message['type'] == 'http.response.start':
             message['headers'].append((b'request-id', TraceCtx.get_id().encode()))
+            self.status_code = message['status']  # 存储状态码
+        elif message['type'] == 'http.response.body':
+            if not message.get('more_body', False):  # 是最后一部分响应体时
+                # 计算请求处理时间
+                duration = round((time.perf_counter() - self.start_time) * 1000)
+                with logger.contextualize(status_code=self.status_code, duration=duration):
+                    logger.log("request", f"{self.client_ip} {self.scope.get('method')} {self.scope.get('path')}")
         return message
 
 

ruoyi-fastapi-backend/utils/log_util.py

@@ -1,16 +1,26 @@
+"""
+@modifier: left666
+@modify_time: 2025/10/9  9:00
+"""
 import os
 import sys
 import time
 from loguru import logger as _logger
-from typing import Dict
-from middlewares.trace_middleware import TraceCtx
 
 
 class LoggerInitializer:
+
+    format_str = (
+        '<green>{time:YYYY-MM-DD HH:mm:ss.S}</green> | '
+        '<cyan>{trace_id}</cyan> | '
+        '<level>{level: <8}</level> | '
+        '<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - '
+        '<level>{message}</level>'
+    )  # 自定义日志格式
+
     def __init__(self):
         self.log_path = os.path.join(os.getcwd(), 'logs')
         self.__ensure_log_directory_exists()
-        self.log_path_error = os.path.join(self.log_path, f'{time.strftime("%Y-%m-%d")}_error.log')
 
     def __ensure_log_directory_exists(self):
         """
@@ -20,36 +30,88 @@ def __ensure_log_directory_exists(self):
             os.mkdir(self.log_path)
 
     @staticmethod
-    def __filter(log: Dict):
+    def __sift_out_common(log) -> bool:
         """
-        自定义日志过滤器，添加trace_id
+        筛选出非request日志，并添加trace_id
         """
-        log['trace_id'] = TraceCtx.get_id()
-        return log
+        if log["level"].name != "request":
+            from middlewares.trace_middleware import TraceCtx
+            log['trace_id'] = TraceCtx.get_id()
+            return True
+        return False
+
+    @staticmethod
+    def __status_code_color(status_code: int) -> str:
+        """根据状态码返回对应的颜色标签"""
+        if status_code < 200:
+            return "blue"  # 1xx 信息响应
+        elif 200 <= status_code < 300:
+            return "green"  # 2xx 成功
+        elif 300 <= status_code < 400:
+            return "cyan"  # 3xx 重定向
+        elif 400 <= status_code < 500:
+            return "yellow"  # 4xx 客户端错误
+        else:
+            return "red"  # 5xx 服务器错误
+
+    def __format_request_log(self, log) -> str:
+        """请求记录格式处理器"""
+        status_code = log["extra"].get("status_code", 0)
+        duration = log["extra"].get("duration", 0)
+        if status_code == 0:
+            return self.format_str
+        color = self.__status_code_color(status_code)
+        return (
+            "<green>{time:YYYY-MM-DD HH:mm:ss.S}</green> |- "
+            "<level>{message}</level>"
+            f"<{color}>  {status_code}</{color}>"
+            f" -| <level>{duration}ms</level>\n"
+        )
 
     def init_log(self):
         """
         初始化日志配置
         """
-        # 自定义日志格式
-        format_str = (
-            '<green>{time:YYYY-MM-DD HH:mm:ss.SSS}</green> | '
-            '<cyan>{trace_id}</cyan> | '
-            '<level>{level: <8}</level> | '
-            '<cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - '
-            '<level>{message}</level>'
+        _logger.level(name="request", no=23, color="<magenta>")  # 自定义等级(介于info与success之间)
+        _logger.remove()  # 移除后重新添加sys.stderr, 目的: 控制台输出与文件日志内容和结构一致
+
+        # 添加请求记录控制台输出
+        _logger.add(
+            sys.stderr,
+            filter=lambda log: log["level"].name == "request",
+            format=self.__format_request_log,
+            colorize=True,  # 终端着色
+            enqueue=True  # 启用异步安全的日志记录
+        )
+        # 添加请求记录文件
+        _logger.add(
+            os.path.join(self.log_path, f'request.log'),
+            rotation="50 MB",  # 文件最大50MB
+            delay=True,  # 延迟到第一条记录消息时再创建文件
+            filter=lambda log: log["level"].name == "request",
+            format=self.__format_request_log,
+            encoding="utf-8",
+            enqueue=True,
+            compression="zip"
+        )
+        # 添加普通日志控制台输出
+        _logger.add(
+            sys.stderr,
+            filter=self.__sift_out_common,
+            format=self.format_str,
+            colorize=True,
+            enqueue=True
         )
-        _logger.remove()
-        # 移除后重新添加sys.stderr, 目的: 控制台输出与文件日志内容和结构一致
-        _logger.add(sys.stderr, filter=self.__filter, format=format_str, enqueue=True)
+        # 添加普通日志文件
         _logger.add(
-            self.log_path_error,
-            filter=self.__filter,
-            format=format_str,
-            rotation='50MB',
-            encoding='utf-8',
+            os.path.join(self.log_path, f'{time.strftime("%Y-%m-%d")}_app.log'),
+            rotation="50 MB",
+            delay=True,
+            filter=self.__sift_out_common,
+            format=self.format_str,
+            encoding="utf-8",
             enqueue=True,
-            compression='zip',
+            compression="zip"
         )
 
         return _logger