Python

云湖机器人 demo 项目指南——快速搭建云湖机器人

本文介绍了云湖社交软件的机器人开发项目。作者体验云湖后，发现其机器人开发简单但文档少，便重构了官方 SDK。项目包含多个文件和文件夹，main.py 是入口，utils 文件夹实现核心逻辑，涵盖七种处理逻辑。通过实例化类、定义路由，实现对不同请求类型的处理。代码使用 match - case 语法，要求 Python 3.10 以上。还展示了消息和指令相关订阅的代码，如 MessageNormal.py 解析消息事件并发送回复消息，实现机器人在聊天场景的自动回复功能，适用于 Python 初学者尝试搭建机器人。

最近听说了一个叫云湖的社交软件，口气挺大，其目标是吊打腾讯。所以下载下来体验了一番，感觉就是个国产 discord。

软件本身没啥意思，但是我发现云湖的机器人开发还比较简单，但文档比较少。于是我就拿官方的无注释 SDK 重构了一份新的 SDK，基本可以开箱即用。

由于我不想把 SDK 弄得太复杂，因此基本没有使用语法糖，一切以清晰易懂为准。

Python 的初学者可以尝试自己搭建一个机器人。

下面是项目详解，这篇文章相对以前的文章来说可能比较硬核，请酌情阅读。

项目结构 

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

❯ tree
.
├── EventDemo.py
├── LICENSE
├── main.py
├── pictures
│   ├── 云湖控制台.png
│   ├── 创建机器人.png
│   ├── 添加指令.png
│   └── 获取API列表.png
├── README.md
└── utils
    ├── BotFollowed.py
    ├── BotUnfollowed.py
    ├── ButtonReportInline.py
    ├── eventParse.py
    ├── GroupJoin.py
    ├── GroupLeave.py
    ├── MessageInstruction.py
    └── MessageNormal.py

3 directories, 16 files

LICENSE 和 README.md/pictures 分别是版权文件（MIT）和说明文档。

main.py 为项目的入口文件，范本是官方 SDK 的 demo。

EventDemo.py 为中间变量结构分析文件，对于项目没有实际作用。由于官方没有给出详细的 event 结构，打印出的 event 不够直观，可以通过一个辅助的 py 文件进行格式化。

utils 文件夹是核心逻辑实现的存在位置，针对不同的用户行为，可以在里面的文件中定义相应的处理逻辑。

一共提供了七种处理逻辑：

  1. BotFollowed.py:关注机器人时触发
  2. BotUnfollowed.py:取消关注机器人时触发
  3. ButtonReportInline.py:按钮点击事件，本 SDK 没有提供这部分处理逻辑，只提供了接口
  4. GroupJoin.py：加入群触发
  5. GroupLeave.py：推出群触发
  6. MessageInstruction.py：收到指令消息时触发
  7. MessageNormal.py:收到消息时触发

可以看到，utils 中还存在一个eventParse.py文件，这个文件用于解析 event 事件。以上 7 个功能可以通过这个模块复用代码。

main.py

接下来我们来逐个解析项目的关键逻辑，在此之前，你最好先阅读项目中的 README 文档，以对该项目有一个初步的了解。

1
2
3

app = Flask(__name__)
sub = Subscription()
openapi = Openapi("your api")

这一段代码实例化了三个类，sub 为 yunhu 模块中的订阅类，openapi 用于匹配你自己的 api，这个 api 可以在控制台中获取。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

@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 以上。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

# 机器人根据内容回复消息
@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)

这一部分定义了一些事件，处理逻辑我们单独写在对应的模块文件中，提高可读性。

1
2
3

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 相关的功能有两个：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

# 机器人根据内容回复消息
@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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

	# 由于不同的 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，典型结构为

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

{
    "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：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20


# 用户关注机器人
{
    "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 方法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

# 关注机器人，与加入群聊的 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 不同，因此理论上这两个函数可以合为一个，主要差别体现在以下行：

1

recvType = "user" if chatType == "bot" else "group"

这行代码决定了机器人是在群里回复还是私聊回复。

机器人运行

填入 api 后，直接运行 main.py 即可。然后将该服务穿透到公网，并将消息订阅接口配置到云湖机器人控制台。

配置消息订阅接口

随后在云湖 app 中进行功能测试。

技术
Licensed under CC BY-NC-SA 4.0
comments powered by Disqus
© 2020 - 2025 菘菓菌
使用 Hugo 构建
主题 Stack 由 Jimmy 设计