QQ机器人开发快速入门（Python）

实验简介

频道机器人简介

频道机器人是基于QQ开放生态的高级扩展服务，通过开放的接口，能够与QQ频道用户实现交互形式丰富的互动。

实验目的

该教程主要是面向新接触QQ频道机器人的开发者，通过教程可以学习到如何通过Python的官方SDK实现一些机器人的基本功能。

实验条件

请先 完成机器人注册和添加，获取机器人相关的信息。相关教程可点击查看 QQ机器人快速注册指南

实验准备

相关概念

机器人SDK： 提供开发者使用的基于OpenAPI的官方SDK，优势主要在于服务稳定性及维护频率高

机器人AppID： 注册机器人后系统分配的唯一ID标识，在完成机器人注册和添加的教程可以获取

机器人Token： 注册机器人后使用OpenAPI系统分配的密钥，在完成机器人注册和添加的教程可以获取，请注意不要外泄。

环境搭建

安装Python3

推荐使用Python3，实验环境已经预安装，可执行下面命令，进行Python版本验证

python3 --version

安装机器人SDK

在终端执行下面命令安装机器人PythonSDK：

pip install qq-bot

同时，由于需要读取 yaml 文件的内容，我们也需要安装 pyyaml

pip install pyyaml

创建项目文件

创建一个 demo 项目文件夹

mkdir /home/demo && cd /home/demo

在 demo 文件夹下创建名为 config.yaml 的配置文件

touch config.yaml

接着，在 demo 文件夹下创建一个名为 robot.py 的文件：

touch robot.py

导入Token 和 AppID

请点击打开 config.yaml 文件，并填入自己机器人的 AppID 和 Token ，注意保存

：

token:
  appid: "123"
  token: "xxxx"

导入依赖包

请点击打开 robot.py ，并在文件中复制导入相关依赖包的代码，注意保存

：

robot.py

import asyncio
import json
import os.path
import threading
from typing import Dict, List

import aiohttp
import qqbot

from qqbot.core.util.yaml_util import YamlUtil
from qqbot.model.message import MessageEmbed, MessageEmbedField, MessageEmbedThumbnail, CreateDirectMessageRequest, \
    MessageArk, MessageArkKv, MessageArkObj, MessageArkObjKv

test_config = YamlUtil.read(os.path.join(os.path.dirname(__file__), "config.yaml"))

设置机器人自动回复普通消息

在 robot.py 文件中添加如下代码，注意保存

：

robot.py

async def _message_handler(event, message: qqbot.Message):
    """
    定义事件回调的处理
    :param event: 事件类型
    :param message: 事件对象（如监听消息是Message对象）
    """
    msg_api = qqbot.AsyncMessageAPI(t_token, False)
    # 打印返回信息
    qqbot.logger.info("event %s" % event + ",receive message %s" % message.content)

    # 发送消息告知用户
    message_to_send = qqbot.MessageSendRequest(content="你好", msg_id=message.id)
    await msg_api.post_message(message.channel_id, message_to_send)


# async的异步接口的使用示例
if __name__ == "__main__":
    t_token = qqbot.Token(test_config["token"]["appid"], test_config["token"]["token"])
    # @机器人后推送被动消息
    qqbot_handler = qqbot.Handler(
        qqbot.HandlerType.AT_MESSAGE_EVENT_HANDLER, _message_handler
    )
    qqbot.async_listen_events(t_token, False, qqbot_handler)

代码运行

在终端命令行输入并执行下列命令，运行机器人

python3 /home/demo/robot.py

这时在频道内 @机器人 hello 指令就可以收到回复了

获取天气数据

首先，在 robot.py 中添加用于获取天气数据的函数，注意保存

：

robot.py

async def get_weather(city_name: str) -> Dict:
    """
    获取天气信息
    :return: 返回天气数据的json对象
    """
    weather_api_url = "http://api.k780.com/?app=weather.today&cityNm=" + city_name + "&appkey=10003&sign=b59bc3ef6191eb9f747dd4e83c99f2a4&format=json"
    async with aiohttp.ClientSession() as session:
        async with session.get(
                url=weather_api_url,
                timeout=5,
        ) as resp:
            content = await resp.text()
            content_json_obj = json.loads(content)
            return content_json_obj

代码说明

上面链接的天气API使用的sign=b59bc3ef6191eb9f747dd4e83c99f2a4可能会过期，如下图指引

请前往天气API地址查看最新的测试sign并替换，或注册账号申请一个免费sign

修改 _message_handler 方法

在 message_handler 方法中，加入调用 get_weather 函数并发送天气的代码。完整 message_handler 的实现如下：

robot.py

async def _message_handler(event, message: qqbot.Message):
    """
    定义事件回调的处理
    :param event: 事件类型
    :param message: 事件对象（如监听消息是Message对象）
    """
    msg_api = qqbot.AsyncMessageAPI(t_token, False)
    # 打印返回信息
    qqbot.logger.info("event %s" % event + ",receive message %s" % message.content)
    # 获取天气数据并发送消息告知用户
    weather_dict = await get_weather("深圳")
    weather_desc = weather_dict['result']['citynm'] + " " \
        + weather_dict['result']['weather'] + " " \
        + weather_dict['result']['days'] + " " \
        + weather_dict['result']['week']
    message_to_send = qqbot.MessageSendRequest(msg_id=message.id, content=weather_desc, image=weather_dict['result']['weather_icon'])
    await msg_api.post_message(message.channel_id, message_to_send)

代码运行

在终端命令行输入并执行下列命令，运行机器人

python3 /home/demo/robot.py

python3 /home/demo/robot.py

效果图如下：

import asyncio
import json
import os.path
import threading
from typing import Dict, List

import aiohttp
import qqbot

from qqbot.core.util.yaml_util import YamlUtil
from qqbot.model.message import MessageEmbed, MessageEmbedField, MessageEmbedThumbnail, CreateDirectMessageRequest, \
    MessageArk, MessageArkKv, MessageArkObj, MessageArkObjKv

test_config = YamlUtil.read(os.path.join(os.path.dirname(__file__), "config.yaml"))

async def _message_handler(event, message: qqbot.Message):
    """
    定义事件回调的处理
    :param event: 事件类型
    :param message: 事件对象（如监听消息是Message对象）
    """
    msg_api = qqbot.AsyncMessageAPI(t_token, False)
    # 打印返回信息
    qqbot.logger.info("event %s" % event + ",receive message %s" % message.content)
    # 获取天气数据并发送消息告知用户
    weather_dict = await get_weather("北京")
    weather_desc = weather_dict['result']['citynm'] + " " \
        + weather_dict['result']['weather'] + " " \
        + weather_dict['result']['days'] + " " \
        + weather_dict['result']['week']
    message_to_send = qqbot.MessageSendRequest(msg_id=message.id, content=weather_desc, image=weather_dict['result']['weather_icon'])
    await msg_api.post_message(message.channel_id, message_to_send)

async def get_weather(city_name: str) -> Dict:
    """
    获取天气信息
    :return: 返回天气数据的json对象
    """
    weather_api_url = "http://api.k780.com/?app=weather.today&cityNm=" + city_name + "&appkey=65849&sign=a54c59baa7a0c590cbf17d699be41b1d&format=json"
    async with aiohttp.ClientSession() as session:
        async with session.get(
                url=weather_api_url,
                timeout=5,
        ) as resp:
            content = await resp.text()
            content_json_obj = json.loads(content)
            return content_json_obj


# async的异步接口的使用示例
if __name__ == "__main__":
    t_token = qqbot.Token(test_config["token"]["appid"], test_config["token"]["token"])
    # @机器人后推送被动消息
    qqbot_handler = qqbot.Handler(
        qqbot.HandlerType.AT_MESSAGE_EVENT_HANDLER, _message_handler
    )
    qqbot.async_listen_events(t_token, False, qqbot_handler)


   await msg_api.post_message(message.channel_id, message_to_send)

设置机器人主动推送消息

上面的教程只实现一个简单的获取天气的功能，但是我们做的是天气机器人，希望实现一个报告天气的功能。一般的天气应用都会在一个特定时间给你推送天气通知，在频道机器人中，你可以通过主动消息来实现这个功能。

在 robot.py 中添加定时发送消息的函数，代码如下：

robot.py

async def send_weather_message_by_time():
    """
    任务描述：每天推送一次普通天气消息（演示方便改为100s定时运行）
    """
    # 获取天气数据
    weather_dict = await get_weather("深圳")
    # 获取频道列表都取首个频道的首个子频道推送
    user_api = qqbot.AsyncUserAPI(t_token, False)
    guilds = await user_api.me_guilds()
    guilds_id = guilds[0].id
    channel_api = qqbot.AsyncChannelAPI(t_token, False)
    channels = await channel_api.get_channels(guilds_id)
    channels_id = channels[0].id
    qqbot.logger.info("channelid %s" % channel_id)
    # 推送消息
    weather = "当前天气是：" + weather_dict['result']['weather']
    send = qqbot.MessageSendRequest(content=weather)
    msg_api = qqbot.AsyncMessageAPI(t_token, False)
    await msg_api.post_message(channels_id, send)
    # 如果需要每天都执行，加上下面两句
    t = threading.Timer(100, await send_weather_message_by_time)
    t.start()

在*main*中添加执行send_weather_message_by_time()的语句：

 # 定时推送主动消息
 send_weather_message_by_time()

编写完毕，注意保存

设置机器人私信

我们希望能提供不同用户不同地方的天气，但是发太多的消息会影响其它的用户。针对这种情况，我们可以通过私信来实现。下面函数中，当我们@机器人hello时收到机器人的私信。

私信中我们不使用ark，而是使用Embed。Embed也是一种结构化消息，它比Ark简单

在 robot.py 添加发送Embed的函数如下：

robot.py

async def send_weather_embed_direct_message(weather_dict, guild_id, user_id):
    """
    被动回复-私信推送天气内嵌消息
    :param user_id: 用户ID
    :param weather_dict: 天气数据字典
    :param guild_id: 发送私信需要的源频道ID
    """
    # 构造消息发送请求数据对象
    embed = MessageEmbed()
    embed.title = weather_dict['result']['citynm'] + " " + weather_dict['result']['weather']
    embed.prompt = "天气消息推送"
    # 构造内嵌消息缩略图
    thumbnail = MessageEmbedThumbnail()
    thumbnail.url = weather_dict['result']['weather_icon']
    embed.thumbnail = thumbnail
    # 构造内嵌消息fields
    embed.fields = [MessageEmbedField(name="当日温度区间：" + weather_dict['result']['temperature']),
                    MessageEmbedField(name="当前温度：" + weather_dict['result']['temperature_curr']),
                    MessageEmbedField(name="最高温度：" + weather_dict['result']['temp_high']),
                    MessageEmbedField(name="最低温度：" + weather_dict['result']['temp_low']),
                    MessageEmbedField(name="当前湿度：" + weather_dict['result']['humidity'])]

    # 通过api发送回复消息
    send = qqbot.MessageSendRequest(embed=embed, content="")
    dms_api = qqbot.AsyncDmsAPI(t_token, False)
    direct_message_guild = await dms_api.create_direct_message(CreateDirectMessageRequest(guild_id, user_id))
    await dms_api.post_direct_message(direct_message_guild.guild_id, send)
    qqbot.logger.info("/私信推送天气内嵌消息 成功")

在_message_handler中调用刚刚添加的函数，使机器人是在私信里给你发送Embed

robot.py

elif "/私信天气" in content:
     # 通过空格区分城市参数
     split = content.split("/私信天气 ")
     weather = await get_weather(split[1])
     await send_weather_embed_direct_message(weather, message.guild_id, message.author.id)

编写完毕，注意保存

在终端命令行输入并执行下列命令，运行机器人

python3 /home/demo/robot.py

在频道中执行下列步骤验证效果：

@机器人后输入“/私信天气城市名”执行
等待几分钟后，到私信面板看看是否有机器人推送过来的天气消息。

01DDC2277EE8A0EE699C8049E38806A7

使用小程序

当用户想要查看全国或者某个省份的天气情况，一次次@机器人就显得十分麻烦，这个时候你可以使用小程序来解决这个问题。了解具体的小程序开发可以看 QQ小程序开发文档，这里只介绍如何通过机器人打开小程序。

机器人打开小程序非常简单，只需要按照下面配置就可以了，不需要增加额外的代码：

4457999744624a70a54a27e919452c89

d29a829800fa4e1f97df42c9da485ccf

配置好后，我们@机器人就可以看到我们设置的服务了，点击就可以打开设置的小程序

企业微信截图4065b3d1f4fa436686ac59d98bebcf09

使用指令

每次@机器人输入指令太麻烦了，有没有简单的方式呢？机器人提供了指令配置，当你输入/时就会产出你配置的指令面板。

配置方式如下：

4457999744624a70a54a27e919452c89 ac4d256d3e314e189fd25a87e4d550a3

配置好后，当我们输入/时，就可以看到配置的面板了

150f1d0da34b4c19ac148dea8afe6171

课后习题

上面已经叙述了机器人的各种功能，下面你可以在这基础上尝试更完整的功能：

请思考

机器人通过天气api拉取默认城市（深圳）的天气，每天主动推送模版消息
机器人通过指令选择“/天气“，输入城市名后，被动推送天气的模版消息
机器人通过指令选择“/私信天气”时，输入城市名后，被动推送私信的天气内嵌消息（建议改成注册需要推送消息）
机器人通过指令选择“天气小程序”，打开天气小程序
机器人通过指令选择下面这些时，推送不同的消息
- /当前天气城市名
- /未来天气城市名
- /穿衣指数城市名
- /出行指数城市名
- /空气质量城市名

完整代码可查看github：天气机器人-Python实现版