feat(log.py):使用LoggerManager管理所有程序日志

.gitignore

@@ -1,4 +1,5 @@
 config/config.yaml
+logs/*.log
 strm/*
 media/*
 __pycache__/*

app/core/__init__.py

@@ -1 +1,2 @@
-from .config import settings
+from .config import settings
+from .log import logger

app/core/config.py

@@ -4,6 +4,7 @@
 
 import logging
 from pathlib import Path
+from datetime import datetime
 from yaml import safe_load
 
 from version import APP_VERSION
@@ -20,13 +21,15 @@ class SettingManager:
     APP_VERSION: int = APP_VERSION
     # 时区
     TZ: str = "Asia/Shanghai"
+    # 开发者模式
+    DEBUG: bool = False
 
     def __init__(self) -> None:
         """
         初始化 SettingManager 对象
         """
         self.__mkdir()
-        self.__init_logging()
+        self.__load_mode()
 
     def __mkdir(self) -> None:
         """
@@ -36,15 +39,18 @@ def __mkdir(self) -> None:
             if not dir_path.exists():
                 dir_path.mkdir(parents=True, exist_ok=True)
 
-    def __init_logging(self):
+        with self.LOG_DIR as dir_path:
+            if not dir_path.exists():
+                dir_path.mkdir(parents=True, exist_ok=True)
+
+    def __load_mode(self) -> None:
         """
-        初始化 loggging 日志模块内容
+        加载模式
         """
         with self.CONFIG.open(mode="r", encoding="utf-8") as file:
-            log_level = safe_load(file).get("Settings").get("log_level") or "INFO"
-
-        formatter = "[%(levelname)s]%(asctime)s - %(message)s"
-        logging.basicConfig(format=formatter, level=getattr(logging, log_level))
+            is_dev = safe_load(file).get("Settings").get("DEV")
+        if is_dev:
+            self.DEBUG = is_dev
 
     @property
     def BASE_DIR(self) -> Path:
@@ -60,12 +66,30 @@ def CONFIG_DIR(self) -> Path:
         """
         return self.BASE_DIR / "config"
 
+    @property
+    def LOG_DIR(self) -> Path:
+        """
+        日志文件路径
+        """
+        return self.BASE_DIR / "logs"
+
     @property
     def CONFIG(self) -> Path:
         """
         配置文件
         """
         return self.CONFIG_DIR / "config.yaml"
+
+    @property
+    def LOG(self) -> Path:
+        """
+        日志文件
+        """
+        if self.DEBUG:
+            return self.LOG_DIR / "dev.log"
+        else:
+            return self.LOG_DIR / f"{datetime.now().strftime("%Y%m%d")}.log"
+
 
     @property
     def AlistServerList(self) -> list[dict[str, any]]:

app/core/log.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+# encoding: utf-8
+
+import logging
+from logging.handlers import RotatingFileHandler
+
+from .config import settings
+
+
+class LoggerManager:
+    """
+    日志管理
+    """
+
+    def __init__(self) -> None:
+        """
+        初始化 LoggerManager 对象
+        """
+        self.__logger = logging.getLogger(__name__)
+        self.__logger.setLevel(logging.DEBUG)
+
+        if settings.DEBUG:
+            ch_level = logging.DEBUG
+            fh_mode = "w"
+        else:
+            ch_level = logging.INFO
+            fh_mode = "a"
+
+        __console_handler = logging.StreamHandler()
+        __console_handler.setLevel(ch_level)
+
+        __file_handler = RotatingFileHandler(
+            filename=settings.LOG,
+            mode=fh_mode,
+            maxBytes=10 * 1024 * 1024,
+            backupCount=5,
+            encoding="utf-8",
+        )
+        __file_handler.setLevel(logging.INFO)
+
+        __formatter = logging.Formatter(fmt="[%(levelname)s]%(asctime)s - %(message)s")
+
+        __console_handler.setFormatter(__formatter)
+        __file_handler.setFormatter(__formatter)
+
+        self.__logger.addHandler(__console_handler)
+        self.__logger.addHandler(__file_handler)
+
+    def __log(self, method: str, msg: str, *args, **kwargs) -> None:
+        """
+        获取模块的logger
+        :param method: 日志方法
+        :param msg: 日志信息
+        """
+        if hasattr(self.__logger, method):
+            getattr(self.__logger, method)(msg, *args, **kwargs)
+
+    def info(self, msg: str, *args, **kwargs) -> None:
+        """
+        重载info方法
+        """
+        self.__log("info", msg, *args, **kwargs)
+
+    def debug(self, msg: str, *args, **kwargs) -> None:
+        """
+        重载debug方法
+        """
+        self.__log("debug", msg, *args, **kwargs)
+
+    def warning(self, msg: str, *args, **kwargs) -> None:
+        """
+        重载warning方法
+        """
+        self.__log("warning", msg, *args, **kwargs)
+
+    def warn(self, msg: str, *args, **kwargs) -> None:
+        """
+        重载warn方法
+        """
+        self.__log("warning", msg, *args, **kwargs)
+
+    def error(self, msg: str, *args, **kwargs) -> None:
+        """
+        重载error方法
+        """
+        self.__log("error", msg, *args, **kwargs)
+
+    def critical(self, msg: str, *args, **kwargs) -> None:
+        """
+        重载critical方法
+        """
+        self.__log("critical", msg, *args, **kwargs)
+
+
+# 初始化公共日志
+logger = LoggerManager()

app/main.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 # encoding: utf-8
 
-import logging
 import asyncio
 from sys import path
 from os.path import dirname
@@ -10,21 +9,21 @@
 from apscheduler.triggers.cron import CronTrigger
 
 path.append(dirname(__file__))
-from core import settings
+from core import settings, logger
 from modules import Alist2Strm
 
 if __name__ == "__main__":
-    logging.info(f"当前的APP版本是：{settings.APP_VERSION}")
+    logger.info(f"AutoFilm启动中，当前的APP版本是：{settings.APP_VERSION}")
 
     scheduler = AsyncIOScheduler()
 
     for server in settings.AlistServerList:
         cron = server.get("cron")
         if cron:
             scheduler.add_job(Alist2Strm(**server).run,trigger=CronTrigger.from_crontab(cron))
-            logging.info(f"{server["id"]}已被添加至后台任务")
+            logger.info(f"{server["id"]}已被添加至后台任务")
         else:
-            logging.warning(f"{server["id"]}未设置Cron")
+            logger.warning(f"{server["id"]}未设置Cron")
 
     scheduler.start()
 

app/modules/alist2strm.py

@@ -1,7 +1,6 @@
 #!/usr/bin/env python3
 # encoding: utf-8
 
-import logging
 from asyncio import to_thread, Semaphore, TaskGroup
 from contextlib import aclosing
 from os import PathLike
@@ -11,6 +10,8 @@
 from aiofile import async_open
 from alist import AlistClient, AlistPath
 
+from core import logger
+
 
 VIDEO_EXTS: Final = frozenset(
     (".mp4", ".mkv", ".flv", ".avi", ".wmv", ".ts", ".rmvb", ".webm")
@@ -79,23 +80,6 @@ def __init__(
         self.overwrite = overwrite
         self._async_semaphore = Semaphore(max_workers)
 
-        logging.debug(
-            "Alist2Strm配置".center(50, "=")
-            + f"""\
-Alist 地址：  {url!r}
-Alist 用户名：{username!r}
-Alist 密码：  {password!r}
-Alist token： {self.token!r}
-Alist 目录：  {self.source_dir!r}
-输出目录：    {self.target_dir!r}
-平铺模式：  {self.flatten_mode}
-下载字幕：    {subtitle}
-下载图片：    {image}
-下载 NFO：    {nfo}
-覆盖：        {self.overwrite}
-最大并发数：  {max_workers}"""
-        )
-
     async def run(self, /):
         """
         处理主体
@@ -130,7 +114,7 @@ async def __file_processer(self, /, path: AlistPath):
         async with self._async_semaphore:
             try:
                 if local_path.exists() and not self.overwrite:
-                    logging.debug(f"跳过文件：{local_path.name!r}")
+                    logger.debug(f"跳过文件：{local_path.name!r}")
                     return
 
                 download_url = path.get_url(token=self.token)
@@ -145,7 +129,7 @@ async def __file_processer(self, /, path: AlistPath):
                         local_path.as_posix(), mode="w", encoding="utf-8"
                     ) as file:
                         await file.write(download_url)
-                    logging.debug(f"创建文件：{local_path!r}")
+                    logger.debug(f"创建文件：{local_path!r}")
                 else:
                     async with (
                         aclosing(
@@ -158,7 +142,7 @@ async def __file_processer(self, /, path: AlistPath):
                         _write = file.write
                         async for chunk in resp.aiter_bytes(1 << 16):
                             await _write(chunk)
-                    logging.debug(f"下载文件：{local_path!r}")
+                    logger.debug(f"下载文件：{local_path!r}")
             except:
-                logging.exception(f"下载失败: {local_path!r}")
+                logger.warning(f"下载失败: {local_path!r}")
                 raise