[Python知识库][“空头计划“第二期 ] QQ机器人(Python ^3.7 + 机器人框架NoneBot-beta2最新版本)实现

前言

临近学校社团招新, 为了给卡哇伊的学弟学妹们准备一些小礼物, 博主和他的好同志们开始策划一些小活动之一 – QQ机器人

正文

步骤详细分析

1. 框架部分

按照博主的理解, QQ机器人的 完整功能框架 有五个部分,
1 ) Python模块( nb-cli ) 中的 机器人框架 ( Nonebot2 )
2 ) 驱动器( httpx )
3 ) 协议适配器( Onebot )
4 ) 启动器( cqhttp )
5 ) 项目和启动器的配置

2. 插件(功能)部分

1 ) 内置插件
2 ) 自定义插件

一. 完整功能框架

注意

首先说一下在 Python3.7及以上版本 的加持下, 机器人框架 (nonebot) 有 beta2 和 a16 两个版本, beta2 是 最新版
不同机器人框架框架的 插件 (你可以理解为机器人的功能) 之间也需要代码迭代更新才能通用
所以博主提前声明，以下的内容是基于 nonebot 的 beta2 版写的，替各位小伙伴提前排坑

帮助文档

1. 机器人框架NoneBot(beta2)帮助文档

创建新项目QQ_Robot

1. 模块 nb-cli 的安装(开发模块包含NoneBot2)

在编译器 (博主推荐 Pycharm )的终端( terminal ) 输入

pip install nb-cli

如下图是成功安装:

2. 驱动器 httpx 的安装

输入nb driver list 查看驱动器列表

nb driver list

如下图:

安装驱动器

1.输入 nb driver 如下图, 选择第三个 Insatll a Builtin Driver 按 回车

nb driver

2.输入httpx 如下图, 回车

httpx

如下图就是安装成功:

3. 协议适配器 Onebot 的安装

输入nb adapter list 查看协议适配器列表

nb adapter list

安装协议适配器

1.输入 nb adapter , 如下图选择第三个 Install a Published Adapter , 回车

nb adapter

2.输入OneBot, 如下图, 回车

OneBot

如下图成功安装:

4.启动器 cqhttp 的安装

cqhttp最新版下载

如下图选择go-cqhttp_windows_amd64.exe, 点击下载
ps: 备用!!(博主接下来教大家如何使用)

下面是博主替不会用 GitHub 的小伙伴们准备的 百度云资源

链接：https://pan.baidu.com/s/1W6Ses_HGygr8lyyNw-b2pQ?pwd=269s 
提取码：269s

5.项目和启动器配置

项目和文件配置

1.在 Pycharm 终端输入nb create , 回车

nb create

2.输入你的项目名称, (随机) , 回车

3. 选择 In a "src" folder , 回车

4.空格选中echo, 回车

5. 空格 选中 OneBot V11 , 回车

如下图, 项目创建完成

用电脑自带的 文件管理器 (快捷键WIN + E)打开项目根目录

提示如下
1.此处点击 复制路径/引用 , 并选择 绝对路径

2.这里 粘贴

进入以下目录

3.新建sq文件夹, 将之前下载好的文件 go-cqhttp_windows_amd64.exe 粘贴到 sq 文件夹里

ps: 因为 启动器 要和项目中的 bot.py 放在同级文件夹

启动器的配置

1 . 双击 EXE文件 go-cqhttp_windows_amd64.exe, 点击更多信息

2.点击仍要运行
ps: Windows自带的安全措施, 我们用的是绿色是开源软件, 放心使用

3.忽略警告, 所有点 确定

4.回到原 启动器 所在位置 sq 文件夹, 双击 运行 go-cqhttp.bat 文件

5.在弹出的命令行窗口, 输入 023 , 选择对应的 HTTP通信 , 正向 Websocket 通信 , 反向 Websocket 通信, 回车
ps:这些是我们机器人开发用到的大部分驱动

023

如下图是成功, 关闭窗口, 回到 sq 文件夹,出现 config.yml

回到 Pycharm 中你的项目目录, 打开 config.yml , 配置如下内容, 全选Ctrl + A , 粘贴Ctrl + V

# go-cqhttp 默认配置文件

account: # 账号相关
  uin:  # QQ账号
  password:  # 密码为空时使用扫码登录
  encrypt: false  # 是否开启密码加密
  status: 0      # 在线状态 请参考 https://github.com/Mrs4s/go-cqhttp/blob/dev/docs/config.md#在线状态
  relogin: # 重连设置
    disabled: false
    delay: 3      # 重连延迟, 单位秒
    interval: 0   # 重连间隔
    max-times: 0  # 最大重连次数, 0为无限制

  # 是否使用服务器下发的新地址进行重连
  # 注意, 此设置可能导致在海外服务器上连接情况更差
  use-sso-address: true

heartbeat:
  disabled: false # 是否开启心跳事件上报
  # 心跳频率, 单位秒
  # -1 为关闭心跳
  interval: 5

message:
  # 上报数据类型
  # 可选: string,array
  post-format: string
  # 是否忽略无效的CQ码, 如果为假将原样发送
  ignore-invalid-cqcode: false
  # 是否强制分片发送消息
  # 分片发送将会带来更快的速度
  # 但是兼容性会有些问题
  force-fragment: false
  # 是否将url分片发送
  fix-url: false
  # 下载图片等请求网络代理
  proxy-rewrite: ''
  # 是否上报自身消息
  report-self-message: false
  # 移除服务端的Reply附带的At
  remove-reply-at: false
  # 为Reply附加更多信息
  extra-reply-data: false

output:
  # 日志等级 trace,debug,info,warn,error
  log-level: warn
  # 是否启用 DEBUG
  debug: false # 开启调试模式

# 默认中间件锚点
default-middlewares: &default
  # 访问密钥, 强烈推荐在公网的服务器设置
  access-token: ''
  # 事件过滤器文件目录
  filter: ''
  # API限速设置
  # 该设置为全局生效
  # 原 cqhttp 虽然启用了 rate_limit 后缀, 但是基本没插件适配
  # 目前该限速设置为令牌桶算法, 请参考:
  # https://baike.baidu.com/item/%E4%BB%A4%E7%89%8C%E6%A1%B6%E7%AE%97%E6%B3%95/6597000?fr=aladdin
  rate-limit:
    enabled: false # 是否启用限速
    frequency: 1  # 令牌回复频率, 单位秒
    bucket: 1     # 令牌桶大小

servers:
  # HTTP 通信设置
  - http:
      # 是否关闭正向HTTP服务器
      disabled: false
      # 服务端监听地址
      host: 127.0.0.1
      # 服务端监听端口
      port: 5701
      # 反向HTTP超时时间, 单位秒
      # 最小值为5，小于5将会忽略本项设置
      timeout: 5
      middlewares:
        <<: *default # 引用默认中间件
      # 反向HTTP POST地址列表
      post:
      #- url: '' # 地址
      #  secret: ''           # 密钥
      #- url: 127.0.0.1:5701 # 地址
      #  secret: ''          # 密钥

  # 正向WS设置
  - ws:
      # 是否禁用正向WS服务器
      disabled: false
      # 正向WS服务器监听地址
      host: 127.0.0.1
      # 正向WS服务器监听端口
      port: 6701
      middlewares:
        <<: *default # 引用默认中间件

  - ws-reverse:
      # 是否禁用当前反向WS服务
      disabled: false
      # 反向WS Universal 地址
      # 注意 设置了此项地址后下面两项将会被忽略
      universal: ws://127.0.0.1:8080/onebot/v11/ws/
      # 反向WS API 地址
      api: ws://your_websocket_api.server
      # 反向WS Event 地址
      event: ws://your_websocket_event.server
      # 重连间隔 单位毫秒
      reconnect-interval: 3000
      middlewares:
        <<: *default # 引用默认中间件
  # pprof 性能分析服务器, 一般情况下不需要启用.
  # 如果遇到性能问题请上传报告给开发者处理
  # 注意: pprof服务不支持中间件、不支持鉴权. 请不要开放到公网
  - pprof:
      # 是否禁用pprof性能分析服务器
      disabled: true
      # pprof服务器监听地址
      host: 127.0.0.1
      # pprof服务器监听端口
      port: 7700

  # 可添加更多
  #- ws-reverse:
  #- ws:
  #- http:
  #- pprof:

database: # 数据库相关设置
  leveldb:
    # 是否启用内置leveldb数据库
    # 启用将会增加10-20MB的内存占用和一定的磁盘空间
    # 关闭将无法使用 撤回 回复 get_msg 等上下文相关功能
    enable: true

打开 .env, 配置如下内容

ENVIRONMENT=prod

打开 bot.py , 配置以下内容

import nonebot
from nonebot.adapters.onebot.v11 import Adapter as ONEBOT_V11Adapter

nonebot.init()

# 加载插件目录，该目录下为各插件
nonebot.load_from_toml("pyproject.toml")

# 测试插件
nonebot.load_builtin_plugin("echo")

app = nonebot.get_asgi()

driver = nonebot.get_driver()
driver.register_adapter(ONEBOT_V11Adapter)

# 加载 nonebot 内置插件
nonebot.load_builtin_plugins()

if __name__ == "__main__":
    nonebot.logger.warning("Always use `nb run` to start the bot instead of manually running!")
    nonebot.run(app="__mp_main__:app")

打开 pyproject.toml , 配置以下内容

[tool.poetry]
name = "bot"
version = "0.1.0"
description = "bot"
authors = []
readme = "README.md"

[tool.poetry.dependencies]
python = "^3.7.3"
nonebot2 = "^2.0.0-beta.1"

[tool.poetry.dev-dependencies]
nb-cli = "^0.6.0"

[tool.nonebot]
# 内置插件全名
plugins = []
# 自定义插件保存位置
plugin_dirs = ["src/plugins"]

[tool.black]
line-length = 88

[tool.isort]
profile = "black"
line_length = 88
skip_gitignore = true

[build-system]
build-backend = "poetry.core.masonry.api"
requires = ["poetry-core>=1.0.0"]

以上所有配置完毕, 完整的项目结构如下

二. 运行部分

测试运行 bot.py

1., 运行项目中的bot.py

弹出以下, 就是运行成功

回到 项目下的 sq 文件夹, 双击打开 go-cqhttp.bat Bat文件

使用 绑定为QQ机器人的QQ号 扫码登陆, 在手机上点击 继续登录

ps:
1. QQ机器人依托于已登记过的QQ账号
2. 此处最好 全屏 , 二维码的加载问题, 有兴趣的小伙伴自行尝试

出现以下为绑定成功:

测试QQ机器人

用任意一个别的QQ账号发送信息到 QQ机器人绑定的QQ账号, 格式为/echo + 空格 + 内容

三. 插件(功能)添加

1.内置插件

NoneBot框架内置插件下载

很多 有趣 的插件( 功能 ) 小伙伴们自行下载, 且官方 内置插件 问题请移步至 官方 解决问题
毕竟博主只是个技术小萌新, 不掐饭, 不下饭

简单介绍一些商店内 有趣 的内置插件的安装

1. wordle 猜单词
   ps: 可以和准备四六级的小伙伴们一起玩耍

点击 点击复制安装指令

回到 Pycharm 在 终端 输入, 回车 , 直接安装即可

出现下图为安装成功

如果安装不成功, 尝试输入 pip install + 插件名全称

将已安装 内置插件 写入 pyproject.toml 中的 plugins 中, 注意 引号 括住, 多个插件之间 英文逗号 隔开

效果图如下:

2. Nonebot2 插件轻量帮助列表
   ps:实用插件管理助手

安装步骤同上

效果如下

2. 自定义插件

简单的分享一个大大的Nonebot2的插件

codeforce 查询插件

import requests
from nonebot import on_command
from nonebot.adapters.onebot.v11 import Bot, Event, Message
import random
from nonebot.adapters.onebot.v11 import MessageSegment
import json
import re
from lxml import etree
from nonebot.matcher import Matcher
from nonebot.adapters import Message
from nonebot.params import Arg, CommandArg, ArgPlainText
import nonebot

def recent_contest():
    headers = {
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/65.0.3325.146 Safari/537.36'
    }
    url = "http://algcontest.rainng.com"
    val = requests.get(url, headers=headers)
    res = json.loads(val.content)
    sum = ""
    i = 0
    for it in res:
        i = i + 1
        sum = sum + '[比赛名称]： '
        sum = sum + it['name']
        sum = sum + '\n[比赛时间]： '
        sum = sum + it['startTime']
        sum = sum + '\n[比赛链接]： '
        sum = sum + it['link']
        sum = sum + '\n'
        if i == 3:
            break
    return sum


Rcontest = on_command("最近比赛", priority=2, block=True)


@Rcontest.handle()
async def Rcontest_(bot: Bot, event: Event):
    try:
        if int(event.get_user_id()) != event.self_id:
            await bot.send(
                event=event,
                message=str(recent_contest())
            )
    except Exception as e:
        await Rcontest.send("最近比赛插件出现故障，请联系Mangata")


def get_least_cf():
    headers = {
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/65.0.3325.146 Safari/537.36'
    }
    url = "http://algcontest.rainng.com"
    val = requests.get(url, headers=headers)
    res = json.loads(val.content)
    sum = ""
    i = 0
    for it in res:
        if it['oj'] == 'CodeForces':
            sum = '找到最近一场CodeForces比赛：\n'
            sum = sum + '[比赛名称]： '
            sum = sum + it['name']
            sum = sum + '\n[比赛时间]： '
            sum = sum + it['startTime']
            sum = sum + '\n[比赛链接]： '
            sum = sum + it['link']
            sum = sum + '\n'
            break
    if sum == "":
        sum = "最近没有Codeforces比赛噢，开始摆烂吧！"
    return sum


codeforces = on_command(cmd="cf", priority=2, block=True)


@codeforces.handle()
async def codeforces_():
    try:
        await codeforces.send(get_least_cf())
    except Exception as e:
        await codeforces.send("最近CF插件出现故障，请联系Mangata")


def honor(num: int):
    if num <= 50:
        return "坚韧黑铁"
    elif num <= 150:
        return "英勇黄铜"
    elif num <= 300:
        return "不屈白银"
    elif num <= 500:
        return "荣耀黄金"
    elif num <= 650:
        return "华贵铂金"
    elif num <= 800:
        return "璀璨钻石"
    elif num <= 1000:
        return "超凡大师"
    elif num <= 1500:
        return "傲世宗师"
    else:
        return "最强王者"


def get_usr(id: int):
    url = "http://acm.mangata.ltd/user/" + str(id)
    ua_headers = {"User-Agent": 'Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 6.0; Trident/4.0)'}
    # 网页代码
    response = requests.get(url=url, headers=ua_headers).text
    # 转换为etree对象
    tree = etree.HTML(response)
    img_lst = tree.xpath('//*[@id="panel"]/div[3]/div/div[1]/div[1]/div[1]/div/div/div[2]/p[2]/text()')
    message = img_lst[0].split(',')
    slove_problem = int(re.findall("\d+", message[0])[0])
    return "\n[当前段位]： [" + honor(slove_problem) + "]"


# Dream OJ 查询用户
DOJ_USER = on_command("DOJ", aliases={'find', '查找', '查找用户'}, priority=2, block=True)


@DOJ_USER.handle()
async def DOJ_USER_(matcher: Matcher, args: Message = CommandArg()):
    plain_text = args.extract_plain_text()  # 首次发送命令时跟随的参数，例：/天气 上海，则args为上海
    if plain_text:
        matcher.set_arg("name", args)  # 如果用户发送了参数则直接赋值


@DOJ_USER.got("name", prompt="请输入你想查询的用户名...")
async def handle_DOJ_USER(name: Message = Arg(), sname: str = ArgPlainText("name")):
    try:
        url = "http://acm.mangata.ltd/api/?"
        headers = {
            'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/65.0.3325.146 Safari/537.36'
        }
        query = "query {user(uname: \"" + sname + "\") {uname,mail,loginat,regat,role,avatarUrl,_id}}"
        url = url + query
        val = requests.get(url, headers=headers)
        res = json.loads(val.content)
        it = res['data']['user']
        ans = ""
        if res['data']['user'] == None:
            ans = ans + "查无此人"
            await DOJ_USER.send(ans)
        else:
            id = it['_id']
            ans = "\n[用户昵称]： [" + it['uname'] + "]"
            ans = ans + str(get_usr(id))
            ans = ans + "\n[上次登陆]： [" + it['loginat'] + "]"
            ans = ans + "\n[注册时间]： [" + it['regat'] + "]"
            ans = ans + "\n[用户身份]： [" + it['role'] + "]"
            qq = str(it['mail'])
            ava = "https://q1.qlogo.cn/g?b=qq&nk="
            if qq.find("@qq.com") == -1:
                ava = "http://acm.mangata.ltd/file/2/12.jpg"
            else:
                qq = qq.strip('@qq.com')
                ava = ava + qq + "&s=160"
            await DOJ_USER.send(MessageSegment.image(ava) + ans)
    except Exception as e:
        await DOJ_USER.send("DOJ用户插件出现故障，请联系Mangata")


# 获取随机题目
def get_problem():
    url = "http://acm.mangata.ltd/api/?"
    headers = {
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/65.0.3325.146 Safari/537.36'
    }
    id = str(random.randint(1, 1786))
    query = "query{problem(id:" + id + "){pid,title,nSubmit,nAccept,difficulty,tag}}"
    url = url + query
    print(url)
    val = requests.get(url, headers=headers)

    res = json.loads(val.content)
    print(res)
    it = res["data"]["problem"]
    if it == None:
        return str("题目已被隐藏请再次输入！")
    else:
        link = "http://acm.mangata.ltd/p/" + it['pid']
        tag = ""
        l = len(it['tag'])
        j = 1
        for i in it['tag']:
            if j < l:
                tag = tag + i + "、"
            else:
                tag = tag + i
            j = j + 1
        ans = "[题目名称]： " + it['title']
        ans = ans + "\n[题目连接]： " + link
        ans = ans + "\n[算法标签]： " + tag
        ans = ans + "\n[总提交数]： " + str(it['nSubmit'])
        ans = ans + "\n[总通过数]： " + str(it['nAccept'])
        ans = ans + "\n[预估难度]： " + str(it['difficulty'])
        ans = ans + "\n骚年快来挑战吧！别忘了写题解噢！"
        return ans


# 每日一题
DOJ_PROBLEM = on_command("随机题目", aliases={'每日一题'}, priority=2, block=True)


@DOJ_PROBLEM.handle()
async def DOJ_PROBLEM_(bot: Bot, event: Event):
    if int(event.get_user_id()) != event.self_id:
        ans = get_problem()
        await bot.send(
            event=event,
            message=ans
        )

自定义插件导入方式

1.在 plugins 目录中新建文件夹, 命名为插件的名称

ps:名称最好全英, 中文可能会出现莫名奇妙的报错

2.在新建的文件夹里创建python文件 __init__

上文中的 帮助 界面可以自己写哦, 模板如下如所示, 添加到目标插件 __init__.py 的代码段里

from nonebot.plugin import PluginMetadata

__plugin_meta__ = PluginMetadata(
    name="插件名称",
    description="描述",
    usage=(
        "提示内容"
    ),
    extra={
        # 额外信息
        "unique_name": "",
        "example": "操作帮助",
        "author": "作者",
        "version": "版本",
    },
)

3.在 __init__.py 文件中粘贴先行检验过的代码

如图就是自定义插件正常运行 SUCCESS

效果如下

爬虫大佬可以自行修改代码, 变成爬取任意网站的比赛信息, 或者是题目信息
所谓 自定义插件 , 肯定是要有一些基础才能自己做出来的, 学习 大大们的插件, 转化成自己的 知识,
也是追求知识和未知的少年们应该为之奋斗的. 所以博主这个 小萌新 就不再说话了, 毕竟小伙伴们有磨砺才能越走越远

结尾

1.总结

暑假放假事情比较多, 博主为了生计奔波劳累, 为了竞赛算法操碎了心, 为心爱的学弟学妹们的熬尽心血, 对大学好朋友们日夜思念, 鲜有时间静下心来思考和总结… 这个是博主在六月出就启动的 空头计划 , 后来有了博主好同志的一些启发, 真正完善了一下内容, 中间部分瑕疵请各位小伙伴们评论区谈论

2.下期目标

网站后端 + 数据库的一些问题, 或者, 比如提速百度网盘, 最强的截图软件, Codeforces比赛助手别的干货教学 , 可能吧, 毕竟博主还得留着肝娶老婆呢, 所以, 真的不是博主托更啊! 这就是原滋原味的 空头计划