nonebot2聊天机器人插件5：加群退群通报与退群次数记录join

2023-07-16 11:31| 来源: 网络整理| 查看: 265

nonebot2聊天机器人插件5：加群退群通报与退群次数记录join_and_leave 1. 插件用途2. 目录结构3. 实现难点与解决方案3.1 读取加群退群信息3.2 数据库操作 4. 代码实现5. 插件配图6. 实际效果7. 下一个插件该插件涉及知识点：读取加群退群信息，数据库操作插件合集：nonebot2聊天机器人插件

该系列为用于QQ群聊天机器人的nonebot2相关插件，不保证完全符合标准规范写法，如有差错和改进余地，欢迎大佬指点修正。前端：nonebot2 后端：go-cqhttp 插件所用语言：python3 前置环境安装过程建议参考零基础2分钟教你搭建QQ机器人——基于nonebot2，但是请注意该教程中的后端版本过旧导致私聊发图异常，需要手动更新go-cqhttp版本。

1. 插件用途

在有用户加群时，自动发送欢迎信息并艾特加群的新人~ 在有用户退群时，发送退群通报，让不是管理员的群友也能得知对方退群的时间点。在有曾经退过群的用户再次加群时，发出通报提示该用户曾经退出过多少次这个群！没错我就是要对那些反复横跳的人进行公开处刑！ bot管理员能够查看指定用户在该群的退群记录，或者把该群历史退群人数最多的用户列出一个排行榜！再次公开处刑！

2. 目录结构

在plugins文件夹中新建一个文件夹join_and_leave，文件夹内目录结构如下：

其中img为用于存储发送的图片文件的文件夹，data为储存数据库文件的文件夹，join_and_leave.py为程序主要代码的位置，config.py用于存储配置项，model.py用于封装与数据库交互的SQL语言操作，__init__.py为程序启动位置。

3. 实现难点与解决方案 3.1 读取加群退群信息

与上一章中读取戳一戳信息类似，加群和退群同样是notice类型的信息，新建一个群，用小号加群和退群进行几次测试后，通过event.get_event_description()就得到了通知信息的大致形式。

description： {'time': 时间戳, 'self_id': 机器人QQ号, 'post_type': 'notice', 'notice_type': 'group_decrease', 'sub_type': 'leave', 'user_id': 退群者QQ号, 'group_id': 群号, 'operator_id': 退群者QQ号}

验证发现：有人加群'notice_type': 'group_increase' 有人自己退群'notice_type': 'group_decrease', 'sub_type': 'leave' 有人被踢出群'notice_type': 'group_decrease', 'sub_type': 'kick' 以类似上一章的方式利用json库转换为字典后就很容易处理。

3.2 数据库操作

为了方便起见，使用轻量的sqlite数据库进行操作，无需登录用户名与密码，只需要移动一个db文件就可以转移数据库。 python3自带sqlite数据库，如何使用SQL语句访问和操作数据库在此不再赘述，如果需要了解可以很容易搜索到相关的教程。文件目录中的db文件不需要自己创建，只需要新建好空文件夹，代码运行后如果不存在数据库，会自动新建数据库文件！

4. 代码实现

Config.py

class Config: # 记录在哪些群组中使用 used_in_group = ["131551175"] # 插件执行优先级 priority = 10

__init__.py

from .join_and_leave import *

join_and_leave.py

from nonebot import on_notice, on_command from nonebot.typing import T_State from nonebot.adapters import Bot, Event from nonebot.adapters.cqhttp import MessageSegment from nonebot.permission import SUPERUSER import json import os from .config import Config from .model import * __plugin_name__ = 'join_and_leave' __plugin_usage__ = '用法：提示有人加群或者退群，并记录此人在该群的历史退群次数。' # 创建数据库 create_db() 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) # 通报加群与退群 join_and_leave = on_notice(priority=Config.priority) @join_and_leave.handle() async def handle_first_receive(bot: Bot, event: Event, state: T_State): 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'] == 'group_increase': # 获取此人是否曾经退群 leave_record = get_exist_data(user_id, group_id) print(leave_record) if len(leave_record) == 0: await join_and_leave.finish( "一只新食材" + MessageSegment.at(values['user_id']) + '跳进了锅里~\n' + send_img("揣手手.jpg")) # 如果此人之前曾经退群过，发出警告 else: await join_and_leave.send( "一只新食材" + MessageSegment.at(values['user_id']) + '跳进了锅里~\n' + send_img("揣手手.jpg")) await join_and_leave.finish( f"警告：记录表明这只新食材曾经退群过{leave_record[0][0]}次！" + send_img("退群.png") + send_img("震撼.jpg")) # 有人退群 elif values['notice_type'] == 'group_decrease': add_data(user_id, group_id) # 自己退群 if values['sub_type'] == 'leave': infos = str(await bot.get_stranger_info(user_id=values['user_id'])) nickname = json.loads(infos.replace("'", '"'))['nickname'] + '(' + str(values['user_id']) + ')' await join_and_leave.finish( nickname + '在这一刻选择了离开我们。\n' + send_img("退群.png") + send_img("震撼.jpg")) # 被踢出群 elif values['sub_type'] == 'kick': infos = str(await bot.get_stranger_info(user_id=values['user_id'])) nickname = json.loads(infos.replace("'", '"'))['nickname'] + '(' + str(values['user_id']) + ')' operator_infos = str(await bot.get_stranger_info(user_id=values['operator_id'])) operator_nickname = json.loads(operator_infos.replace("'", '"'))['nickname'] + '(' + str( values['operator_id']) + ')' await join_and_leave.finish('超级可爱的' + operator_nickname + '面无表情地把' + \ nickname + '一脚踢出了这个世界！\n' + send_img("踢人.jpg")) # 管理员专属指令 # 查询命令 copyer_helper = on_command("退群帮助", priority=Config.priority) @copyer_helper.handle() async def handle_first_receive(bot: Bot, event: Event, state: T_State): await copyer_helper.finish('''退群指令仅限bot管理员使用：退群次数查询 [QQ号]——查询指定用户在当前群的退群次数退群次数排行——查询当前群的退群次数最多的人''') get_leave_data = on_command("退群次数查询", permission=SUPERUSER, priority=Config.priority) @get_leave_data.handle() async def handle_first_receive(bot: Bot, event: Event, state: T_State): ids = event.get_session_id() # 如果这是一条群聊信息 if ids.startswith("group"): _, group_id, user_id = event.get_session_id().split("_") if group_id in Config.used_in_group: search_id = str(event.get_message()).strip() leave_record = get_exist_data(search_id , group_id) infos = str(await bot.get_stranger_info(user_id=search_id)) nickname = json.loads(infos.replace("'", '"'))['nickname'] + '(' + str(search_id) + ')' if len(leave_record) == 0: await join_and_leave.finish(nickname+"没有退出过该群。") else: await join_and_leave.finish(f"{nickname}退出过该群{leave_record[0][0]}次。") get_most_leave_user = on_command("退群次数排行", permission=SUPERUSER, priority=Config.priority) @get_most_leave_user.handle() async def handle_first_receive(bot: Bot, event: Event, state: T_State): ids = event.get_session_id() # 如果这是一条群聊信息 if ids.startswith("group"): _, group_id, _ = event.get_session_id().split("_") if group_id in Config.used_in_group: leave_record = search_most_leave(group_id) if len(leave_record) == 0: await get_most_leave_user.finish("该群暂无退群数据") else: result = '前五位退群最多的人是：\n' for user_id, leave in leave_record: infos = str(await bot.get_stranger_info(user_id=user_id)) nickname = json.loads(infos.replace("'", '"'))['nickname'] + '(' + str(user_id) + ')' result += nickname + f'\n退群:{leave}次\n\n' await get_most_leave_user.finish(result[:-1]+send_img('震撼.jpg'))

model.py

import sqlite3 import os # 数据库文件的路径 db_path = os.path.split(os.path.realpath(__file__))[0] + '/data/join_and_leave.db' # 如果不存在的话，创建数据库，并且创建数据表 def create_db(): conn = sqlite3.connect(db_path) try: create_tb_cmd = ''' CREATE TABLE IF NOT EXISTS leave_record (user_id INT8, group_id INT8, leave INT); ''' conn.execute(create_tb_cmd) except: pass conn.commit() conn.close() # 查询指定人退群次数，返回退群次数 def get_exist_data(user_id, group_id): conn = sqlite3.connect(db_path) insert_cmd = f'SELECT leave from leave_record where user_id == {user_id} and group_id = {group_id}' cursor = list(conn.execute(insert_cmd)) conn.close() # 如果不存在，则返回空列表 # 否则返回退群次数 return cursor # 添加退群用户 def add_data(user_id, group_id): leave = get_exist_data(user_id, group_id) # 如果用户数据还不存在 if not leave: cmd = f'INSERT INTO leave_record (user_id,group_id,leave) VALUES ({user_id}, {group_id}, {1});' # 否则计数+1 else: cmd = f'UPDATE leave_record SET leave = {leave[0][0]+1} WHERE user_id == {user_id} and group_id = {group_id};' conn = sqlite3.connect(db_path) conn.execute(cmd) conn.commit() conn.close() # 返回该群退群次数最高的人，返回用户qq号与退群次数 def search_most_leave(group_id): conn = sqlite3.connect(db_path) insert_cmd = f'SELECT user_id,leave from leave_record WHERE group_id = {group_id} order by -leave limit 5' cursor = list(conn.execute(insert_cmd)) conn.close() return cursor 5. 插件配图

揣手手.jpg 请添加图片描述

踢人.jpg 请添加图片描述

退群.png 请添加图片描述

震撼.jpg 请添加图片描述

6. 实际效果

请添加图片描述

7. 下一个插件

nonebot2聊天机器人插件6：复读机博弈论ban_copyer

【本文地址】

公司简介

联系我们