- 1. 插件用途
- 2. 目录结构
- 3. 实现难点与解决方案
- 3.1 随机触发
- 3.2 返回随机信息
- 3.3 禁言用户
- 3.4 json转换
- 3.5 获取用户昵称
- 3.6 异步休眠
- 4. 代码实现
- 5. 插件配图
- 6. 实际效果
- 7. 下一个插件
该插件涉及知识点：随机触发，返回随机信息，禁言用户，json转换，获取用户昵称，异步休眠
插件合集：nonebot2聊天机器人插件
该系列为用于QQ群聊天机器人的nonebot2相关插件，不保证完全符合标准规范写法，如有差错和改进余地，欢迎大佬指点修正。
前端：nonebot2
后端：go-cqhttp
插件所用语言：python3
前置环境安装过程建议参考零基础2分钟教你搭建QQ机器人——基于nonebot2，但是请注意该教程中的后端版本过旧导致私聊发图异常，需要手动更新go-cqhttp版本。
在不使用任意指定命令的情况下，随机对群聊中涉及到的关键词或者针对bot的戳一戳信息进行响应，可以任意自定义触发的逻辑，冷却时间与触发概率，同时，能够在一组配置好的应答中随机选择一个返回。
程序能够配备一个超级用户组，超级用户的发言无视触发概率和冷却时间，必定触发响应的反应，而且对超级用户的响应不会更新冷却时间。
响应可以返回多条信息，并且自定义返回间隔。
在plugins文件夹中新建一个文件夹chat，文件夹内目录结构如下：
|-chat
|-img
|-所有在信息发送中用到的图片
|-__init__.py
|-chat.py
|-config.py
|-response_for_surper_user.py
|-response_for_all_time.py
|-response_for_common_user.py
其中img为用于存储发送的图片文件的文件夹，chat.py为程序主要代码的位置，config.py用于存储配置项，__init__.py为程序启动位置。
程序首先执行response_for_surper_user.py中针对超级用户的响应，如果没有发现任何可以用于返回的响应，且发言者不是超级用户，再执行response_for_all_time.py中无视cd和触发时间的响应主要是用于禁言那些发送违规信息的用户。
如果这两步都没有返回结果，那么再执行最后的常规响应流程response_for_common_user.py，依次检索代码中所有写好的响应逻辑。
用于指定触发概率，利用random库生成一个0-99之间的随机数，然后判断大小即可。
from random import randint
# 以指定概率p返回True或者False
# 用于随机决定是否要回应
# 默认值为配置文件中的默认聊天响应概率
def random_response(p = Config.p_chat_response):
return randint(0, 99) < p
此处使用了random库从列表中随机抽取的方法，返回随机结果。
from random import choice
responses = [send_img('震惊1.jpg'),
send_img('震惊2.png'),
send_img('我不懂但我大受震撼.png')]
return choice(responses)
nonebot2的禁言没有直接的函数，需要调用后端的api来实现，因此需要将事件响应器中的bot参数传递到需要使用的地方。
在权限不足以禁言对方时，会抛出异常，需要处理一下，否则后续执行会中断。
解除禁言的方式为将该用户禁言0秒。
# 尝试禁言60秒
try:
await bot.set_group_ban(group_id=group_id, user_id=user_id, duration=60)
# 如果对方是管理员，那就假装无事发生
except:
pass
响应戳一戳信息需要先理解信息格式，通过在后台可以发现，bot读取到的戳一戳信息如下：
{'time': 时间戳, 'self_id': botQQ号, 'post_type': 'notice', 'notice_type': 'notify', 'sub_type': 'poke', 'user_id': 戳bot的人QQ号, 'group_id': 群号, 'target_id': botQQ号, 'sender_id': 戳bot的人QQ号}
尝试使用event.get_message()发现出现报错，notice信息不含msg。
于是使用event.get_event_description()获取完整信息，并且考虑将该字符串直接以类似读取json的方式转换为字典，从而快速获取相应的信息。
注：此处有两点非常重要！
1、json库读取的key字符串需要用双引号标注，而得到的信息中是单引号，因此需要先将所有的单引号替换为双引号再进行转换！
2、获取的QQ号是没有引号的，因此它们在转换后会被作为int变量放入字典，当你使用配置文件中字符串类型的QQ号与之进行匹配时，一定要记得做类型转换！
最终，读取相关信息的代码如下：
import json
description = event.get_event_description()
values = json.loads(description.replace("'", '"'))
此时values是一个包含所有属性名与对应值的字典。
获取指定用户的昵称同样需要调用后端的api来实现，并且需要用到json转换来读取返回的信息。
具体实现过程如下：
import json
infos = str(await bot.get_stranger_info(user_id=values['user_id']))
nickname = json.loads(infos.replace("'", '"'))['nickname'] + '(' + str(values['user_id']) + ')'
如果用户QQ号是123456，昵称是你好，那么nickname的最终返回结果就是你好(123456)。
在异步函数执行中使用time.sleep()会使得执行卡在这里什么都不做，为了高效起见，对于异步中的休眠使用asyncio.sleep()，不过由于该函数是异步的，因此需要使用await前缀。
import asyncio
# 此行必须写在异步函数内部
await asyncio.sleep(1)
此处仅给出一些触发事件的写法样例，具体内容可以自行添加。
毕竟把全部的判断逻辑都展示出来就很容易被群里那帮搞事的人玩坏
Config.py
class Config:
# 记录在哪些群组中使用
used_in_group = ["131551175"]
# 插件执行优先级
priority = 10
# 接话冷却时间（秒），在这段时间内不会连续两次接话
chat_cd = 15
# 戳一戳冷却时间（秒）
notice_cd = 900
# 机器人QQ号
bot_id = "123456789"
# 管理员QQ号，管理员无视冷却cd和触发概率
super_uid = ["673321342"]
# 聊天回复概率，用百分比表示，0-100%
p_chat_response = 60
# 戳一戳回复概率，用百分比表示，0-100%
p_poke_response = 20
# 默认禁言时间，每多戳一次会在默认禁言时间上翻倍
default_ban_time = 60
__init__.py
from .chat import *
chat.py
from nonebot import on_message, on_notice
from nonebot.typing import T_State
from nonebot.adapters import Bot, Event
from .config import Config
from time import time
import os
from nonebot.adapters.cqhttp import MessageSegment
import json
from collections import Counter
from random import randint
from .response_for_surper_user import *
from .response_for_common_user import *
from .response_for_all_time import *
__plugin_name__ = 'chat'
__plugin_usage__ = '用法： 日常聊天中响应关键词与戳一戳。'
img_path = 'file:///' + os.path.split(os.path.realpath(__file__))[0] + '/img/'
# 发送图片时用到的函数, 返回发送图片所用的编码字符串
def send_img(img_name):
global img_path
return MessageSegment.image(img_path + img_name)
# 记录上一次响应时间
last_response = {}
last_notice_response = {}
# 初始化时间戳， 初始化为开机时间-cd时间
init_last_response = time() - Config.chat_cd
init_last_notice_response = time() - Config.notice_cd
for group_id in Config.used_in_group:
last_response[group_id] = init_last_response
last_notice_response[group_id] = init_last_notice_response
# 判断是否过了响应cd的函数，默认使用配置文件中的cd
# 如果已经超过了最短响应间隔，返回True
def cool_down(group_id, cd = Config.chat_cd):
global last_response
return time() - last_response[group_id] > cd
# 以指定概率p返回True或者False
# 用于随机决定是否要回应
# 默认值为配置文件中的默认聊天响应概率
def random_response(p = Config.p_chat_response):
return randint(0, 99) < p
chat = on_message(priority=Config.priority)
# 针对聊天信息
@chat.handle()
async def handle_first_receive(bot: Bot, event: Event, state: T_State):
# 上次响应时间
global last_response
ids = event.get_session_id()
# 只对于群聊信息进行响应
if ids.startswith("group"):
# 拆解得到群号与用户号
_, group_id, user_id = event.get_session_id().split("_")
# 只对位于启用列表内的群组和非bot自身发送的信息进行响应
if group_id in Config.used_in_group and user_id != Config.bot_id:
# 获取信息文本
msg = str(event.get_message()).strip().replace('\r\n', '').replace('\n', '').replace(' ', '')
# 对信息内的所有文本进行出现频率统计
words = Counter(msg)
# 默认基础禁言时间
default_ban_time = Config.default_ban_time
# 1. 执行超级用户信息处理
# 超级用户无视冷却cd，也不会重置冷却cd
if user_id in Config.super_uid:
response = await get_response_for_surper_user(bot, chat, msg, user_id, group_id, default_ban_time, last_response,
words, send_img, cool_down, random_response)
# 如果回应不为空
if response:
# 发送响应字符串
await chat.finish(response)
# 2. 执行无冷却cd的违禁信息检查，忽略超级用户
if user_id not in Config.super_uid:
response = await get_response_for_all_time(bot, chat, msg, user_id, group_id, default_ban_time, last_response,
words, send_img, cool_down, random_response)
# 如果回应不为空
if response:
# 发送响应字符串
await chat.finish(response)
# 3. 执行普通用户信息处理，如果超级用户的响应为空，也要进入这一步
# 用户是否为超级用户
is_super_user = user_id in Config.super_uid
response = await get_response_for_common_user(bot, chat, msg, user_id, group_id, default_ban_time, last_response,
words, send_img, cool_down, random_response, is_super_user)
# 如果回应不为空
if response:
# 只有普通用户信息处理需要更新最近响应时间
if user_id not in Config.super_uid:
last_response[group_id] = time()
# 发送响应字符串
await chat.finish(response)
# --------以下信息用于对bot的戳一戳响应-------------
# 记录上一次戳机器人的nickname
last_notice_nickname = {}
# 记录cd内再次戳之后的吐槽次数
response = 0
# poke_ban_list[群组id][QQ号]得到被封禁次数
# 每次禁言默认事件*2^已经被封禁次数
poke_ban_list = {}
# 初始化
for group_id in Config.used_in_group:
poke_ban_list[group_id] = {}
# 针对戳一戳
chat_notice = on_notice(priority=Config.priority)
@chat_notice.handle()
async def handle_first_receive(bot: Bot, event: Event, state: T_State):
global last_notice_response
global last_notice_nickname
global response
try:
ids = event.get_session_id()
except:
pass
# 如果读取正常没有出错，因为有些notice格式不支持session
else:
# 如果这是一条群聊信息
if ids.startswith("group"):
_, group_id, user_id = event.get_session_id().split("_")
# 只对列表中的群使用
if group_id in Config.used_in_group:
description = event.get_event_description()
values = json.loads(description.replace("'", '"'))
# 如果被戳的是机器人
if values['notice_type'] == 'notify' and values['sub_type'] == 'poke' and str(
values['target_id']) == Config.bot_id:
if user_id in Config.super_uid:
await chat_notice.finish("如果是你的话，想戳多少次都可以哦~" + MessageSegment.image(img_path + '坏心思.jpg'))
# 如果不在响应cd
elif time() - last_notice_response[group_id] >= Config.notice_cd:
if randint(0, 99) < Config.p_poke_response:
last_notice_response[group_id] = time()
infos = str(await bot.get_stranger_info(user_id=values['user_id']))
nickname = json.loads(infos.replace("'", '"'))['nickname'] + '(' + str(
values['user_id']) + ')'
last_notice_nickname[group_id] = nickname
response = 0
# 清空ban列表
poke_ban_list[group_id] == {}
await chat_notice.finish(
nickname + "谢谢你戳了我，我自由了，现在你是新的群机器人了~" + MessageSegment.image(img_path + '坏心思.jpg'))
else:
if response == 0:
if randint(0, 99) < Config.p_poke_response:
response += 1
poke_ban_list[group_id][user_id] = 1
await chat_notice.finish("都说了新的群机器人已经是" + last_notice_nickname[
group_id] + "了呀，还戳我干什么？" + MessageSegment.image(img_path + '坏心思.jpg'))
elif response == 1:
if randint(0, 99) < Config.p_poke_response:
response += 1
if user_id in poke_ban_list[group_id]:
poke_ban_list[group_id][user_id] += 1
else:
poke_ban_list[group_id][user_id] = 1
await chat_notice.finish(
"去戳" + last_notice_nickname[group_id] + "呀！再这样就不理你们了！" + MessageSegment.image(
img_path + '坏心思.jpg'))
elif response == 2:
if randint(0, 99) < Config.p_poke_response:
response += 1
# 如果戳过了，那么每戳一次就把禁言时间翻倍
if user_id in poke_ban_list[group_id]:
poke_ban_list[group_id][user_id] += 1
else:
poke_ban_list[group_id][user_id] = 1
try:
await bot.set_group_ban(group_id=group_id, user_id=user_id,
duration=Config.default_ban_time * (
2 ** (poke_ban_list[group_id][user_id] - 1)))
except:
pass
await chat_notice.finish("mdzz！再戳我报警了！" + MessageSegment.image(img_path + '报警.jpg'))
else:
# 如果戳过了，那么每戳一次就把禁言时间翻倍
if user_id in poke_ban_list[group_id]:
poke_ban_list[group_id][user_id] += 1
else:
poke_ban_list[group_id][user_id] = 1
try:
await bot.set_group_ban(group_id=group_id, user_id=user_id,
duration=Config.default_ban_time * (
2 ** (poke_ban_list[group_id][user_id] - 1)))
except:
pass
response_for_surper_user.py
# 用于处理超级用户信息的函数
async def get_response_for_surper_user(bot, chat, msg, user_id, group_id, default_ban_time, last_response,
words, send_img, cool_down, random_response):
if "星岚贴贴" in msg or "贴贴星岚" in msg:
return "贴贴~♡" + send_img('贴贴.jpg')
if "抱星岚" in msg or "星岚抱" in msg:
return "好耶~抱住你蹭蹭~"
if "我只离开了几分钟" in msg:
return "你们就搞出这种大新闻！这像话吗？！"
response_for_all_time.py
from collections import Counter
# 用于处理任何情况下都需要进行判断的函数
# 主要用于对一些违禁用户进行禁言处理
async def get_response_for_all_time(bot, chat, msg, user_id, group_id, default_ban_time, last_response,
words, send_img, cool_down, random_response):
msg = msg.replace('阿', '啊')
words = Counter(msg)
# 针对恶臭发言用户
if "哼" in words and "啊" in words:
if words["哼"] >= 2 and words["啊"] >= 3:
# 尝试禁言
try:
await bot.set_group_ban(group_id=group_id, user_id=user_id, duration=default_ban_time)
# 如果对方是管理员，那就假装无事发生
except:
pass
return "恶臭，死吧！" + send_img('杀戮的欲望.jpg')
# 针对发病用户
if "嘿" in words:
if words["嘿"] >= 5:
# 尝试禁言
try:
await bot.set_group_ban(group_id=group_id, user_id=user_id, duration=default_ban_time)
# 如果对方是管理员，那就假装无事发生
except:
pass
return "禁止发病！" + send_img('杀戮的欲望.jpg')
response_for_common_user.py
from random import choice
import asyncio
# 用于处理普通用户信息的函数
async def get_response_for_common_user(bot, chat, msg, user_id, group_id, default_ban_time, last_response,
words, send_img, cool_down, random_response, is_super_user):
# 使用默认冷却cd与默认概率的响应
if (cool_down(group_id) and random_response()) or is_super_user:
# 随机返回响应
if "抱星岚" in msg or "星岚抱" in msg:
responses = ["这就送你一发恒星核心温度的抱抱~",
"倒也不是不行，但是你真的能承受住吗？",
"为什么会有人想要拥抱真空呢？"]
return choice(responses)
# 发送多条信息并且在过程中休眠
if "自爆" in msg:
await chat.send("自爆程序即将启动，倒计时：")
await asyncio.sleep(1)
await chat.send("3")
await asyncio.sleep(1)
await chat.send("2")
await asyncio.sleep(1)
await chat.send("1")
await asyncio.sleep(1)
return r'开始执行rm -rf /*'
# 添加响应概率不同的事件
if (cool_down(group_id) and random_response(p=20)) or is_super_user:
if "？" == msg or "?" == msg:
responses = ['？',
'¿',
'？？',
'？？？',
'当我打出问号的时候，不是我有问题，而是我觉得你有问题',
send_img('问号1.jpg'),
send_img('问号2.jpg')]
return choice(responses)
坏心思.jpg
报警.jpg
杀戮的欲望.jpg
问号1.jpg
问号2.jpg
nonebot2聊天机器人插件5：加群退群通报与退群次数记录join_and_leave
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）