refactor: 重构会话封禁功能处理逻辑

pkg/boot/app.py

@@ -28,6 +28,9 @@ class Application:
     def __init__(self):
         pass
 
+    async def initialize(self):
+        await self.im_mgr.initialize()
+
     async def run(self):
         # TODO make it async
         plugin_host.initialize_plugins()

pkg/boot/boot.py

@@ -50,7 +50,10 @@ async def make_app() -> app.Application:
     # 生成标识符
     identifier.init()
 
-    cfg_mgr = await config.load_config()
+    cfg_mgr = await config.load_python_module_config(
+        "config.py",
+        "config-template.py"
+    )
     context.set_config_manager(cfg_mgr)
     cfg = cfg_mgr.data
 
@@ -63,7 +66,10 @@ async def make_app() -> app.Application:
         if overrided:
             qcg_logger.info("以下配置项已使用 override.json 覆盖：" + ",".join(overrided))
     
-    tips_mgr = await config.load_tips()
+    tips_mgr = await config.load_python_module_config(
+        "tips.py",
+        "tips-custom-template.py"
+    )
 
     # 初始化文字转图片
     from pkg.utils import text2img
@@ -121,4 +127,5 @@ async def make_app() -> app.Application:
 
 async def main():
     app_inst = await make_app()
+    await app_inst.initialize()
     await app_inst.run()

pkg/boot/config.py

@@ -4,11 +4,11 @@ from ..config import manager as config_mgr
 from ..config.impls import pymodule
 
 
-async def load_config() -> config_mgr.ConfigManager:
+async def load_python_module_config(config_name: str, template_name: str) -> config_mgr.ConfigManager:
-    """加载配置文件"""
+    """加载Python模块配置文件"""
     cfg_inst = pymodule.PythonModuleConfigFile(
-        "config.py",
+        config_name,
-        "config-template.py"
+        template_name
     )
 
     cfg_mgr = config_mgr.ConfigManager(cfg_inst)
@@ -17,19 +17,6 @@ async def load_config() -> config_mgr.ConfigManager:
     return cfg_mgr
 
 
-async def load_tips() -> config_mgr.ConfigManager:
-    """加载提示文件"""
-    tips_inst = pymodule.PythonModuleConfigFile(
-        "tips.py",
-        "tips-custom-template.py"
-    )
-
-    tips_mgr = config_mgr.ConfigManager(tips_inst)
-    await tips_mgr.load_config()
-
-    return tips_mgr
-
-
 async def override_config_manager(cfg_mgr: config_mgr.ConfigManager) -> list[str]:
     override_json = json.load(open("override.json", "r", encoding="utf-8"))
     overrided = []
@@ -39,5 +26,5 @@ async def override_config_manager(cfg_mgr: config_mgr.ConfigManager) -> list[str
         if key in config:
             config[key] = override_json[key]
             overrided.append(key)
-            
+
     return overrided

pkg/qqbot/banlist.py

@@ -1,50 +0,0 @@
-from ..utils import context
-
-
-def is_banned(launcher_type: str, launcher_id: int, sender_id: int) -> bool:
-    if not context.get_qqbot_manager().enable_banlist:
-        return False
-    
-    result = False
-    
-    if launcher_type == 'group':
-        # 检查是否显式声明发起人QQ要被person忽略
-        if sender_id in context.get_qqbot_manager().ban_person:
-            result = True
-        else:
-            for group_rule in context.get_qqbot_manager().ban_group:
-                if type(group_rule) == int:
-                    if group_rule == launcher_id:  # 此群群号被禁用
-                        result = True
-                elif type(group_rule) == str:
-                    if group_rule.startswith('!'):
-                        # 截取!后面的字符串作为表达式，判断是否匹配
-                        reg_str = group_rule[1:]
-                        import re
-                        if re.match(reg_str, str(launcher_id)):  # 被豁免，最高级别
-                            result = False
-                            break
-                    else:
-                        # 判断是否匹配regexp
-                        import re
-                        if re.match(group_rule, str(launcher_id)):  # 此群群号被禁用
-                            result = True
-                    
-    else:
-        # ban_person, 与群规则相同
-        for person_rule in context.get_qqbot_manager().ban_person:
-            if type(person_rule) == int:
-                if person_rule == launcher_id:
-                    result = True
-            elif type(person_rule) == str:
-                if person_rule.startswith('!'):
-                    reg_str = person_rule[1:]
-                    import re
-                    if re.match(reg_str, str(launcher_id)):
-                        result = False
-                        break
-                else:
-                    import re
-                    if re.match(person_rule, str(launcher_id)):
-                        result = True
-    return result

pkg/qqbot/bansess/bansess.py

@@ -0,0 +1,71 @@
+# 处理对会话的禁用配置
+# 过去的 banlist
+from __future__ import annotations
+import re
+
+from ...boot import app
+from ...boot import config as config_util
+from ...config import manager as cfg_mgr
+
+
+class SessionBanManager:
+    
+    ap: app.Application = None
+
+    banlist_mgr: cfg_mgr.ConfigManager
+
+    def __init__(self, ap: app.Application):
+        self.ap = ap
+
+    async def initialize(self):
+        self.banlist_mgr = await config_util.load_python_module_config(
+            "banlist.py",
+            "res/templates/banlist-template.py"
+        )
+
+    async def is_banned(
+        self, launcher_type: str, launcher_id: int, sender_id: int
+    ) -> bool:
+        if not self.banlist_mgr.data['enable']:
+            return False
+        
+        result = False
+
+        if launcher_type == 'group':
+            if not self.banlist_mgr.data['enable_group']:  # 未启用群聊响应
+                result = True
+            # 检查是否显式声明发起人QQ要被person忽略
+            elif sender_id in self.banlist_mgr.data['person']:
+                result = True
+            else:
+                for group_rule in self.banlist_mgr.data['group']:
+                    if type(group_rule) == int:
+                        if group_rule == launcher_id:
+                            result = True
+                    elif type(group_rule) == str:
+                        if group_rule.startswith('!'):
+                            reg_str = group_rule[1:]
+                            if re.match(reg_str, str(launcher_id)):
+                                result = False
+                                break
+                        else:
+                            if re.match(group_rule, str(launcher_id)):
+                                result = True
+        elif launcher_type == 'person':
+            if not self.banlist_mgr.data['enable_private']:
+                result = True
+            else:
+                for person_rule in self.banlist_mgr.data['person']:
+                    if type(person_rule) == int:
+                        if person_rule == launcher_id:
+                            result = True
+                    elif type(person_rule) == str:
+                        if person_rule.startswith('!'):
+                            reg_str = person_rule[1:]
+                            if re.match(reg_str, str(launcher_id)):
+                                result = False
+                                break
+                        else:
+                            if re.match(person_rule, str(launcher_id)):
+                                result = True
+        return result

pkg/qqbot/manager.py

@@ -20,6 +20,7 @@ from ..plugin import models as plugin_models
 import tips as tips_custom
 from ..qqbot import adapter as msadapter
 from . import resprule
+from .bansess import bansess
 
 from ..boot import app
 
@@ -42,11 +43,24 @@ class QQBotManager:
     ban_person = []
     ban_group = []
 
+    # modern
+    ap: app.Application = None
+
+    bansess_mgr: bansess.SessionBanManager = None
+
     def __init__(self, first_time_init=True, ap: app.Application = None):
         config = context.get_config_manager().data
 
+        self.ap = ap
+        self.bansess_mgr = bansess.SessionBanManager(ap)
+
         self.timeout = config['process_message_timeout']
         self.retry = config['retry_times']
+    
+    async def initialize(self):
+        await self.bansess_mgr.initialize()
+
+        config = context.get_config_manager().data
 
         logging.debug("Use adapter:" + config['msg_source_adapter'])
         if config['msg_source_adapter'] == 'yirimirai':
@@ -160,19 +174,6 @@ class QQBotManager:
 
         self.unsubscribe_all = unsubscribe_all
 
-        # 加载禁用列表
-        if os.path.exists("banlist.py"):
-            import banlist
-            self.enable_banlist = banlist.enable
-            self.ban_person = banlist.person
-            self.ban_group = banlist.group
-            logging.info("加载禁用列表: person: {}, group: {}".format(self.ban_person, self.ban_group))
-
-            if hasattr(banlist, "enable_private"):
-                self.enable_private = banlist.enable_private
-            if hasattr(banlist, "enable_group"):
-                self.enable_group = banlist.enable_group
-
         config = context.get_config_manager().data
         if os.path.exists("sensitive.json") \
                 and config['sensitive_word_filter'] is not None \
@@ -222,6 +223,11 @@ class QQBotManager:
         """
         私聊群聊通用消息处理方法
         """
+        # 检查bansess
+        if await self.bansess_mgr.is_banned(launcher_type, launcher_id, sender_id):
+            self.ap.logger.info("根据禁用列表忽略{}_{}的消息".format(launcher_type, launcher_id))
+            return []
+
         if mirai.Image in message_chain:
             return []
         elif sender_id == self.bot_account_id:

pkg/qqbot/process.py

@@ -1,4 +1,5 @@
 # 此模块提供了消息处理的具体逻辑的接口
+from __future__ import annotations
 import asyncio
 import time
 import traceback
@@ -6,12 +7,6 @@ import traceback
 import mirai
 import logging
 
-# 这里不使用动态引入config
-# 因为在这里动态引入会卡死程序
-# 而此模块静态引用config与动态引入的表现一致
-# 已弃用，由于超时时间现已动态使用
-# import config as config_init_import
-
 from ..qqbot import ratelimit
 from ..qqbot import command, message
 from ..openai import session as openai_session
@@ -20,9 +15,9 @@ from ..utils import context
 from ..plugin import host as plugin_host
 from ..plugin import models as plugin_models
 from ..qqbot import ignore
-from ..qqbot import banlist
 from ..qqbot import blob
 import tips as tips_custom
+from ..boot import app
 
 processing = []
 
@@ -45,11 +40,6 @@ async def process_message(launcher_type: str, launcher_id: int, text_message: st
     reply = []
     session_name = "{}_{}".format(launcher_type, launcher_id)
 
-    # 检查发送方是否被禁用
-    if banlist.is_banned(launcher_type, launcher_id, sender_id):
-        logging.info("根据禁用列表忽略{}_{}的消息".format(launcher_type, launcher_id))
-        return []
-
     if ignore.ignore(text_message):
         logging.info("根据忽略规则忽略消息: {}".format(text_message))
         return []