PhoneWork/docs/feishu/development_guide.md

飞书 Bot 开发经验总结

基于 lark-oapi Python SDK 的实战经验，涵盖接收消息、发送消息、文件上传、WebSocket 长连接等。

---

1. SDK 初始化

import lark_oapi as lark

client = (
    lark.Client.builder()
    .app_id(APP_ID)
    .app_secret(APP_SECRET)
    .log_level(lark.LogLevel.WARNING)
    .build()
)

client 是全局单例，线程安全，可在模块级别创建。

---

2. 接收消息：WebSocket 长连接模式

飞书推荐使用 WebSocket 长连接（而非 HTTP 回调），省去公网暴露和回调验证。

2.1 事件注册

from lark_oapi.api.im.v1 import P2ImMessageReceiveV1

handler = (
    lark.EventDispatcherHandler.builder("", "")  # 加密 key 和验证 token，长连接模式留空
    .register_p2_im_message_receive_v1(on_message)  # 注册消息回调
    .build()
)

2.2 启动 WebSocket 客户端

ws_client = lark.ws.Client(
    APP_ID,
    APP_SECRET,
    event_handler=handler,
    log_level=lark.LogLevel.INFO,
)
ws_client.start()  # 阻塞调用

2.3 关键坑点：事件循环

lark_oapi.ws.client 在 import 时捕获当前事件循环。如果你的主程序用了 asyncio（如 uvicorn）， 必须在 WebSocket 线程中创建新的事件循环并替换：

import lark_oapi.ws.client as _lark_ws_client

thread_loop = asyncio.new_event_loop()
asyncio.set_event_loop(thread_loop)
_lark_ws_client.loop = thread_loop  # 关键：替换 SDK 内部持有的 loop

2.4 断线重连

ws_client.start() 在连接断开时会直接返回（不抛异常），需要自己包裹重连循环：

backoff = 1.0
while True:
    try:
        ws_client.start()
    except Exception:
        pass
    time.sleep(backoff)
    backoff = min(backoff * 2, 60.0)

2.5 解析收到的消息

def on_message(data: P2ImMessageReceiveV1):
    event = data.event
    message = event.message
    sender = event.sender

    chat_id = message.chat_id          # 会话 ID（群/私聊）
    msg_type = message.message_type    # "text", "image", "file", ...
    content = json.loads(message.content)  # 消息内容是 JSON 字符串
    text = content.get("text", "")

    # 发送者信息
    open_id = sender.sender_id.open_id  # 用户唯一标识

注意：

• message.content 是 JSON 字符串，不是纯文本，需要 json.loads
• 群聊中 @bot 的消息会包含 @xxx 前缀，需要用正则清除：re.sub(r"@\S+\s*", "", text)
• 飞书遇到网络问题会在 ~60s 内重发相同消息，需要做去重（按 (user_id, content) + 时间窗口）

---

3. 发送消息

所有发送都通过 client.im.v1.message.create，通过 msg_type 和 content 区分消息类型。

3.1 receive_id_type 的选择

值	含义	用途
"chat_id"	会话 ID	回复到当前会话（群聊或私聊）
"open_id"	用户 open_id	主动私聊某个用户（如通知）
"user_id"	用户 user_id	同上，另一种 ID 体系

经验：回复消息用 chat_id，主动推送通知用 open_id。

3.2 发送纯文本

content = json.dumps({"text": "Hello"}, ensure_ascii=False)
request = (
    CreateMessageRequest.builder()
    .receive_id_type("chat_id")
    .request_body(
        CreateMessageRequestBody.builder()
        .receive_id(chat_id)
        .msg_type("text")
        .content(content)
        .build()
    )
    .build()
)
response = client.im.v1.message.create(request)

3.3 发送卡片消息（Markdown）

卡片消息是飞书中展示富文本的主要方式。msg_type 为 "interactive"，content 为卡片 JSON。

JSON 2.0 Markdown 卡片（推荐）

card = {
    "schema": "2.0",
    "body": {
        "elements": [
            {
                "tag": "markdown",
                "content": "**粗体** *斜体* `code`\n- 列表项\n```python\nprint('hello')\n```",
            }
        ]
    }
}
content = json.dumps(card, ensure_ascii=False)
# msg_type = "interactive"

JSON 1.0 卡片（旧版，带标题栏）

card = {
    "config": {"wide_screen_mode": True},
    "header": {
        "title": {"tag": "plain_text", "content": "标题"},
        "template": "turquoise",  # 标题栏颜色
    },
    "elements": [
        {"tag": "div", "text": {"tag": "lark_md", "content": "**markdown** 内容"}},
    ],
}

JSON 1.0 vs 2.0 的区别：

• 1.0 用 lark_md tag + div 容器，2.0 直接用 markdown tag
• 2.0 支持标准 CommonMark，1.0 有较多语法限制
• 2.0 支持表格、标题（# ## ###），1.0 不支持
• 需要标题栏时用 1.0 的 header；纯内容展示用 2.0 更简洁

3.4 发送文件

两步流程：先上传文件获取 file_key，再发送文件消息。

from lark_oapi.api.im.v1 import CreateFileRequest, CreateFileRequestBody

# Step 1: 上传
with open(path, "rb") as f:
    req = (
        CreateFileRequest.builder()
        .request_body(
            CreateFileRequestBody.builder()
            .file_type("stream")  # "stream", "opus", "mp4", "pdf", "doc", "xls", "ppt"
            .file_name(file_name)
            .file(f)
            .build()
        )
        .build()
    )
    resp = client.im.v1.file.create(req)
    file_key = resp.data.file_key

# Step 2: 发送
content = json.dumps({"file_key": file_key}, ensure_ascii=False)
# msg_type = "file"

---

4. Markdown 语法速查（JSON 2.0 卡片）

JSON 2.0 支持标准 CommonMark（除 HTMLBlock），外加飞书扩展语法。

标准语法

效果	语法
粗体	**text**
斜体	*text*
删除线	~~text~~
行内代码	`code`
代码块	```python\ncode\n```
链接	[text](url)
图片	
有序列表	1. item (4 空格缩进为子项)
无序列表	- item
引用	> text
标题	# ~ ###### (1-6 级)
分割线	--- 或 <hr>
表格	标准 Markdown 表格语法

飞书扩展语法

效果	语法
@某人	<at id=open_id></at>
@所有人	<at id=all></at>
彩色文本	<font color='red'>红色</font>
标签	<text_tag color='blue'>标签</text_tag>
飞书表情	:DONE: :THUMBSUP:
带图标链接	<link icon='chat_outlined' url='...'>文本</link>
电话链接	[显示文本](tel://号码) (仅移动端)

特殊字符转义

Markdown 特殊字符（* ~ > < [ ] ( ) # : + " ' 等）需要 HTML 实体转义才能原样显示：

字符	转义
*	*
<	&#60;
>	&#62;
~	&sim;
`	&#96;

完整对照参考 HTML 转义标准。

---

5. 长消息处理

飞书单条文本消息有长度限制（约 4000 字符），卡片消息整体 JSON 约 28KB。超长内容需要拆分：

MAX_TEXT_LEN = 3900

def split_message(text: str) -> list[str]:
    """在换行符处分割，避免截断代码块或段落中间。"""
    if len(text) <= MAX_TEXT_LEN:
        return [text]
    parts = []
    remaining = text
    while remaining:
        if len(remaining) <= MAX_TEXT_LEN:
            parts.append(remaining)
            break
        chunk = remaining[:MAX_TEXT_LEN]
        last_newline = chunk.rfind("\n")
        if last_newline > MAX_TEXT_LEN // 2:
            chunk = remaining[:last_newline + 1]
        parts.append(chunk)
        remaining = remaining[len(chunk):]
    return parts

多段发送时加 await asyncio.sleep(0.3) 避免飞书限流。

---

6. 异步兼容

lark-oapi 的 SDK 方法全是同步的（阻塞 I/O），在 asyncio 环境中需要用 run_in_executor：

loop = asyncio.get_running_loop()
response = await loop.run_in_executor(
    None,
    lambda: client.im.v1.message.create(request),
)

---

7. 错误处理

所有 API 调用检查 response.success()：

if not response.success():
    logger.error("code=%s msg=%s", response.code, response.msg)

常见错误码：

• 权限不足：机器人未添加到群聊，或未开通对应 API 权限
• receive_id 无效：ID 类型与 receive_id_type 不匹配
• 内容过长：消息体超出限制

---

8. 实用模式总结

场景	msg_type	发送函数
系统通知、短消息	text	send_text
LLM/AI 回复（含代码、列表）	interactive	send_markdown (JSON 2.0 卡片)
结构化信息（带标题栏）	interactive	send_card (JSON 1.0 卡片)
文件传输	file	send_file (先上传再发送)
主动推送给用户	同上	receive_id_type="open_id"
回复当前会话	同上	receive_id_type="chat_id"