fz-mirrors/AstrBot
1
0
Fork 0
mirror of https://github.com/AstrBotDevs/AstrBot synced 2026-07-15 17:30:13 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: GOCQ适配QQ频道

Browse Source
This commit is contained in:
Soulter
2023-05-20 15:30:07 +08:00
parent 6894900e46
commit 4b158a1c89
3 changed files with 107 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

71
cores/qqbot/core.py
View File

@@ -18,7 +18,8 @@ from nakuru import (
CQHTTP,
GroupMessage,
GroupMemberIncrease,
FriendMessage
FriendMessage,
GuildMessage
)
from nakuru.entities.components import Plain,At
from model.command.command import Command
@@ -313,17 +314,32 @@ def initBot(cfg, prov):
if os.path.exists("cmd_config.json"):
with open("cmd_config.json", 'r', encoding='utf-8') as f:
cmd_config = json.load(f)
global admin_qq
global admin_qq, admin_qqchan
if "admin_qq" in cmd_config:
admin_qq = cmd_config['admin_qq']
gu.log("管理者QQ号: " + admin_qq, gu.LEVEL_INFO)
else:
admin_qq = input("[System] 请输入管理者QQ号(管理者QQ号才能使用update/plugin等指令): ")
gu.log("未设置管理者QQ号(管理者才能使用update/plugin等指令)", gu.LEVEL_WARNING)
admin_qq = input("请输入管理者QQ号(必须设置): ")
gu.log("管理者QQ号设置为: " + admin_qq, gu.LEVEL_INFO, fg=gu.FG_COLORS['yellow'])
cmd_config['admin_qq'] = admin_qq
with open("cmd_config.json", 'w', encoding='utf-8') as f:
json.dump(cmd_config, f, indent=4)
f.flush()
if "admin_qqchan" in cmd_config:
admin_qqchan = cmd_config['admin_qqchan']
gu.log("管理者频道用户号: " + admin_qqchan, gu.LEVEL_INFO)
else:
gu.log("未设置管理者QQ频道用户号(管理者才能使用update/plugin等指令)", gu.LEVEL_WARNING)
admin_qqchan = input("请输入管理者频道用户号(不是QQ号, 可以先回车跳过然后在频道发送指令!myid获取): ")
if admin_qqchan == "":
gu.log("跳过设置管理者频道用户号", gu.LEVEL_INFO, fg=gu.FG_COLORS['yellow'])
else:
gu.log("管理者频道用户号设置为: " + admin_qqchan, gu.LEVEL_INFO, fg=gu.FG_COLORS['yellow'])
cmd_config['admin_qqchan'] = admin_qqchan
with open("cmd_config.json", 'w', encoding='utf-8') as f:
json.dump(cmd_config, f, indent=4)
f.flush()
global gocq_app, gocq_loop
gocq_loop = asyncio.new_event_loop()
gocq_bot = QQ(True, gocq_loop)
@@ -441,7 +457,7 @@ def oper_msg(message,
role = "member" # 角色
hit = False # 是否命中指令
command_result = () # 调用指令返回的结果
global admin_qq, cached_plugins, gocq_bot
global admin_qq, admin_qqchan, cached_plugins, gocq_bot
if platform == PLATFORM_QQCHAN:
gu.log(f"接收到消息：{message.content}", gu.LEVEL_INFO, tag="QQ频道")
@@ -450,9 +466,9 @@ def oper_msg(message,
global qqchan_loop
if platform == PLATFORM_GOCQ:
if isinstance(message.message[0], Plain):
gu.log(f"接收到消息：{message.message[0].text}", gu.LEVEL_INFO, tag="QQ")
gu.log(f"接收到消息：{message.message[0].text}", gu.LEVEL_INFO, tag="GOCQ")
elif isinstance(message.message[0], At):
gu.log(f"接收到消息：{message.message[1].text}", gu.LEVEL_INFO, tag="QQ")
gu.log(f"接收到消息：{message.message[1].text}", gu.LEVEL_INFO, tag="GOCQ")
user_id = message.user_id
user_name = message.user_id
@@ -498,20 +514,24 @@ def oper_msg(message,
qq_msg = str(message.message[1].text).strip()
else:
return
session_id = message.group_id
# 适配GO-CQHTTP的频道功能
if message.type == "GuildMessage":
session_id = message.channel_id
else:
session_id = message.group_id
else:
qq_msg = message.message[0].text
session_id = message.user_id
role = "member"
if str(message.sender.user_id) == admin_qq:
gu.log("检测到管理员身份", gu.LEVEL_INFO, tag="QQ")
if str(message.sender.user_id) == admin_qq or str(message.sender.tiny_id) == admin_qqchan:
gu.log("检测到管理员身份", gu.LEVEL_INFO, tag="GOCQ")
role = "admin"
if qq_msg == "":
send_message(platform, message, f"Hi~", msg_ref=msg_ref, gocq_loop=gocq_loop, qqchannel_bot=qqchannel_bot, gocq_bot=gocq_bot)
return
logf.write("[QQBOT] "+ qq_msg+'\n')
logf.write("[GOCQBOT] "+ qq_msg+'\n')
logf.flush()
# 关键词回复
@@ -636,7 +656,7 @@ def oper_msg(message,
with open("keyword.json", "r", encoding="utf-8") as f:
keywords = json.load(f)
# QQ昵称
# 昵称
if command == "nick":
with open("cmd_config.json", "r", encoding="utf-8") as f:
global nick_qq
@@ -733,7 +753,7 @@ class gocqClient():
global nick_qq
# 将nick_qq转换为元组
if nick_qq == None:
nick_qq = ("ai",)
nick_qq = ("ai","!","！")
if isinstance(nick_qq, str):
nick_qq = (nick_qq,)
if isinstance(nick_qq, list):
@@ -765,4 +785,29 @@ class gocqClient():
global nick_qq
await app.sendGroupMessage(source.group_id, [
Plain(text=f"欢迎加入本群！\n欢迎给https://github.com/Soulter/QQChannelChatGPT项目一个Star😊~\n@我输入help查看帮助~\n我叫{nick_qq}, 你也可以以【{nick_qq}+问题】的格式来提醒我并问我问题哦~\n")
])
])
@gocq_app.receiver("GuildMessage")
async def _(app: CQHTTP, source: GuildMessage):
# gu.log(str(source), gu.LEVEL_INFO, max_len=9999)
global nick_qq
if nick_qq == None:
nick_qq = ("ai","!","！")
if isinstance(nick_qq, str):
nick_qq = (nick_qq,)
if isinstance(nick_qq, list):
nick_qq = tuple(nick_qq)
if isinstance(source.message[0], Plain):
if source.message[0].text.startswith(nick_qq):
_len = 0
for i in nick_qq:
if source.message[0].text.startswith(i):
_len = len(i)
source.message[0].text = source.message[0].text[_len:].strip()
new_sub_thread(oper_msg, (source, True, None, PLATFORM_GOCQ))
if isinstance(source.message[0], At):
if source.message[0].tiny_id == source.self_tiny_id:
new_sub_thread(oper_msg, (source, True, None, PLATFORM_GOCQ))
else:
return

11
model/command/command.py
View File

@@ -51,8 +51,19 @@ class Command:
if self.command_start_with(message, "plugin"):
return True, self.plugin_oper(message, role, cached_plugins)
if self.command_start_with(message, "myid"):
return True, self.get_my_id(message_obj, platform)
return False, None
def get_my_id(self, message_obj, platform):
if platform == "gocq":
if message_obj.type == "GuildMessage":
return True, f"你的频道id是{str(message_obj.sender.tiny_id)}", "plugin"
else:
return True, f"你的QQ是{str(message_obj.sender.user_id)}", "plugin"
def plugin_reload(self, cached_plugins: dict, target: str = None, all: bool = False):
plugins = self.get_plugin_modules()
fail_rec = ""

44
model/platform/qq.py
View File

@@ -1,13 +1,17 @@
from nakuru.entities.components import Plain, At, Image
from util import general_utils as gu
import asyncio
from nakuru import (
CQHTTP,
GuildMessage
)
class QQ:
def __init__(self, is_start: bool, gocq_loop = None) -> None:
self.is_start = is_start
self.gocq_loop = gocq_loop
def run_bot(self, gocq):
self.client = gocq
self.client: CQHTTP = gocq
self.client.run()
def get_msg_loop(self):
@@ -18,12 +22,12 @@ class QQ:
res,
image_mode: bool = False):
if not self.is_start:
raise Exception("管理员未启动QQ平台")
raise Exception("管理员未启动GOCQ平台")
"""
res可以是一个数组, 也就是gocq的消息链。
插件开发者请使用send方法, 可以不用直接调用这个方法。
"""
gu.log("回复QQ消息: "+str(res), level=gu.LEVEL_INFO, tag="QQ", max_len=30)
gu.log("回复GOCQ消息: "+str(res), level=gu.LEVEL_INFO, tag="GOCQ", max_len=40)
if isinstance(source, int):
source = {
@@ -31,9 +35,14 @@ class QQ:
"group_id": source
}
# 回复消息链
if isinstance(res, list) and len(res) > 0:
await self.client.sendGroupMessage(source.group_id, res)
return
if source.type == "GuildMessage":
await self.client.sendGuildChannelMessage(source.guild_id, source.channel_id, res)
return
else:
await self.client.sendGroupMessage(source.group_id, res)
return
# 通过消息链处理
if not image_mode:
@@ -46,6 +55,10 @@ class QQ:
await self.client.sendFriendMessage(source.user_id, [
Plain(text=res)
])
elif source.type == "GuildMessage":
await self.client.sendGuildChannelMessage(source.guild_id, source.channel_id, [
Plain(text=res)
])
else:
if source.type == "GroupMessage":
await self.client.sendGroupMessage(source.group_id, [
@@ -58,15 +71,34 @@ class QQ:
Plain(text="好的，我根据你的需要为你生成了一张图片😊"),
Image.fromURL(url=res)
])
elif source.type == "GuildMessage":
await self.client.sendGuildChannelMessage(source.guild_id, source.channel_id, [
Plain(text="好的，我根据你的需要为你生成了一张图片😊"),
Image.fromURL(url=res)
])
def send(self,
to,
res):
res,
):
'''
提供给插件的发送QQ消息接口, 不用在外部await。
参数说明：第一个参数可以是消息对象，也可以是QQ群号。第二个参数是消息内容（消息内容可以是消息链列表，也可以是纯文字信息）。
'''
try:
asyncio.run_coroutine_threadsafe(self.send_qq_msg(to, res), self.gocq_loop).result()
except BaseException as e:
raise e
def send_guild(self,
message_obj,
res,
):
'''
提供给插件的发送GOCQ QQ频道消息接口, 不用在外部await。
参数说明：第一个参数必须是消息对象, 第二个参数是消息内容（消息内容可以是消息链列表，也可以是纯文字信息）。
'''
try:
asyncio.run_coroutine_threadsafe(self.send_qq_msg(message_obj, res), self.gocq_loop).result()
except BaseException as e:
raise e