🦄 refactor: optimize log output

backend/common/log.py

@@ -2,53 +2,120 @@
 # -*- coding: utf-8 -*-
 from __future__ import annotations
 
+import inspect
+import logging
 import os
 
-from typing import TYPE_CHECKING
-
 from loguru import logger
 
 from backend.core import path_conf
 from backend.core.conf import settings
 
-if TYPE_CHECKING:
-    import loguru
-
-
-class Logger:
-    def __init__(self):
-        self.log_path = path_conf.LOG_DIR
-
-    def log(self) -> loguru.Logger:
-        if not os.path.exists(self.log_path):
-            os.mkdir(self.log_path)
-
-        # 日志文件
-        log_stdout_file = os.path.join(self.log_path, settings.LOG_STDOUT_FILENAME)
-        log_stderr_file = os.path.join(self.log_path, settings.LOG_STDERR_FILENAME)
-
-        # loguru 日志: https://loguru.readthedocs.io/en/stable/api/logger.html#loguru._logger.Logger.add
-        log_config = dict(rotation='10 MB', retention='15 days', compression='tar.gz', enqueue=True)
-        # stdout
-        logger.add(
-            log_stdout_file,
-            level='INFO',
-            filter=lambda record: record['level'].name == 'INFO' or record['level'].no <= 25,
-            **log_config,
-            backtrace=False,
-            diagnose=False,
-        )
-        # stderr
-        logger.add(
-            log_stderr_file,
-            level='ERROR',
-            filter=lambda record: record['level'].name == 'ERROR' or record['level'].no >= 30,
-            **log_config,
-            backtrace=True,
-            diagnose=True,
-        )
-
-        return logger
-
-
-log = Logger().log()
+
+class InterceptHandler(logging.Handler):
+    """
+    Default handler from examples in loguru documentaion.
+    See https://loguru.readthedocs.io/en/stable/overview.html#entirely-compatible-with-standard-logging
+    """
+
+    def emit(self, record: logging.LogRecord):
+        # Get corresponding Loguru level if it exists
+        try:
+            level = logger.level(record.levelname).name
+        except ValueError:
+            level = record.levelno
+
+        # Find caller from where originated the logged message
+        # frame, depth = logging.currentframe(), 2
+        # print(frame, depth)
+        # while frame and frame.f_code and frame.f_code.co_filename == logging.__file__:
+        #     frame = frame.f_back
+        #     depth += 1
+
+        # Find caller from where originated the logged message.
+        # 换成inspect获取
+        frame, depth = inspect.currentframe(), 0
+        while frame and (depth == 0 or frame.f_code.co_filename == logging.__file__):
+            frame = frame.f_back
+            depth += 1
+
+        logger.opt(depth=depth, exception=record.exc_info).log(level, record.getMessage())
+
+
+def setup_logging(log_level: str = 'INFO'):
+    """
+    from https://pawamoy.github.io/posts/unify-logging-for-a-gunicorn-uvicorn-app/
+    https://github.com/pawamoy/pawamoy.github.io/issues/17
+    """
+
+    from sys import stderr, stdout
+
+    logger.configure(handlers=[{'sink': stdout, 'serialize': 0, 'level': log_level}])
+    logger.configure(handlers=[{'sink': stderr, 'serialize': 0, 'level': log_level}])
+
+    # intercept everything at the root logger
+    logging.root.handlers = [InterceptHandler()]
+    logging.root.setLevel(log_level)
+
+    # remove every other logger's handlers
+    # and propagate to root logger
+    # noinspection PyUnresolvedReferences
+
+    # https://segmentfault.com/a/1190000042197982
+    # uvicorn的两种方式启动,命令行正常,代码方式启动uvicorn.access可能会缺失
+    # 如果缺失则添加
+    # if "uvicorn.error" in logging.root.manager.loggerDict.keys() and "uvicorn.access" not in logging.root.manager.loggerDict.keys():
+    #     uvicorn_access_logger = logging.getLogger("uvicorn.access")
+    #     uvicorn_access_logger.handlers = []
+
+    for name in logging.root.manager.loggerDict.keys():
+        # if 'uvicorn' in name :
+        logging.getLogger(name).handlers = []
+        # propagate是可选项，其默认是为1，表示消息将会传递给高层次logger的handler，通常我们需要指定其值为0
+        # 设置1，传给loguru
+
+        # logging.info(f"{logging.getLogger(name)}, {logging.getLogger(name).propagate}")
+        if 'uvicorn.access' in name or 'watchfiles.main' in name:
+            logging.getLogger(name).propagate = False
+        else:
+            logging.getLogger(name).propagate = True
+
+        logging.debug(f'{logging.getLogger(name)}, {logging.getLogger(name).propagate}')
+
+    logger.remove()
+    logger.configure(handlers=[{'sink': stdout, 'serialize': 0, 'level': log_level}])
+    logger.configure(handlers=[{'sink': stderr, 'serialize': 0, 'level': log_level}])
+
+
+def set_customize_logfile():
+    log_path = path_conf.LOG_DIR
+    if not os.path.exists(log_path):
+        os.mkdir(log_path)
+
+    # 日志文件
+    log_stdout_file = os.path.join(log_path, settings.LOG_STDOUT_FILENAME)
+    log_stderr_file = os.path.join(log_path, settings.LOG_STDERR_FILENAME)
+
+    # loguru 日志: https://loguru.readthedocs.io/en/stable/api/logger.html#loguru._logger.Logger.add
+    log_config = dict(rotation='10 MB', retention='15 days', compression='tar.gz', enqueue=True)
+    # stdout
+    logger.add(
+        log_stdout_file,
+        level='INFO',
+        filter=lambda record: record['level'].name == 'INFO' or record['level'].no <= 25,
+        **log_config,
+        backtrace=False,
+        diagnose=False,
+    )
+    # stderr
+    logger.add(
+        log_stderr_file,
+        level='ERROR',
+        filter=lambda record: record['level'].name == 'ERROR' or record['level'].no >= 30,
+        **log_config,
+        backtrace=True,
+        diagnose=True,
+    )
+
+
+log = logger

backend/core/registrar.py

@@ -56,6 +56,9 @@ def register_app():
         lifespan=register_init,
     )
 
+    # 日志
+    register_logger(app)
+
     # 静态文件
     register_static_file(app)
 
@@ -74,6 +77,18 @@ def register_app():
     return app
 
 
+def register_logger(app: FastAPI) -> None:
+    """
+    处理日志的【】配置初始化操作
+    :param app:
+    :return:
+    """
+    from backend.common.log import set_customize_logfile, setup_logging
+
+    setup_logging('INFO')
+    set_customize_logfile()
+
+
 def register_static_file(app: FastAPI):
     """
     静态文件交互开发模式, 生产使用 nginx 静态资源服务

backend/main.py

@@ -13,6 +13,11 @@
     # 如果你喜欢在 IDE 中进行 DEBUG，main 启动方法会很有帮助
     # 如果你喜欢通过日志方式进行调试，建议使用 fastapi cli 方式启动服务
     try:
-        uvicorn.run(app=f'{Path(__file__).stem}:app', reload=True)
+        config = uvicorn.Config(app=f'{Path(__file__).stem}:app', reload=True)
+        server = uvicorn.Server(config)
+        from backend.common.log import setup_logging
+
+        setup_logging('INFO')
+        server.run()
     except Exception as e:
         raise e