频道机器人简介
频道机器人是基于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 项目文件夹
在 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效果图如下：
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在频道中执行下列步骤验证效果：
- @机器人后输入“/私信天气 城市名”执行
- 等待几分钟后，到私信面板看看是否有机器人推送过来的天气消息。
使用小程序
当用户想要查看全国或者某个省份的天气情况，一次次@机器人就显得十分麻烦，这个时候你可以使用小程序来解决这个问题。了解具体的小程序开发可以看 QQ小程序开发文档，这里只介绍如何通过机器人打开小程序。
机器人打开小程序非常简单，只需要按照下面配置就可以了，不需要增加额外的代码：
配置好后，我们@机器人就可以看到我们设置的服务了，点击就可以打开设置的小程序
使用指令
每次@机器人输入指令太麻烦了，有没有简单的方式呢？机器人提供了指令配置，当你输入/时就会产出你配置的指令面板。
配置方式如下：
配置好后，当我们输入/时，就可以看到配置的面板了
完整代码可查看github： 天气机器人-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)目前讲的内容已经足够制作大部分游戏的修改器了，主要看你对API的熟悉程度和逆向调试的经验
再深入讲的话就是系统内核层了，比如SSDT hook什么的，而一般单机游戏不会用到内核层的保护（但是大部分网游会），而且调试和修改内核层的东西比较麻烦，很容易蓝屏，所以不打算讲内核的东西
本章介绍怎么hook D3D函数，实现在游戏画面中显示自己的文字，阅读之前最好补习下D3D编程的知识，本章以D3D9、32位程序为例，编译程序需要安装DirectX 9 SDK
虚函数表hook
要在D3D程序里绘制自己的东西一般要hook IDirect3DDevice9的EndScene函数，调用这个函数时原程序的绘制已经完成，可以轮到我们绘制了
可以用inline hook来hook这个函数，但是流行的做法是虚函数表hook
先来看看EndScene是怎么被调用的：
虚函数表hook的原理就是修改虚函数表中的函数地址，使它指向我们的函数（是不是很像IAT hook）
这里说明一下一般成员函数的调用约定默认是thiscall，也就是从ecx寄存器传入this指针，但是也可以手动声明成其他调用约定，比如stdcall，这时就从堆栈传入this指针，相当于this指针是函数的第一个参数。
D3D所有成员函数的调用约定都是__stdcall，这就方便了我们获取this指针，直接声明成第一个参数就行了（否则要用内联汇编取ecx的值）
虚函数表hook实现
BOOL hookVTable(void* pInterface, int index, void* hookFunction, void** oldAddress)
{
void** address = &(*(void***)pInterface)[index];
if (address == NULL)
return FALSE;
// 保存原函数地址
if (oldAddress != NULL)
*oldAddress = *address;
// 修改虚函数表中地址为hookFunction
DWORD oldProtect, oldProtect2;
VirtualProtect(address, sizeof(DWORD), PAGE_READWRITE, &oldProtect);
*address = hookFunction;
VirtualProtect(address, sizeof(DWORD), oldProtect, &oldProtect2);
return TRUE;
}
BOOL unhookVTable(void* pInterface, int index, void* oldAddress)
{
// 修改回原函数地址
return hookVTable(pInterface, index, oldAddress, NULL);
} （其实就是拿IATHook改的）
hook EndScene
方法1
为了在程序初始化前hook我们要在主线程运行前远线程注入，首先hook Direct3DCreate9，然后是CreateDevice和EndScene
typedef IDirect3D9* (WINAPI* Direct3DCreate9Type)(UINT SDKVersion);
typedef HRESULT(STDMETHODCALLTYPE* CreateDeviceType)(IDirect3D9* thiz, UINT Adapter, D3DDEVTYPE DeviceType, HWND hFocusWindow, DWORD BehaviorFlags, D3DPRESENT_PARAMETERS* pPresentationParameters, IDirect3DDevice9** ppReturnedDeviceInterface);
typedef HRESULT(STDMETHODCALLTYPE* EndSceneType)(IDirect3DDevice9* thiz);
Direct3DCreate9Type RealDirect3DCreate9 = NULL;
CreateDeviceType RealCreateDevice = NULL;
EndSceneType RealEndScene = NULL;
IDirect3D9* g_d3d9 = NULL;
IDirect3DDevice9* g_device = NULL;
HRESULT STDMETHODCALLTYPE MyEndScene(IDirect3DDevice9* thiz)
{
// 这里放我们的绘制代码
return RealEndScene(thiz);
}
HRESULT STDMETHODCALLTYPE MyCreateDevice(IDirect3D9* thiz, UINT Adapter, D3DDEVTYPE DeviceType, HWND hFocusWindow, DWORD BehaviorFlags, D3DPRESENT_PARAMETERS* pPresentationParameters, IDirect3DDevice9** ppReturnedDeviceInterface)
{
unhookVTable(g_d3d9, 16, RealCreateDevice);
HRESULT res = RealCreateDevice(thiz, Adapter, DeviceType, hFocusWindow, BehaviorFlags, pPresentationParameters, ppReturnedDeviceInterface);
g_device = *ppReturnedDeviceInterface;
hookVTable(g_device, 42, MyEndScene, &RealEndScene); // EndScene是IDirect3DDevice9第43个函数
return res;
}
IDirect3D9* WINAPI MyDirect3DCreate9(UINT SDKVersion)
{
unhookIAT(GetModuleHandle(NULL), "d3d9.dll", "Direct3DCreate9");
g_d3d9 = RealDirect3DCreate9(SDKVersion);
hookVTable(g_d3d9, 16, MyCreateDevice, &RealCreateDevice); // CreateDevice是IDirect3D9第17个函数
return g_d3d9;
}
BOOL APIENTRY DllMain( HMODULE hModule,
DWORD ul_reason_for_call,
LPVOID lpReserved
)
{
switch (ul_reason_for_call)
{
case DLL_PROCESS_ATTACH:
hookIAT(GetModuleHandle(NULL), "d3d9.dll", "Direct3DCreate9", MyDirect3DCreate9, &RealDirect3DCreate9);
break;
case DLL_PROCESS_DETACH:
if (g_device != NULL && RealEndScene != NULL)
unhookVTable(g_device, 42, RealEndScene);
break;
case DLL_THREAD_ATTACH:
case DLL_THREAD_DETACH:
break;
}
return TRUE;
}方法2
这个方法不需要在主线程运行前注入，CE和直播软件OBS也是用了这个方法，所以推荐使用
理论上同一个类的虚函数地址是一样的，所以我们可以自己创建一个IDirect3DDevice9，然后就能获取虚函数地址了
经测试不同IDirect3DDevice9的虚函数表指针不一样，所以只能用inline hook，不能用虚函数表hook
void* endSceneAddr = NULL;
BYTE endSceneOldCode[sizeof(JmpCode)];
HRESULT STDMETHODCALLTYPE MyEndScene(IDirect3DDevice9* thiz)
{
// 这里放我们的绘制代码
unhook(endSceneAddr, endSceneOldCode);
HRESULT hr = thiz->EndScene();
hook(endSceneAddr, MyEndScene, endSceneOldCode);
return hr;
}
DWORD WINAPI initHookThread(LPVOID dllMainThread)
{
// 等待DllMain（LoadLibrary线程）结束
WaitForSingleObject(dllMainThread, INFINITE);
CloseHandle(dllMainThread);
// 创建一个窗口用于初始化D3D
WNDCLASSEX wc = {};
wc.cbSize = sizeof(wc);
wc.style = CS_OWNDC;
wc.hInstance = GetModuleHandle(NULL);
wc.lpfnWndProc = DefWindowProc;
wc.lpszClassName = _T("DummyWindow");
if (RegisterClassEx(&wc) == 0)
{
MessageBox(NULL, _T("注册窗口类失败"), _T(""), MB_OK);
return 0;
}
HWND hwnd = CreateWindowEx(0, wc.lpszClassName, _T(""), WS_OVERLAPPEDWINDOW, 0, 0, 640, 480, NULL, NULL, wc.hInstance, NULL);
if (hwnd == NULL)
{
MessageBox(NULL, _T("创建窗口失败"), _T(""), MB_OK);
return 0;
}
// 初始化D3D
IDirect3D9* d3d9 = Direct3DCreate9(D3D_SDK_VERSION);
if (d3d9 == NULL)
{
MessageBox(NULL, _T("创建D3D失败"), _T(""), MB_OK);
DestroyWindow(hwnd);
return 0;
}
D3DPRESENT_PARAMETERS pp = {};
pp.Windowed = TRUE;
pp.SwapEffect = D3DSWAPEFFECT_COPY;
IDirect3DDevice9* device;
if (FAILED(d3d9->CreateDevice(D3DADAPTER_DEFAULT, D3DDEVTYPE_HAL, hwnd,
D3DCREATE_SOFTWARE_VERTEXPROCESSING, &pp, &device)))
{
MessageBox(NULL, _T("创建设备失败"), _T(""), MB_OK);
d3d9->Release();
DestroyWindow(hwnd);
return 0;
}
// hook EndScene
endSceneAddr = (*(void***)device)[42]; // EndScene是IDirect3DDevice9第43个函数
hook(endSceneAddr, MyEndScene, endSceneOldCode);
// 释放
d3d9->Release();
device->Release();
DestroyWindow(hwnd);
return 0;
}
BOOL APIENTRY DllMain(HMODULE hModule,
DWORD ul_reason_for_call,
LPVOID lpReserved
)
{
switch (ul_reason_for_call)
{
case DLL_PROCESS_ATTACH:
// 取当前线程句柄
HANDLE curThread;
if (!DuplicateHandle(GetCurrentProcess(), GetCurrentThread(), GetCurrentProcess(), &curThread, SYNCHRONIZE, FALSE, 0))
return FALSE;
// DllMain中不能使用COM组件，所以要在另一个线程初始化
CloseHandle(CreateThread(NULL, 0, initHookThread, curThread, 0, NULL));
break;
case DLL_PROCESS_DETACH:
if (endSceneAddr != NULL)
unhook(endSceneAddr, endSceneOldCode);
break;
case DLL_THREAD_ATTACH:
case DLL_THREAD_DETACH:
break;
}
return TRUE;
}绘制文字
这部分就直接用D3D的API了，不懂的去看看D3D教程..
HRESULT STDMETHODCALLTYPE MyEndScene(IDirect3DDevice9* thiz)
{
static RECT rect = { 0, 0, 200, 200 };
g_font->DrawText(NULL, _T("Hello World"), -1, &rect, DT_TOP | DT_LEFT, D3DCOLOR_XRGB(255, 0, 0));
return RealEndScene(thiz);
}
HRESULT STDMETHODCALLTYPE MyCreateDevice(IDirect3D9* thiz, UINT Adapter, D3DDEVTYPE DeviceType, HWND hFocusWindow, DWORD BehaviorFlags, D3DPRESENT_PARAMETERS* pPresentationParameters, IDirect3DDevice9** ppReturnedDeviceInterface)
{
unhookVTable(g_d3d9, 16, RealCreateDevice);
HRESULT res = RealCreateDevice(thiz, Adapter, DeviceType, hFocusWindow, BehaviorFlags, pPresentationParameters, ppReturnedDeviceInterface);
g_device = *ppReturnedDeviceInterface;
D3DXFONT_DESC d3dFont = {};
d3dFont.Height = 25;
d3dFont.Width = 12;
d3dFont.Weight = 500;
d3dFont.Italic = FALSE;
d3dFont.CharSet = DEFAULT_CHARSET;
wcscpy_s(d3dFont.FaceName, L"Times New Roman");
D3DXCreateFontIndirect(g_device, &d3dFont, &g_font);
// 测试中不知道为什么第一次调用DrawText后device的虚函数表会恢复，没办法只好在hook前调用一次
static RECT rect = { 0, 0, 200, 200 };
g_font->DrawText(NULL, _T("Hello World"), -1, &rect, DT_TOP | DT_LEFT, D3DCOLOR_XRGB(255, 0, 0));
hookVTable(g_device, 42, MyEndScene, &RealEndScene); // EndScene是IDirect3DDevice9第43个函数
return res;
}效果
注意左上角出现了我们绘制的"Hello World"
hook D3D函数可以实现在游戏中显示自己的UI（播放Bad Apple），或者在绘制某些东西的时候禁用Z轴缓冲实现透视
原文链接：https://blog.csdn.net/xfgryujk/article/details/52739486