mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-15 17:30:13 +08:00
Merge branch 'master' into dev
This commit is contained in:
Soulter
@@ -1,540 +0,0 @@
|
||||
import json
|
||||
from util import general_utils as gu
|
||||
has_git = True
|
||||
try:
|
||||
import git.exc
|
||||
from git.repo import Repo
|
||||
except BaseException as e:
|
||||
gu.log("你正运行在无Git环境下，暂时将无法使用插件、热更新功能。")
|
||||
has_git = False
|
||||
import os
|
||||
import sys
|
||||
import requests
|
||||
from model.provider.provider import Provider
|
||||
import json
|
||||
import util.plugin_util as putil
|
||||
import shutil
|
||||
import importlib
|
||||
from util.cmd_config import CmdConfig as cc
|
||||
from model.platform.qq import QQ
|
||||
import stat
|
||||
from nakuru.entities.components import (
|
||||
Plain,
|
||||
Image
|
||||
)
|
||||
from PIL import Image as PILImage
|
||||
from cores.qqbot.global_object import GlobalObject, AstrMessageEvent
|
||||
from pip._internal import main as pipmain
|
||||
|
||||
from .adapter.protocol_adapter import UnifiedBotCompatibleLayer
|
||||
import asyncio
|
||||
|
||||
PLATFORM_QQCHAN = 'qqchan'
|
||||
PLATFORM_GOCQ = 'gocq'
|
||||
|
||||
# 指令功能的基类，通用的（不区分语言模型）的指令就在这实现
|
||||
class Command:
|
||||
def __init__(self, provider: Provider, global_object: GlobalObject = None, unified_bot_compatible_layer: UnifiedBotCompatibleLayer = None):
|
||||
self.provider = provider
|
||||
self.global_object = global_object
|
||||
self.unified_bot_compatible_layer = unified_bot_compatible_layer
|
||||
|
||||
|
||||
async def check_command(self,
|
||||
message,
|
||||
session_id: str,
|
||||
role,
|
||||
platform,
|
||||
message_obj):
|
||||
# UBCL
|
||||
await self.unified_bot_compatible_layer.check_commands(message, message_obj)
|
||||
|
||||
# 插件
|
||||
cached_plugins = self.global_object.cached_plugins
|
||||
ame = AstrMessageEvent(
|
||||
message_str=message,
|
||||
message_obj=message_obj,
|
||||
gocq_platform=self.global_object.platform_qq,
|
||||
qq_sdk_platform=self.global_object.platform_qqchan,
|
||||
platform=platform,
|
||||
role=role,
|
||||
global_object=self.global_object,
|
||||
session_id = session_id
|
||||
)
|
||||
for k, v in cached_plugins.items():
|
||||
try:
|
||||
hit, res = v["clsobj"].run(ame)
|
||||
if hit:
|
||||
return True, res
|
||||
except TypeError as e:
|
||||
# 参数不匹配，尝试使用旧的参数方案
|
||||
try:
|
||||
hit, res = v["clsobj"].run(message, role, platform, message_obj, self.global_object.platform_qq)
|
||||
if hit:
|
||||
return True, res
|
||||
except BaseException as e:
|
||||
gu.log(f"{k}插件加载出现问题，原因: {str(e)}\n已安装插件: {cached_plugins.keys}\n如果你没有相关装插件的想法, 请直接忽略此报错, 不影响其他功能的运行。", level=gu.LEVEL_WARNING)
|
||||
except BaseException as e:
|
||||
gu.log(f"{k}插件加载出现问题，原因: {str(e)}\n已安装插件: {cached_plugins.keys}\n如果你没有相关装插件的想法, 请直接忽略此报错, 不影响其他功能的运行。", level=gu.LEVEL_WARNING)
|
||||
|
||||
if self.command_start_with(message, "nick"):
|
||||
return True, self.set_nick(message, platform, role)
|
||||
if self.command_start_with(message, "plugin"):
|
||||
return True, self.plugin_oper(message, role, cached_plugins, platform)
|
||||
if self.command_start_with(message, "myid") or self.command_start_with(message, "!myid"):
|
||||
return True, self.get_my_id(message_obj)
|
||||
if self.command_start_with(message, "nconf") or self.command_start_with(message, "newconf"):
|
||||
return True, self.get_new_conf(message, role)
|
||||
if self.command_start_with(message, "web"): # 网页搜索
|
||||
return True, self.web_search(message)
|
||||
if self.command_start_with(message, "keyword"):
|
||||
return True, self.keyword(message_obj, role)
|
||||
|
||||
return False, None
|
||||
|
||||
def web_search(self, message):
|
||||
if message == "web on":
|
||||
self.global_object.web_search = True
|
||||
return True, "已开启网页搜索", "web"
|
||||
elif message == "web off":
|
||||
self.global_object.web_search = False
|
||||
return True, "已关闭网页搜索", "web"
|
||||
return True, f"网页搜索功能当前状态: {self.global_object.web_search}", "web"
|
||||
|
||||
def get_my_id(self, message_obj):
|
||||
return True, f"你的ID：{str(message_obj.sender.tiny_id)}", "plugin"
|
||||
|
||||
def get_new_conf(self, message, role):
|
||||
if role != "admin":
|
||||
return False, f"你的身份组{role}没有权限使用此指令。", "newconf"
|
||||
l = message.split(" ")
|
||||
if len(l) <= 1:
|
||||
obj = cc.get_all()
|
||||
p = gu.create_text_image("【cmd_config.json】", json.dumps(obj, indent=4, ensure_ascii=False))
|
||||
return True, [Image.fromFileSystem(p)], "newconf"
|
||||
|
||||
def get_plugin_modules(self):
|
||||
plugins = []
|
||||
try:
|
||||
if os.path.exists("addons/plugins"):
|
||||
plugins = putil.get_modules("addons/plugins")
|
||||
return plugins
|
||||
elif os.path.exists("QQChannelChatGPT/addons/plugins"):
|
||||
plugins = putil.get_modules("QQChannelChatGPT/addons/plugins")
|
||||
return plugins
|
||||
else:
|
||||
return None
|
||||
except BaseException as e:
|
||||
raise e
|
||||
|
||||
def plugin_reload(self, cached_plugins: dict, target: str = None, all: bool = False):
|
||||
plugins = self.get_plugin_modules()
|
||||
fail_rec = ""
|
||||
if plugins is None:
|
||||
return False, "未找到任何插件模块"
|
||||
|
||||
for plugin in plugins:
|
||||
try:
|
||||
p = plugin['module']
|
||||
root_dir_name = plugin['pname']
|
||||
if p not in cached_plugins or p == target or all:
|
||||
module = __import__("addons.plugins." + root_dir_name + "." + p, fromlist=[p])
|
||||
if p in cached_plugins:
|
||||
module = importlib.reload(module)
|
||||
cls = putil.get_classes(p, module)
|
||||
obj = getattr(module, cls[0])()
|
||||
try:
|
||||
info = obj.info()
|
||||
if 'name' not in info or 'desc' not in info or 'version' not in info or 'author' not in info:
|
||||
fail_rec += f"载入插件{p}失败，原因: 插件信息不完整\n"
|
||||
continue
|
||||
if isinstance(info, dict) == False:
|
||||
fail_rec += f"载入插件{p}失败，原因: 插件信息格式不正确\n"
|
||||
continue
|
||||
except BaseException as e:
|
||||
fail_rec += f"调用插件{p} info失败, 原因: {str(e)}\n"
|
||||
continue
|
||||
cached_plugins[info['name']] = {
|
||||
"module": module,
|
||||
"clsobj": obj,
|
||||
"info": info,
|
||||
"name": info['name'],
|
||||
"root_dir_name": root_dir_name,
|
||||
}
|
||||
except BaseException as e:
|
||||
fail_rec += f"加载{p}插件出现问题，原因 {str(e)}\n"
|
||||
if fail_rec == "":
|
||||
return True, None
|
||||
else:
|
||||
return False, fail_rec
|
||||
|
||||
'''
|
||||
插件指令
|
||||
'''
|
||||
def plugin_oper(self, message: str, role: str, cached_plugins: dict, platform: str):
|
||||
if not has_git:
|
||||
return False, "你正在运行在无Git环境下，暂时将无法使用插件、热更新功能。", "plugin"
|
||||
l = message.split(" ")
|
||||
if len(l) < 2:
|
||||
p = gu.create_text_image("【插件指令面板】", "安装插件: \nplugin i 插件Github地址\n卸载插件: \nplugin d 插件名 \n重载插件: \nplugin reload\n查看插件列表：\nplugin l\n更新插件: plugin u 插件名\n")
|
||||
return True, [Image.fromFileSystem(p)], "plugin"
|
||||
else:
|
||||
ppath = ""
|
||||
if os.path.exists("addons/plugins"):
|
||||
ppath = "addons/plugins"
|
||||
elif os.path.exists("QQChannelChatGPT/addons/plugins"):
|
||||
ppath = "QQChannelChatGPT/addons/plugins"
|
||||
else:
|
||||
return False, "未找到插件目录", "plugin"
|
||||
if l[1] == "i":
|
||||
if role != "admin":
|
||||
return False, f"你的身份组{role}没有权限安装插件", "plugin"
|
||||
try:
|
||||
# 删除末尾的/
|
||||
if l[2].endswith("/"):
|
||||
l[2] = l[2][:-1]
|
||||
# 得到url的最后一段
|
||||
d = l[2].split("/")[-1]
|
||||
# 转换非法字符：-
|
||||
d = d.replace("-", "_")
|
||||
# 创建文件夹
|
||||
plugin_path = os.path.join(ppath, d)
|
||||
if os.path.exists(plugin_path):
|
||||
shutil.rmtree(plugin_path)
|
||||
os.mkdir(plugin_path)
|
||||
Repo.clone_from(l[2],to_path=plugin_path,branch='master')
|
||||
|
||||
# 读取插件的requirements.txt
|
||||
if os.path.exists(os.path.join(plugin_path, "requirements.txt")):
|
||||
mm = pipmain(['install', '-r', os.path.join(plugin_path, "requirements.txt")])
|
||||
if mm != 0:
|
||||
return False, "插件依赖安装失败，需要您手动pip安装对应插件的依赖。", "plugin"
|
||||
# 加载没缓存的插件
|
||||
ok, err = self.plugin_reload(cached_plugins, target=d)
|
||||
if ok:
|
||||
return True, "插件拉取并载入成功~", "plugin"
|
||||
else:
|
||||
# if os.path.exists(plugin_path):
|
||||
# shutil.rmtree(plugin_path)
|
||||
return False, f"插件拉取载入失败。\n跟踪: \n{err}", "plugin"
|
||||
except BaseException as e:
|
||||
return False, f"拉取插件失败，原因: {str(e)}", "plugin"
|
||||
elif l[1] == "d":
|
||||
if role != "admin":
|
||||
return False, f"你的身份组{role}没有权限删除插件", "plugin"
|
||||
if l[2] not in cached_plugins:
|
||||
return False, "未找到该插件", "plugin"
|
||||
|
||||
try:
|
||||
root_dir_name = cached_plugins[l[2]]["root_dir_name"]
|
||||
self.remove_dir(os.path.join(ppath, root_dir_name))
|
||||
del cached_plugins[l[2]]
|
||||
return True, "插件卸载成功~", "plugin"
|
||||
except BaseException as e:
|
||||
return False, f"卸载插件失败，原因: {str(e)}", "plugin"
|
||||
elif l[1] == "u":
|
||||
if l[2] not in cached_plugins:
|
||||
return False, "未找到该插件", "plugin"
|
||||
root_dir_name = cached_plugins[l[2]]["root_dir_name"]
|
||||
plugin_path = os.path.join(ppath, root_dir_name)
|
||||
try:
|
||||
repo = Repo(path = plugin_path)
|
||||
repo.remotes.origin.pull()
|
||||
# 读取插件的requirements.txt
|
||||
if os.path.exists(os.path.join(plugin_path, "requirements.txt")):
|
||||
mm = pipmain(['install', '-r', os.path.join(plugin_path, "requirements.txt")])
|
||||
if mm != 0:
|
||||
return False, "插件依赖安装失败，需要您手动pip安装对应插件的依赖。", "plugin"
|
||||
|
||||
ok, err = self.plugin_reload(cached_plugins, target=l[2])
|
||||
if ok:
|
||||
return True, "\n更新插件成功!!", "plugin"
|
||||
else:
|
||||
return False, "更新插件成功，但是重载插件失败。\n问题跟踪: \n"+err, "plugin"
|
||||
except BaseException as e:
|
||||
return False, "更新插件失败, 请使用plugin i指令覆盖安装", "plugin"
|
||||
|
||||
elif l[1] == "l":
|
||||
try:
|
||||
plugin_list_info = "\n".join([f"{k}: \n名称: {v['info']['name']}\n简介: {v['info']['desc']}\n版本: {v['info']['version']}\n作者: {v['info']['author']}\n" for k, v in cached_plugins.items()])
|
||||
p = gu.create_text_image("【已激活插件列表】", plugin_list_info + "\n使用plugin v 插件名 查看插件帮助\n")
|
||||
return True, [Image.fromFileSystem(p)], "plugin"
|
||||
except BaseException as e:
|
||||
return False, f"获取插件列表失败，原因: {str(e)}", "plugin"
|
||||
elif l[1] == "v":
|
||||
try:
|
||||
if l[2] in cached_plugins:
|
||||
info = cached_plugins[l[2]]["info"]
|
||||
p = gu.create_text_image(f"【插件信息】", f"名称: {info['name']}\n{info['desc']}\n版本: {info['version']}\n作者: {info['author']}\n\n帮助:\n{info['help']}")
|
||||
return True, [Image.fromFileSystem(p)], "plugin"
|
||||
else:
|
||||
return False, "未找到该插件", "plugin"
|
||||
except BaseException as e:
|
||||
return False, f"获取插件信息失败，原因: {str(e)}", "plugin"
|
||||
elif l[1] == "reload":
|
||||
if role != "admin":
|
||||
return False, f"你的身份组{role}没有权限重载插件", "plugin"
|
||||
for plugin in cached_plugins:
|
||||
try:
|
||||
print(f"更新插件 {plugin} 依赖...")
|
||||
plugin_path = os.path.join(ppath, cached_plugins[plugin]["root_dir_name"])
|
||||
if os.path.exists(os.path.join(plugin_path, "requirements.txt")):
|
||||
mm = pipmain(['install', '-r', os.path.join(plugin_path, "requirements.txt"), "--quiet"])
|
||||
if mm != 0:
|
||||
return False, "插件依赖安装失败，需要您手动pip安装对应插件的依赖。", "plugin"
|
||||
except BaseException as e:
|
||||
print(f"插件{plugin}依赖安装失败，原因: {str(e)}")
|
||||
try:
|
||||
ok, err = self.plugin_reload(cached_plugins, all = True)
|
||||
if ok:
|
||||
return True, "\n重载插件成功~", "plugin"
|
||||
else:
|
||||
# if os.path.exists(plugin_path):
|
||||
# shutil.rmtree(plugin_path)
|
||||
return False, f"插件重载失败。\n跟踪: \n{err}", "plugin"
|
||||
except BaseException as e:
|
||||
return False, f"插件重载失败，原因: {str(e)}", "plugin"
|
||||
|
||||
elif l[1] == "dev":
|
||||
if role != "admin":
|
||||
return False, f"你的身份组{role}没有权限开发者模式", "plugin"
|
||||
return True, "cached_plugins: \n" + str(cached_plugins), "plugin"
|
||||
|
||||
def remove_dir(self, file_path):
|
||||
while 1:
|
||||
if not os.path.exists(file_path):
|
||||
break
|
||||
try:
|
||||
shutil.rmtree(file_path)
|
||||
except PermissionError as e:
|
||||
err_file_path = str(e).split("\'", 2)[1]
|
||||
if os.path.exists(err_file_path):
|
||||
os.chmod(err_file_path, stat.S_IWUSR)
|
||||
|
||||
'''
|
||||
nick: 存储机器人的昵称
|
||||
'''
|
||||
def set_nick(self, message: str, platform: str, role: str = "member"):
|
||||
if role != "admin":
|
||||
return True, "你无权使用该指令 :P", "nick"
|
||||
if platform == PLATFORM_GOCQ:
|
||||
l = message.split(" ")
|
||||
if len(l) == 1:
|
||||
return True, "【设置机器人昵称】示例：\n支持多昵称\nnick 昵称1 昵称2 昵称3", "nick"
|
||||
nick = l[1:]
|
||||
cc.put("nick_qq", nick)
|
||||
self.global_object.nick = tuple(nick)
|
||||
return True, f"设置成功！现在你可以叫我这些昵称来提问我啦~", "nick"
|
||||
elif platform == PLATFORM_QQCHAN:
|
||||
nick = message.split(" ")[2]
|
||||
return False, "QQ频道平台不支持为机器人设置昵称。", "nick"
|
||||
|
||||
def general_commands(self):
|
||||
return {
|
||||
"help": "帮助",
|
||||
"keyword": "设置关键词/关键指令回复",
|
||||
"update": "更新面板",
|
||||
"update latest": "更新到最新版本",
|
||||
"update r": "重启机器人",
|
||||
"reset": "重置会话",
|
||||
"nick": "设置机器人昵称",
|
||||
"plugin": "插件安装、卸载和重载",
|
||||
"web on/off": "启动或关闭网页搜索能力",
|
||||
"/bing": "切换到bing模型",
|
||||
"/gpt": "切换到OpenAI ChatGPT API",
|
||||
"/revgpt": "切换到网页版ChatGPT",
|
||||
}
|
||||
|
||||
def help_messager(self, commands: dict, platform: str, cached_plugins: dict = None):
|
||||
try:
|
||||
resp = requests.get("https://soulter.top/channelbot/notice.json").text
|
||||
notice = json.loads(resp)["notice"]
|
||||
except BaseException as e:
|
||||
notice = ""
|
||||
msg = "# Help Center\n## 指令列表\n"
|
||||
# msg = "Github项目名QQChannelChatGPT, 有问题提交issue, 欢迎Star\n【指令列表】\n"
|
||||
for key, value in commands.items():
|
||||
msg += f"`{key}` - {value}\n"
|
||||
# plugins
|
||||
if cached_plugins != None:
|
||||
plugin_list_info = "\n".join([f"`{k}` {v['info']['name']}\n{v['info']['desc']}\n" for k, v in cached_plugins.items()])
|
||||
if plugin_list_info.strip() != "":
|
||||
msg += "\n## 插件列表\n> 使用plugin v 插件名 查看插件帮助\n"
|
||||
msg += plugin_list_info
|
||||
msg += notice
|
||||
|
||||
try:
|
||||
# p = gu.create_text_image("【Help Center】", msg)
|
||||
p = gu.create_markdown_image(msg)
|
||||
return [Image.fromFileSystem(p)]
|
||||
except BaseException as e:
|
||||
gu.log(str(e))
|
||||
finally:
|
||||
return msg
|
||||
|
||||
# 接受可变参数
|
||||
def command_start_with(self, message: str, *args):
|
||||
for arg in args:
|
||||
if message.startswith(arg) or message.startswith('/'+arg):
|
||||
return True
|
||||
return False
|
||||
|
||||
# keyword: 关键字
|
||||
def keyword(self, message_obj, role: str):
|
||||
if role != "admin":
|
||||
return True, "你没有权限使用该指令", "keyword"
|
||||
|
||||
plain_text = ""
|
||||
image_url = ""
|
||||
|
||||
for comp in message_obj.message:
|
||||
if isinstance(comp, Plain):
|
||||
plain_text += comp.text
|
||||
elif isinstance(comp, Image) and image_url == "":
|
||||
if comp.url is None:
|
||||
image_url = comp.file
|
||||
else:
|
||||
image_url = comp.url
|
||||
|
||||
l = plain_text.split(" ")
|
||||
|
||||
if len(l) < 3 and image_url == "":
|
||||
return True, """
|
||||
【设置关键词回复】示例：
|
||||
1. keyword hi 你好
|
||||
当发送hi的时候会回复你好
|
||||
2. keyword /hi 你好
|
||||
当发送/hi时会回复你好
|
||||
3. keyword d hi
|
||||
删除hi关键词的回复
|
||||
4. keyword hi <图片>
|
||||
当发送hi时会回复图片
|
||||
""", "keyword"
|
||||
|
||||
del_mode = False
|
||||
if l[1] == "d":
|
||||
del_mode = True
|
||||
|
||||
try:
|
||||
if os.path.exists("keyword.json"):
|
||||
with open("keyword.json", "r", encoding="utf-8") as f:
|
||||
keyword = json.load(f)
|
||||
if del_mode:
|
||||
# 删除关键词
|
||||
if l[2] not in keyword:
|
||||
return False, "该关键词不存在", "keyword"
|
||||
else: del keyword[l[2]]
|
||||
else:
|
||||
keyword[l[1]] = {
|
||||
"plain_text": " ".join(l[2:]),
|
||||
"image_url": image_url
|
||||
}
|
||||
else:
|
||||
if del_mode:
|
||||
return False, "该关键词不存在", "keyword"
|
||||
keyword = {
|
||||
l[1]: {
|
||||
"plain_text": " ".join(l[2:]),
|
||||
"image_url": image_url
|
||||
}
|
||||
}
|
||||
with open("keyword.json", "w", encoding="utf-8") as f:
|
||||
json.dump(keyword, f, ensure_ascii=False, indent=4)
|
||||
f.flush()
|
||||
if del_mode:
|
||||
return True, "删除成功: "+l[2], "keyword"
|
||||
if image_url == "":
|
||||
return True, "设置成功: "+l[1]+" "+" ".join(l[2:]), "keyword"
|
||||
else:
|
||||
return True, [Plain("设置成功: "+l[1]+" "+" ".join(l[2:])), Image.fromURL(image_url)], "keyword"
|
||||
except BaseException as e:
|
||||
return False, "设置失败: "+str(e), "keyword"
|
||||
|
||||
def update(self, message: str, role: str):
|
||||
if not has_git:
|
||||
return False, "你正在运行在无Git环境下，暂时将无法使用插件、热更新功能。", "update"
|
||||
if role != "admin":
|
||||
return True, "你没有权限使用该指令", "keyword"
|
||||
l = message.split(" ")
|
||||
if len(l) == 1:
|
||||
# 得到本地版本号和最新版本号
|
||||
try:
|
||||
repo = Repo()
|
||||
except git.exc.InvalidGitRepositoryError:
|
||||
repo = Repo(path="QQChannelChatGPT")
|
||||
now_commit = repo.head.commit
|
||||
|
||||
# 得到远程3条commit列表, 包含commit信息
|
||||
origin = repo.remotes.origin
|
||||
origin.fetch()
|
||||
commits = list(repo.iter_commits('master', max_count=3))
|
||||
commits_log = ''
|
||||
index = 1
|
||||
for commit in commits:
|
||||
if commit.message.endswith("\n"):
|
||||
commits_log += f"[{index}] {commit.message}-----------\n"
|
||||
else:
|
||||
commits_log += f"[{index}] {commit.message}\n-----------\n"
|
||||
index+=1
|
||||
remote_commit_hash = origin.refs.master.commit.hexsha[:6]
|
||||
|
||||
return True, f"当前版本: {now_commit.hexsha[:6]}\n最新版本: {remote_commit_hash}\n\n3条commit(非最新):\n{str(commits_log)}\n使用update latest更新至最新版本\n", "update"
|
||||
else:
|
||||
if l[1] == "latest":
|
||||
pash_tag = ""
|
||||
try:
|
||||
try:
|
||||
repo = Repo()
|
||||
except git.exc.InvalidGitRepositoryError:
|
||||
repo = Repo(path="QQChannelChatGPT")
|
||||
pash_tag = "QQChannelChatGPT"+os.sep
|
||||
repo.remotes.origin.pull()
|
||||
|
||||
try:
|
||||
origin = repo.remotes.origin
|
||||
origin.fetch()
|
||||
commits = list(repo.iter_commits('master', max_count=1))
|
||||
commit_log = commits[0].message
|
||||
except BaseException as e:
|
||||
commit_log = "无法获取commit信息"
|
||||
|
||||
tag = "update"
|
||||
if len(l) == 3 and l[2] == "r":
|
||||
tag = "update latest r"
|
||||
|
||||
return True, f"更新成功。新版本内容: \n{commit_log}\nps:重启后生效。输入update r重启（重启指令不返回任何确认信息）。", tag
|
||||
|
||||
except BaseException as e:
|
||||
return False, "更新失败: "+str(e), "update"
|
||||
if l[1] == "r":
|
||||
py = sys.executable
|
||||
os.execl(py, py, *sys.argv)
|
||||
|
||||
def reset(self):
|
||||
return False
|
||||
|
||||
def set(self):
|
||||
return False
|
||||
|
||||
def unset(self):
|
||||
return False
|
||||
|
||||
def key(self):
|
||||
return False
|
||||
|
||||
def help(self):
|
||||
return False
|
||||
|
||||
def status(self):
|
||||
return False
|
||||
|
||||
def token(self):
|
||||
return False
|
||||
|
||||
def his(self):
|
||||
return False
|
||||
|
||||
def draw(self):
|
||||
return False
|
||||
|
||||
|
||||
266
model/command/internal_handler.py
Normal file
266
model/command/internal_handler.py
Normal file
@@ -0,0 +1,266 @@
|
||||
import aiohttp
|
||||
|
||||
from model.command.manager import CommandManager
|
||||
from model.plugin.manager import PluginManager
|
||||
from type.message_event import AstrMessageEvent
|
||||
from type.command import CommandResult
|
||||
from type.types import Context
|
||||
from type.config import VERSION
|
||||
from SparkleLogging.utils.core import LogManager
|
||||
from logging import Logger
|
||||
from nakuru.entities.components import Image
|
||||
|
||||
logger: Logger = LogManager.GetLogger(log_name='astrbot')
|
||||
|
||||
|
||||
class InternalCommandHandler:
|
||||
def __init__(self, manager: CommandManager, plugin_manager: PluginManager) -> None:
|
||||
self.manager = manager
|
||||
self.plugin_manager = plugin_manager
|
||||
|
||||
self.manager.register("help", "查看帮助", 10, self.help)
|
||||
self.manager.register("wake", "设置机器人唤醒词", 10, self.set_nick)
|
||||
self.manager.register("update", "更新 AstrBot", 10, self.update)
|
||||
self.manager.register("plugin", "插件管理", 10, self.plugin)
|
||||
self.manager.register("reboot", "重启 AstrBot", 10, self.reboot)
|
||||
self.manager.register("websearch", "网页搜索开关", 10, self.web_search)
|
||||
self.manager.register("t2i", "文本转图片开关", 10, self.t2i_toggle)
|
||||
self.manager.register("myid", "获取你在此平台上的ID", 10, self.myid)
|
||||
self.manager.register("provider", "查看和切换当前使用的 LLM 资源来源", 10, self.provider)
|
||||
|
||||
def provider(self, message: AstrMessageEvent, context: Context):
|
||||
if len(context.llms) == 0:
|
||||
return CommandResult().message("当前没有加载任何 LLM 资源。")
|
||||
|
||||
tokens = self.manager.command_parser.parse(message.message_str)
|
||||
|
||||
if tokens.len == 1:
|
||||
ret = "## 当前载入的 LLM 资源\n"
|
||||
for idx, llm in enumerate(context.llms):
|
||||
ret += f"{idx}. {llm.llm_name}"
|
||||
if llm.origin:
|
||||
ret += f" (来源: {llm.origin})"
|
||||
if context.message_handler.provider == llm.llm_instance:
|
||||
ret += " (当前使用)"
|
||||
ret += "\n"
|
||||
|
||||
ret += "\n使用 provider <序号> 切换 LLM 资源。"
|
||||
return CommandResult().message(ret)
|
||||
else:
|
||||
try:
|
||||
idx = int(tokens.get(1))
|
||||
if idx >= len(context.llms):
|
||||
return CommandResult().message("provider: 无效的序号。")
|
||||
context.message_handler.set_provider(context.llms[idx].llm_instance)
|
||||
return CommandResult().message(f"已经成功切换到 LLM 资源 {context.llms[idx].llm_name}。")
|
||||
except BaseException as e:
|
||||
return CommandResult().message("provider: 参数错误。")
|
||||
|
||||
def set_nick(self, message: AstrMessageEvent, context: Context):
|
||||
message_str = message.message_str
|
||||
if message.role != "admin":
|
||||
return CommandResult().message("你没有权限使用该指令。")
|
||||
l = message_str.split(" ")
|
||||
if len(l) == 1:
|
||||
return CommandResult().message(f"设置机器人唤醒词。以唤醒词开头的消息会唤醒机器人处理，起到 @ 的效果。\n示例：wake 昵称。当前唤醒词有：{context.nick}")
|
||||
nick = l[1].strip()
|
||||
if not nick:
|
||||
return CommandResult().message("wake: 请指定唤醒词。")
|
||||
context.config_helper.put("nick_qq", nick)
|
||||
context.nick = tuple(nick)
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=True,
|
||||
message_chain=f"已经成功将唤醒词设定为 {nick}。",
|
||||
)
|
||||
|
||||
def update(self, message: AstrMessageEvent, context: Context):
|
||||
tokens = self.manager.command_parser.parse(message.message_str)
|
||||
if message.role != "admin":
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=False,
|
||||
message_chain="你没有权限使用该指令",
|
||||
)
|
||||
update_info = context.updator.check_update(None, None)
|
||||
if tokens.len == 1:
|
||||
ret = ""
|
||||
if not update_info:
|
||||
ret = f"当前已经是最新版本 v{VERSION}。"
|
||||
else:
|
||||
ret = f"发现新版本 {update_info.version}，更新内容如下:\n---\n{update_info.body}\n---\n- 使用 /update latest 更新到最新版本。\n- 使用 /update vX.X.X 更新到指定版本。"
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=False,
|
||||
message_chain=ret,
|
||||
)
|
||||
else:
|
||||
if tokens.get(1) == "latest":
|
||||
try:
|
||||
context.updator.update()
|
||||
return CommandResult().message(f"已经成功更新到最新版本 v{update_info.version}。要应用更新，请重启 AstrBot。输入 /reboot 即可重启")
|
||||
except BaseException as e:
|
||||
return CommandResult().message(f"更新失败。原因：{str(e)}")
|
||||
elif tokens.get(1).startswith("v"):
|
||||
try:
|
||||
context.updator.update(version=tokens.get(1))
|
||||
return CommandResult().message(f"已经成功更新到版本 v{tokens.get(1)}。要应用更新，请重启 AstrBot。输入 /reboot 即可重启")
|
||||
except BaseException as e:
|
||||
return CommandResult().message(f"更新失败。原因：{str(e)}")
|
||||
else:
|
||||
return CommandResult().message("update: 参数错误。")
|
||||
|
||||
def reboot(self, message: AstrMessageEvent, context: Context):
|
||||
if message.role != "admin":
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=False,
|
||||
message_chain="你没有权限使用该指令",
|
||||
)
|
||||
context.updator._reboot(3, context)
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=True,
|
||||
message_chain="AstrBot 将在 3s 后重启。",
|
||||
)
|
||||
|
||||
def plugin(self, message: AstrMessageEvent, context: Context):
|
||||
tokens = self.manager.command_parser.parse(message.message_str)
|
||||
if tokens.len == 1:
|
||||
ret = "# 插件指令面板 \n- 安装插件: `plugin i 插件Github地址`\n- 卸载插件: `plugin d 插件名`\n- 查看插件列表：`plugin l`\n - 更新插件: `plugin u 插件名`\n"
|
||||
return CommandResult().message(ret)
|
||||
|
||||
if tokens.get(1) == "l":
|
||||
plugin_list_info = ""
|
||||
for plugin in context.cached_plugins:
|
||||
plugin_list_info += f"- `{plugin.metadata.plugin_name}` By {plugin.metadata.author}: {plugin.metadata.desc}\n"
|
||||
if plugin_list_info.strip() == "":
|
||||
return CommandResult().message("plugin v: 没有找到插件。")
|
||||
return CommandResult().message(plugin_list_info)
|
||||
|
||||
elif tokens.get(1) == "d":
|
||||
if message.role != "admin":
|
||||
return CommandResult().message("plugin d: 你没有权限使用该指令。")
|
||||
if tokens.len == 2:
|
||||
return CommandResult().message("plugin d: 请指定要卸载的插件名。")
|
||||
plugin_name = tokens.get(2)
|
||||
try:
|
||||
self.plugin_manager.uninstall_plugin(plugin_name)
|
||||
except BaseException as e:
|
||||
return CommandResult().message(f"plugin d: 卸载插件失败。原因：{str(e)}")
|
||||
return CommandResult().message(f"plugin d: 已经成功卸载插件 {plugin_name}。")
|
||||
|
||||
elif tokens.get(1) == "i":
|
||||
if message.role != "admin":
|
||||
return CommandResult().message("plugin i: 你没有权限使用该指令。")
|
||||
if tokens.len == 2:
|
||||
return CommandResult().message("plugin i: 请指定要安装的插件的 Github 地址，或者前往可视化面板安装。")
|
||||
plugin_url = tokens.get(2)
|
||||
try:
|
||||
self.plugin_manager.install_plugin(plugin_url)
|
||||
except BaseException as e:
|
||||
return CommandResult().message(f"plugin i: 安装插件失败。原因：{str(e)}")
|
||||
return CommandResult().message("plugin i: 已经成功安装插件。")
|
||||
|
||||
elif tokens.get(1) == "u":
|
||||
if message.role != "admin":
|
||||
return CommandResult().message("plugin u: 你没有权限使用该指令。")
|
||||
if tokens.len == 2:
|
||||
return CommandResult().message("plugin u: 请指定要更新的插件名。")
|
||||
plugin_name = tokens.get(2)
|
||||
try:
|
||||
self.plugin_manager.update_plugin(plugin_name)
|
||||
except BaseException as e:
|
||||
return CommandResult().message(f"plugin u: 更新插件失败。原因：{str(e)}")
|
||||
return CommandResult().message(f"plugin u: 已经成功更新插件 {plugin_name}。")
|
||||
|
||||
return CommandResult().message("plugin: 参数错误。")
|
||||
|
||||
async def help(self, message: AstrMessageEvent, context: Context):
|
||||
notice = ""
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get("https://soulter.top/channelbot/notice.json") as resp:
|
||||
notice = (await resp.json())["notice"]
|
||||
except BaseException as e:
|
||||
logger.warn("An error occurred while fetching astrbot notice. Never mind, it's not important.")
|
||||
|
||||
msg = "# Help Center\n## 指令列表\n"
|
||||
for key, value in self.manager.commands_handler.items():
|
||||
if value.plugin_metadata:
|
||||
msg += f"- `{key}` ({value.plugin_metadata.plugin_name}): {value.description}\n"
|
||||
else: msg += f"- `{key}`: {value.description}\n"
|
||||
# plugins
|
||||
if context.cached_plugins != None:
|
||||
plugin_list_info = ""
|
||||
for plugin in context.cached_plugins:
|
||||
plugin_list_info += f"- `{plugin.metadata.plugin_name}` {plugin.metadata.desc}\n"
|
||||
if plugin_list_info.strip() != "":
|
||||
msg += "\n## 插件列表\n> 使用plugin v 插件名 查看插件帮助\n"
|
||||
msg += plugin_list_info
|
||||
msg += notice
|
||||
|
||||
return CommandResult().message(msg)
|
||||
|
||||
def web_search(self, message: AstrMessageEvent, context: Context):
|
||||
l = message.message_str.split(' ')
|
||||
if len(l) == 1:
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=True,
|
||||
message_chain=f"网页搜索功能当前状态: {context.web_search}",
|
||||
)
|
||||
elif l[1] == 'on':
|
||||
context.web_search = True
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=True,
|
||||
message_chain="已开启网页搜索",
|
||||
)
|
||||
elif l[1] == 'off':
|
||||
context.web_search = False
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=True,
|
||||
message_chain="已关闭网页搜索",
|
||||
)
|
||||
else:
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=False,
|
||||
message_chain="参数错误",
|
||||
)
|
||||
|
||||
def t2i_toggle(self, message: AstrMessageEvent, context: Context):
|
||||
p = context.t2i_mode
|
||||
if p:
|
||||
context.config_helper.put("qq_pic_mode", False)
|
||||
context.t2i_mode = False
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=True,
|
||||
message_chain="已关闭文本转图片模式。",
|
||||
)
|
||||
context.config_helper.put("qq_pic_mode", True)
|
||||
context.t2i_mode = True
|
||||
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=True,
|
||||
message_chain="已开启文本转图片模式。",
|
||||
)
|
||||
|
||||
def myid(self, message: AstrMessageEvent, context: Context):
|
||||
try:
|
||||
user_id = str(message.message_obj.sender.user_id)
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=True,
|
||||
message_chain=f"你在此平台上的ID：{user_id}",
|
||||
)
|
||||
except BaseException as e:
|
||||
return CommandResult(
|
||||
hit=True,
|
||||
success=False,
|
||||
message_chain=f"在 {message.platform} 上获取你的ID失败，原因: {str(e)}",
|
||||
)
|
||||
142
model/command/manager.py
Normal file
142
model/command/manager.py
Normal file
@@ -0,0 +1,142 @@
|
||||
import heapq
|
||||
import inspect
|
||||
import traceback
|
||||
from typing import Dict
|
||||
from type.types import Context
|
||||
from type.plugin import PluginMetadata
|
||||
from type.message_event import AstrMessageEvent
|
||||
from type.command import CommandResult
|
||||
from type.register import RegisteredPlugins
|
||||
from model.command.parser import CommandParser
|
||||
from model.plugin.command import PluginCommandBridge
|
||||
from SparkleLogging.utils.core import LogManager
|
||||
from logging import Logger
|
||||
from dataclasses import dataclass
|
||||
|
||||
logger: Logger = LogManager.GetLogger(log_name='astrbot')
|
||||
|
||||
@dataclass
|
||||
class CommandMetadata():
|
||||
inner_command: bool
|
||||
plugin_metadata: PluginMetadata
|
||||
handler: callable
|
||||
use_regex: bool = False
|
||||
ignore_prefix: bool = False
|
||||
description: str = ""
|
||||
|
||||
class CommandManager():
|
||||
def __init__(self):
|
||||
self.commands = []
|
||||
self.commands_handler: Dict[str, CommandMetadata] = {}
|
||||
self.command_parser = CommandParser()
|
||||
|
||||
def register(self,
|
||||
command: str,
|
||||
description: str,
|
||||
priority: int,
|
||||
handler: callable,
|
||||
use_regex: bool = False,
|
||||
ignore_prefix: bool = False,
|
||||
plugin_metadata: PluginMetadata = None,
|
||||
):
|
||||
'''
|
||||
优先级越高，越先被处理。
|
||||
|
||||
use_regex: 是否使用正则表达式匹配指令。
|
||||
'''
|
||||
if command in self.commands_handler:
|
||||
raise ValueError(f"Command {command} already exists.")
|
||||
if not handler:
|
||||
raise ValueError(f"Handler of {command} is None.")
|
||||
|
||||
heapq.heappush(self.commands, (-priority, command))
|
||||
self.commands_handler[command] = CommandMetadata(
|
||||
inner_command=plugin_metadata == None,
|
||||
plugin_metadata=plugin_metadata,
|
||||
handler=handler,
|
||||
use_regex=use_regex,
|
||||
ignore_prefix=ignore_prefix,
|
||||
description=description
|
||||
)
|
||||
if plugin_metadata:
|
||||
logger.info(f"已注册 {plugin_metadata.author}/{plugin_metadata.plugin_name} 的指令 {command}。")
|
||||
else:
|
||||
logger.info(f"已注册指令 {command}。")
|
||||
|
||||
def register_from_pcb(self, pcb: PluginCommandBridge):
|
||||
for request in pcb.plugin_commands_waitlist:
|
||||
plugin = None
|
||||
for registered_plugin in pcb.cached_plugins:
|
||||
if registered_plugin.metadata.plugin_name == request.plugin_name:
|
||||
plugin = registered_plugin
|
||||
break
|
||||
if not plugin:
|
||||
logger.warning(f"插件 {request.plugin_name} 未找到，无法注册指令 {request.command_name}。")
|
||||
else:
|
||||
self.register(command=request.command_name,
|
||||
description=request.description,
|
||||
priority=request.priority,
|
||||
handler=request.handler,
|
||||
use_regex=request.use_regex,
|
||||
ignore_prefix=request.ignore_prefix,
|
||||
plugin_metadata=plugin.metadata)
|
||||
self.plugin_commands_waitlist = []
|
||||
|
||||
async def check_command_ignore_prefix(self, message_str: str) -> bool:
|
||||
for _, command in self.commands:
|
||||
command_metadata = self.commands_handler[command]
|
||||
if command_metadata.ignore_prefix:
|
||||
trig = False
|
||||
if self.commands_handler[command].use_regex:
|
||||
trig = self.command_parser.regex_match(message_str, command)
|
||||
else:
|
||||
trig = message_str.startswith(command)
|
||||
if trig:
|
||||
return True
|
||||
return False
|
||||
|
||||
async def scan_command(self, message_event: AstrMessageEvent, context: Context) -> CommandResult:
|
||||
message_str = message_event.message_str
|
||||
for _, command in self.commands:
|
||||
trig = False
|
||||
if self.commands_handler[command].use_regex:
|
||||
trig = self.command_parser.regex_match(message_str, command)
|
||||
else:
|
||||
trig = message_str.startswith(command)
|
||||
if trig:
|
||||
logger.info(f"触发 {command} 指令。")
|
||||
command_result = await self.execute_handler(command, message_event, context)
|
||||
if not command_result:
|
||||
continue
|
||||
if command_result.hit:
|
||||
return command_result
|
||||
|
||||
async def execute_handler(self,
|
||||
command: str,
|
||||
message_event: AstrMessageEvent,
|
||||
context: Context) -> CommandResult:
|
||||
command_metadata = self.commands_handler[command]
|
||||
handler = command_metadata.handler
|
||||
# call handler
|
||||
try:
|
||||
if inspect.iscoroutinefunction(handler):
|
||||
command_result = await handler(message_event, context)
|
||||
else:
|
||||
command_result = handler(message_event, context)
|
||||
|
||||
if not isinstance(command_result, CommandResult):
|
||||
raise ValueError(f"Command {command} handler should return CommandResult.")
|
||||
|
||||
context.metrics_uploader.command_stats[command] += 1
|
||||
|
||||
return command_result
|
||||
except BaseException as e:
|
||||
logger.error(traceback.format_exc())
|
||||
|
||||
if not command_metadata.inner_command:
|
||||
text = f"执行 {command}/({command_metadata.plugin_metadata.plugin_name} By {command_metadata.plugin_metadata.author}) 指令时发生了异常。{e}"
|
||||
logger.error(text)
|
||||
else:
|
||||
text = f"执行 {command} 指令时发生了异常。{e}"
|
||||
logger.error(text)
|
||||
return CommandResult().message(text)
|
||||
@@ -1,289 +0,0 @@
|
||||
from model.command.command import Command
|
||||
from model.provider.openai_official import ProviderOpenAIOfficial
|
||||
from cores.qqbot.personality import personalities
|
||||
|
||||
from model.platform.qq import QQ
|
||||
from util import general_utils as gu
|
||||
from cores.qqbot.global_object import GlobalObject
|
||||
from .adapter.protocol_adapter import UnifiedBotCompatibleLayer
|
||||
|
||||
class CommandOpenAIOfficial(Command):
|
||||
def __init__(self, provider: ProviderOpenAIOfficial, global_object: GlobalObject):
|
||||
self.provider = provider
|
||||
self.cached_plugins = {}
|
||||
self.global_object = global_object
|
||||
self.personality_str = ""
|
||||
self.unified_bot_compatible_layer = UnifiedBotCompatibleLayer(self.global_object.platform_qqchan)
|
||||
super().__init__(provider, global_object, self.unified_bot_compatible_layer)
|
||||
|
||||
async def check_command(self,
|
||||
message: str,
|
||||
session_id: str,
|
||||
role: str,
|
||||
platform: str,
|
||||
message_obj):
|
||||
self.platform = platform
|
||||
hit, res = await super().check_command(
|
||||
message,
|
||||
session_id,
|
||||
role,
|
||||
platform,
|
||||
message_obj
|
||||
)
|
||||
|
||||
if hit:
|
||||
return True, res
|
||||
if self.command_start_with(message, "reset", "重置"):
|
||||
return True, self.reset(session_id, message)
|
||||
elif self.command_start_with(message, "his", "历史"):
|
||||
return True, self.his(message, session_id)
|
||||
elif self.command_start_with(message, "token"):
|
||||
return True, self.token(session_id)
|
||||
elif self.command_start_with(message, "gpt"):
|
||||
return True, self.gpt()
|
||||
elif self.command_start_with(message, "status"):
|
||||
return True, self.status()
|
||||
elif self.command_start_with(message, "count"):
|
||||
return True, self.count()
|
||||
elif self.command_start_with(message, "help", "帮助"):
|
||||
return True, self.help()
|
||||
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, role)
|
||||
elif self.command_start_with(message, "画", "draw"):
|
||||
return True, self.draw(message)
|
||||
elif self.command_start_with(message, "key"):
|
||||
return True, self.key(message)
|
||||
elif self.command_start_with(message, "switch"):
|
||||
return True, self.switch(message)
|
||||
|
||||
return False, None
|
||||
|
||||
def help(self):
|
||||
commands = super().general_commands()
|
||||
commands['画'] = '画画'
|
||||
commands['key'] = '添加OpenAI key'
|
||||
commands['set'] = '人格设置面板'
|
||||
commands['gpt'] = '查看gpt配置信息'
|
||||
commands['status'] = '查看key使用状态'
|
||||
commands['token'] = '查看本轮会话token'
|
||||
return True, super().help_messager(commands, self.platform, self.global_object.cached_plugins), "help"
|
||||
|
||||
|
||||
def reset(self, session_id: str, message: str = "reset"):
|
||||
if self.provider is None:
|
||||
return False, "未启动OpenAI ChatGPT语言模型.", "reset"
|
||||
l = message.split(" ")
|
||||
if len(l) == 1:
|
||||
self.provider.forget(session_id)
|
||||
return True, "重置成功", "reset"
|
||||
if len(l) == 2 and l[1] == "p":
|
||||
self.provider.forget(session_id)
|
||||
if self.personality_str != "":
|
||||
self.set(self.personality_str, session_id) # 重新设置人格
|
||||
return True, "重置成功", "reset"
|
||||
|
||||
def his(self, message: str, session_id: str):
|
||||
if self.provider is None:
|
||||
return False, "未启动OpenAI ChatGPT语言模型.", "his"
|
||||
#分页，每页5条
|
||||
msg = ''
|
||||
size_per_page = 3
|
||||
page = 1
|
||||
if message[4:]:
|
||||
page = int(message[4:])
|
||||
# 检查是否有过历史记录
|
||||
if session_id not in self.provider.session_dict:
|
||||
msg = f"历史记录为空"
|
||||
return True, msg, "his"
|
||||
l = self.provider.session_dict[session_id]
|
||||
max_page = len(l)//size_per_page + 1 if len(l)%size_per_page != 0 else len(l)//size_per_page
|
||||
p = self.provider.get_prompts_by_cache_list(self.provider.session_dict[session_id], divide=True, paging=True, size=size_per_page, page=page)
|
||||
return True, f"历史记录如下：\n{p}\n第{page}页 | 共{max_page}页\n*输入/his 2跳转到第2页", "his"
|
||||
|
||||
def token(self, session_id: str):
|
||||
if self.provider is None:
|
||||
return False, "未启动OpenAI ChatGPT语言模型.", "token"
|
||||
return True, f"会话的token数: {self.provider.get_user_usage_tokens(self.provider.session_dict[session_id])}\n系统最大缓存token数: {self.provider.max_tokens}", "token"
|
||||
|
||||
def gpt(self):
|
||||
if self.provider is None:
|
||||
return False, "未启动OpenAI ChatGPT语言模型.", "gpt"
|
||||
return True, f"OpenAI GPT配置:\n {self.provider.chatGPT_configs}", "gpt"
|
||||
|
||||
def status(self):
|
||||
if self.provider is None:
|
||||
return False, "未启动OpenAI ChatGPT语言模型.", "status"
|
||||
chatgpt_cfg_str = ""
|
||||
key_stat = self.provider.get_key_stat()
|
||||
index = 1
|
||||
max = 9000000
|
||||
gg_count = 0
|
||||
total = 0
|
||||
tag = ''
|
||||
for key in key_stat.keys():
|
||||
sponsor = ''
|
||||
total += key_stat[key]['used']
|
||||
if key_stat[key]['exceed']:
|
||||
gg_count += 1
|
||||
continue
|
||||
if 'sponsor' in key_stat[key]:
|
||||
sponsor = key_stat[key]['sponsor']
|
||||
chatgpt_cfg_str += f" |-{index}: {key[-8:]} {key_stat[key]['used']}/{max} {sponsor}{tag}\n"
|
||||
index += 1
|
||||
return True, f"⭐使用情况({str(gg_count)}个已用):\n{chatgpt_cfg_str}", "status"
|
||||
|
||||
def count(self):
|
||||
if self.provider is None:
|
||||
return False, "未启动OpenAI ChatGPT语言模型。", "reset"
|
||||
guild_count, guild_msg_count, guild_direct_msg_count, session_count = self.provider.get_stat()
|
||||
return True, f"【本指令部分统计可能已经过时】\n当前会话数: {len(self.provider.session_dict)}\n共有频道数: {guild_count} \n共有消息数: {guild_msg_count}\n私信数: {guild_direct_msg_count}\n历史会话数: {session_count}", "count"
|
||||
|
||||
def key(self, message: str):
|
||||
if self.provider is None:
|
||||
return False, "未启动OpenAI ChatGPT语言模型.", "reset"
|
||||
l = message.split(" ")
|
||||
if len(l) == 1:
|
||||
msg = "感谢您赞助key，key为官方API使用，请以以下格式赞助:\n/key xxxxx"
|
||||
return True, msg, "key"
|
||||
key = l[1]
|
||||
if self.provider.check_key(key):
|
||||
self.provider.append_key(key)
|
||||
return True, f"*★,°*:.☆(￣▽￣)/$:*.°★* 。\n该Key被验证为有效。感谢你的赞助~"
|
||||
else:
|
||||
return True, "该Key被验证为无效。也许是输入错误了，或者重试。", "key"
|
||||
|
||||
def switch(self, message: str):
|
||||
'''
|
||||
切换账号
|
||||
'''
|
||||
l = message.split(" ")
|
||||
if len(l) == 1:
|
||||
_, ret, _ = self.status()
|
||||
curr_ = self.provider.get_curr_key()
|
||||
if curr_ is None:
|
||||
ret += "当前您未选择账号。输入/switch <账号序号>切换账号。"
|
||||
else:
|
||||
ret += f"当前您选择的账号为：{curr_[-8:]}。输入/switch <账号序号>切换账号。"
|
||||
return True, ret, "switch"
|
||||
elif len(l) == 2:
|
||||
try:
|
||||
key_stat = self.provider.get_key_stat()
|
||||
index = int(l[1])
|
||||
if index > len(key_stat) or index < 1:
|
||||
return True, "账号序号不合法。", "switch"
|
||||
else:
|
||||
try:
|
||||
new_key = list(key_stat.keys())[index-1]
|
||||
ret = self.provider.check_key(new_key)
|
||||
self.provider.set_key(new_key)
|
||||
except BaseException as e:
|
||||
return True, "账号切换失败，原因: " + str(e), "switch"
|
||||
return True, f"账号切换成功。", "switch"
|
||||
except BaseException as e:
|
||||
return True, "未知错误: "+str(e), "switch"
|
||||
else:
|
||||
return True, "参数过多。", "switch"
|
||||
|
||||
def unset(self, session_id: str):
|
||||
if self.provider is None:
|
||||
return False, "未启动OpenAI ChatGPT语言模型.", "unset"
|
||||
self.provider.curr_personality = {}
|
||||
self.provider.forget(session_id)
|
||||
return True, "已清除人格并重置历史记录。", "unset"
|
||||
|
||||
def set(self, message: str, session_id: str):
|
||||
if self.provider is None:
|
||||
return False, "未启动OpenAI ChatGPT语言模型.", "set"
|
||||
l = message.split(" ")
|
||||
if len(l) == 1:
|
||||
return True, f"【人格文本由PlexPt开源项目awesome-chatgpt-pr \
|
||||
ompts-zh提供】\n设置人格: \n/set 人格名。例如/set 编剧\n人格列表: /set list\n人格详细信息: \
|
||||
/set view 人格名\n自定义人格: /set 人格文本\n重置会话(清除人格): /reset\n重置会话(保留人格): /reset p\n【当前人格】: {str(self.provider.curr_personality)}", "set"
|
||||
elif l[1] == "list":
|
||||
msg = "人格列表：\n"
|
||||
for key in personalities.keys():
|
||||
msg += f" |-{key}\n"
|
||||
msg += '\n\n*输入/set view 人格名查看人格详细信息'
|
||||
msg += '\n*不定时更新人格库，请及时更新本项目。'
|
||||
return True, msg, "set"
|
||||
elif l[1] == "view":
|
||||
if len(l) == 2:
|
||||
return True, "请输入/set view 人格名", "set"
|
||||
ps = l[2].strip()
|
||||
if ps in personalities:
|
||||
msg = f"人格{ps}的详细信息：\n"
|
||||
msg += f"{personalities[ps]}\n"
|
||||
else:
|
||||
msg = f"人格{ps}不存在"
|
||||
return True, msg, "set"
|
||||
else:
|
||||
ps = l[1].strip()
|
||||
if ps in personalities:
|
||||
self.provider.curr_personality = {
|
||||
'name': ps,
|
||||
'prompt': personalities[ps]
|
||||
}
|
||||
self.provider.session_dict[session_id] = []
|
||||
new_record = {
|
||||
"user": {
|
||||
"role": "user",
|
||||
"content": personalities[ps],
|
||||
},
|
||||
"AI": {
|
||||
"role": "assistant",
|
||||
"content": "好的，接下来我会扮演这个角色。"
|
||||
},
|
||||
'type': "personality",
|
||||
'usage_tokens': 0,
|
||||
'single-tokens': 0
|
||||
}
|
||||
self.provider.session_dict[session_id].append(new_record)
|
||||
self.personality_str = message
|
||||
return True, f"人格{ps}已设置。", "set"
|
||||
else:
|
||||
self.provider.curr_personality = {
|
||||
'name': '自定义人格',
|
||||
'prompt': ps
|
||||
}
|
||||
new_record = {
|
||||
"user": {
|
||||
"role": "user",
|
||||
"content": ps,
|
||||
},
|
||||
"AI": {
|
||||
"role": "assistant",
|
||||
"content": "好的，接下来我会扮演这个角色。"
|
||||
},
|
||||
'type': "personality",
|
||||
'usage_tokens': 0,
|
||||
'single-tokens': 0
|
||||
}
|
||||
self.provider.session_dict[session_id] = []
|
||||
self.provider.session_dict[session_id].append(new_record)
|
||||
self.personality_str = message
|
||||
return True, f"自定义人格已设置。 \n人格信息: {ps}", "set"
|
||||
|
||||
def draw(self, message):
|
||||
if self.provider is None:
|
||||
return False, "未启动OpenAI ChatGPT语言模型.", "draw"
|
||||
if message.startswith("/画"):
|
||||
message = message[2:]
|
||||
elif message.startswith("画"):
|
||||
message = message[1:]
|
||||
try:
|
||||
# 画图模式传回3个参数
|
||||
img_url = self.provider.image_chat(message)
|
||||
return True, img_url, "draw"
|
||||
except Exception as e:
|
||||
if 'exceeded' in str(e):
|
||||
return f"OpenAI API错误。原因：\n{str(e)} \n超额了。可自己搭建一个机器人(Github仓库：QQChannelChatGPT)"
|
||||
return False, f"图片生成失败: {e}", "draw"
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
186
model/command/openai_official_handler.py
Normal file
186
model/command/openai_official_handler.py
Normal file
@@ -0,0 +1,186 @@
|
||||
from model.command.manager import CommandManager
|
||||
from type.message_event import AstrMessageEvent
|
||||
from type.command import CommandResult
|
||||
from type.types import Context
|
||||
from SparkleLogging.utils.core import LogManager
|
||||
from logging import Logger
|
||||
from nakuru.entities.components import Image
|
||||
from model.provider.openai_official import ProviderOpenAIOfficial, MODELS
|
||||
from util.personality import personalities
|
||||
from util.io import download_image_by_url
|
||||
|
||||
logger: Logger = LogManager.GetLogger(log_name='astrbot')
|
||||
|
||||
|
||||
class OpenAIOfficialCommandHandler():
|
||||
def __init__(self, manager: CommandManager) -> None:
|
||||
self.manager = manager
|
||||
|
||||
self.provider = None
|
||||
|
||||
self.manager.register("reset", "重置会话", 10, self.reset)
|
||||
self.manager.register("his", "查看历史记录", 10, self.his)
|
||||
self.manager.register("status", "查看当前状态", 10, self.status)
|
||||
self.manager.register("switch", "切换账号", 10, self.switch)
|
||||
self.manager.register("unset", "清除个性化人格设置", 10, self.unset)
|
||||
self.manager.register("set", "设置个性化人格", 10, self.set)
|
||||
self.manager.register("draw", "调用 DallE 模型画图", 10, self.draw)
|
||||
self.manager.register("model", "切换模型", 10, self.model)
|
||||
self.manager.register("画", "调用 DallE 模型画图", 10, self.draw)
|
||||
|
||||
def set_provider(self, provider):
|
||||
self.provider = provider
|
||||
|
||||
async def reset(self, message: AstrMessageEvent, context: Context):
|
||||
tokens = self.manager.command_parser.parse(message.message_str)
|
||||
if tokens.len == 1:
|
||||
await self.provider.forget(message.session_id, keep_system_prompt=True)
|
||||
return CommandResult().message("重置成功")
|
||||
elif tokens.get(1) == 'p':
|
||||
await self.provider.forget(message.session_id)
|
||||
|
||||
async def model(self, message: AstrMessageEvent, context: Context):
|
||||
tokens = self.manager.command_parser.parse(message.message_str)
|
||||
if tokens.len == 1:
|
||||
ret = await self._print_models()
|
||||
return CommandResult().message(ret)
|
||||
model = tokens.get(1)
|
||||
if model.isdigit():
|
||||
try:
|
||||
models = await self.provider.get_models()
|
||||
except BaseException as e:
|
||||
logger.error(f"获取模型列表失败: {str(e)}")
|
||||
return CommandResult().message("获取模型列表失败，无法使用编号切换模型。可以尝试直接输入模型名来切换，如 gpt-4o。")
|
||||
models = list(models)
|
||||
if int(model) <= len(models) and int(model) >= 1:
|
||||
model = models[int(model)-1]
|
||||
self.provider.set_model(model.id)
|
||||
return CommandResult().message(f"模型已设置为 {model.id}")
|
||||
else:
|
||||
self.provider.set_model(model)
|
||||
return CommandResult().message(f"模型已设置为 {model} (自定义)")
|
||||
|
||||
async def _print_models(self):
|
||||
try:
|
||||
models = await self.provider.get_models()
|
||||
except BaseException as e:
|
||||
return "获取模型列表失败: " + str(e)
|
||||
i = 1
|
||||
ret = "OpenAI GPT 类可用模型"
|
||||
for model in models:
|
||||
ret += f"\n{i}. {model.id}"
|
||||
i += 1
|
||||
ret += "\nTips: 使用 /model 模型名/编号，即可实时更换模型。如目标模型不存在于上表，请输入模型名。"
|
||||
logger.debug(ret)
|
||||
return ret
|
||||
|
||||
def his(self, message: AstrMessageEvent, context: Context):
|
||||
tokens = self.manager.command_parser.parse(message.message_str)
|
||||
size_per_page = 3
|
||||
page = 1
|
||||
if tokens.len == 2:
|
||||
try:
|
||||
page = int(tokens.get(1))
|
||||
except BaseException as e:
|
||||
return CommandResult().message("页码格式错误")
|
||||
contexts, total_num = self.provider.dump_contexts_page(message.session_id, size_per_page, page=page)
|
||||
t_pages = total_num // size_per_page + 1
|
||||
return CommandResult().message(f"历史记录如下：\n{contexts}\n第 {page} 页 | 共 {t_pages} 页\n*输入 /his 2 跳转到第 2 页")
|
||||
|
||||
def status(self, message: AstrMessageEvent, context: Context):
|
||||
keys_data = self.provider.get_keys_data()
|
||||
ret = "OpenAI Key"
|
||||
for k in keys_data:
|
||||
status = "🟢" if keys_data[k] else "🔴"
|
||||
ret += "\n|- " + k[:8] + " " + status
|
||||
|
||||
conf = self.provider.get_configs()
|
||||
ret += "\n当前模型: " + conf['model']
|
||||
if conf['model'] in MODELS:
|
||||
ret += "\n最大上下文窗口: " + str(MODELS[conf['model']]) + " tokens"
|
||||
|
||||
if message.session_id in self.provider.session_memory and len(self.provider.session_memory[message.session_id]):
|
||||
ret += "\n你的会话上下文: " + str(self.provider.session_memory[message.session_id][-1]['usage_tokens']) + " tokens"
|
||||
|
||||
return CommandResult().message(ret)
|
||||
|
||||
async def switch(self, message: AstrMessageEvent, context: Context):
|
||||
'''
|
||||
切换账号
|
||||
'''
|
||||
tokens = self.manager.command_parser.parse(message.message_str)
|
||||
if tokens.len == 1:
|
||||
_, ret, _ = self.status()
|
||||
curr_ = self.provider.get_curr_key()
|
||||
if curr_ is None:
|
||||
ret += "当前您未选择账号。输入/switch <账号序号>切换账号。"
|
||||
else:
|
||||
ret += f"当前您选择的账号为：{curr_[-8:]}。输入/switch <账号序号>切换账号。"
|
||||
return CommandResult().message(ret)
|
||||
elif tokens.len == 2:
|
||||
try:
|
||||
key_stat = self.provider.get_keys_data()
|
||||
index = int(tokens.get(1))
|
||||
if index > len(key_stat) or index < 1:
|
||||
return CommandResult().message("账号序号错误。")
|
||||
else:
|
||||
try:
|
||||
new_key = list(key_stat.keys())[index-1]
|
||||
self.provider.set_key(new_key)
|
||||
except BaseException as e:
|
||||
return CommandResult().message("切换账号未知错误: "+str(e))
|
||||
return CommandResult().message("切换账号成功。")
|
||||
except BaseException as e:
|
||||
return CommandResult().message("切换账号错误。")
|
||||
else:
|
||||
return CommandResult().message("参数过多。")
|
||||
|
||||
def unset(self, message: AstrMessageEvent, context: Context):
|
||||
self.provider.curr_personality = {}
|
||||
self.provider.forget(message.session_id)
|
||||
return CommandResult().message("已清除个性化设置。")
|
||||
|
||||
|
||||
def set(self, message: AstrMessageEvent, context: Context):
|
||||
l = message.message_str.split(" ")
|
||||
if len(l) == 1:
|
||||
return CommandResult().message("- 设置人格: \nset 人格名。例如 set 编剧\n- 人格列表: set list\n- 人格详细信息: set view 人格名\n- 自定义人格: set 人格文本\n- 重置会话(清除人格): reset\n- 重置会话(保留人格): reset p\n\n【当前人格】: " + str(self.provider.curr_personality['prompt']))
|
||||
elif l[1] == "list":
|
||||
msg = "人格列表：\n"
|
||||
for key in personalities.keys():
|
||||
msg += f"- {key}\n"
|
||||
msg += '\n\n*输入 set view 人格名 查看人格详细信息'
|
||||
return CommandResult().message(msg)
|
||||
elif l[1] == "view":
|
||||
if len(l) == 2:
|
||||
return CommandResult().message("请输入人格名")
|
||||
ps = l[2].strip()
|
||||
if ps in personalities:
|
||||
msg = f"人格{ps}的详细信息：\n"
|
||||
msg += f"{personalities[ps]}\n"
|
||||
else:
|
||||
msg = f"人格{ps}不存在"
|
||||
return CommandResult().message(msg)
|
||||
else:
|
||||
ps = "".join(l[1:]).strip()
|
||||
if ps in personalities:
|
||||
self.provider.curr_personality = {
|
||||
'name': ps,
|
||||
'prompt': personalities[ps]
|
||||
}
|
||||
self.provider.personality_set(self.provider.curr_personality, message.session_id)
|
||||
return CommandResult().message(f"人格已设置。 \n人格信息: {ps}")
|
||||
else:
|
||||
self.provider.curr_personality = {
|
||||
'name': '自定义人格',
|
||||
'prompt': ps
|
||||
}
|
||||
self.provider.personality_set(self.provider.curr_personality, message.session_id)
|
||||
return CommandResult().message(f"人格已设置。 \n人格信息: {ps}")
|
||||
|
||||
async def draw(self, message: AstrMessageEvent, context: Context):
|
||||
message = message.message_str.removeprefix("画")
|
||||
img_url = await self.provider.image_generate(message)
|
||||
return CommandResult(
|
||||
message_chain=[Image.fromURL(img_url)],
|
||||
)
|
||||
25
model/command/parser.py
Normal file
25
model/command/parser.py
Normal file
@@ -0,0 +1,25 @@
|
||||
import re
|
||||
|
||||
class CommandTokens():
|
||||
def __init__(self) -> None:
|
||||
self.tokens = []
|
||||
self.len = 0
|
||||
|
||||
def get(self, idx: int):
|
||||
if idx >= self.len:
|
||||
return None
|
||||
return self.tokens[idx].strip()
|
||||
|
||||
class CommandParser():
|
||||
def __init__(self):
|
||||
pass
|
||||
|
||||
def parse(self, message: str):
|
||||
cmd_tokens = CommandTokens()
|
||||
cmd_tokens.tokens = message.split(" ")
|
||||
cmd_tokens.len = len(cmd_tokens.tokens)
|
||||
return cmd_tokens
|
||||
|
||||
def regex_match(self, message: str, command: str) -> bool:
|
||||
return re.search(command, message, re.MULTILINE) is not None
|
||||
|
||||
@@ -1,134 +0,0 @@
|
||||
from model.command.command import Command
|
||||
from model.provider.rev_chatgpt import ProviderRevChatGPT
|
||||
from model.platform.qq import QQ
|
||||
from cores.qqbot.personality import personalities
|
||||
from cores.qqbot.global_object import GlobalObject
|
||||
|
||||
class CommandRevChatGPT(Command):
|
||||
def __init__(self, provider: ProviderRevChatGPT, global_object: GlobalObject):
|
||||
self.provider = provider
|
||||
self.cached_plugins = {}
|
||||
self.global_object = global_object
|
||||
self.personality_str = ""
|
||||
super().__init__(provider, global_object)
|
||||
|
||||
async def check_command(self,
|
||||
message: str,
|
||||
session_id: str,
|
||||
role: str,
|
||||
platform: str,
|
||||
message_obj):
|
||||
self.platform = platform
|
||||
hit, res = await super().check_command(
|
||||
message,
|
||||
session_id,
|
||||
role,
|
||||
platform,
|
||||
message_obj
|
||||
)
|
||||
|
||||
if hit:
|
||||
return True, res
|
||||
if self.command_start_with(message, "help", "帮助"):
|
||||
return True, self.help()
|
||||
elif self.command_start_with(message, "reset"):
|
||||
return True, self.reset(session_id, message)
|
||||
elif self.command_start_with(message, "update"):
|
||||
return True, self.update(message, role)
|
||||
elif self.command_start_with(message, "set"):
|
||||
return True, self.set(message, session_id)
|
||||
elif self.command_start_with(message, "switch"):
|
||||
return True, self.switch(message, session_id)
|
||||
return False, None
|
||||
|
||||
def reset(self, session_id, message: str):
|
||||
l = message.split(" ")
|
||||
if len(l) == 1:
|
||||
self.provider.forget(session_id)
|
||||
return True, "重置完毕。", "reset"
|
||||
if len(l) == 2 and l[1] == "p":
|
||||
self.provider.forget(session_id)
|
||||
ret = self.provider.text_chat(self.personality_str)
|
||||
return True, f"重置完毕（保留人格）。\n\n{ret}", "reset"
|
||||
|
||||
def set(self, message: str, session_id: str):
|
||||
l = message.split(" ")
|
||||
if len(l) == 1:
|
||||
return True, f"设置人格: \n/set 人格名或人格文本。例如/set 编剧\n人格列表: /set list\n人格详细信息: \
|
||||
/set view 人格名\n重置会话(清除人格): /reset\n重置会话(保留人格): /reset p", "set"
|
||||
elif l[1] == "list":
|
||||
msg = "人格列表：\n"
|
||||
for key in personalities.keys():
|
||||
msg += f" |-{key}\n"
|
||||
msg += '\n\n*输入/set view 人格名查看人格详细信息'
|
||||
msg += '\n*不定时更新人格库，请及时更新本项目。'
|
||||
return True, msg, "set"
|
||||
elif l[1] == "view":
|
||||
if len(l) == 2:
|
||||
return True, "请输入/set view 人格名", "set"
|
||||
ps = l[2].strip()
|
||||
if ps in personalities:
|
||||
msg = f"人格【{ps}】详细信息：\n"
|
||||
msg += f"{personalities[ps]}\n"
|
||||
else:
|
||||
msg = f"人格【{ps}】不存在。"
|
||||
return True, msg, "set"
|
||||
else:
|
||||
ps = l[1].strip()
|
||||
if ps in personalities:
|
||||
self.reset(session_id, "reset")
|
||||
self.personality_str = personalities[ps]
|
||||
ret = self.provider.text_chat(self.personality_str, session_id)
|
||||
return True, f"人格【{ps}】已设置。\n\n{ret}", "set"
|
||||
else:
|
||||
self.reset(session_id, "reset")
|
||||
self.personality_str = ps
|
||||
ret = self.provider.text_chat(ps, session_id)
|
||||
return True, f"人格信息已设置。\n\n{ret}", "set"
|
||||
|
||||
def switch(self, message: str, session_id: str):
|
||||
'''
|
||||
切换账号
|
||||
'''
|
||||
l = message.split(" ")
|
||||
rev_chatgpt = self.provider.get_revchatgpt()
|
||||
if len(l) == 1:
|
||||
ret = "当前账号：\n"
|
||||
index = 0
|
||||
curr_ = None
|
||||
for revstat in rev_chatgpt:
|
||||
index += 1
|
||||
ret += f"[{index}]. {revstat['id']}\n"
|
||||
# if session_id in revstat['user']:
|
||||
# curr_ = revstat['id']
|
||||
for user in revstat['user']:
|
||||
if session_id == user['id']:
|
||||
curr_ = revstat['id']
|
||||
break
|
||||
if curr_ is None:
|
||||
ret += "当前您未选择账号。输入/switch <账号序号>切换账号。"
|
||||
else:
|
||||
ret += f"当前您选择的账号为：{curr_}。输入/switch <账号序号>切换账号。"
|
||||
return True, ret, "switch"
|
||||
elif len(l) == 2:
|
||||
try:
|
||||
index = int(l[1])
|
||||
if index > len(self.provider.rev_chatgpt) or index < 1:
|
||||
return True, "账号序号不合法。", "switch"
|
||||
else:
|
||||
# pop
|
||||
for revstat in self.provider.rev_chatgpt:
|
||||
if session_id in revstat['user']:
|
||||
revstat['user'].remove(session_id)
|
||||
# append
|
||||
self.provider.rev_chatgpt[index - 1]['user'].append(session_id)
|
||||
return True, f"切换账号成功。当前账号为：{self.provider.rev_chatgpt[index - 1]['id']}", "switch"
|
||||
except BaseException:
|
||||
return True, "账号序号不合法。", "switch"
|
||||
else:
|
||||
return True, "参数过多。", "switch"
|
||||
|
||||
def help(self):
|
||||
commands = super().general_commands()
|
||||
commands['set'] = '设置人格'
|
||||
return True, super().help_messager(commands, self.platform, self.global_object.cached_plugins), "help"
|
||||
@@ -1,52 +0,0 @@
|
||||
from model.command.command import Command
|
||||
from model.provider.rev_edgegpt import ProviderRevEdgeGPT
|
||||
import asyncio
|
||||
from model.platform.qq import QQ
|
||||
from cores.qqbot.global_object import GlobalObject
|
||||
|
||||
class CommandRevEdgeGPT(Command):
|
||||
def __init__(self, provider: ProviderRevEdgeGPT, global_object: GlobalObject):
|
||||
self.provider = provider
|
||||
self.cached_plugins = {}
|
||||
self.global_object = global_object
|
||||
super().__init__(provider, global_object)
|
||||
|
||||
def check_command(self,
|
||||
message: str,
|
||||
session_id: str,
|
||||
role: str,
|
||||
platform: str,
|
||||
message_obj):
|
||||
self.platform = platform
|
||||
|
||||
hit, res = super().check_command(
|
||||
message,
|
||||
session_id,
|
||||
role,
|
||||
platform,
|
||||
message_obj
|
||||
)
|
||||
|
||||
if hit:
|
||||
return True, res
|
||||
if self.command_start_with(message, "reset"):
|
||||
return True, self.reset()
|
||||
elif self.command_start_with(message, "help"):
|
||||
return True, self.help()
|
||||
elif self.command_start_with(message, "update"):
|
||||
return True, self.update(message, role)
|
||||
|
||||
return False, None
|
||||
|
||||
def reset(self, loop = None):
|
||||
if self.provider is None:
|
||||
return False, "未启动Bing语言模型.", "reset"
|
||||
res = asyncio.run_coroutine_threadsafe(self.provider.forget(), loop).result()
|
||||
print(res)
|
||||
if res:
|
||||
return res, "重置成功", "reset"
|
||||
else:
|
||||
return res, "重置失败", "reset"
|
||||
|
||||
def help(self):
|
||||
return True, super().help_messager(super().general_commands(), self.platform, self.global_object.cached_plugins), "help"
|
||||
Reference in New Issue
Block a user