✨ 将cd，block，count限制复原配置文件

zhenxun/builtin_plugins/hooks/_auth_checker.py

@@ -88,7 +88,7 @@ def unblock(
     @classmethod
     async def check(
         cls,
-        module: str,
+        module_path: str,
         user_id: str,
         group_id: str | None,
         channel_id: str | None,
@@ -106,17 +106,17 @@ async def check(
         异常:
             IgnoredException: IgnoredException
         """
-        if limit_model := cls.cd_limit.get(module):
+        if limit_model := cls.cd_limit.get(module_path):
             await cls.__check(limit_model, user_id, group_id, channel_id, session)
-        if limit_model := cls.block_limit.get(module):
+        if limit_model := cls.block_limit.get(module_path):
             await cls.__check(limit_model, user_id, group_id, channel_id, session)
-        if limit_model := cls.count_limit.get(module):
+        if limit_model := cls.count_limit.get(module_path):
             await cls.__check(limit_model, user_id, group_id, channel_id, session)
 
     @classmethod
     async def __check(
         cls,
-        limit_model: Limit,
+        limit_model: Limit | None,
         user_id: str,
         group_id: str | None,
         channel_id: str | None,
@@ -291,12 +291,14 @@ async def auth_limit(self, plugin: PluginInfo, session: EventSession):
         if not group_id:
             group_id = channel_id
             channel_id = None
-        limit_list: list[PluginLimit] = await plugin.plugin_limit.all()  # type: ignore
+        limit_list: list[PluginLimit] = await plugin.plugin_limit.filter(
+            status=True
+        ).all()  # type: ignore
         for limit in limit_list:
             LimitManage.add_limit(limit)
         if user_id:
             await LimitManage.check(
-                plugin.module, user_id, group_id, channel_id, session
+                plugin.module_path, user_id, group_id, channel_id, session
             )
 
     async def auth_plugin(

zhenxun/builtin_plugins/init/init_config.py

@@ -1,16 +1,16 @@
 from pathlib import Path
 
 import nonebot
-from nonebot import get_loaded_plugins
-from nonebot.drivers import Driver
-from nonebot.plugin import Plugin
 from ruamel.yaml import YAML
+from nonebot.plugin import Plugin
+from nonebot.drivers import Driver
+from nonebot import get_loaded_plugins
 from ruamel.yaml.comments import CommentedMap
 
+from zhenxun.services.log import logger
 from zhenxun.configs.config import Config
-from zhenxun.configs.path_config import DATA_PATH
 from zhenxun.configs.utils import RegisterConfig
-from zhenxun.services.log import logger
+from zhenxun.configs.path_config import DATA_PATH
 
 _yaml = YAML(pure=True)
 _yaml.allow_unicode = True
@@ -72,15 +72,14 @@ def _generate_simple_config():
                     Config.set_config(module, k, _data[module][k])
                 _tmp_data[module][k] = Config.get_config(module, k)
             except AttributeError as e:
-                raise AttributeError(f"{e}\n" + "可能为config.yaml配置文件填写不规范")
+                raise AttributeError(f"{e}\n可能为config.yaml配置文件填写不规范") from e
     Config.save()
     temp_file = DATA_PATH / "temp_config.yaml"
     # 重新生成简易配置文件
     try:
         with open(temp_file, "w", encoding="utf8") as wf:
-            # yaml.dump(_tmp_data, wf, Dumper=yaml.RoundTripDumper, allow_unicode=True)
             _yaml.dump(_tmp_data, wf)
-        with open(temp_file, "r", encoding="utf8") as rf:
+        with open(temp_file, encoding="utf8") as rf:
             _data = _yaml.load(rf)
         # 添加注释
         for module in _data.keys():
@@ -93,7 +92,7 @@ def _generate_simple_config():
         with SIMPLE_CONFIG_FILE.open("w", encoding="utf8") as wf:
             _yaml.dump(_data, wf)
     except Exception as e:
-        logger.error(f"生成简易配置注释错误...", e=e)
+        logger.error("生成简易配置注释错误...", e=e)
     if temp_file.exists():
         temp_file.unlink()
 

zhenxun/builtin_plugins/init/init_plugin.py

@@ -21,6 +21,8 @@
     PluginLimitType,
 )
 
+from .manager import manager
+
 _yaml = YAML(pure=True)
 _yaml.allow_unicode = True
 _yaml.indent = 2
@@ -148,23 +150,25 @@ async def _():
     #     ["name", "author", "version", "admin_level", "plugin_type"],
     #     10,
     # )
-    if limit_list:
-        limit_create = []
-        plugins = []
-        if module_path_list := [limit.module_path for limit in limit_list]:
-            plugins = await PluginInfo.filter(module_path__in=module_path_list).all()
-        if plugins:
-            for limit in limit_list:
-                if lmt := [p for p in plugins if p.module_path == limit.module_path]:
-                    plugin = lmt[0]
-                    limit_type_list = [
-                        _limit.limit_type for _limit in await plugin.plugin_limit.all()  # type: ignore
-                    ]
-                    if limit.limit_type not in limit_type_list:
-                        limit.plugin = plugin
-                        limit_create.append(limit)
-        if limit_create:
-            await PluginLimit.bulk_create(limit_create, 10)
+    # for limit in limit_list:
+    #     limit_create = []
+    #     plugins = []
+    #     if module_path_list := [limit.module_path for limit in limit_list]:
+    #         plugins = await PluginInfo.get_plugins(module_path__in=module_path_list)
+    #     if plugins:
+    #         for limit in limit_list:
+    #             if lmt := [p for p in plugins if p.module_path == limit.module_path]:
+    #                 plugin = lmt[0]
+    #                 """不在数据库中"""
+    #                 limit_type_list = [
+    #                     _limit.limit_type
+    #                     for _limit in await plugin.plugin_limit.all()  # type: ignore
+    #                 ]
+    #                 if limit.limit_type not in limit_type_list:
+    #                     limit.plugin = plugin
+    #                     limit_create.append(limit)
+    #     if limit_create:
+    #         await PluginLimit.bulk_create(limit_create, 10)
     if task_list:
         module_dict = {
             t[1]: t[0] for t in await TaskInfo.all().values_list("id", "module")
@@ -195,10 +199,18 @@ async def _():
     await data_migration()
     await PluginInfo.filter(module_path__in=load_plugin).update(load_status=True)
     await PluginInfo.filter(module_path__not_in=load_plugin).update(load_status=False)
+    manager.init()
+    if limit_list:
+        for limit in limit_list:
+            if not manager.exist(limit.module_path, limit.limit_type):
+                """不存在，添加"""
+                manager.add(limit.module_path, limit)
+    manager.save_file()
+    await manager.load_to_db()
 
 
 async def data_migration():
-    await limit_migration()
+    # await limit_migration()
     await plugin_migration()
     await group_migration()