refactor: 独立消息处理函数

pkg/qqbot/manager.py

@@ -10,18 +10,14 @@ from mirai import At, GroupMessage, MessageEvent, Mirai, Plain, StrangerMessage,
 import config
 import pkg.openai.session
 import pkg.openai.manager
-from func_timeout import func_set_timeout, FunctionTimedOut
+from func_timeout import FunctionTimedOut
-import datetime
 import logging
 
 import pkg.qqbot.filter
-
+import pkg.qqbot.process as processor
-help_text = config.help_message
 
 inst = None
 
-processing = []
-
 
 # 并行运行
 def go(func, args=()):
@@ -54,7 +50,6 @@ def check_response_rule(text: str) -> (bool, str):
 
 # 控制QQ消息输入输出的类
 class QQBotManager:
-    timeout = 60
     retry = 3
 
     bot = None
@@ -93,6 +88,9 @@ class QQBotManager:
                 )
             )
 
+        else:
+            raise Exception("未知的适配器类型")
+
         @bot.on(FriendMessage)
         async def on_friend_message(event: FriendMessage):
             go(self.on_person_message, (event,))
@@ -110,163 +108,6 @@ class QQBotManager:
         global inst
         inst = self
 
-    # 统一的消息处理函数
-    @func_set_timeout(timeout)
-    def process_message(self, launcher_type: str, launcher_id: int, text_message: str) -> str:
-        global processing
-        reply = ''
-        session_name = "{}_{}".format(launcher_type, launcher_id)
-
-        pkg.openai.session.get_session(session_name).acquire_response_lock()
-
-        try:
-            if session_name in processing:
-                pkg.openai.session.get_session(session_name).release_response_lock()
-                return "[bot]err:正在处理中，请稍后再试"
-
-            processing.append(session_name)
-
-            try:
-
-                if text_message.startswith('!') or text_message.startswith("！"):  # 指令
-                    try:
-                        logging.info(
-                            "[{}]发起指令:{}".format(session_name, text_message[:min(20, len(text_message))] + (
-                                "..." if len(text_message) > 20 else "")))
-
-                        cmd = text_message[1:].strip().split(' ')[0]
-
-                        params = text_message[1:].strip().split(' ')[1:]
-                        if cmd == 'help':
-                            reply = "[bot]" + help_text
-                        elif cmd == 'reset':
-                            pkg.openai.session.get_session(session_name).reset(explicit=True)
-                            reply = "[bot]会话已重置"
-                        elif cmd == 'last':
-                            result = pkg.openai.session.get_session(session_name).last_session()
-                            if result is None:
-                                reply = "[bot]没有前一次的对话"
-                            else:
-                                datetime_str = datetime.datetime.fromtimestamp(result.create_timestamp).strftime(
-                                    '%Y-%m-%d %H:%M:%S')
-                                reply = "[bot]已切换到前一次的对话：\n创建时间:{}\n".format(
-                                    datetime_str) + result.prompt[
-                                                    :min(100,
-                                                         len(result.prompt))] + \
-                                        ("..." if len(result.prompt) > 100 else "#END#")
-                        elif cmd == 'next':
-                            result = pkg.openai.session.get_session(session_name).next_session()
-                            if result is None:
-                                reply = "[bot]没有后一次的对话"
-                            else:
-                                datetime_str = datetime.datetime.fromtimestamp(result.create_timestamp).strftime(
-                                    '%Y-%m-%d %H:%M:%S')
-                                reply = "[bot]已切换到后一次的对话：\n创建时间:{}\n".format(
-                                    datetime_str) + result.prompt[
-                                                    :min(100,
-                                                         len(result.prompt))] + \
-                                        ("..." if len(result.prompt) > 100 else "#END#")
-                        elif cmd == 'prompt':
-                            reply = "[bot]当前对话所有内容：\n" + pkg.openai.session.get_session(session_name).prompt
-                        elif cmd == 'list':
-                            pkg.openai.session.get_session(session_name).persistence()
-                            page = 0
-
-                            if len(params) > 0:
-                                try:
-                                    page = int(params[0])
-                                except ValueError:
-                                    pass
-
-                            results = pkg.openai.session.get_session(session_name).list_history(page=page)
-                            if len(results) == 0:
-                                reply = "[bot]第{}页没有历史会话".format(page)
-                            else:
-                                reply = "[bot]历史会话 第{}页：\n".format(page)
-                                current = -1
-                                for i in range(len(results)):
-                                    # 时间(使用create_timestamp转换) 序号 部分内容
-                                    datetime_obj = datetime.datetime.fromtimestamp(results[i]['create_timestamp'])
-                                    reply += "#{} 创建:{} {}\n".format(i + page * 10,
-                                                                       datetime_obj.strftime("%Y-%m-%d %H:%M:%S"),
-                                                                       results[i]['prompt'][
-                                                                       :min(20, len(results[i]['prompt']))])
-                                    if results[i]['create_timestamp'] == pkg.openai.session.get_session(
-                                            session_name).create_timestamp:
-                                        current = i + page * 10
-
-                                reply += "\n以上信息倒序排列"
-                                if current != -1:
-                                    reply += ",当前会话是 #{}\n".format(current)
-                                else:
-                                    reply += ",当前处于全新会话或不在此页"
-                        elif cmd == 'usage':
-                            api_keys = pkg.openai.manager.get_inst().key_mgr.api_key
-                            reply = "[bot]api-key使用情况:(阈值:{})\n\n".format(
-                                pkg.openai.manager.get_inst().key_mgr.api_key_usage_threshold)
-
-                            using_key_name = ""
-                            for api_key in api_keys:
-                                reply += "{}:\n - {}字 {}%\n".format(api_key,
-                                                                     pkg.openai.manager.get_inst().key_mgr.get_usage(
-                                                                         api_keys[api_key]),
-                                                                     round(
-                                                                         pkg.openai.manager.get_inst().key_mgr.get_usage(
-                                                                             api_keys[
-                                                                                 api_key]) / pkg.openai.manager.get_inst().key_mgr.api_key_usage_threshold * 100,
-                                                                         3))
-                                if api_keys[api_key] == pkg.openai.manager.get_inst().key_mgr.using_key:
-                                    using_key_name = api_key
-                            reply += "\n当前使用:{}".format(using_key_name)
-                    except Exception as e:
-                        self.notify_admin("{}指令执行失败:{}".format(session_name, e))
-                        logging.exception(e)
-                        reply = "[bot]err:{}".format(e)
-                else:  # 消息
-                    logging.info("[{}]发送消息:{}".format(session_name, text_message[:min(20, len(text_message))] + (
-                        "..." if len(text_message) > 20 else "")))
-
-                    session = pkg.openai.session.get_session(session_name)
-                    try:
-                        prefix = "[GPT]" if hasattr(config, "show_prefix") and config.show_prefix else ""
-                        reply = prefix + session.append(text_message)
-                    except openai.error.APIConnectionError as e:
-                        self.notify_admin("{}会话调用API失败:{}".format(session_name, e))
-                        reply = "[bot]err:调用API失败，请重试或联系作者，或等待修复"
-                    except openai.error.RateLimitError as e:
-                        # 尝试切换api-key
-                        current_tokens_amt = pkg.openai.manager.get_inst().key_mgr.get_usage(pkg.openai.manager.get_inst().key_mgr.get_using_key())
-                        pkg.openai.manager.get_inst().key_mgr.set_current_exceeded()
-                        switched, name = pkg.openai.manager.get_inst().key_mgr.auto_switch()
-
-                        if not switched:
-                            self.notify_admin("API调用额度超限({}),请向OpenAI账户充值或在config.py中更换api_key".format(current_tokens_amt))
-                            reply = "[bot]err:API调用额度超额，请联系作者，或等待修复"
-                        else:
-                            openai.api_key = pkg.openai.manager.get_inst().key_mgr.get_using_key()
-                            self.notify_admin("API调用额度超限({}),已切换到{}".format(current_tokens_amt, name))
-                            reply = "[bot]err:API调用额度超额，已自动切换，请重新发送消息"
-                    except openai.error.InvalidRequestError as e:
-                        self.notify_admin("{}API调用参数错误:{}\n\n这可能是由于config.py中的prompt_submit_length参数或"
-                                          "completion_api_params中的max_tokens参数数值过大导致的，请尝试将其降低".format(
-                            session_name, e))
-                        reply = "[bot]err:API调用参数错误，请联系作者，或等待修复"
-                    except Exception as e:
-                        logging.exception(e)
-                        reply = "[bot]err:{}".format(e)
-
-                logging.info(
-                    "回复[{}]消息:{}".format(session_name,
-                                             reply[:min(100, len(reply))] + ("..." if len(reply) > 100 else "")))
-                reply = self.reply_filter.process(reply)
-
-            finally:
-                processing.remove(session_name)
-        finally:
-            pkg.openai.session.get_session(session_name).release_response_lock()
-
-            return reply
-
     def send(self, event, msg):
         asyncio.run(self.bot.send(event, msg))
 
@@ -286,7 +127,7 @@ class QQBotManager:
                 failed = 0
                 for i in range(self.retry):
                     try:
-                        reply = self.process_message('person', event.sender.id, str(event.message_chain))
+                        reply = processor.process_message('person', event.sender.id, str(event.message_chain))
                         break
                     except FunctionTimedOut:
                         pkg.openai.session.get_session('person_{}'.format(event.sender.id)).release_response_lock()
@@ -312,13 +153,11 @@ class QQBotManager:
             if At(self.bot.qq) in event.message_chain:
                 event.message_chain.remove(At(self.bot.qq))
 
-            processing.append("group_{}".format(event.sender.id))
-
             # 超时则重试，重试超过次数则放弃
             failed = 0
             for i in range(self.retry):
                 try:
-                    replys = self.process_message('group', event.group.id,
+                    replys = processor.process_message('group', event.group.id,
                                                   str(event.message_chain).strip() if text is None else text)
                     break
                 except FunctionTimedOut:

pkg/qqbot/process.py

@@ -0,0 +1,176 @@
+# 此模块提供了消息处理的具体逻辑的接口
+import datetime
+
+import pkg.qqbot.manager as manager
+from func_timeout import func_set_timeout
+import logging
+import openai
+
+import config
+
+import pkg.openai.session
+import pkg.openai.manager
+
+processing = []
+
+
+@func_set_timeout(config.process_message_timeout)
+def process_message(launcher_type: str, launcher_id: int, text_message: str) -> str:
+    global processing
+
+    mgr = pkg.openai.manager.get_inst()
+
+    reply = ''
+    session_name = "{}_{}".format(launcher_type, launcher_id)
+
+    pkg.openai.session.get_session(session_name).acquire_response_lock()
+
+    try:
+        if session_name in processing:
+            pkg.openai.session.get_session(session_name).release_response_lock()
+            return "[bot]err:正在处理中，请稍后再试"
+
+        processing.append(session_name)
+
+        try:
+
+            if text_message.startswith('!') or text_message.startswith("！"):  # 指令
+                try:
+                    logging.info(
+                        "[{}]发起指令:{}".format(session_name, text_message[:min(20, len(text_message))] + (
+                            "..." if len(text_message) > 20 else "")))
+
+                    cmd = text_message[1:].strip().split(' ')[0]
+
+                    params = text_message[1:].strip().split(' ')[1:]
+                    if cmd == 'help':
+                        reply = "[bot]" + config.help_message
+                    elif cmd == 'reset':
+                        pkg.openai.session.get_session(session_name).reset(explicit=True)
+                        reply = "[bot]会话已重置"
+                    elif cmd == 'last':
+                        result = pkg.openai.session.get_session(session_name).last_session()
+                        if result is None:
+                            reply = "[bot]没有前一次的对话"
+                        else:
+                            datetime_str = datetime.datetime.fromtimestamp(result.create_timestamp).strftime(
+                                '%Y-%m-%d %H:%M:%S')
+                            reply = "[bot]已切换到前一次的对话：\n创建时间:{}\n".format(
+                                datetime_str) + result.prompt[
+                                                :min(100,
+                                                     len(result.prompt))] + \
+                                    ("..." if len(result.prompt) > 100 else "#END#")
+                    elif cmd == 'next':
+                        result = pkg.openai.session.get_session(session_name).next_session()
+                        if result is None:
+                            reply = "[bot]没有后一次的对话"
+                        else:
+                            datetime_str = datetime.datetime.fromtimestamp(result.create_timestamp).strftime(
+                                '%Y-%m-%d %H:%M:%S')
+                            reply = "[bot]已切换到后一次的对话：\n创建时间:{}\n".format(
+                                datetime_str) + result.prompt[
+                                                :min(100,
+                                                     len(result.prompt))] + \
+                                    ("..." if len(result.prompt) > 100 else "#END#")
+                    elif cmd == 'prompt':
+                        reply = "[bot]当前对话所有内容：\n" + pkg.openai.session.get_session(session_name).prompt
+                    elif cmd == 'list':
+                        pkg.openai.session.get_session(session_name).persistence()
+                        page = 0
+
+                        if len(params) > 0:
+                            try:
+                                page = int(params[0])
+                            except ValueError:
+                                pass
+
+                        results = pkg.openai.session.get_session(session_name).list_history(page=page)
+                        if len(results) == 0:
+                            reply = "[bot]第{}页没有历史会话".format(page)
+                        else:
+                            reply = "[bot]历史会话 第{}页：\n".format(page)
+                            current = -1
+                            for i in range(len(results)):
+                                # 时间(使用create_timestamp转换) 序号 部分内容
+                                datetime_obj = datetime.datetime.fromtimestamp(results[i]['create_timestamp'])
+                                reply += "#{} 创建:{} {}\n".format(i + page * 10,
+                                                                   datetime_obj.strftime("%Y-%m-%d %H:%M:%S"),
+                                                                   results[i]['prompt'][
+                                                                   :min(20, len(results[i]['prompt']))])
+                                if results[i]['create_timestamp'] == pkg.openai.session.get_session(
+                                        session_name).create_timestamp:
+                                    current = i + page * 10
+
+                            reply += "\n以上信息倒序排列"
+                            if current != -1:
+                                reply += ",当前会话是 #{}\n".format(current)
+                            else:
+                                reply += ",当前处于全新会话或不在此页"
+                    elif cmd == 'usage':
+                        api_keys = pkg.openai.manager.get_inst().key_mgr.api_key
+                        reply = "[bot]api-key使用情况:(阈值:{})\n\n".format(
+                            pkg.openai.manager.get_inst().key_mgr.api_key_usage_threshold)
+
+                        using_key_name = ""
+                        for api_key in api_keys:
+                            reply += "{}:\n - {}字 {}%\n".format(api_key,
+                                                                 pkg.openai.manager.get_inst().key_mgr.get_usage(
+                                                                     api_keys[api_key]),
+                                                                 round(
+                                                                     pkg.openai.manager.get_inst().key_mgr.get_usage(
+                                                                         api_keys[
+                                                                             api_key]) / pkg.openai.manager.get_inst().key_mgr.api_key_usage_threshold * 100,
+                                                                     3))
+                            if api_keys[api_key] == pkg.openai.manager.get_inst().key_mgr.using_key:
+                                using_key_name = api_key
+                        reply += "\n当前使用:{}".format(using_key_name)
+                except Exception as e:
+                    mgr.notify_admin("{}指令执行失败:{}".format(session_name, e))
+                    logging.exception(e)
+                    reply = "[bot]err:{}".format(e)
+            else:  # 消息
+                logging.info("[{}]发送消息:{}".format(session_name, text_message[:min(20, len(text_message))] + (
+                    "..." if len(text_message) > 20 else "")))
+
+                session = pkg.openai.session.get_session(session_name)
+                try:
+                    prefix = "[GPT]" if hasattr(config, "show_prefix") and config.show_prefix else ""
+                    reply = prefix + session.append(text_message)
+                except openai.error.APIConnectionError as e:
+                    mgr.notify_admin("{}会话调用API失败:{}".format(session_name, e))
+                    reply = "[bot]err:调用API失败，请重试或联系作者，或等待修复"
+                except openai.error.RateLimitError as e:
+                    # 尝试切换api-key
+                    current_tokens_amt = pkg.openai.manager.get_inst().key_mgr.get_usage(
+                        pkg.openai.manager.get_inst().key_mgr.get_using_key())
+                    pkg.openai.manager.get_inst().key_mgr.set_current_exceeded()
+                    switched, name = pkg.openai.manager.get_inst().key_mgr.auto_switch()
+
+                    if not switched:
+                        mgr.notify_admin("API调用额度超限({}),请向OpenAI账户充值或在config.py中更换api_key".format(
+                            current_tokens_amt))
+                        reply = "[bot]err:API调用额度超额，请联系作者，或等待修复"
+                    else:
+                        openai.api_key = pkg.openai.manager.get_inst().key_mgr.get_using_key()
+                        mgr.notify_admin("API调用额度超限({}),已切换到{}".format(current_tokens_amt, name))
+                        reply = "[bot]err:API调用额度超额，已自动切换，请重新发送消息"
+                except openai.error.InvalidRequestError as e:
+                    mgr.notify_admin("{}API调用参数错误:{}\n\n这可能是由于config.py中的prompt_submit_length参数或"
+                                     "completion_api_params中的max_tokens参数数值过大导致的，请尝试将其降低".format(
+                        session_name, e))
+                    reply = "[bot]err:API调用参数错误，请联系作者，或等待修复"
+                except Exception as e:
+                    logging.exception(e)
+                    reply = "[bot]err:{}".format(e)
+
+            logging.info(
+                "回复[{}]消息:{}".format(session_name,
+                                         reply[:min(100, len(reply))] + ("..." if len(reply) > 100 else "")))
+            reply = mgr.reply_filter.process(reply)
+
+        finally:
+            processing.remove(session_name)
+    finally:
+        pkg.openai.session.get_session(session_name).release_response_lock()
+
+        return reply