一、核心能力总览

• WebSocket 长连接，内网可运行，无需公网 IP / 域名

• 30 秒自动心跳保活，断线自动指数退避重连

• 支持监听：文本 / 图片 / 文件 / 视频 / 语音 / 引用消息

• 支持：被动回复、@指定人、主动推送、流式打字机回复

• 支持：文件 / 图片 / 语音 / 视频自动下载 + AES 解密

• 支持：群聊 @某人、@all、欢迎语、模板卡片交互

二、消息结构通用字段

{
  "msgid": "消息ID",
  "aibotid": "机器人ID",
  "chatid": "群ID/单聊会话ID",
  "chattype": "single | group",
  "from": {"userid": "发送人ID"},
  "msgtype": "text | image | file | video | voice",
  "text": {"content": "用户输入内容"},
  "quote": { "msgtype": "...", "...": "..." }
}

常用取值：

• chattype='single'：单聊

• chattype='group'：群聊

• from.userid：发送人企业微信账号

• chatid：群 ID（群聊才存在）

• quote：引用消息（引用才存在）

三、支持监听的所有消息类型

1）监听所有消息

@ws_client.on('message')
async def handle_msg(frame):
    body = frame["body"]

2）仅文本消息

@ws_client.on('message.text')

3）仅图片消息

@ws_client.on('message.image')

4）仅文件消息

@ws_client.on('message.file')

5）仅视频消息

@ws_client.on('message.video')

6）仅语音消息

@ws_client.on('message.voice')

7）引用消息（通用解析）

在任意消息内判断：

if "quote" in body:
    quote = body["quote"]
    msgtype = quote["msgtype"]  # text/image/file/video/voice

四、支持的回复方式

1）普通文本回复

await ws_client.reply(frame, {
    "msgtype": "text",
    "text": {"content": "收到"}
})

2）群聊 @指定人回复

userid = frame["body"]["from"]["userid"]
await ws_client.reply(frame, {
    "msgtype": "text",
    "text": {
        "content": f"<@{userid}> 收到",
        "mentioned_list": [userid]
    }
})

3）群聊 @所有人

"mentioned_list": ["@all"]

4）流式打字机效果回复

from aibot import generate_req_id
stream_id = generate_req_id("stream")
await ws_client.reply_stream(frame, stream_id, "思考中...", finish=False)
await ws_client.reply_stream(frame, stream_id, "完成", finish=True)

5）主动推送消息（私聊 / 群聊）

# 主动发私信
await ws_client.push_user(userid="xxx", body={...})

# 主动发群聊
await ws_client.push_group(chatid="xxx", body={...})

五、文件 / 媒体下载（全类型通用）

# 从消息/引用中获取
url = ...
aeskey = ...

data, filename = await ws_client.download_file(url, aeskey)
with open(filename, 'wb') as f:
    f.write(data)

支持：

• 图片 image

• 文件 file

• 视频 video

• 语音 voice

六、系统事件监听

@ws_client.on('connected')        # 已连接
@ws_client.on('authenticated')    # 认证成功（心跳启动）
@ws_client.on('disconnected')     # 断开
@ws_client.on('reconnecting')     # 重连中
@ws_client.on('error')            # 异常

七、业务事件监听

@ws_client.on('event.enter_chat')          # 用户进入会话
@ws_client.on('event.template_card_event') # 卡片按钮点击
@ws_client.on('event.feedback_event')      # 用户反馈

八、完整可运行 “全能模板”

from aibot import WSClient, WSClientOptions

BOT_ID = ""
BOT_SECRET = ""

ws_client = WSClient(WSClientOptions(
    bot_id=BOT_ID,
    secret=BOT_SECRET,
    heartbeat_interval=30000,
    max_reconnect_attempts=-1
))

@ws_client.on('authenticated')
def on_auth():
    print("✅ 机器人已上线")

# 文本消息
@ws_client.on('message.text')
async def on_text(frame):
    body = frame["body"]
    userid = body["from"]["userid"]
    content = body["text"]["content"]
    is_group = body["chattype"] == "group"

    # 回复
    await ws_client.reply(frame, {
        "msgtype": "text",
        "text": {
            "content": f"<@{userid}> 你说：{content}" if is_group else f"你说：{content}",
            "mentioned_list": [userid] if is_group else []
        }
    })

# 引用消息解析
@ws_client.on('message')
async def on_all(frame):
    body = frame["body"]
    if "quote" in body:
        print("📎 引用消息类型：", body["quote"]["msgtype"])

# 启动
if __name__ == "__main__":
    ws_client.run()

九、一句话总结你现在能做的所有事

• ✅ 长连接 + 心跳保活 + 自动重连

• ✅ 监听文本、图片、文件、视频、语音

• ✅ 解析引用消息（任意类型）

• ✅ 单聊 / 群聊识别

• ✅ 回复、@指定人、@all、流式回复

• ✅ 主动推送消息

• ✅ 下载解密图片 / 文件 / 视频 / 语音

• ✅ 监听进入会话、卡片点击事件