程序師世界是廣大編程愛好者互助、分享、學習的平台,程序師世界有你更精彩!
首頁
編程語言
C語言|JAVA編程
Python編程
網頁編程
ASP編程|PHP編程
JSP編程
數據庫知識
MYSQL數據庫|SqlServer數據庫
Oracle數據庫|DB2數據庫
您现在的位置: 程式師世界 >> 編程語言 >  >> 更多編程語言 >> Python

手把手QQ機器人制作教程,根據官方接口進行開發,基於Python語言制作的詳細教程(更新中)

編輯:Python

第 1 課、注冊 QQ 開放平台賬戶

QQ開放平台官方地址:https://q.qq.com/#/app/bot

QQ開放平台包含:QQ機器人、QQ小程序、QQ小游戲,我們這邊選擇QQ機器人。

機器人類型:設置私域機器人或者公域機器人,當然公域機器人對於服務器的要求過高,我們這邊選擇私域機器人進行開發。

特別注意在選擇沙箱頻道的時候,你必須提前創建個測試頻道,必須<20人的頻道。

例如我這邊創建了迷你小喚QQ機器人的測試頻道。

創建完成後就能夠進入你剛剛創建QQ機器人的後台,展示如下所示:

然後點擊發布流程中的代碼開發即可查看API文檔,進行編寫我們服務器端的程序代碼。

點擊開發設置能夠查看 BotAppID、機器人令牌、機器人密鑰,後續服務器端開發需要。

第 2 課、QQ 頻道機器人接口說明

QQ頻道機器人提供兩類接口:

第一種:基於 REST 風格的 OPENAPI。

傳統風格 描述形式:http://localhost/user/getById?id=1
Rest 風格 描述形式:http://localhost/user/1

顯而易見,Rest風格優點:隱藏資源的訪問行為,無法通過地址得知對資源是哪種操作;簡化書寫。

第二種:基於 WebSocket 提供實時事件通知的事件中心。

這種的話,和野生的nonebot框架采用的類似,都是基於協議端與服務器端的長連接模式。

博客推薦:Python基於cq-http協議端,使用nonebot2框架制作屬於自己的智能機器人

第 3 課、SDK 開發基礎庫安裝

QQ 小程序的開發 SDK 官方提供了三種語言,分別是:GO語言、JS語言、Python語言,分別對應如下 3 基礎庫:

golang: https://github.com/tencent-connect/botgo(opens new window)
nodejs: https://github.com/tencent-connect/bot-node-sdk(opens new window)
python: https://github.com/tencent-connect/botpy

我們選擇 Python 語言進行開發:

GITHUB:https://github.com/tencent-connect/botpy

安裝 qq-botpy 第三方庫

pip install qq-botpy

使用方法:需要使用的地方import botpy

import botpy

兼容提示:原機器人的老版本qq-bot仍然可以使用,但新接口的支持上會逐漸暫停,此次升級不會影響線上使用的機器人

第 4 課、botpy 快速入門案例測試

版本介紹:

首先介紹下版本問題,老版本是 qq-bot,新版本是 botpy,當然隨著新版本迭代升級,老版本將會陸續的停用相關api,但是騰訊雲的案例是 基於 老版本 qq-bot。

第 ① 步:通過繼承實現 bot.Client,實現自己的機器人 Client

第 ② 步:實現機器人相關事件的處理方法,如 on_at_message_create, 詳細的事件監聽列表,請參考 事件監聽.md

如下,是定義機器人被@的後自動回復:

import botpy
from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
await message.reply(content=f"機器人{
self.robot.name}收到你的@消息了: {
message.content}")

注意:每個事件會下發具體的數據對象,如message相關事件是message.Message的對象 (部分事件透傳了後台數據,暫未實現對象緩存)

第 ③ 步:設置機器人需要監聽的事件通道,並啟動 client

import botpy
from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
await self.api.post_message(channel_id=message.channel_id, content="content")
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid="12345", token="xxxx")

也可以通過預設置的類型,設置需要監聽的事件通道

import botpy
intents = botpy.Intents.none()
intents.public_guild_messages=True

第 5 課、騰訊雲官方模擬實驗

環境搭建

安裝Python3,推薦使用Python3,實驗環境已經預安裝,可執行下面命令,進行Python版本驗證。

python3 --version

安裝機器人SDK

在終端執行下面命令安裝機器人PythonSDK:

pip install qq-bot

同時,由於需要讀取 yaml 文件的內容,我們也需要安裝 pyyaml

pip install pyyaml

創建項目文件

創建一個 demo 項目文件夾

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

在 demo 文件夾下創建名為 config.yaml 的配置文件

touch config.yaml

接著,在 demo 文件夾下創建一個名為 robot.py 的文件:

touch robot.py

導入Token 和 AppID

請點擊打開 config.yaml 文件,並填入自己機器人的 AppID 和 Token ,注意保存:

token:
appid: "自己QQ小程序AppID"
token: "自己QQ小程序token"

導入依賴包

請點擊打開 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 文件中添加如下代碼,注意保存:

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 指令就可以收到回復了

第 6 課、騰訊雲官方案例:天氣

代碼說明

上面鏈接的天氣API使用的sign=b59bc3ef6191eb9f747dd4e83c99f2a4可能會過期,如下圖指引:

請前往天氣API地址查看最新的測試sign並替換,或注冊賬號申請一個免費sign

天氣免費API:https://www.nowapi.com/api/weather.today


修改 _message_handler 方法

在 _message_handler 方法中,加入調用 get_weather 函數並發送天氣的代碼。完整 _message_handler 的實現如下:

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)

接口函數 get_weather(city_name: str) -> Dict:

async def get_weather(city_name: str) -> Dict:
weather_api_url = "http://api.k780.com/?app=weather.today&appkey=10003&sign=b59bc3ef6191eb9f747dd4e83c99f2a4&format=json&cityNm=" + city_name
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

代碼運行

在終端命令行輸入並執行下列命令,運行機器人

python3 /home/demo/robot.py

效果圖如下:

第 7 課、騰訊雲官方案例:主動推送

設置機器人主動推送消息

上面的教程只實現一個簡單的獲取天氣的功能,但是我們做的是天氣機器人,希望實現一個報告天氣的功能。一般的天氣應用都會在一個特定時間給你推送天氣通知,在頻道機器人中,你可以通過主動消息來實現這個功能。

在 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()

編寫完畢,注意保存

在終端命令行輸入並執行下列命令,運行機器人

python3 /home/demo/robot.py

效果圖如下:

第 8 課、騰訊雲官方案例:指令回復

設置機器人指令回復ark消息

提供給個人開發者的 Ark 有3種,這裡使用 23 號 Ark。其它 Ark 見 消息模板

消息模板:https://bot.q.qq.com/wiki/develop/api/openapi/message/message_template.html

先在 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)

編寫完畢,注意保存

代碼運行,在終端命令行輸入並執行下列命令,運行機器人

python3 /home/demo/robot.py

@機器人後輸入“/天氣 城市名”,效果如下圖:

設置機器人私信

我們希望能提供不同用戶不同地方的天氣,但是發太多的消息會影響其它的用戶。針對這種情況,我們可以通過私信來實現。下面函數中,當我們@機器人hello時收到機器人的私信。

私信中我們不使用ark,而是使用Embed。Embed也是一種結構化消息,它比Ark簡單
在 robot.py 添加發送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

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

在頻道中執行下列步驟驗證效果:

@機器人後輸入“/私信天氣 城市名”執行
等待幾分鐘後,到私信面板看看是否有機器人推送過來的天氣消息。

第 9 課、騰訊雲官方案例:小程序

當用戶想要查看全國或者某個省份的天氣情況,一次次@機器人就顯得十分麻煩,這個時候你可以使用小程序來解決這個問題。了解具體的小程序開發可以看 QQ小程序開發文檔,這裡只介紹如何通過機器人打開小程序。

機器人打開小程序非常簡單,只需要按照下面配置就可以了,不需要增加額外的代碼:

第 10 課、Pycharm SSH 連接服務器

為了方便使用我們也可以,直接使用Python Pycharm SSH 連接我們的的小程序服務器端:這邊我們現在寶塔面板下載 id_rsa 密鑰文件,當然你也可以直接 host + password 進行連接。


然後連接配置後,選擇IDE 版本 和 項目工程的目錄(對應服務器的目錄)


然後需要等待大概10分鐘左右讓它安裝好環境,然後會自動打開,可見如下所示已經是剛才配置的服務器端的項目目錄了。

在服務器端項目目錄編寫測試代碼,其中 appid 和 token 換成你自己小程序的數據:

import botpy
from botpy.types.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
await self.api.post_message(channel_id=message.channel_id, content="content")
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid={
appid}, token={
token})

報錯提示: from typing import TypedDict ImportError: cannot import name ‘TypedDict’

如果出現如上的問題:說明你的 python 版本和 botpy 版本不匹配:

官方說明:botpy 更新包的話需要添加 --upgrade 兼容版本:python3.8+

第 11 課、Python 讀取 YAML 文件

在框架搭建中,一些配置文件(例如:數據庫賬號、url,日志級別等…)會經常的用到,所以我們需要一個存放這些參數的文件,並從中讀取和寫入。當然,存放參數除了用yaml,像ini文件、json文件、excel文件也是可以的,看個人選擇。本文就介紹yaml文件的操作方法。

安裝 pyyaml 模塊:

pip install pyyaml

讀取YAML代碼如下:

import yaml
file = open('test.yaml', 'r', encoding="utf-8")
# 使用文件對象作為參數
data = yaml.load(file)
print(data)

結果運行出現警告,原因是YAML 5.1版本後棄用了 yaml.load(file) 這個用法,因為覺得很不安全,5.1 版本之後就修改了需要指定 Loader,通過默認加載​​器(FullLoader)禁止執行任意函數,該 load 函數也變得更加安全。建議代碼如下:

import yaml
file = open('test.yaml', 'r', encoding="utf-8")
# 讀取文件中的所有數據
file_data = file.read()
file.close()
# 指定Loader
data = yaml.load(file_data,Loader=yaml.FullLoader)
print(data)

上述例子和之前的例子我用了兩種不同的讀取方式,第一種為文件對象,第二種為文件中的所有字符串數據,兩種方法都可以。

同時大家也發現了:yaml讀取出來的數據結構不是由pyyaml模塊讀取代碼決定的,而是yaml文件中的內容決定了它的結構,比如現在舉例的YAML是字典裡面嵌套列表,從文件讀取出來後,也是這種數據結構形式。

load_all() 方法使用代碼如下:

import yaml
file = open('test.yaml', 'r', encoding="utf-8")
#datas為load_all()方法返回的迭代器對象
datas = yaml.load_all(file,Loader=yaml.FullLoader)
for data in datas:
print(data)

本例中的 YAML 文件使用分為兩段,load_all()方法會創建一個迭代器對象,分開存放兩段數據。

使用 dump() 方法將一個 python 對象生成 yaml 格式的文檔

import yaml
data = {
 'name':'喚醒手腕', 'age':20 , 'sex':'男' }
yaml_file = open('test.yaml','w')
# yaml_file作為dump()第二個參數,是一個可寫狀態的文件對象
yaml.dump(data, yaml_file ,default_flow_style=False,encoding='utf-8',allow_unicode=True)
yaml_file.flush()
yaml_file.close()

第 12 課、API 教程:用戶對象 User

用戶對象中所涉及的 ID 類數據,都僅在機器人場景流通,與真實的 ID 無關。請不要理解為真實的 ID

注意:union_openid 是關聯應用的 openid,如果你不是分布式場景,其實是無所謂的。

union_openid 與 union_user_account 只有在單獨拉取 member 信息的時候才會提供,在其他的事件中所攜帶的 user 對象,均無這兩個字段的內容。

獲取當前用戶信息

import botpy
import yaml
from botpy.message import Message
config = yaml.load(open('config.yaml', encoding="utf-8").read(), Loader=yaml.FullLoader)
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
user = await self.api.me()
botpy.logger.info(user)
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid=config['token']['appid'], token=config['token']['token'])

日志打印 QQ 機器人信息

{

"id": "11586990140073229091",
"username": "gitsub",
"avatar": "https://thirdqq.qlogo.cn/g?b=oidb&k=M5TibpXicS7Jt4z89BZxiamAA&s=100&t=1641802698",
"union_openid": "74F138F7F3AF68C4B8E8325013FCA295",
"union_user_account": ""
}

第 13 課、API 教程:發消息 / 回復

單純回復發送文本消息

消息回復簡單調用 message 事件的 reply 方法:

await message.reply(content="content")

官方回復消息案例代碼:

# -*- coding: utf-8 -*-
import asyncio
import os
import botpy
from botpy import logging
from botpy.ext.cog_yaml import read
from botpy.message import Message
test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
_log = logging.get_logger()
class MyClient(botpy.Client):
async def on_ready(self):
_log.info(f"robot 「{
self.robot.name}」 on_ready!")
async def on_at_message_create(self, message: Message):
_log.info(message.author.avatar)
if "sleep" in message.content:
await asyncio.sleep(10)
_log.info(message.author.username)
await message.reply(content=f"機器人{
self.robot.name}收到你的@消息了: {
message.content}")
if __name__ == "__main__":
# 通過預設置的類型,設置需要監聽的事件通道
# intents = botpy.Intents.none()
# intents.public_guild_messages=True
# 通過kwargs,設置需要監聽的事件通道
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid=test_config["appid"], token=test_config["token"])

首先分析下 message 事件對象的信息:


案例代碼中 message.content 是獲取消息的內容

運行結果:


回復其他類型的消息

import botpy
from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
await self.api.post_message(channel_id="xxxx", content="xxx", msg_id="xxxx", embed=embed)
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid={
appid}, token={
token})

官方 api 發送消息:

注意事項:

發送圖片案例:回復對方的頭像

在第 12 課介紹了,user 對象的相關屬性,用戶頭像就是:message.author.avatar

import os
import botpy
from botpy.ext.cog_yaml import read
from botpy.message import Message
test_config = read(os.path.join(os.path.dirname(__file__), "config.example.yaml"))
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
await message.reply(image=message.author.avatar)
if __name__ == "__main__":
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid=test_config["appid"], token=test_config["token"])

運行結果展示:

內嵌格式:進行 @ 某人 / 全體成員、內嵌表情

功能描述:利用 content 字段發送內嵌格式的消息。

內嵌格式僅在 content 中會生效,在 Ark 和 Embed 中不生效。


表情:https://bot.q.qq.com/wiki/develop/api/openapi/emoji/model.html

await message.reply(content='你在干嘛呀?<emoji:4>')

小補充:事件 id :message.event_id

事件 id 區分不同的事件,每次都是唯一不同的

await message.reply(content='你在干嘛呀?<emoji:4>,{}'.format(message.event_id))


回復消息內容:

async def on_at_message_create(self, message: Message):
# 構造消息發送請求數據對象
message_reference = Reference(message_id=message.id)
# 通過api發送回復消息
await self.api.post_message(
channel_id=message.channel_id,
content="<emoji:4>這是一條引用消息",
msg_id=message.id,
message_reference=message_reference,
)

消息中包含的URL需要進行配置

第 14 課、API 教程:語音房上麥

START 0 開始播放操作、PAUSE 1 暫停播放操作、RESUME 2 繼續播放操作、STOP 3 停止播放操作

案例代碼測試:

import botpy
from botpy.audio import Audio
class MyClient(botpy.Client):
async def on_audio_start(self, audio: Audio):
audio_control={

"audio_url":"url",
"text":"簡單愛-周傑倫",
"status":"START",
}
await self.api.update_audio(audio.channel_id, audio_control=audio_control)
intents = botpy.Intents(audio_action=True)
client = MyClient(intents=intents)
client.run(appid={
appid}, token={
token})

[botpy] 接口請求異常,請求連接: https://api.sgroup.qq.com/channels/9368829/mic, 錯誤代碼: 401, 返回內容: {‘message’: ‘check app privilege not pass’, ‘code’: 11253}, trace_id:adfcc85b0e21b5c2f4817abe82d9c15d

特別注意:ErrorCheckAppPrivilegeNotPass 檢查應用權限不通過,該機器人應用未獲得調用該接口的權限,需要向平台申請

第 15 課、API 教程:注冊指令 handler

注冊指令handler 模式編程

# -*- coding: utf-8 -*-
import os
import botpy
from botpy import logging, BotAPI
from botpy.ext.command_util import Commands
from botpy.message import Message
from botpy.ext.cog_yaml import read
test_config = read(os.path.join(os.path.dirname(__file__), "config.yaml"))
_log = logging.get_logger()
@Commands(name=("你好", "hello"))
async def hello(api: BotAPI, message: Message, params=None):
_log.info(params)
# 第一種用reply發送消息
await message.reply(content=params)
# 第二種用api.post_message發送消息
await api.post_message(channel_id=message.channel_id, content=params, msg_id=message.id)
return True
@Commands("晚安")
async def good_night(api: BotAPI, message: Message, params=None):
_log.info(params)
# 第一種用reply發送消息
await message.reply(content=params)
# 第二種用api.post_message發送消息
await api.post_message(channel_id=message.channel_id, content=params, msg_id=message.id)
return True
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
# 注冊指令handler
handlers = [
hello,
good_night,
]
for handler in handlers:
if await handler(api=self.api, message=message):
return
if __name__ == "__main__":
# 通過預設置的類型,設置需要監聽的事件通道
# intents = botpy.Intents.none()
# intents.public_guild_messages=True
# 通過kwargs,設置需要監聽的事件通道
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid=test_config["appid"], token=test_config["token"])

第 16 課、API 教程:頻道對象 guild

獲取頻道的信息:

import botpy
from botpy.audio import Audio
from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
guild = await self.api.get_guild(guild_id=message.guild_id)
await self.api.post_message(channel_id=message.channel_id, content=str(guild), msg_id=message.event_id)
intents = botpy.Intents(audio_action=True, public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid="", token="")

運行結果:

獲取頻道的子頻道信息:

async def on_at_message_create(self, message: Message):
channels = await self.api.get_channels(guild_id=message.guild_id)
await self.api.post_message(channel_id=message.channel_id, content=str(channels), msg_id=message.event_id)


channel 頻道對象信息

第 17 課、API 教程:發送按鈕消息

功能描述

通過在 MessageSendRequest 中指定 keyboard 字段發送帶按鈕的消息,支持 keyboard 模版 和 自定義 keyboard 兩種請求格式。

要求操作人在該子頻道具有發送消息和 對應消息按鈕組件 的權限。請求參數 keyboard 模版 和 自定義 keyboard 只能單一傳值。

keyboard 模版

調用前需要先申請消息按鈕組件模板,這一步會得到一個模板 id,在請求時填在 keyboard 字段上。申請消息按鈕組件模板需要提供響應的 json,具體格式參考 InlineKeyboard。

注意:僅 markdown 消息支持消息按鈕。

import botpy
from botpy.message import Message
from botpy.types.message import MarkdownPayload, MessageMarkdownParams
class MyClient(botpy.Client):
async def handle_send_markdown_by_template(self, channel_id, msg_id):
params = [
MessageMarkdownParams(key="title", values=["標題"]),
MessageMarkdownParams(key="content", values=["為了成為一名合格的巫師,請務必閱讀頻道公告", "藏館黑色魔法書"]),
]
markdown = MarkdownPayload(template_id=65, params=params)
# 通過api發送回復消息
await self.api.post_message(channel_id, markdown=markdown, msg_id=msg_id)
async def handle_send_markdown_by_content(self, channel_id, msg_id):
markdown = MarkdownPayload(content="# 標題 \n## 簡介很開心 \n內容")
# 通過api發送回復消息
await self.api.post_message(channel_id, markdown=markdown, msg_id=msg_id)
async def on_at_message_create(self, message: Message):
await message.reply(content=f"機器人{
self.robot.name}收到你的@消息了: {
message.content}")
await self.handle_send_markdown_by_template(message.channel_id, message.id)
await self.handle_send_markdown_by_content(message.channel_id, message.id)
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid={
appid}, token={
token})

第 18 課、API 教程:禁言(非管理員)

import botpy
from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
await self.api.mute_member(guild_id="xxxx", user_id="xxxx", mute_end_timestamp="xxxx", mute_seconds="xxxx")
intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)
client.run(appid={
appid}, token={
token})

self.api.mute_member 禁言 非管理員

第 19 課、API 教程:查看錯誤碼

錯誤碼:查看 - 錯誤碼

第 20 課、API 教程:事件監聽

通過繼承實現 bot.Client, 實現自己的機器人Client,在 MyClient 類中調用函數實現處理事件的功能。

class MyClient(botpy.Client):

監聽的事件(公域消息事件的監聽)

首先需要訂閱事件public_guild_messages

intents = botpy.Intents(public_guild_messages=True)
client = MyClient(intents=intents)

監聽 @ 機器人 和 頻道刪除消息

from botpy.message import Message
class MyClient(botpy.Client):
async def on_at_message_create(self, message: Message):
""" 此處為處理該事件的代碼 """
async def on_public_message_delete(self, message: Message):
""" 此處為處理該事件的代碼 """

全部消息監聽,而不只是 at 機器人的消息

首先需要訂閱事件guild_messages

intents = botpy.Intents(guild_messages=True)
client = MyClient(intents=intents)

from botpy.message import Message
class MyClient(botpy.Client):
async def on_message_create(self, message: Message):
""" 此處為處理該事件的代碼 """
async def on_message_delete(self, message: Message):
""" 此處為處理該事件的代碼 """

頻道成員事件的監聽

首先需要訂閱事件guild_members

intents = botpy.Intents(guild_members=True)
client = MyClient(intents=intents)

首先需要訂閱事件guild_members
intents = botpy.Intents(guild_members=True)
client = MyClient(intents=intents)

互動事件的監聽

首先需要訂閱事件interaction

intents = botpy.Intents(interaction=True)
client = MyClient(intents=intents)


注:需要引入Interaction

from botpy.interaction import Interaction
class MyClient(botpy.Client):
async def on_interaction_create(self, interaction: Interaction):
""" 此處為處理該事件的代碼 """

論壇事件的監聽

僅 私域 機器人能夠設置此 intents

首先需要訂閱事件forums

intents = botpy.Intents(forums=True)
client = MyClient(intents=intents)

注:需要引入Thread、Post、Reply和AuditResult

from botpy.forum import Thread
from botpy.types.forum import Post, Reply, AuditResult
class MyClient(botpy.Client):
async def on_forum_thread_create(self, thread: Thread):
""" 此處為處理該事件的代碼 """
async def on_forum_thread_update(self, thread: Thread):
""" 此處為處理該事件的代碼 """
async def on_forum_thread_delete(self, thread: Thread):
""" 此處為處理該事件的代碼 """
async def on_forum_post_create(self, post: Post):
""" 此處為處理該事件的代碼 """
async def on_forum_post_delete(self, post: Post):
""" 此處為處理該事件的代碼 """
async def on_forum_reply_create(self, reply: Reply):
""" 此處為處理該事件的代碼 """
async def on_forum_reply_delete(self, reply: Reply):
""" 此處為處理該事件的代碼 """
async def on_forum_publish_audit_result(self, auditresult: AuditResult):
""" 此處為處理該事件的代碼 """

  1. 上一篇文章:
  2. 下一篇文章:
Copyright © 程式師世界 All Rights Reserved