编辑
2024-03-03
技术资料
0
请注意,本文编写于 78 天前,最后修改于 78 天前,其中某些信息可能已经过时。

目录

频道机器人简介
实验目的
实验条件
相关概念
环境搭建
安装机器人SDK
创建项目文件
导入依赖包
设置机器人自动回复普通消息
代码运行
获取天气数据
代码说明
代码运行
设置机器人主动推送消息
设置机器人私信
使用小程序

频道机器人简介

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

实验目的

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

实验条件

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

相关概念

机器人SDK: 提供开发者使用的基于OpenAPI的官方SDK,优势主要在于服务稳定性及维护频率高 机器人AppID: 注册机器人后系统分配的唯一ID标识,在完成机器人注册和添加的教程可以获取 机器人Token: 注册机器人后使用OpenAPI系统分配的密钥,在完成机器人注册和添加的教程可以获取,请注意不要外泄。

环境搭建

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

python
python3 --version

安装机器人SDK

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

python
pip install qq-bot

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

python
pip install pyyaml

创建项目文件

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

python
touch config.yaml

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

python
touch robot.py

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

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

导入依赖包

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

robot.py

python
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

python
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)

代码运行

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

python
python3 /home/demo/robot.py

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

获取天气数据

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

robot.py

python
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 可能会过期,如下图指引 image.png 请前往天气API地址查看最新的测试sign并替换,或注册账号申请一个免费sign

image.png

修改 _message_handler 方法 在 _message_handler 方法中,加入调用 get_weather 函数并发送天气的代码。完整 _message_handler 的实现如下:

robot.py

python
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)

代码运行

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

ptyhon
python3 /home/demo/robot.py

效果图如下:

ptyhon
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

ptyhon
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

python
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

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

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

image.png

使用小程序

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

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

image.png

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

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

配置方式如下:

image.png

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

image.png

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

python
#!/usr/bin/env python3 # -*- coding: utf-8 -*- import asyncio import json import os.path import time from multiprocessing import Process from typing import Dict, List import aiohttp import qqbot import schedule 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")) public_channel_id = "" async def _message_handler(event, message: qqbot.Message): """ 定义事件回调的处理 :param event: 事件类型 :param message: 事件对象(如监听消息是Message对象) """ msg_api = qqbot.AsyncMessageAPI(t_token, False) # 打印返回信息 content = message.content qqbot.logger.info("event %s" % event + ",receive message %s" % content) # 根据指令触发不同的推送消息 if "/天气 " in content: split = content.split("/天气 ") weather = await get_weather(split[1]) await send_weather_ark_message(weather, message.channel_id, message.id) 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) if "/当前天气 " in content: split = content.split("/当前天气 ") weather = await get_weather(split[1]) await send_weather_ark_message(weather, message.channel_id, message.id) elif "/未来天气 " in content: split = content.split("/未来天气 ") future_weather = await get_future_weather(split[1]) await send_future_weather_ark_message(future_weather, message.channel_id, message.id) elif "/空气质量 " in content: split = content.split("/空气质量 ") aqi_dict = await get_aqi(split[1]) await send_aqi_ark_message(aqi_dict, message.channel_id, message.id) elif "/穿衣指数 " in content: split = content.split("/穿衣指数 ") weather_life_dict = await get_weather_life_index(split[1]) await send_clothes_ark_message(weather_life_dict, message.channel_id, message.id) elif "/紫外线指数 " in content: split = content.split("/紫外线指数 ") weather_life_dict = await get_weather_life_index(split[1].strip()) await send_uv_ark_message(weather_life_dict, message.channel_id, message.id) async def _create_weather_ark_obj_list(weather_dict) -> List[MessageArkObj]: obj_list = [MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value=weather_dict['result']['citynm'] + " " + weather_dict['result']['weather'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="当日温度区间:" + weather_dict['result']['temperature'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="当前温度:" + weather_dict['result']['temperature_curr'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="当前湿度:" + weather_dict['result']['humidity'])])] return obj_list async def _create_future_weather_ark_obj_list(weather_dict) -> List[MessageArkObj]: obj_list = [MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value=weather_dict['result'][0]['citynm'] + "未来三天天气预报")]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="明天:" + weather_dict['result'][1]['weather'] + ", " + weather_dict['result'][1]['temperature'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="后天:" + weather_dict['result'][2]['weather'] + ", " + weather_dict['result'][2]['temperature'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="外后天:" + weather_dict['result'][3]['weather'] + ", " + weather_dict['result'][3]['temperature'])])] return obj_list async def _create_clothes_ark_obj_list(life_index_dic) -> List[MessageArkObj]: obj_list = [MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="城市:" + life_index_dic['result'][0]['citynm'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="体感:" + life_index_dic['result'][0]['lifeindex_ct_attr'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="建议:" + life_index_dic['result'][0]['lifeindex_ct_dese'])])] return obj_list async def _create_uv_ark_obj_list(life_index_dic) -> List[MessageArkObj]: obj_list = [MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="城市:" + life_index_dic['result'][0]['citynm'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="紫外线指数:" + life_index_dic['result'][0]['lifeindex_uv_attr'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="建议:" + life_index_dic['result'][0]['lifeindex_uv_dese'])])] return obj_list async def _create_aqi_ark_obj_list(aqi_dict) -> List[MessageArkObj]: obj_list = [MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="城市:" + aqi_dict['result']['citynm'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="空气质量:" + aqi_dict['result']['aqi_levnm'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="PM2.5:" + aqi_dict['result']['aqi_scope'])]), MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value="建议:" + aqi_dict['result']['aqi_remark'])])] return obj_list async def send_weather_ark_message(weather_dict, channel_id, message_id): """ 被动回复-子频道推送模版消息 :param channel_id: 回复消息的子频道ID :param message_id: 回复消息ID :param weather_dict:天气消息 """ # 构造消息发送请求数据对象 ark = MessageArk() # 模版ID=23 ark.template_id = 23 ark.kv = [MessageArkKv(key="#DESC#", value="描述"), MessageArkKv(key="#PROMPT#", value="提示消息"), MessageArkKv(key="#LIST#", obj=await _create_weather_ark_obj_list(weather_dict))] # 通过api发送回复消息 send = qqbot.MessageSendRequest(content="", ark=ark, msg_id=message_id) msg_api = qqbot.AsyncMessageAPI(t_token, False) await msg_api.post_message(channel_id, send) 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("/私信推送天气内嵌消息 成功") async def send_clothes_ark_message(life_index_dict, channel_id, message_id): """ 被动回复-子频道推送穿衣指数 :param channel_id: 回复消息的子频道ID :param message_id: 回复消息ID :param life_index_dict:天气消息 """ # 构造消息发送请求数据对象 ark = MessageArk() # 模版ID=23 ark.template_id = 23 ark.kv = [MessageArkKv(key="#DESC#", value="描述"), MessageArkKv(key="#PROMPT#", value="提示消息"), MessageArkKv(key="#LIST#", obj=await _create_clothes_ark_obj_list(life_index_dict))] # 通过api发送回复消息 send = qqbot.MessageSendRequest(content="", ark=ark, msg_id=message_id) msg_api = qqbot.AsyncMessageAPI(t_token, False) await msg_api.post_message(channel_id, send) async def send_uv_ark_message(life_index_dict, channel_id, message_id): """ 被动回复-子频道推送紫外线指数 :param channel_id: 回复消息的子频道ID :param message_id: 回复消息ID :param life_index_dict:天气消息 """ # 构造消息发送请求数据对象 ark = MessageArk() # 模版ID=23 ark.template_id = 23 ark.kv = [MessageArkKv(key="#DESC#", value="描述"), MessageArkKv(key="#PROMPT#", value="提示消息"), MessageArkKv(key="#LIST#", obj=await _create_uv_ark_obj_list(life_index_dict))] # 通过api发送回复消息 send = qqbot.MessageSendRequest(content="", ark=ark, msg_id=message_id) msg_api = qqbot.AsyncMessageAPI(t_token, False) await msg_api.post_message(channel_id, send) async def send_aqi_ark_message(aqi_dict, channel_id, message_id): """ 被动回复-子频道推送 PM2.5 空气质量指数 :param channel_id: 回复消息的子频道ID :param message_id: 回复消息ID :param aqi_dict:空气质量数据 """ # 构造消息发送请求数据对象 ark = MessageArk() # 模版ID=23 ark.template_id = 23 ark.kv = [MessageArkKv(key="#DESC#", value="描述"), MessageArkKv(key="#PROMPT#", value="提示消息"), MessageArkKv(key="#LIST#", obj=await _create_aqi_ark_obj_list(aqi_dict))] # 通过api发送回复消息 send = qqbot.MessageSendRequest(content="", ark=ark, msg_id=message_id) msg_api = qqbot.AsyncMessageAPI(t_token, False) await msg_api.post_message(channel_id, send) async def send_future_weather_ark_message(future_weather_dict, channel_id, message_id): """ 被动回复-子频道推送未来三天天气 :param channel_id: 回复消息的子频道ID :param message_id: 回复消息ID :param future_weather_dict:空气质量数据 """ # 构造消息发送请求数据对象 ark = MessageArk() # 模版ID=23 ark.template_id = 23 ark.kv = [MessageArkKv(key="#DESC#", value="描述"), MessageArkKv(key="#PROMPT#", value="提示消息"), MessageArkKv(key="#LIST#", obj=await _create_future_weather_ark_obj_list(future_weather_dict))] # 通过api发送回复消息 send = qqbot.MessageSendRequest(content="", ark=ark, msg_id=message_id) msg_api = qqbot.AsyncMessageAPI(t_token, False) await msg_api.post_message(channel_id, send) async def get_weather(city_name: str) -> Dict: """ 获取天气信息 :return: 返回天气数据的json对象 返回示例 { "success":"1", "result":{ "weaid":"1", "days":"2022-03-04", "week":"星期五", "cityno":"beijing", "citynm":"北京", "cityid":"101010100", "temperature":"13℃/-1℃", "temperature_curr":"10℃", "humidity":"17%", "aqi":"98", "weather":"扬沙转晴", "weather_curr":"扬沙", "weather_icon":"http://api.k780.com/upload/weather/d/30.gif", "weather_icon1":"", "wind":"西北风", "winp":"4级", "temp_high":"13", "temp_low":"-1", "temp_curr":"10", "humi_high":"0", "humi_low":"0", "weatid":"31", "weatid1":"", "windid":"7", "winpid":"4", "weather_iconid":"30" } } """ weather_api_url = "http://api.k780.com/?app=weather.today&cityNm=" + city_name + "&appkey=00000&sign=wdnmdzijiquhuaqianshenqing&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 def get_future_weather(city_name: str) -> Dict: """ 获取未来几天的天气信息 :return: 返回天气数据的json对象 返回示例(返回值过长,部分省略) { "success": "1", "result": [{ "weaid": "1", "days": "2014-07-30", "week": "星期三", "cityno": "beijing", "citynm": "北京", "cityid": "101010100", "temperature": "23℃/11℃", /*温度*/ "humidity": "0%/0%", /*湿度,后期气像局未提供,如有需要可使用weather.today接口 */ "weather": "多云转晴", "weather_icon": "http://api.k780.com/upload/weather/d/1.gif", /*气象图标(白天) 全部气象图标下载*/ "weather_icon1": "http://api.k780.com/upload/weather/d/0.gif", /*气象图标(夜间) 全部气象图标下载*/ "wind": "微风", /*风向*/ "winp": "小于3级", /*风力*/ "temp_high": "31", /*最高温度*/ "temp_low": "24", /*最低温度*/ "humi_high": "0", /*湿度栏位已不再更新*/ "humi_low": "0",/*湿度栏位已不再更新*/ "weatid": "2", /*白天天气ID,可对照weather.wtype接口中weaid*/ "weatid1": "1", /*夜间天气ID,可对照weather.wtype接口中weaid*/ "windid": "1", /*风向ID(暂无对照表)*/ "winpid": "2" /*风力ID(暂无对照表)*/ "weather_iconid": "1", /*气象图标编号(白天),对应weather_icon 1.gif*/ "weather_iconid1": "0" /*气象图标编号(夜间),对应weather_icon1 0.gif*/ }, ...... """ weather_api_url = "http://api.k780.com/?app=weather.future&cityNm=" + city_name + "&appkey=00000&sign=nmsl&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 def get_weather_life_index(citi_name: str) -> Dict: """ 获取生活指数 :return: 返回天气数据的json对象 返回示例 { success: "1", result: { 2017-04-17: { weaid: "1", days: "2017-04-17", week_1: "星期一", simcode: "beijing", citynm: "北京", cityid: "101010100", lifeindex_uv_id: "101", lifeindex_uv_typeno: "uv", lifeindex_uv_typenm: "紫外线指数", lifeindex_uv_attr: "弱", lifeindex_uv_dese: "辐射较弱,涂擦SPF12-15、PA+护肤品。", lifeindex_gm_id: "111", lifeindex_gm_typeno: "gm", lifeindex_gm_typenm: "感冒指数", lifeindex_gm_attr: "少发", lifeindex_gm_dese: "无明显降温,感冒机率较低。", lifeindex_ct_id: "108", lifeindex_ct_typeno: "ct", lifeindex_ct_typenm: "穿衣指数", lifeindex_ct_attr: "较舒适", lifeindex_ct_dese: "建议穿薄外套或牛仔裤等服装。", lifeindex_xc_id: "112", lifeindex_xc_typeno: "xc", lifeindex_xc_typenm: "洗车指数", lifeindex_xc_attr: "较适宜", lifeindex_xc_dese: "无雨且风力较小,易保持清洁度。", lifeindex_yd_id: "114", lifeindex_yd_typeno: "yd", lifeindex_yd_typenm: "运动指数", lifeindex_yd_attr: "较适宜", lifeindex_yd_dese: "风力稍强,推荐您进行室内运动。", lifeindex_kq_id: "109", lifeindex_kq_typeno: "kq", lifeindex_kq_typenm: "空气污染扩散指数", lifeindex_kq_attr: "良", lifeindex_kq_dese: "气象条件有利于空气污染物扩散。" }, ... """ weather_api_url = "http://api.k780.com/?app=weather.lifeindex&cityNm=" + citi_name + "&appkey=00000&sign=0nmsl&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 def get_aqi(citi_name: str) -> Dict: """ 获取空气质量(aqi)数据 :return: 返回空气质量数据的json对象 返回示例 { success: "1", result: { "success": "1", "result": { "weaid": "180", "cityno": "gdzhongshan", "citynm": "中山", "cityid": "101281701", "aqi": "18", "aqi_scope": "0-50", "aqi_levid": "1", "aqi_levnm": "优", "aqi_remark": "参加户外活动呼吸清新空气" } """ weather_api_url = "http://api.k780.com/?app=weather.pm25&cityNm=" + citi_name + "&appkey=0000&sign=nmsl&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 def set_schedule_task(): schedule.every(10).seconds.do(send_weather_message_by_time) while True: schedule.run_pending() time.sleep(1) def send_weather_message_by_time(): """ 任务描述:每天推送一次普通天气消息 """ loop = asyncio.get_event_loop() token = qqbot.Token(test_config["token"]["appid"], test_config["token"]["token"]) # 获取频道列表,取首个频道的首个子频道推送 global public_channel_id if not public_channel_id: user_api = qqbot.AsyncUserAPI(token, False) guild_id = loop.run_until_complete(user_api.me_guilds())[0].id channel_api = qqbot.AsyncChannelAPI(token, False) public_channel_id = loop.run_until_complete(channel_api.get_channels(guild_id))[0].id # 获取天气数据 weather_dict = loop.run_until_complete(get_weather("深圳")) # 推送消息 content = "当日温度区间:" + weather_dict['result']['temperature'] send = qqbot.MessageSendRequest(content=content) msg_api = qqbot.AsyncMessageAPI(token, False) loop.run_until_complete(msg_api.post_message("2568610", send)) # async的异步接口的使用示例 if __name__ == "__main__": # 定时推送主动消息 Process(target=set_schedule_task).start() # @机器人后推送被动消息 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)

原文链接:https://cloud.tencent.com/developer/article/2068473