From 15c8f0b6f798afb37471d322ab09738a6f348f6e Mon Sep 17 00:00:00 2001 From: Soulter <37870767+Soulter@users.noreply.github.com> Date: Sat, 8 Apr 2023 23:58:46 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20update=E5=92=8C=E5=88=87=E6=8D=A2?= =?UTF-8?q?=E6=A8=A1=E5=9E=8B=E6=8C=87=E4=BB=A4=E4=BB=A5=E5=8F=8Akeyword?= =?UTF-8?q?=E6=8C=87=E4=BB=A4=E7=8E=B0=E5=9C=A8=E4=BB=85=E5=8F=AF=E7=AE=A1?= =?UTF-8?q?=E7=90=86=E5=91=98=E4=BD=BF=E7=94=A8=EF=BC=9B=20fix:=20?= =?UTF-8?q?=E4=BF=AE=E5=A4=8Dkeyword=E6=8C=87=E4=BB=A4=E5=9C=A8=E4=BD=BF?= =?UTF-8?q?=E7=94=A8=E5=AE=98=E6=96=B9=E6=A8=A1=E5=9E=8B=E7=9A=84=E6=97=B6?= =?UTF-8?q?=E5=80=99=E4=BC=9A=E8=A2=AB=E8=AF=86=E5=88=AB=E4=B8=BA=E2=80=9C?= =?UTF-8?q?=E8=B5=9E=E5=8A=A9key=E2=80=9D=E7=9A=84=E6=8C=87=E4=BB=A4?= =?UTF-8?q?=E7=9A=84=E9=97=AE=E9=A2=98=20#80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cores/qqbot/core.py | 95 +++++++++++++++++------- model/command/command.py | 8 +- model/command/command_openai_official.py | 10 +-- model/command/command_rev_chatgpt.py | 6 +- model/command/command_rev_edgegpt.py | 6 +- 5 files changed, 86 insertions(+), 39 deletions(-) diff --git a/cores/qqbot/core.py b/cores/qqbot/core.py index d046514a5..671415972 100644 --- a/cores/qqbot/core.py +++ b/cores/qqbot/core.py @@ -351,7 +351,7 @@ def save_provider_preference(chosen_provider): ''' 处理消息 ''' -def oper_msg(message, at=False, msg_ref = None): +def oper_msg(message: Message, at=False, msg_ref = None): global session_dict, provider print("[QQBOT] 接收到消息："+ str(message.content)) qq_msg = '' @@ -360,7 +360,8 @@ def oper_msg(message, at=False, msg_ref = None): user_name = message.author.username global chosen_provider, reply_prefix, keywords hit = False # 是否命中指令 - command_result = () + command_result = () # 调用指令返回的结果 + role = "member" # 角色 # 检查发言频率 if not check_frequency(user_id): @@ -371,6 +372,7 @@ def oper_msg(message, at=False, msg_ref = None): logf.flush() if at: + # 在频道内 # 过滤@ qq_msg = message.content lines = qq_msg.splitlines() @@ -382,7 +384,15 @@ def oper_msg(message, at=False, msg_ref = None): session_id = user_id else: session_id = message.channel_id + + # 得到身份 + if "2" in message.member.roles or "4" in message.member.roles or "5" in message.member.roles: + print("[System] 检测到管理员身份") + role = "admin" + else: + role = "member" else: + # 私信 qq_msg = message.content session_id = user_id @@ -406,26 +416,57 @@ def oper_msg(message, at=False, msg_ref = None): return # 检查是否是更换语言模型的请求 + temp_switch = "" if qq_msg.startswith('/bing'): - chosen_provider = REV_EDGEGPT - save_provider_preference(chosen_provider) - send_qq_msg(message, f"已切换至【{chosen_provider}】", msg_ref=msg_ref) - return + l = qq_msg.split(' ') + if len(l) >= 1 and l[1] != "": + # 临时对话模式，先记录下之前的语言模型，回答完毕后再切回 + temp_switch = chosen_provider + chosen_provider = REV_EDGEGPT + qq_msg = l[1] + else: + if role != "admin": + send_qq_msg(message, f"你没有权限更换语言模型。", msg_ref=msg_ref) + return + chosen_provider = REV_EDGEGPT + save_provider_preference(chosen_provider) + send_qq_msg(message, f"已切换至【{chosen_provider}】", msg_ref=msg_ref) + return elif qq_msg.startswith('/gpt'): - chosen_provider = OPENAI_OFFICIAL - save_provider_preference(chosen_provider) - send_qq_msg(message, f"已切换至【{chosen_provider}】", msg_ref=msg_ref) - return + l = qq_msg.split(' ') + if len(l) >= 1 and l[1] != "": + # 临时对话模式，先记录下之前的语言模型，回答完毕后再切回 + temp_switch = chosen_provider + chosen_provider = OPENAI_OFFICIAL + qq_msg = l[1] + else: + if role != "admin": + send_qq_msg(message, f"你没有权限更换语言模型。", msg_ref=msg_ref) + return + chosen_provider = OPENAI_OFFICIAL + save_provider_preference(chosen_provider) + send_qq_msg(message, f"已切换至【{chosen_provider}】", msg_ref=msg_ref) + return elif qq_msg.startswith('/revgpt'): - chosen_provider = REV_CHATGPT - save_provider_preference(chosen_provider) - send_qq_msg(message, f"已切换至【{chosen_provider}】", msg_ref=msg_ref) - return + l = qq_msg.split(' ') + if len(l) >= 1 and l[1] != "": + # 临时对话模式，先记录下之前的语言模型，回答完毕后再切回 + temp_switch = chosen_provider + chosen_provider = REV_CHATGPT + qq_msg = l[1] + else: + if role != "admin": + send_qq_msg(message, f"你没有权限更换语言模型。", msg_ref=msg_ref) + return + chosen_provider = REV_CHATGPT + save_provider_preference(chosen_provider) + send_qq_msg(message, f"已切换至【{chosen_provider}】", msg_ref=msg_ref) + return chatgpt_res = "" if chosen_provider == OPENAI_OFFICIAL: - hit, command_result = command_openai_official.check_command(qq_msg, session_id, user_name) + hit, command_result = command_openai_official.check_command(qq_msg, session_id, user_name, role) print(f"{hit} {command_result}") # hit: 是否触发了指令. if not hit: @@ -438,15 +479,13 @@ def oper_msg(message, at=False, msg_ref = None): print("[System-Err] OpenAI API错误。原因如下:\n"+str(e)) if 'exceeded' in str(e): send_qq_msg(message, f"OpenAI API错误。原因：\n{str(e)} \n超额了。可自己搭建一个机器人(Github仓库：QQChannelChatGPT)") - return else: f_res = re.sub(r'(https|http)?:\/\/(\w|\.|\/|\?|\=|\&|\%)*\b', '[被隐藏的链接]', str(e), flags=re.MULTILINE) f_res = f_res.replace(".", "·") send_qq_msg(message, f"OpenAI API错误。原因如下：\n{f_res} \n前往官方频道反馈~") - return elif chosen_provider == REV_CHATGPT: - hit, command_result = command_rev_chatgpt.check_command(qq_msg) + hit, command_result = command_rev_chatgpt.check_command(qq_msg, role) if not hit: try: chatgpt_res = str(rev_chatgpt.text_chat(qq_msg)) @@ -455,14 +494,12 @@ def oper_msg(message, at=False, msg_ref = None): except BaseException as e: print("[System-Err] Rev ChatGPT API错误。原因如下:\n"+str(e)) send_qq_msg(message, f"Rev ChatGPT API错误。原因如下: \n{str(e)} \n前往官方频道反馈~") - return elif chosen_provider == REV_EDGEGPT: - hit, command_result = command_rev_edgegpt.check_command(qq_msg, client.loop) + hit, command_result = command_rev_edgegpt.check_command(qq_msg, client.loop, role) if not hit: try: if rev_edgegpt.is_busy(): send_qq_msg(message, f"[RevBing] 正忙，请稍后再试",msg_ref=msg_ref) - return else: res, res_code = asyncio.run_coroutine_threadsafe(rev_edgegpt.text_chat(qq_msg), client.loop).result() if res_code == 0: # bing不想继续话题，重置会话后重试。 @@ -471,14 +508,20 @@ def oper_msg(message, at=False, msg_ref = None): res, res_code = asyncio.run_coroutine_threadsafe(rev_edgegpt.text_chat(qq_msg), client.loop).result() if res_code == 0: # bing还是不想继续话题，大概率说明提问有问题。 send_qq_msg(message, f"Bing仍然不想继续话题, 请检查您的提问。", msg_ref=msg_ref) - return - chatgpt_res = str(res) - if REV_EDGEGPT in reply_prefix: - chatgpt_res = reply_prefix[REV_EDGEGPT] + chatgpt_res + else: + chatgpt_res = str(res) + if REV_EDGEGPT in reply_prefix: + chatgpt_res = reply_prefix[REV_EDGEGPT] + chatgpt_res except BaseException as e: print("[System-Err] Rev NewBing API错误。原因如下:\n"+str(e)) send_qq_msg(message, f"Rev NewBing API错误。原因如下：\n{str(e)} \n前往官方频道反馈~") - return + + # 切换回原来的语言模型 + if temp_switch != "": + chosen_provider = temp_switch + + if chatgpt_res == "": + return # 指令回复 if hit: diff --git a/model/command/command.py b/model/command/command.py index 2e4ac8083..0015f0a22 100644 --- a/model/command/command.py +++ b/model/command/command.py @@ -27,7 +27,9 @@ class Command: return True return False - def keyword(self, message: str): + def keyword(self, message: str, role: str): + if role != "admin": + return True, "你没有权限使用该指令", "keyword" if len(message.split(" ")) != 3: return True, "【设置关键词/关键指令回复】示例：\nkeyword hi 你好\n当发送hi的时候会回复你好\nkeyword /hi 你好\n当发送/hi时会回复你好", "keyword" @@ -45,7 +47,9 @@ class Command: except BaseException as e: return False, "设置失败: "+str(e), "keyword" - def update(self, message: str): + def update(self, message: str, role: str): + if role != "admin": + return True, "你没有权限使用该指令", "keyword" l = message.split(" ") if len(l) == 1: # 得到本地版本号和最新版本号 diff --git a/model/command/command_openai_official.py b/model/command/command_openai_official.py index 951f08b17..b17e3eac5 100644 --- a/model/command/command_openai_official.py +++ b/model/command/command_openai_official.py @@ -6,7 +6,7 @@ class CommandOpenAIOfficial(Command): def __init__(self, provider: ProviderOpenAIOfficial): self.provider = provider - def check_command(self, message: str, session_id: str, user_name: str): + def check_command(self, message: str, session_id: str, user_name: str, role): if self.command_start_with(message, "reset", "重置"): return True, self.reset(session_id) elif self.command_start_with(message, "his", "历史"): @@ -21,18 +21,18 @@ class CommandOpenAIOfficial(Command): return True, self.count() elif self.command_start_with(message, "help", "帮助"): return True, self.help() - elif self.command_start_with(message, "key"): - return True, self.key(message, user_name) elif self.command_start_with(message, "unset"): return True, self.unset(session_id) elif self.command_start_with(message, "set"): return True, self.set(message, session_id) elif self.command_start_with(message, "update"): - return True, self.update(message) + return True, self.update(message, role) elif self.command_start_with(message, "画"): return True, self.draw(message) elif self.command_start_with(message, "keyword"): - return True, self.keyword(message) + return True, self.keyword(message, role) + elif self.command_start_with(message, "key"): + return True, self.key(message, user_name) return False, None diff --git a/model/command/command_rev_chatgpt.py b/model/command/command_rev_chatgpt.py index 85fe78a69..cb2f61de0 100644 --- a/model/command/command_rev_chatgpt.py +++ b/model/command/command_rev_chatgpt.py @@ -5,13 +5,13 @@ class CommandRevChatGPT(Command): def __init__(self, provider: ProviderRevChatGPT): self.provider = provider - def check_command(self, message: str): + def check_command(self, message: str, role): if self.command_start_with(message, "help", "帮助"): return True, self.help() elif self.command_start_with(message, "update"): - return True, self.update(message) + return True, self.update(message, role) elif self.command_start_with(message, "keyword"): - return True, self.keyword(message) + return True, self.keyword(message, role) return False, None def help(self): diff --git a/model/command/command_rev_edgegpt.py b/model/command/command_rev_edgegpt.py index c9b66c5e3..06eabcd71 100644 --- a/model/command/command_rev_edgegpt.py +++ b/model/command/command_rev_edgegpt.py @@ -5,15 +5,15 @@ class CommandRevEdgeGPT(Command): def __init__(self, provider: ProviderRevEdgeGPT): self.provider = provider - def check_command(self, message: str, loop): + def check_command(self, message: str, loop, role): if self.command_start_with(message, "reset"): return True, self.reset(loop) elif self.command_start_with(message, "help"): return True, self.help() elif self.command_start_with(message, "update"): - return True, self.update(message) + return True, self.update(message, role) elif self.command_start_with(message, "keyword"): - return True, self.keyword(message) + return True, self.keyword(message, role) return False, None def reset(self, loop):