laoxong/AstrBot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: update和切换模型指令以及keyword指令现在仅可管理员使用；

Browse Source 
fix: 修复keyword指令在使用官方模型的时候会被识别为“赞助key”的指令的问题 #80
This commit is contained in:
Soulter
2023-04-08 23:58:46 +08:00
parent 9bca158174
commit 15c8f0b6f7
5 changed files with 86 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cores/qqbot/core.py
+69 -26
View File
@@ -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:
model/command/command.py
+6 -2
View File
@@ -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:
# 得到本地版本号和最新版本号
model/command/command_openai_official.py
+5 -5
View File
@@ -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
model/command/command_rev_chatgpt.py
+3 -3
View File
@@ -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):
model/command/command_rev_edgegpt.py
+3 -3
View File
@@ -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):