【Python】【QQ频道】部署一个天气查询机器人
本文最后更新于 2024-02-22,文章内容可能已经过时
1. 开发前的准备
如下图,点击 频道机器人开发官网 ,在官网页面点击 立即注册
有两种主体类型,可以根据自己的实际情况进行选择,这里介绍个人开发者的流程,企业详细流程见 企业主体入驻
提交完成后,需要等待审核。如果审核通过,就会发送邮件到你的邮箱。
在完成邮箱、手机号等认证后,就可以进入 QQ机器人管理界面,如下图所示:
点击 生成BotAppID 进入机器人配置界面(如下图):
这里面有两个选项需要注意(标红部分):沙箱频道ID 是指你创建的频道的ID,需要注意的是,如果你之前创建的频道人数超过限制,就需要创建另一个频道;机器人类型有两种,一种是私域机器人,一种是公域机器人。简单来说,私域机器人只能在你自己的频道使用,而公域机器人可以在所有频道使用。
点击 提交审核 ,审核完成后就能看到如下界面:
点击 查看详情 就可以看到你的 BotAppID
、BotToken
、BotSecret
了,注意这个信息不要泄露。
现在,点击频道右上角「…」—>点击「频道设置」—>点击「机器人」—>添加测试机器人,就可以将机器人添加到自己的频道了。不过此时机器人还没有任何的功能,下面手把手教你用 python 写一个机器人服务
2. 环境搭建
安装 Python3
linux
在命令行输入 python --version
查看是否已经安装过 Python3
。如果像下面一样,显示的版本为 Python 3.x.x
,则请跳过安装环节。
python3
Python 3.9.10
下面开始安装 Python3
:
首先安装一些必要的依赖包
yum -y install zlib-devel bzip2-devel openssl-devel ncurses-devel sqlite-devel readline-devel tk-devel gdbm-devel libpcap-devel xz-devel libffi-devel gcc
接着,下载 Python3
安装包,并解压缩安装包
wget https://www.python.org/ftp/python/3.10.0/Python-3.10.0.tgz
tar -zxvf Python-3.10.0.tgz
然后,指定 Python3
安装路径
cd Python-3.7.1
./configure --prefix=/root/python37
接着,执行安装 Python3
make
make install
然后,为 Python3
和 pip3
添加软链接。软链接类似于 windows 的快捷方式,当你在终端输入 python3
时会使用你指定的 python
地址
ln -s /root/python310/bin/python3.10 /usr/bin/python3
ln -s /root/python310/bin/pip3 /usr/bin/pip3
注意: 这里的 /root/python37/
是我的 Python3
安装路径,和之前下载的安装包放在同一个位置。运行前请确认一下你的安装路径是否和我一样。
最后,在命令行输入 python3 --version
指令检验是否安装完成,如果安装成功,会打印出 python
的版本号
python3 --version
mac
先打开 Terminal,安装 Homebrew。(可先用 brew -v
查看是否已经安装 Homebrew)
/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/master/install.sh)"
安装 Homebrew 后,将 Homebrew 目录插入到 PATH
环境变量顶部。你可以通过在 ~/.profile
文件底部添加以下行来执行此操作
export PATH="/usr/local/opt/python/libexec/bin:$PATH"
现在,可以用 Homebrew 安装 Python3
了
brew install python
在命令行输入 python --version 指令检验是否安装完成,如果安装成功,会打印出 python 的版本号
python --version
windows
在 https://www.python.org/downloads/windows/ 下载 executable installer (一般选 64-bit 的),下载完成后,运行 exe 安装包即可
注意: 记得勾选 Add Python 3.x to Path
选项
安装成功后,打开命令提示符窗口,输入 python
,如果安装成功,会打印出 python
的版本号
C:\> python
安装机器人 SDK
还未安装过机器人 SDK 的同学请运行:
pip install qq-bot
已经安装过机器人 SDK 的同学请运行:(本 demo 要求 SDK 版本大于 v0.7.4)
pip install qq-bot --upgrade
同时,由于需要读取 yaml
文件的内容,我们也需要安装 pyyaml
pip install pyyaml
创建项目
创建一个 demo 项目文件夹
mkdir demo
cd demo
接着,在 demo
文件夹下创建名为 config.yaml
的配置文件,填入自己的 BotAppID
和 Bot token
,内容类似下面所示。 也可直接下载 github 仓库里的 config.example.yaml
文件,然后自己修改后缀名和内容
token:
appid: "123"
token: "xxxx"
接着,在 demo
文件夹下创建一个名为 robot.py
的文件:
- 在Linux和mac上,你需要使用
touch robot.py
创建一个名为robot.py
的文件。 - 在windows上,你可以右键–>创建txt文件–>重命名为
robot.py
最后,打开 robot.py
文件,在开头导入相关的包:
在Linux和mac上你需要使用 vim robot.py
编辑 robot.py
文件,键盘输入 i
,把文件变成可编辑状态,复制粘贴下面代码。esc
键退出,键盘输入 :wq
保持退出。
在windows上,你需要使用文本编辑器打开文件,并复制粘贴下面的代码,ctrl+s
保存文件。
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"))
3. 机器人自动回复普通消息
在 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 robot.py
运行机器人。 这时在频道内 @机器人 hello
指令就可以收到回复了
4. 获取天气数据
天气机器人最重要的就是提供天气的数据,这里是使用的 https://www.nowapi.com/api/weather.today
的Api。
首先,在 robot.py
中添加用于获取天气数据的函数
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=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
然后,修改 _message_handler
函数,加入如下代码调用 get_weather
函数并发送天气
# 获取天气数据并发送消息告知用户
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)
效果图如下:
5. 机器人主动推送消息
上面的教程只实现一个简单的获取天气的功能,但是我们做的是天气机器人,希望实现一个报告天气的功能。一般的天气应用都会在一个特定时间给你推送天气通知,在频道机器人中,你可以通过主动消息来实现这个功能。代码如下:
在 robot.py
中定义一个全局变量,用于记录定时推送消息的子频道 ID,并添加定时发送消息的函数
public_channel_id = ""
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))
在 __main__
中添加如下语句,执行 send_weather_message_by_time()
# 定时推送主动消息
Process(target=set_schedule_task).start()
运行该代码,效果如下图
6. 机器人指令回复ark消息
提供给个人开发者的 Ark
有3种,这里使用 24 号 Ark
。其它 Ark
见消息模板
先在 robot.py
中添加发送ark的函数
async def _create_ark_obj_list(weather_dict) -> List[MessageArkObj]:
obj_list = [MessageArkObj(obj_kv=[MessageArkObjKv(key="desc", value=weather_dict['result']['citynm'] + " " + weather_dict['result']['weather'] + " " + weather_dict['result']['days'] + " " + weather_dict['result']['week'])]),
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 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_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)
再修改 _message_handler
函数发送ark
# 根据指令触发不同的推送消息
content = message.content
if "/天气" in content:
# 通过空格区分城市参数
split = content.split("/天气 ")
weather = await get_weather(split[1])
await send_weather_ark_message(weather, message.channel_id, message.id)
效果如下图:
7. 机器人私信
我们希望能提供不同用户不同地方的天气,但是发太多的消息会影响其它的用户。针对这种情况,我们可以通过私信来实现。下面函数中,当我们@机器人hello时收到机器人的私信。
私信中我们不使用ark,而是使用 Embed
。Embed
也是一种结构化消息,它比Ark简单,发送 Embed
的函数如下:
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
content = message.content
if "/私信天气" in content:
# 通过空格区分城市参数
split = content.split("/私信天气 ")
weather = await get_weather(split[1])
await send_weather_embed_direct_message(weather, message.guild_id, message.author.id)
效果图如下:
8. 使用小程序
当用户想要查看全国或者某个省份的天气情况,一次次@机器人就显得十分麻烦,这个时候你可以使用小程序来解决这个问题。了解具体的小程序开发可以看QQ小程序开发文档,这里只介绍如何通过机器人打开小程序。
机器人打开小程序非常简单,只需要按照下面配置就可以了,不需要增加额外的代码:
配置好后,我们@机器人就可以看到我们设置的服务了,点击就可以打开设置的小程序
9. 使用指令
每次@机器人输入指令太麻烦了,有没有简单的方式呢?机器人提供了指令配置,当你输入 /
时就会产出你配置的指令面板。配置方式如下:
配置好后,当我们输入 /
时,就可以看到配置的面板了
需要注意,点击指令后输入的内容增加了一个
/
,上面的例子就变成了@天气机器人-测试中 /天气
10. 最佳实践
上面已经叙述了机器人的各种功能,下面把这些功能都整合起来:
- 机器人通过天气api拉取默认城市(深圳)的天气,每天主动推送模版消息
- 机器人通过指令选择“/天气“,输入城市名后,被动推送天气的模版消息
- 机器人通过指令选择“/私信天气”时,被动推送私信的天气内嵌消息(建议改成注册需要推送消息)
- 机器人通过指令选择“/当前天气、/未来天气、/穿衣指数、/紫外线指数、/空气质量”时,被动推送模版消息
- 机器人通过指令选择“全国天气小程序”,打开天气小程序
整合完之后的完整代码如下:
#!/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=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
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=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
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=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
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=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
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)
完整代码看 天气机器人-Python实现版
- 感谢你赐予我前进的力量