main. … 阅读全文
main.py 为项目的入口文件，范本是官方 SDK 的 demo。
EventDemo.py 为中间变量结构分析文件，对于项目没有实际作用。由于官方没有给出详细的 event 结构，打印出的 event 不够直观，可以通过一个辅助的 py 文件进行格式化。
utils 文件夹是核心逻辑实现的存在位置，针对不同的用户行为，可以在里面的文件中定义相应的处理逻辑。
一共提供了七种处理逻辑：
- BotFollowed.py:关注机器人时触发
- BotUnfollowed.py:取消关注机器人时触发
- ButtonReportInline.py:按钮点击事件，本 SDK 没有提供这部分处理逻辑，只提供了接口
- GroupJoin.py：加入群触发
- GroupLeave.py：推出群触发
- MessageInstruction.py：收到指令消息时触发
- MessageNormal.py:收到消息时触发
可以看到，utils 中还存在一个eventParse.py文件，这个文件用于解析 event 事件。以上 7 个功能可以通过这个模块复用代码。
main.py
接下来我们来逐个解析项目的关键逻辑，在此之前，你最好先阅读项目中的 README 文档，以对该项目有一个初步的了解。
app = Flask(__name__)
sub = Subscription()
openapi = Openapi("your api")
这一段代码实例化了三个类，sub 为 yunhu 模块中的订阅类，openapi 用于匹配你自己的 api，这个 api 可以在控制台中获取。
@app.route("/sub", methods=["POST", "GET"])
def subRoute():
match request.method:
case "POST":
print("POST")
sub.listen(request)
return "success"
# 网页端可测试服务器是否正常
case "GET":
return "连接通畅"
case _:
return "failed"
这段代码定义了一个 flask 路由，他接受 POST 和 GET 类型的请求。POST 请求用于响应云湖的订阅消息，GET 请求在这里仅仅用于测试服务器的连通性。
假设服务器正常，你可以在浏览器中中成功访问 http://ip:port/sub。
这里使用了 match-case 来替代 if-else，因此你的 python 版本必须为 3.10 以上。
# 机器人根据内容回复消息
@sub.onMessageNormal
def onMessageNormalHander(event):
# 这里写与用户交流的逻辑
return MessageNormal.onMessageNormalTest(event=event, openapi=openapi)
# 接收到指令时触发
@sub.onMessageInstruction
def onMessageInstructionHandler(event):
return MessageInstruction.onMessageInstructionTest(event=event, openapi=openapi)
# 进入群的时候，触发
@sub.onGroupJoin
def onGroupJoinHandler(event):
return GroupJoin.onGroupJoinTest(event=event, openapi=openapi)
# 离开群的时候触发
@sub.onGroupLeave
def onGroupLeaveHandler(event):
return GroupLeave.onGroupLeaveTest(event=event, openapi=openapi)
# 关注机器人触发
@sub.onBotFollowed
def onBotFollowedHandler(event):
return BotFollowed.onBotFollowedTest(event=event, openapi=openapi)
# 取关机器人触发
@sub.onBotUnfollowed
def onBotUnfollowedHandler(event):
return BotUnfollowed.onBotUnfollowedTest(event=event, openapi=openapi)
# 机器人设置消息事件
@sub.onButtonReportInline
def onButtonReportInlineHandler(event):
return ButtonReportInline.onButtonReportInlineTest(event=event, openapi=openapi)
这一部分定义了一些事件，处理逻辑我们单独写在对应的模块文件中，提高可读性。
if __name__ == "__main__":
app.run("0.0.0.0", 7888)
# app.run("0.0.0.0", 8857)
这段代码启动了一个 flask 实例。监听所有 IPv4 地址。注意，云湖似乎并不支持 IPv6 服务器，因此，没有公网 IP 的服务器需要进行内网穿透，而国内节点的内网穿透对 http 协议存在限制，因此你需要自行解决 https 问题，或者使用海外节点代理 http。
Message 相关订阅
与 Message 相关的功能有两个：
# 机器人根据内容回复消息
@sub.onMessageNormal
def onMessageNormalHander(event):
# 这里写与用户交流的逻辑
return MessageNormal.onMessageNormalTest(event=event, openapi=openapi)
# 接收到指令时触发
@sub.onMessageInstruction
def onMessageInstructionHandler(event):
return MessageInstruction.onMessageInstructionTest(event=event, openapi=openapi)
首先来看 utils/MessageNormal.py
from .eventParse import messageEventParse
# NormalMessage 测试用例
def onMessageNormalTest(event, openapi):
print("根据内容回复消息")
print(event)
chatType, senderNickname, recvId, recvType = messageEventParse(event)
res = openapi.sendMessage(
recvId,
recvType,
"text",
{"text": f"{recvType} 中给 {senderNickname} 的消息回复"},
)
print(res.content)
content = {
"text": "机器人批量回复普通消息",
"buttons": [
{
"text": "复制内容 1",
"actionType": 2,
"value": "复制内容 1",
},
{
"text": "复制内容 2",
"actionType": 2,
"value": "复制内容 2",
},
],
}
res2 = openapi.batchSendMessage([recvId], recvType, "text", content)
return 0
这段代码首先解析消息事件，通过 messageEventParse 函数提取后续回复消息需要的信息。根据解析结果发送一条简单的文本回复消息。然后批量发送一条包含文本和按钮的复杂消息。
通过这种方式，代码实现了对消息事件的处理和回复，适用于机器人在聊天场景中的自动回复功能。
预期行为：在群里发消息，机器人在群里回复你；私聊机器人，机器人也私聊回复你。当然，这个功能的实现与 EventParse 也有关系，稍后再说。
Message 指令相关订阅
来看看 utils/MessageInstruction.py
from .eventParse import messageEventParse
# NormalInstruction 测试用例
def onMessageInstructionTest(event, openapi):
print("根据指令回复消息")
print(event)
chatType, senderNickname, recvId, recvType = messageEventParse(event)
# 根据指令回复
commandID = event["message"]["commandId"]
match commandID:
# 根据指令触发回复
case 1359:
res = openapi.sendMessage(
recvId,
recvType,
"text",
{"text": f"指令 {commandID} 触发的 {recvType} 消息回复"},
)
return 0
与前面单纯的消息不同，指令消息需要你根据用户的需求进行不同的处理。因此我们使用 match-case 结构来匹配用户指令 commandID，并作出相应的回答。
增加指令只需要在 match-case 块新增指令 ID 及触发逻辑即可。
群和机器人相关订阅消息
群和机器人虽然是两个不同的功能，但在我看来，它们本质上都是群，只不过机器人这个“群”的消息是私密的，别人无法看见，因此两者的解析和处理逻辑类似。
这里就不贴所有的代码了，来看看 BotFollowed.py
from .eventParse import followedEventParse
def onBotFollowedTest(event, openapi):
print("用户关注机器人")
print(event)
chatType, senderNickname, recvId, recvType = followedEventParse(event)
res = openapi.sendMessage(
recvId,
recvType,
"text",
{"text": f"感谢 {senderNickname} 关注"},
)
print(res.content)
return 0
逻辑很简单，首先使用 followedEventParse 函数解析 event，获取到必要的信息，然后发送一条私聊消息，感谢关注。
加入群的行为与关注机器人类似，不过欢迎的消息会发在群里，而不是私聊。
event 解析模块
最后来看看 utils/eventParse.py
# 由于不同的 POST 获取到的结构不同，因此要将不同类型分开
# 消息类型
def messageEventParse(event):
# 初始化一些必要信息
chatType = event["chat"]["chatType"]
senderNickname = event["sender"]["senderNickname"]
# 首先确定是群消息还是用户消息
match chatType:
# 用户：userID
case "bot":
recvId = event["sender"]["senderId"]
recvType = "user"
# 群：groupID
case "group":
recvId = event["chat"]["chatId"]
recvType = "group"
return chatType, senderNickname, recvId, recvType
第一个函数，messageEventParse 用于解析消息订阅所产生的 event，典型结构为
{
"sender": {
"senderId": "",
"senderType": "user",
"senderUserLevel": "owner",
"senderNickname": "",
},
"chat": {"chatId": "200196062", "chatType": "group"},
"message": {
"msgId": "396dc28329b6402482310d214e923334",
"parentId": "",
"sendTime": 1745809972234,
"chatId": "200196062",
"chatType": "group",
"contentType": "text",
"content": {"text": "测试指令"},
"instructionId": 1359,
"instructionName": "测试指令",
"commandId": 1359,
"commandName": "测试指令",
},
}
{
"sender": {
"senderId": "",
"senderType": "user",
"senderUserLevel": "owner",
"senderNickname": "",
},
"chat": {"chatId": "29677771", "chatType": "bot"},
"message": {
"msgId": "bd398171aefd4ca49b946021e4f82380",
"parentId": "",
"sendTime": 1745810154818,
"chatId": "29677771",
"chatType": "bot",
"contentType": "text",
"content": {"text": "测试指令"},
"instructionId": 1359,
"instructionName": "测试指令",
"commandId": 1359,
"commandName": "测试指令",
},
}
可见，上面为群消息，下面为私聊消息，你看出区别了吗？主要是 chatType 不同。因此我们通过 chatType 来首先判断这条消息是群消息还是私聊消息，从而确定机器人应该群发消息还是私聊回复消息。
不过这个逻辑并不绝对，例如，当用户在群内索要资源时，机器人完成可以私发给它，从而避免在群内刷屏。该部分逻辑需要根据功能进行大幅修改。
下面来看看关注和加群的 event：
# 用户关注机器人
{
"time": 1745825366482,
"chatId": "29677771",
"chatType": "bot",
"userId": "2356866",
"nickname": "",
"avatarUrl": "https://chat-storage1.jwznb.com/defalut-avatars/Mary%20Cassatt.png?sign=d1f62445e42a27bf1da25bfb866efba5&t=680f3c66",
}
# 新用户加入群
{
"time": 1745825787222,
"chatId": "200196062",
"chatType": "group",
"userId": "2356866",
"nickname": "",
"avatarUrl": "https://chat-storage1.jwznb.com/defalut-avatars/Mary%20Cassatt.png?sign=62bf56734f173596debb608c6b5d81f1&t=680f3e0b",
}
可以看到，两者结构类似，但相比 Message 的 event 结构简单了许多，因此解析方式也不同。对于这类 event，需要在utils/eventParse.py中重写 parse 方法
# 关注机器人，与加入群聊的 event 结构类似
def followedEventParse(event):
print(event)
chatType = event["chatType"]
recvId = event["userId"]
recvType = "user" if chatType == "bot" else "group"
senderNickname = event["nickname"]
return chatType, senderNickname, recvId, recvType
# 加入群，指向 followedEventParse 函数
groupEventParse = followedEventParse
可以看到，两个函数仅有 recvType 不同，因此理论上这两个函数可以合为一个，主要差别体现在以下行：
recvType = "user" if chatType == "bot" else "group"
这行代码决定了机器人是在群里回复还是私聊回复。
机器人运行
填入 api 后，直接运行 main.py 即可。然后将该服务穿透到公网，并将消息订阅接口配置到云湖机器人控制台。
随后在云湖 app 中进行功能测试。 收起