python实现qq频道机器人开发

2022-08-09 09:31:02 浏览数 (1)

Post Views: 7

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

实验简介
频道机器人简介

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

实验目的

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

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

实验准备

相关概念

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

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

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

环境搭建

安装Python3

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

代码语言：javascript复制
python3 --version
安装机器人SDK

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

代码语言：javascript复制
pip install qq-bot

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

代码语言：javascript复制
pip install pyyaml
创建项目文件

创建一个 demo 项目文件夹

代码语言：javascript复制
mkdir /home/demo && cd /home/demo

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

代码语言：javascript复制
touch config.yaml

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

代码语言：javascript复制
touch robot.py
导入Token 和 AppID

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

：

代码语言：javascript复制
token:
  appid: "123"
  token: "xxxx"
导入依赖包

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

：

robot.py

代码语言：javascript复制
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

代码语言：javascript复制
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)
代码运行

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

代码语言：javascript复制
python3 /home/demo/robot.py

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

获取天气数据

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

：

robot.py

代码语言：javascript复制
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

代码语言：javascript复制
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)
代码运行

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

代码语言：javascript复制
python3 /home/demo/robot.py

python3 /home/demo/robot.py

效果图如下：

代码语言：javascript复制
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

代码语言：javascript复制
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()**的语句：

代码语言：javascript复制
 # 定时推送主动消息
 send_weather_message_by_time()

编写完毕，注意保存

设置机器人私信

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

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

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

robot.py

代码语言：javascript复制
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

代码语言：javascript复制
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)

编写完毕，注意保存

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

代码语言：javascript复制
python3 /home/demo/robot.py

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

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

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

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

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

使用指令

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

配置方式如下：

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

课后习题

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

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

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

代码语言：javascript复制
#!/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)
机器人 小程序 python 微信

0 人点赞