ErisPulse YunhuUserAdapter

云湖用户账户驱动适配器 —— 用个人账户驱动 Bot。

基于云湖用户账户协议实现的 ErisPulse 适配器，支持多账户、WebSocket 长连接与自动重连，提供文本 / 图片 / 视频 / 文件 / 语音 / Markdown / HTML / A2UI 等完整消息收发能力。

安装

pip install ErisPulse-YunhuUserAdapter

使用方法

基础使用

from ErisPulse.Core import adapter

# 获取适配器实例
yunhu = adapter.get("yunhu_user")

# 发送文本消息
await yunhu.Send.To("user", "user_id").Text("Hello World!")

# 发送到群聊
await yunhu.Send.To("group", "group_id").Text("群消息")

# 发送图片
await yunhu.Send.To("group", "group_id").Image("https://example.com/image.jpg")

多账户使用

from ErisPulse.Core import adapter

yunhu = adapter.get("yunhu_user")

# 使用指定账户发送消息
await yunhu.Send.Using("default").To("group", "group_id").Text("来自默认账户的消息")

# 使用第二个账户发送消息
await yunhu.Send.Using("account2").To("user", "user_id").Text("来自账户2的消息")

# 或者使用 Account 方法（与 Using 等效）
await yunhu.Send.Account("account2").To("group", "group_id").Text("消息内容")

链式修饰

# 回复消息
await yunhu.Send.To("group", "group_id").Reply("message_id").Text("回复内容")

await yunhu.Send.To("group", "group_id").Reply("message_id").At("user_id").Text("@用户并回复")

# 因为用户账户较为特殊，即便你不是管理员，也可以@全体 故这里只会发送一个艾特全体的文本
# 这是一个伪@全体
await yunhu.Send.To("group", "group_id").Reply("message_id").AtAll().Text("@全体并回复")

# 添加按钮
buttons = [
    [
        {"text": "确认", "actionType": 3, "value": "confirm"},
        {"text": "取消", "actionType": 3, "value": "cancel"}
    ]
]
await yunhu.Send.To("group", "group_id").Buttons(buttons).Text("请选择")

# 组合使用
await yunhu.Send.Using("default").To("group", "group_id").Reply("msg_id").Buttons(buttons).Text("回复消息")

消息类型

# 文本消息
await yunhu.Send.To("user", "user_id").Text("纯文本消息")

# Markdown 消息
await yunhu.Send.To("group", "group_id").Markdown("**粗体** 和 *斜体*")

# HTML 消息
await yunhu.Send.To("group", "group_id").Html("<b>加粗</b> 和 <i>斜体</i>")

# A2UI 消息
await yunhu.Send.To("group", "group_id").A2ui({"type": "card", "content": "..."})

# 编辑消息
await yunhu.Send.To("group", "group_id").Edit("message_id", "新内容")

# 编辑消息（指定内容类型）
await yunhu.Send.To("group", "group_id").Edit("message_id", "新内容", content_type="markdown")

# 撤回消息
await yunhu.Send.To("group", "group_id").Recall("message_id")

# 以下媒体消息均支持URL和二进制传入

# 图片消息（URL）
await yunhu.Send.To("group", "group_id").Image("https://example.com/image.jpg")

# 视频消息
await yunhu.Send.To("group", "group_id").Video("https://example.com/video.mp4")

# 文件消息
await yunhu.Send.To("group", "group_id").File("https://example.com/document.pdf")

# 语音消息
await yunhu.Send.To("group", "group_id").Audio("https://example.com/audio.mp3")

# 表情消息（URL 或 表情ID 或 二进制数据）
await yunhu.Send.To("group", "group_id").Face("https://example.com/sticker.png")

OneBot12 原始消息

# 发送 OneBot12 格式消息段
ob12_message = [
    {"type": "text", "data": {"text": "Hello"}},
    {"type": "image", "data": {"file_id": "image_id"}}
]
await yunhu.Send.To("group", "group_id").Raw_ob12(ob12_message)

配置

在 config.toml 中配置：

[YunhuUserAdapter]
ws_reconnect_interval = 30
ws_timeout = 70

[YunhuUserAdapter.accounts.default]
# 登录方式二选一：直接填 token，或填邮箱+密码（两者同时配置时优先使用 token）
email = "your_email@example.com"
password = "your_password"
token = ""
platform = "windows"
device_id = ""
enabled = true

配置说明

事件监听

from ErisPulse.Core.Event import message, command, notice

# 消息事件
@message.on_message()
async def message_handler(event):
    if event["platform"] == "yunhu_user":
        await event.reply("Echo: " + event.get_alt_message())

# 命令处理
@command("hello", help="发送问候")
async def hello_command(event):
    await event.reply("你好！我是云湖用户机器人。")

事件格式

消息事件

{
    "id": "event_id",
    "time": 1234567890,
    "type": "message",
    "detail_type": "group",  # 或 "private"
    "platform": "yunhu_user",
    "self": {
        "platform": "yunhu_user",
        "user_id": "your_user_id"
    },
    "message": [
        {"type": "text", "data": {"text": "消息内容"}}
    ],
    "alt_message": "消息内容",
    "user_id": "sender_user_id",
    "user_nickname": "发送者昵称",
    "group_id": "group_id",   # 群消息时包含
    "yunhu_user_raw": {...},  # 原始事件数据
    "yunhu_user_raw_type": "push_message"  # 原始事件类型
}

平台特有字段

所有云湖用户特有字段都以 yunhu_user_ 前缀标识：

• yunhu_user_raw: 原始事件数据（包含完整的云湖协议数据）
• yunhu_user_raw_type: 原始事件类型（如 push_message、edit_message 等）

API 响应格式

适配器的 call_api 方法返回符合 OneBot12 的响应格式：

{
    "status": "ok",  # 或 "failed"
    "retcode": 0,  # 0 表示成功，其他为错误码
    "data": {...},  # 响应数据
    "message_id": "message_id",  # 消息ID
    "message": "",  # 错误信息（成功时为空）
    "yunhu_user_raw": {...}  # 原始响应数据
}

事件处理范围

适配器支持了以下云湖 WebSocket 事件：

其他事件类型（如 heartbeat_ack、draft_input 等）会被忽略。

OneBot12 支持的 detail_type

适配器将云湖消息转换为以下 OneBot12 类型：

开发状态

• 基础适配器结构
• 多账户支持（AccountConfigClass 声明式配置）
• WebSocket 连接和消息接收（使用 sdk.client.ws_connect）
• 自动重连机制
• 事件转换（OneBot12 格式）
• 消息发送（文本、图片、视频、文件、语音、Markdown、HTML、A2UI、表情）
• 链式修饰（Reply、Buttons、At、AtAll）
• 消息管理（Edit、Recall）
• 用户邮箱登录
• Raw_ob12 原始消息发送
• sdk.client 统一 HTTP 客户端（移除直接 aiohttp 依赖）
• sdk.client.ws_connect 统一 WebSocket 客户端（移除直接 websockets 依赖）

致谢

• yhchatAPI
• yh_user_sdk

相关链接

• ErisPulse
• ErisPulse-YunhuAdapter