refactor: 完成会话管理相关指令

pkg/qqbot/cmds/mgr.py

@@ -5,6 +5,7 @@ import copy
 import pkgutil
 import traceback
 import types
+import json
 
 
 __command_list__ = {}
@@ -257,7 +258,7 @@ def execute(context: Context) -> list:
     while True:
         try:
             logging.debug('执行指令: {}'.format(path))
-            node = node[path]
+            node = __command_list__[path]
 
             # 检查权限
             if ctx.privilege < node['privilege']:

pkg/qqbot/cmds/session/default.py

@@ -0,0 +1,73 @@
+from ..mgr import AbstractCommandNode, Context
+
+
+@AbstractCommandNode.register(
+    parent=None,
+    name="default",
+    description="操作情景预设",
+    usage="!default\n!default [指定情景预设为默认]",
+    aliases=[],
+    privilege=1
+)
+class DefaultCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        import pkg.openai.session
+        session_name = ctx.session_name
+        params = ctx.params
+        reply = []
+        import config
+
+        if len(params) == 0:
+            # 输出目前所有情景预设
+            import pkg.openai.dprompt as dprompt
+            reply_str = "[bot]当前所有情景预设({}模式):\n\n".format(config.preset_mode)
+
+            prompts = dprompt.mode_inst().list()
+
+            for key in prompts:
+                pro = prompts[key]
+                reply_str += "名称: {}".format(key)
+
+                for r in pro:
+                    reply_str += "\n   - [{}]: {}".format(r['role'], r['content'])
+
+                reply_str += "\n\n"
+
+            reply_str += "\n当前默认情景预设:{}\n".format(dprompt.mode_inst().get_using_name())
+            reply_str += "请使用 !default <情景预设名称> 来设置默认情景预设"
+            reply = [reply_str]
+        elif params[0] != "set":
+            reply = ["[bot]err: 已弃用，请使用!default set <情景预设名称> 来设置默认情景预设"]
+        else:
+            return False, []
+
+        return True, reply
+
+
+@AbstractCommandNode.register(
+    parent=DefaultCommand,
+    name="set",
+    description="设置默认情景预设",
+    usage="!default set <情景预设名称>",
+    aliases=[],
+    privilege=2
+)
+class DefaultSetCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        reply = []
+
+        if len(ctx.crt_params) == 0:
+            reply = ["[bot]err: 请指定情景预设名称"]
+        elif len(ctx.crt_params) > 0:
+            import pkg.openai.dprompt as dprompt
+            try:
+                full_name = dprompt.mode_inst().set_using_name(ctx.crt_params[0])
+                reply = ["[bot]已设置默认情景预设为:{}".format(full_name)]
+            except Exception as e:
+                reply = ["[bot]err: {}".format(e)]
+        else:
+            reply = ["[bot]err: 仅管理员可设置默认情景预设"]
+
+        return True, reply

pkg/qqbot/cmds/session/del.py

@@ -0,0 +1,52 @@
+from ..mgr import AbstractCommandNode, Context
+import datetime
+
+
+@AbstractCommandNode.register(
+    parent=None,
+    name="del",
+    description="删除当前会话的历史记录",
+    usage="!del <序号>\n!del all",
+    aliases=[],
+    privilege=1
+)
+class DelCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        import pkg.openai.session
+        session_name = ctx.session_name
+        params = ctx.params
+        reply = []
+        if len(params) == 0:
+            reply = ["[bot]参数不足, 格式: !del <序号>\n可以通过!list查看序号"]
+        else:
+            if params[0] == 'all':
+                return False, []
+            elif params[0].isdigit():
+                if pkg.openai.session.get_session(session_name).delete_history(int(params[0])):
+                    reply = ["[bot]已删除历史会话 #{}".format(params[0])]
+                else:
+                    reply = ["[bot]没有历史会话 #{}".format(params[0])]
+            else:
+                reply = ["[bot]参数错误, 格式: !del <序号>\n可以通过!list查看序号"]
+
+        return True, reply
+
+
+@AbstractCommandNode.register(
+    parent=DelCommand,
+    name="all",
+    description="删除当前会话的全部历史记录",
+    usage="!del all",
+    aliases=[],
+    privilege=1
+)
+class DelAllCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        import pkg.openai.session
+        session_name = ctx.session_name
+        reply = []
+        pkg.openai.session.get_session(session_name).delete_all_history()
+        reply = ["[bot]已删除所有历史会话"]
+        return True, reply

pkg/qqbot/cmds/session/delhst.py

@@ -0,0 +1,50 @@
+from ..mgr import AbstractCommandNode, Context
+
+
+@AbstractCommandNode.register(
+    parent=None,
+    name="delhst",
+    description="删除指定会话的所有历史记录",
+    usage="!delhst <会话名称>\n!delhst all",
+    aliases=[],
+    privilege=2
+)
+class DelHistoryCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        import pkg.openai.session
+        import pkg.utils.context
+        params = ctx.params
+        reply = []
+        if len(params) == 0:
+            reply = [
+            "[bot]err:请输入要删除的会话名: group_<群号> 或者 person_<QQ号>, 或使用 !delhst all 删除所有会话的历史记录"]
+        else:
+            if params[0] == 'all':
+                return False, []
+            else:
+                if pkg.utils.context.get_database_manager().delete_all_history(params[0]):
+                    reply = ["[bot]已删除会话 {} 的所有历史记录".format(params[0])]
+                else:
+                    reply = ["[bot]未找到会话 {} 的历史记录".format(params[0])]
+
+        return True, reply
+    
+
+@AbstractCommandNode.register(
+    parent=DelHistoryCommand,
+    name="all",
+    description="删除所有会话的全部历史记录",
+    usage="!delhst all",
+    aliases=[],
+    privilege=2
+)
+class DelAllHistoryCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        import pkg.utils.context
+        reply = []
+        pkg.utils.context.get_database_manager().delete_all_session_history()
+        reply = ["[bot]已删除所有会话的历史记录"]
+        return True, reply
+    

pkg/qqbot/cmds/session/last.py

@@ -0,0 +1,28 @@
+from ..mgr import AbstractCommandNode, Context
+import datetime
+
+
+@AbstractCommandNode.register(
+    parent=None,
+    name="last",
+    description="切换前一次对话",
+    usage="!last",
+    aliases=[],
+    privilege=1
+)
+class LastCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        import pkg.openai.session
+        session_name = ctx.session_name
+
+        reply = []
+        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)]
+
+        return True, reply

pkg/qqbot/cmds/session/list.py

@@ -0,0 +1,67 @@
+from ..mgr import AbstractCommandNode, Context
+import datetime
+import json
+
+
+@AbstractCommandNode.register(
+    parent=None,
+    name='list',
+    description='列出当前会话的所有历史记录',
+    usage='!list\n!list [页数]',
+    aliases=[],
+    privilege=1
+)
+class ListCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        import pkg.openai.session
+        session_name = ctx.session_name
+        params = ctx.params
+        reply = []
+
+        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_str = "[bot]历史会话 第{}页：\n".format(page)
+            current = -1
+            for i in range(len(results)):
+                # 时间(使用create_timestamp转换) 序号 部分内容
+                datetime_obj = datetime.datetime.fromtimestamp(results[i]['create_timestamp'])
+                msg = ""
+                try:
+                    msg = json.loads(results[i]['prompt'])
+                except json.decoder.JSONDecodeError:
+                    msg = pkg.openai.session.reset_session_prompt(session_name, results[i]['prompt'])
+                    # 持久化
+                    pkg.openai.session.get_session(session_name).persistence()
+                if len(msg) >= 2:
+                    reply_str += "#{} 创建:{} {}\n".format(i + page * 10,
+                                                        datetime_obj.strftime("%Y-%m-%d %H:%M:%S"),
+                                                        msg[0]['content'])
+                else:
+                    reply_str += "#{} 创建:{} {}\n".format(i + page * 10,
+                                                        datetime_obj.strftime("%Y-%m-%d %H:%M:%S"),
+                                                        "无内容")
+                if results[i]['create_timestamp'] == pkg.openai.session.get_session(
+                        session_name).create_timestamp:
+                    current = i + page * 10
+
+            reply_str += "\n以上信息倒序排列"
+            if current != -1:
+                reply_str += ",当前会话是 #{}\n".format(current)
+            else:
+                reply_str += ",当前处于全新会话或不在此页"
+
+        reply = [reply_str]
+
+        return True, reply

pkg/qqbot/cmds/session/next.py

@@ -0,0 +1,28 @@
+from ..mgr import AbstractCommandNode, Context
+import datetime
+
+
+@AbstractCommandNode.register(
+    parent=None,
+    name="next",
+    description="切换后一次对话",
+    usage="!next",
+    aliases=[],
+    privilege=1
+)
+class NextCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        import pkg.openai.session
+        session_name = ctx.session_name
+        reply = []
+
+        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)]
+
+        return True, reply

pkg/qqbot/cmds/session/prompt.py

@@ -0,0 +1,32 @@
+from ..mgr import AbstractCommandNode, Context
+import datetime
+
+
+@AbstractCommandNode.register(
+    parent=None,
+    name="prompt",
+    description="获取当前会话的前文",
+    usage="!prompt",
+    aliases=[],
+    privilege=1
+)
+class PromptCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        import pkg.openai.session
+        session_name = ctx.session_name
+        params = ctx.params
+        reply = []
+
+        msgs = ""
+        session: list = pkg.openai.session.get_session(session_name).prompt
+        for msg in session:
+            if len(params) != 0 and params[0] in ['-all', '-a']:
+                msgs = msgs + "{}: {}\n\n".format(msg['role'], msg['content'])
+            elif len(msg['content']) > 30:
+                msgs = msgs + "[{}]: {}...\n\n".format(msg['role'], msg['content'][:30])
+            else:
+                msgs = msgs + "[{}]: {}\n\n".format(msg['role'], msg['content'])
+        reply = ["[bot]当前对话所有内容：\n{}".format(msgs)]
+
+        return True, reply

pkg/qqbot/cmds/session/resend.py

@@ -0,0 +1,30 @@
+from ..mgr import AbstractCommandNode, Context
+import datetime
+
+
+@AbstractCommandNode.register(
+    parent=None,
+    name="resend",
+    description="重新获取上一次问题的回复",
+    usage="!resend",
+    aliases=[],
+    privilege=1
+)
+class ResendCommand(AbstractCommandNode):
+    @classmethod
+    def process(cls, ctx: Context) -> tuple[bool, list]:
+        import pkg.openai.session
+        import config
+        session_name = ctx.session_name
+        reply = []
+
+        session = pkg.openai.session.get_session(session_name)
+        to_send = session.undo()
+
+        mgr = pkg.utils.context.get_qqbot_manager()
+
+        reply = pkg.qqbot.message.process_normal_message(to_send, mgr, config,
+                                                        ctx.launcher_type, ctx.launcher_id,
+                                                        ctx.sender_id)
+
+        return True, reply