自制QQ机器人插件笔记[nonebot2部署于ubuntu系统服务器]

75 阅读 0 评论 50 点赞

我是靠谱客的博主微笑衬衫，最近开发中收集的这篇文章主要介绍自制QQ机器人插件笔记[nonebot2部署于ubuntu系统服务器]，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

前言

目前机器人框架有很多,框架内也有很多插件实现各种各样的功能.但是当我们想要的功能没有理想的插件可以实现时,也就需要自己编写一个插件.很多框架只支持易语言编写的插件,但是我个人认为易语言的编程体验十分不好,于是最终采用可以用python编写插件的nonebot框架+go-cqhttp方案

目标功能

为了管理一个千人QQ群,需要一个可靠且便利的方式来记录群员的违规行为,经过讨论,管理员们决定采用类似机动车驾驶证的记分制度.本文章即为实现记分并踢出记分达到指定分数的群员.

nonebot2

nonebot的官方文档已经给出安装方法,在这里只大概记录一下.官方文档

首先安装需要的库[需要Python3.7以上]

pip install nb-cli==0.5.3

最新版本为0.6.4,但是这个版本安装nonebot时会有一些本人不会解决的麻烦,所以尝试了旧版本,可以正常运行

在机器人目录内安装

nb

选择Create a New Project--输入项目名--选择第一项新建文件夹--载入 nonebot 内置插件--选择 cqhttp插件

此时就会生成机器人需要的文件,启动机器人只需要启动bot.py程序

python3 bot.py

Go-cqhttp

这就是一个QQ客户端,首先在github下载合适的文件go-cqhttp下载

下载后启动:

./go-cqhttp

会生成其配置文件,需要修改的是config.yml文件:

xxxxxxxxxxxxxxxxx
    uin:      # 机器人QQ账号[密码不写,登陆时扫码登陆]
xxxxxxxxxxxxxxxxx
    universal: ws://127.0.0.1:8080/cqhttp/ws
xxxxxxxxxxxxxxxxx

再次启动go-cqhttp

加载插件

在机器人目录的plugins目录内,新建lab.py作为插件,并修改bot.py以加载插件

nonebot.init()

# load your plugin here

nonebot.run()

将# load your plugin here替换为

nonebot.load_plugins("triority/plugins/lab.py")

自行替换插件目录

编写插件

插件的编辑方法在官方文档中已经有所说明,在下面直接给出文章写到这时已经写完的代码

这是我写的第一个插件,用于在QQ群记分

from nonebot.adapters.cqhttp import Bot, Event
from nonebot.plugin import on_message
import numpy as np
import datetime
#白名单ops
ops=['']
#记分
scores=np.load('scores.npy',allow_pickle=True).item()

#回复部分
reply = on_message(priority=100)
@reply.handle()
async def reply_handle(bot: Bot, event: Event):
    try:
        #获取聊天内容user_msg
        user_msg = str(event.get_message()).strip()
        #获取发消息的人user,群号qun
        msg = str(event.get_session_id())
        user = msg.split("_")[2]
        qun = msg.split("_")[1]
        now = datetime.datetime.now()
        #消息空格分割列表words
        words = user_msg.split(' ')
        if qun=='':
            if user_msg[0:2]=='查分':
                if words[1] in scores:
                    await reply.finish(words[1]+' 的记分为 '+str(scores[words[1]][0]))
                else:
                    await reply.finish(words[1]+' 无记分记录')
            if user in ops :
                if user_msg[0:2]=='记分':
                    if int(words[1])>0:
                        if words[2] in scores:
                            scores[words[2]][0]=scores[words[2]][0]+int(words[1])
                            scores[words[2]].append([now.strftime("%Y-%m-%d %H:%M:%S"),words[1],words[3]])
                        else:
                            scores[words[2]]=[int(words[1]),[now.strftime("%Y-%m-%d %H:%M:%S"),words[1],words[3]]]
                        np.save('scores.npy', scores)
                        await reply.finish(words[2]+' 当前记分为 '+str(scores[words[2]][0]))
                    else:
                        await reply.finish('加分数不能为非正数!')
                if user_msg[0:2]=='清零':
                    if words[1] in scores:
                        del scores[words[1]]
                        np.save('scores.npy', scores)
                        await reply.finish(words[1]+' 记分已清空')
                    else:
                        await reply.finish(words[1]+' 无记分记录')
                if user_msg[0:2]=='明细':
                    if words[1] in scores:
                        reason='总记分'+str(scores[words[1]][0])+':n'
                        for i in range(1,len(scores[words[1]])):
                            reason=reason+str(i)+'.'+scores[words[1]][i][0]+'n  记分数:'+scores[words[1]][i][1]+'n  理由:'+scores[words[1]][i][2]+'n'
                        await reply.finish(words[1]+' 记分明细如下:n'+reason)
                    else:
                        await reply.finish(words[1]+' 无记分记录')
                if user_msg[0:2]=='穷举':
                    await reply.finish('全部记录如下:'+str(scores))
    except KeyError:
        await reply.finish()

这是第二个,用于在QQ群中发送rcon命令

from rcon import MCRcon
from nonebot.adapters.cqhttp import Bot, Event
from nonebot.plugin import on_message
 
def rcon_send(ip,port,key,command):
    with MCRcon(ip, key, port ) as mcr:
        resp = mcr.command(command)
        return resp
 
servers={'name':["ip",port,'key',['op']]
         }
 
reply = on_message(priority=100)
 
@reply.handle()
async def reply_handle(bot: Bot, event: Event):
    try:
        #获取聊天内容user_msg
        user_msg = str(event.get_message()).strip()
        #获取发消息的人user,群号qun
        msg = str(event.get_session_id())
        user = msg.split("_")[2]
        qun = msg.split("_")[1]
        # 消息空格分割列表words
        words = user_msg.split(' ')
        if words[0] == 'rcon':
            if words[1] not in servers:
                await reply.finish('服务器未被记录')
            elif user in servers[words[1]][3]:
                try:
                    ret = rcon_send(servers[words[1]][0], servers[words[1]][1], servers[words[1]][2], words[2])
                    await reply.finish(ret)
                except:
                    await reply.finish('无法连接至服务器rcon')
            else:
                await reply.finish('您没有连接服务器的权限')
    except KeyError:
        await reply.finish()