♻ 重写模块管理器

core/admin/models.py

@@ -5,4 +5,4 @@ from sqlmodel import SQLModel, Field
 
 class Admin(SQLModel, table=True):
     id: Optional[int] = Field(default=None, primary_key=True)
-    user_id: int = Field()
+    user_id: int = Field()

main.py

@@ -38,7 +38,7 @@ def main() -> None:
     # 传入服务并启动
     Log.info("正在启动服务")
     services = ServicesManager(mysql, redis, browser)
-    services.refresh_list("./core/*")
+    services.refresh_list("core/*")
     services.import_module()
     services.add_service()
 

models/base.py

@@ -1,6 +1,7 @@
 import imghdr
+import os
 from enum import Enum
-from typing import Union
+from typing import Union, Optional
 
 from models.baseobject import BaseObject
 
@@ -64,3 +65,29 @@ class GameItem(BaseObject):
 
     __slots__ = ("name", "type", "value", "item_id")
 
+
+class ModuleInfo:
+
+    def __init__(self, file_name: Optional[str] = None, plugin_name: Optional[str] = None,
+                 relative_path: Optional[str] = None):
+        self.relative_path = relative_path
+        self.module_name = plugin_name
+        self.file_name = file_name
+        if file_name is None:
+            if relative_path is None:
+                raise ValueError("file_name 和 relative_path 都不能为空")
+            self.file_name = os.path.basename(relative_path)
+        if plugin_name is None:
+            self.module_name, _ = os.path.splitext(self.file_name)
+
+    @property
+    def package_path(self) -> str:
+        if self.relative_path is None:
+            return ""
+        if os.path.isdir(self.relative_path):
+            return self.relative_path.replace(os.sep, ".")
+        root, _ = os.path.splitext(self.relative_path)
+        return root.replace(os.sep, ".")
+
+    def __str__(self):
+        return self.module_name

utils/helpers.py

@@ -7,9 +7,9 @@ import genshin
 import httpx
 from genshin import Client, types
 from httpx import UnsupportedProtocol
-from service.cookies.services import CookiesService
-from service.user.services import UserService
 
+from core.cookies.services import CookiesService
+from core.user.services import UserService
 from logger import Log
 from models.base import RegionEnum
 

utils/job/manager.py

@@ -1,12 +1,9 @@
-import os
+from typing import List
-from glob import glob
-from importlib import import_module
-from os import path
-from typing import List, Union
 
 from telegram.ext import Application
 
 from logger import Log
+from utils.manager import ModulesManager
 
 JobsClass: List[object] = []
 
@@ -23,43 +20,12 @@ def listener_jobs_class():
     return decorator
 
 
-class JobsManager:
+class JobsManager(ModulesManager):
     def __init__(self):
+        super().__init__()
         self.job_list: List[str] = []  # 用于存储文件名称
         self.exclude_list: List[str] = []
-
+        self.manager_name = "定时任务管理器"
-    def refresh_list(self, plugin_paths):
-        self.job_list.clear()
-        plugin_paths = glob(plugin_paths)
-        for plugin_path in plugin_paths:
-            if plugin_path.startswith('__'):
-                continue
-            module_name = path.basename(path.normpath(plugin_path))
-            root, ext = os.path.splitext(module_name)
-            if ext == ".py":
-                self.job_list.append(root)
-
-    def add_exclude(self, exclude: Union[str, List[str]]):
-        if isinstance(exclude, str):
-            self.exclude_list.append(exclude)
-        elif isinstance(exclude, list):
-            self.exclude_list.extend(exclude)
-        else:
-            raise TypeError
-
-    def import_module(self):
-        for job_name in self.job_list:
-            if job_name not in self.exclude_list:
-                try:
-                    import_module(f"jobs.{job_name}")
-                except ImportError as exc:
-                    Log.warning(f"Job模块 {job_name} 导入失败", exc)
-                except ImportWarning as exc:
-                    Log.warning(f"Job模块 {job_name} 加载成功但有警告", exc)
-                except BaseException as exc:
-                    Log.warning(f"Job模块 {job_name} 加载失败", exc)
-                else:
-                    Log.debug(f"Job模块 {job_name} 加载成功")
 
     @staticmethod
     def add_handler(application: Application):

utils/job/register.py

@@ -8,7 +8,7 @@ def register_job(application: Application):
     Log.info("正在加载Job管理器")
     jobs_manager = JobsManager()
 
-    jobs_manager.refresh_list("./jobs/*")
+    jobs_manager.refresh_list("jobs/*")
 
     # 忽略内置模块
     jobs_manager.add_exclude(["base"])

utils/manager.py

@@ -0,0 +1,55 @@
+import os
+from glob import glob
+from importlib import import_module
+from os import path
+from typing import List, Union
+
+from logger import Log
+from models.base import ModuleInfo
+
+
+class ModulesManager:
+    def __init__(self):
+        self.manager_name: str = "模块管理器"
+        self.modules_list: List[ModuleInfo] = []  # 用于存储文件名称
+        self.exclude_list: List[str] = []
+
+    def clear(self):
+        self.modules_list.clear()
+
+    def refresh_list(self, pathname: str):
+        path_list = glob(pathname)
+        for temp_path in path_list:
+            if temp_path.startswith('__'):
+                continue
+            if os.path.isdir(temp_path):
+                self.modules_list.append(ModuleInfo(relative_path=temp_path))
+            else:
+                module_name = path.basename(path.normpath(temp_path))
+                root, ext = os.path.splitext(module_name)
+                if ext == ".py":
+                    self.modules_list.append(ModuleInfo(relative_path=temp_path))
+
+    def add_exclude(self, exclude: Union[str, List[str]]):
+        if isinstance(exclude, str):
+            self.exclude_list.append(exclude)
+        elif isinstance(exclude, list):
+            self.exclude_list.extend(exclude)
+        else:
+            raise TypeError
+
+    def import_module(self):
+        module_name_list: List[str] = []
+        for module_info in self.modules_list:
+            if module_info.module_name not in self.exclude_list:
+                try:
+                    import_module(f"{module_info.package_path}")
+                except ImportError as exc:
+                    Log.warning(f"{self.manager_name}加载 {module_info} 失败", exc)
+                except ImportWarning as exc:
+                    Log.warning(f"{self.manager_name}加载 {module_info} 成功但有警告", exc)
+                except BaseException as exc:
+                    Log.warning(f"{self.manager_name}加载 {module_info} 失败", exc)
+                else:
+                    module_name_list.append(module_info.module_name)
+        Log.info(self.manager_name + "加载模块: " + ", ".join(module_name_list))

utils/plugins/manager.py

@@ -1,12 +1,9 @@
-import os
+from typing import List
-from glob import glob
-from importlib import import_module
-from os import path
-from typing import List, Union
 
 from telegram.ext import Application
 
 from logger import Log
+from utils.manager import ModulesManager
 
 PluginsClass: List[object] = []
 
@@ -23,46 +20,11 @@ def listener_plugins_class():
     return decorator
 
 
-class PluginsManager:
+class PluginsManager(ModulesManager):
+
     def __init__(self):
-        self.plugin_list: List[str] = []  # 用于存储文件名称
+        super().__init__()
-        self.exclude_list: List[str] = []
+        self.manager_name = "插件管理器"
-
-    def refresh_list(self, plugin_paths):
-        plugin_paths = glob(plugin_paths)
-        for plugin_path in plugin_paths:
-            if plugin_path.startswith('__'):
-                continue
-            if os.path.isdir(plugin_path):
-                plugin_path = os.path.basename(plugin_path)
-                self.plugin_list.append(plugin_path)
-                continue
-            module_name = path.basename(path.normpath(plugin_path))
-            root, ext = os.path.splitext(module_name)
-            if ext == ".py":
-                self.plugin_list.append(root)
-
-    def add_exclude(self, exclude: Union[str, List[str]]):
-        if isinstance(exclude, str):
-            self.exclude_list.append(exclude)
-        elif isinstance(exclude, list):
-            self.exclude_list.extend(exclude)
-        else:
-            raise TypeError
-
-    def import_module(self):
-        for plugin_name in self.plugin_list:
-            if plugin_name not in self.exclude_list:
-                try:
-                    import_module(f"plugins.{plugin_name}")
-                except ImportError as exc:
-                    Log.warning(f"插件 {plugin_name} 导入失败", exc)
-                except ImportWarning as exc:
-                    Log.warning(f"插件 {plugin_name} 加载成功但有警告", exc)
-                except BaseException as exc:
-                    Log.warning(f"插件 {plugin_name} 加载失败", exc)
-                else:
-                    Log.debug(f"插件 {plugin_name} 加载成功")
 
     @staticmethod
     def add_handler(application: Application):

utils/plugins/register.py

@@ -33,7 +33,7 @@ def register_plugin_handlers(application: Application):
     Log.info("正在加载插件管理器")
     plugins_manager = PluginsManager()
 
-    plugins_manager.refresh_list("./plugins/genshin/*")
+    plugins_manager.refresh_list("plugins/genshin/*")
 
     # 忽略内置模块
     # plugins_manager.add_exclude(["start", "auth", "inline", "errorhandler"])

utils/service/manager.py

@@ -1,12 +1,10 @@
 import inspect
-import os
+from typing import List, Dict
-from glob import glob
-from importlib import import_module
-from typing import List, Union, Dict
 
 from logger import Log
 from models.types import Func
 from utils.aiobrowser import AioBrowser
+from utils.manager import ModulesManager
 from utils.mysql import MySQL
 from utils.redisdb import RedisDB
 
@@ -26,43 +24,15 @@ def listener_service():
     return decorator
 
 
-class ServicesManager:
+class ServicesManager(ModulesManager):
     def __init__(self, mysql: MySQL, redis: RedisDB, browser: AioBrowser):
+        super().__init__()
         self.browser = browser
         self.redis = redis
         self.mysql = mysql
         self.services_list: List[str] = []
         self.exclude_list: List[str] = []
-
+        self.manager_name = "核心服务管理器"
-    def refresh_list(self, app_paths):
-        self.services_list.clear()
-        app_paths = glob(app_paths)
-        for app_path in app_paths:
-            if os.path.isdir(app_path):
-                app_path = os.path.basename(app_path)
-                self.services_list.append(app_path)
-
-    def add_exclude(self, exclude: Union[str, List[str]]):
-        if isinstance(exclude, str):
-            self.exclude_list.append(exclude)
-        elif isinstance(exclude, list):
-            self.exclude_list.extend(exclude)
-        else:
-            raise TypeError
-
-    def import_module(self):
-        for services_name in self.services_list:
-            if services_name not in self.exclude_list:
-                try:
-                    import_module(f"apps.{services_name}")
-                except ImportError as exc:
-                    Log.warning(f"Service模块 {services_name} 导入失败", exc)
-                except ImportWarning as exc:
-                    Log.warning(f"Service模块 {services_name} 加载成功但有警告", exc)
-                except Exception as exc:
-                    Log.warning(f"Service模块 {services_name} 加载失败", exc)
-                else:
-                    Log.debug(f"Service模块 {services_name} 加载成功")
 
     def add_service(self):
         for func in ServicesFunctions: