✨ 指定bot的功能管理 (#1706)

zhenxun/builtin_plugins/__init__.py

@@ -9,6 +9,8 @@
 from nonebot.drivers import Driver
 from tortoise.exceptions import OperationalError
 
+from zhenxun.utils.platform import PlatformUtils
+
 require("nonebot_plugin_apscheduler")
 require("nonebot_plugin_alconna")
 require("nonebot_plugin_session")
@@ -19,6 +21,7 @@
 from zhenxun.services.log import logger
 from zhenxun.models.sign_user import SignUser
 from zhenxun.models.goods_info import GoodsInfo
+from zhenxun.models.bot_console import BotConsole
 from zhenxun.models.user_console import UserConsole
 from zhenxun.utils.decorator.shop import shop_register
 from zhenxun.models.bot_connect_log import BotConnectLog
@@ -33,6 +36,10 @@ async def _(bot: Bot):
     await BotConnectLog.create(
         bot_id=bot.self_id, platform=bot.adapter, connect_time=datetime.now(), type=1
     )
+    if not await BotConsole.exists(bot_id=bot.self_id):
+        await BotConsole.create(
+            bot_id=bot.self_id, platform=PlatformUtils.get_platform(bot)
+        )
 
 
 @driver.on_bot_disconnect

zhenxun/builtin_plugins/hooks/_auth_checker.py

@@ -11,6 +11,7 @@
 from zhenxun.configs.config import Config
 from zhenxun.utils.message import MessageUtils
 from zhenxun.models.level_user import LevelUser
+from zhenxun.models.bot_console import BotConsole
 from zhenxun.models.plugin_info import PluginInfo
 from zhenxun.models.plugin_limit import PluginLimit
 from zhenxun.models.user_console import UserConsole
@@ -244,6 +245,7 @@ async def auth(
                         if not plugin.limit_superuser:
                             cost_gold = 0
                             raise IsSuperuserException()
+                    await self.auth_bot(plugin, bot.self_id)
                     await self.auth_group(plugin, session, message)
                     await self.auth_admin(plugin, session)
                     await self.auth_plugin(plugin, session, event)
@@ -278,6 +280,16 @@ async def auth(
         if is_ignore:
             raise IgnoredException("权限检测 ignore")
 
+    async def auth_bot(self, plugin: PluginInfo, bot_id: str):
+        """机器人权限
+
+        参数:
+            plugin: PluginInfo
+            bot_id: bot_id
+        """
+        if await BotConsole.is_block_plugin(plugin.module, bot_id):
+            raise IgnoredException("机器人权限检测 ignore")
+
     async def auth_limit(self, plugin: PluginInfo, session: EventSession):
         """插件限制
 

zhenxun/builtin_plugins/superuser/bot_manage/__init__.py

@@ -0,0 +1,30 @@
+import os
+import time
+
+from nonebot.rule import to_me
+from nonebot.utils import run_sync
+from nonebot.permission import SUPERUSER
+from nonebot.plugin import PluginMetadata
+from nonebot_plugin_session import EventSession
+from nonebot_plugin_apscheduler import scheduler
+from nonebot_plugin_alconna import Alconna, on_alconna
+
+from zhenxun.services.log import logger
+from zhenxun.utils.enum import PluginType
+from zhenxun.utils.message import MessageUtils
+from zhenxun.configs.path_config import TEMP_PATH
+from zhenxun.configs.utils import PluginExtraData
+from zhenxun.utils.utils import ResourceDirManager
+
+__plugin_meta__ = PluginMetadata(
+    name="Bot管理",
+    description="指定bot对象的功能/被动开关和状态",
+    usage="""
+    清理临时数据
+    """.strip(),
+    extra=PluginExtraData(
+        author="",
+        version="0.1",
+        plugin_type=PluginType.SUPERUSER,
+    ).dict(),
+)

zhenxun/models/bot_console.py

@@ -0,0 +1,60 @@
+from tortoise import fields
+
+from zhenxun.services.db_context import Model
+
+
+class BotConsole(Model):
+    id = fields.IntField(pk=True, generated=True, auto_increment=True)
+    """自增id"""
+    bot_id = fields.CharField(255, unique=True, description="bot_id")
+    """bot_id"""
+    status = fields.BooleanField(default=True, description="Bot状态")
+    """Bot状态"""
+    block_plugin = fields.TextField(default="", description="禁用插件")
+    """禁用插件"""
+    block_task = fields.TextField(default="", description="禁用被动技能")
+    """禁用被动技能"""
+    create_time = fields.DatetimeField(auto_now_add=True, description="创建时间")
+    """创建时间"""
+    platform = fields.CharField(255, null=True, description="平台")
+    """平台"""
+
+    class Meta:  # type: ignore
+        table = "bot_console"
+        table_description = "Bot数据表"
+
+    @classmethod
+    async def get_bot_status(cls, bot_id: str) -> bool:
+        result = await cls.get_or_none(bot_id=bot_id)
+        return result.status if result else False
+
+    @classmethod
+    async def set_block_plugin(cls, bot_id: str, module: str):
+        bot_data, _ = await cls.get_or_create(bot_id=bot_id)
+        if f"<{module}," not in bot_data.block_plugin:
+            bot_data.block_plugin += f"<{module},"
+        await bot_data.save(update_fields=["block_plugin"])
+
+    @classmethod
+    async def set_unblock_plugin(cls, bot_id: str, module: str):
+        bot_data, _ = await cls.get_or_create(bot_id=bot_id)
+        if f"<{module}," in bot_data.block_plugin:
+            bot_data.block_plugin = bot_data.block_plugin.replace(f"<{module},", "")
+        await bot_data.save(update_fields=["block_plugin"])
+
+    @classmethod
+    async def set_block_task(cls, bot_id: str, task: str):
+        bot_data, _ = await cls.get_or_create(bot_id=bot_id)
+        if f"<{task}," not in bot_data.block_task:
+            bot_data.block_plugin += f"<{task},"
+        await bot_data.save(update_fields=["block_task"])
+
+    @classmethod
+    async def is_block_plugin(cls, bot_id: str, task: str) -> bool:
+        bot_data, _ = await cls.get_or_create(bot_id=bot_id)
+        return f"<{task}," in bot_data.block_plugin
+
+    @classmethod
+    async def is_block_task(cls, bot_id: str, task: str) -> bool:
+        bot_data, _ = await cls.get_or_create(bot_id=bot_id)
+        return f"<{task}," in bot_data.block_task

zhenxun/models/group_console.py

@@ -37,8 +37,8 @@ class GroupConsole(Model):
         default="", description="超级用户禁用插件"
     )
     """超级用户禁用插件"""
-    block_task = fields.TextField(default="", description="禁用插件")
-    """禁用插件"""
+    block_task = fields.TextField(default="", description="禁用被动技能")
+    """禁用被动技能"""
     superuser_block_task = fields.TextField(default="", description="超级用户禁用被动")
     """超级用户禁用被动"""
     platform = fields.CharField(255, default="qq", description="所属平台")