飞书机器人 (lark_bot)

pywayne.lark_bot 是应用机器人侧的主能力封装，覆盖主动发消息、引用回复、消息编辑、撤回、reaction、置顶、群管理、资源上传下载、历史消息查询等常见飞书 IM 能力。

如果把整个模块按对象分层，可以理解为：

• TextContent: 文本格式工具，解决 @、加粗、链接等轻量文本拼接

• PostContent: rich_text 富文本构造器，适合做结构化文本、代码块、Markdown 表格降级

• CardContentV2: schema 2.0 卡片构造器，适合大段 Markdown 和轻交互展示

• LarkBot: 统一 API 封装，负责真正和飞书 OpenAPI 通信

能力概览

LarkBot 当前重点覆盖这些能力：

• 主动发送消息

  • 文本

  • 图片

  • 音频

  • 媒体

  • 文件

  • rich_text 富文本

  • card 卡片

  • 分享群聊

  • 分享用户

  • 系统消息

  • Markdown 自动路由发送

• 消息后处理

  • 引用回复

  • 线程内回复

  • 转发消息

  • 撤回消息

  • 获取单条消息

  • 获取历史消息列表

  • 编辑文本 / rich_text / card 消息

  • 原位更新 card 卡片

  • 查询已读用户

  • 加急消息

  • 加 / 删 / 查 reaction

  • 置顶 / 取消置顶 / 查询置顶

• 群管理

  • 创建群聊

  • 删除群聊

  • 修改群名称 / 描述 / 头像 / 群主

  • 拉人入群

  • 移出成员

  • 设置 / 取消管理员

  • 转让群主

  • 获取群公告

  • patch 群公告

• 资源与基础信息

  • 上传 / 下载图片

  • 上传 / 下载文件

  • 下载消息资源

  • 批量下载消息内资源

  • 查询用户信息

  • 查询群列表

  • 通过群名找群 ID

  • 取群成员

  • 通过成员名找 open_id

  • 获取群名和用户名

重点能力

如果你想快速抓住这个模块最核心的扩展能力，优先关注这几组：

• 消息生命周期补齐

  • reply_message

  • forward_message

  • recall_message

  • get_message

  • get_message_list

  • edit_text_message

  • edit_post_message

  • edit_card_message

• 消息状态操作

  • get_message_read_users

  • urgent_message

  • add_reaction / delete_reaction / list_reactions

  • pin_message / unpin_message / list_pinned_messages

• 流式卡片回复

  • reply_streaming_card

  • update_streaming_card

  • recolor_streaming_card

  • stream_reply_card

  • astream_reply_card

• 群管理增强

  • create_chat / update_chat / delete_chat

  • add_members_to_chat / remove_members_from_chat

  • set_chat_admin / transfer_chat_owner

  • get_chat_announcement / set_chat_announcement

• 批量发消息

  • batch_send_message 支持按用户 / 部门触达

如果你在做机器人业务闭环，最常见的组合一般是：

• 收到用户消息 -> reply_message 引用回复

• 长任务开始 -> reply_streaming_card -> 多次 update_streaming_card -> 完成后 recolor_streaming_card(..., template="green")

• 告警消息 -> forward_message 给值班人 -> urgent_message 加急

• 重要结论 -> reply_message -> pin_message

• 任务处理中 -> add_reaction -> 完成 / 失败后 delete_reaction

• 项目群初始化 -> create_chat -> add_members_to_chat -> set_chat_admin -> send_card_to_chat

快速开始

from pywayne.lark_bot import LarkBot

bot = LarkBot(app_id="cli_xxx", app_secret="sec_xxx")

bot.send_text_to_user("ou_xxx", "你好")
bot.send_text_to_chat("oc_xxx", "群消息测试")

TextContent

TextContent 适合快速拼出飞书文本消息里常见的富文本标记。

常用方法：

• make_at_all_pattern()

• make_at_someone_pattern(someone_open_id, username, id_type)

• make_bold_pattern(content)

• make_italian_pattern(content)

• make_underline_pattern(content)

• make_delete_line_pattern(content)

• make_url_pattern(url, text)

示例：构造一条带 @ 和链接的文本

from pywayne.lark_bot import LarkBot, TextContent

bot = LarkBot(app_id="cli_xxx", app_secret="sec_xxx")

msg = (
    TextContent.make_at_someone_pattern("ou_xxx", "Wayne", "open_id")
    + " "
    + TextContent.make_bold_pattern("发布完成")
    + "，请查看 "
    + TextContent.make_url_pattern("https://example.com", "详情页")
)
bot.send_text_to_chat("oc_xxx", msg)

PostContent

PostContent 适合构造复杂 rich_text 富文本，尤其适合：

• 多段结构化说明

• 一行混排文字 / 链接 / @ / 图片

• 代码块

• Markdown 分块发送

• Markdown 表格降级为等宽代码块

常用方法：

• make_text_content

• make_link_content

• make_at_content

• make_image_content

• make_media_content

• make_emoji_content

• make_hr_content

• make_code_block_content

• make_markdown_content

• add_markdown

• add_content_in_line

• add_contents_in_line

• add_content_in_new_line

• add_contents_in_new_line

• list_emoji_types

示例：混排 rich_text 消息

from pywayne.lark_bot import LarkBot, PostContent

bot = LarkBot(app_id="cli_xxx", app_secret="sec_xxx")

post = PostContent(title="发布通知")
post.add_contents_in_new_line([
    post.make_text_content("构建完成", styles=["bold"]),
    post.make_emoji_content("OK"),
])
post.add_contents_in_new_line([
    post.make_link_content("查看 Jenkins", "https://jenkins.example.com"),
    post.make_at_content("ou_xxx"),
])
post.add_content_in_new_line(
    post.make_code_block_content("bash", "deploy.sh --env prod")
)

bot.send_rich_text_to_chat("oc_xxx", post.get_content())

示例：Markdown 表格安全降级

md = """
## 回归结果

| 模块 | 结果 | 负责人 |
| --- | --- | --- |
| 登录 | pass | Alice |
| 支付 | pass | Bob |
| 推荐 | running | Carol |
"""

post = PostContent(title="测试日报")
post.add_markdown(md, table_as="code_block", max_chunk_bytes=8000)
bot.send_rich_text_to_chat("oc_xxx", post.get_content())

CardContentV2

CardContentV2 是面向 schema 2.0 card 卡片的轻量构造器，适合大段 Markdown 公告、日报、状态展示。

常用方法：

• add_markdown

• add_hr

• add_image

• get_card

• list_header_templates

示例：日报卡片

from pywayne.lark_bot import CardContentV2

card = CardContentV2(title="日报", template="blue")
card.add_markdown("# 今日进展\n\n- 接口联调完成\n- 修复 3 个问题\n- 代码已合并")
card.add_hr()
card.add_image("img_xxx")

bot.send_card_to_chat("oc_xxx", card.get_card())

示例：查看常用卡片头部模板色

from pywayne.tools import wayne_print

wayne_print(CardContentV2.list_header_templates(), color="cyan")

LarkBot 类

class LarkBot(app_id: str, app_secret: str)

主应用机器人封装。

发送消息

文本

• send_text_to_user(user_open_id, text)

• send_text_to_chat(chat_id, text)

bot.send_text_to_user("ou_xxx", "私聊你好")
bot.send_text_to_chat("oc_xxx", "群里你好")

图片

• upload_image(image_path)

• send_image_to_user(user_open_id, image_key)

• send_image_to_chat(chat_id, image_key)

• download_image(image_key, image_save_path)

image_key = bot.upload_image("/tmp/report.png")
bot.send_image_to_chat("oc_xxx", image_key)

音频 / 媒体 / 文件

• upload_file(file_path, file_type="stream")

• send_audio_to_user / send_audio_to_chat

• send_media_to_user / send_media_to_chat

• send_file_to_user / send_file_to_chat

• download_file(file_key, file_save_path)

opus_key = bot.upload_file("/tmp/voice.opus", file_type="opus")
bot.send_audio_to_chat("oc_xxx", opus_key)

mp4_key = bot.upload_file("/tmp/demo.mp4", file_type="mp4")
bot.send_media_to_chat("oc_xxx", mp4_key)

pdf_key = bot.upload_file("/tmp/spec.pdf", file_type="pdf")
bot.send_file_to_chat("oc_xxx", pdf_key)

rich_text 富文本

• send_rich_text_to_user

• send_rich_text_to_chat

post = PostContent(title="值班提醒")
post.add_content_in_new_line(post.make_text_content("今晚 20:00 发布", styles=["bold"]))
bot.send_rich_text_to_chat("oc_xxx", post.get_content())

card 卡片

• send_card_to_user

• send_card_to_chat

bot.send_card_to_chat(
    "oc_xxx",
    {
        "header": {"title": {"content": "状态卡片", "tag": "plain_text"}},
        "elements": [{"tag": "markdown", "content": "**服务正常**"}]
    }
)

分享消息

• share_chat_to_user / share_chat_to_chat

• share_user_to_user / share_user_to_chat

系统消息

• send_system_message_to_user

适合做系统通知类场景，不走普通文本样式。

推荐入口：send_markdown_message_to_chat

send_markdown_message_to_chat(chat_id: str, md_text: str, *, title: str = '', prefer: str = 'card_v2', table_fallback: str = 'code_block', max_message_bytes: int | None = None)

这是推荐的高层发送入口，适合绝大多数“我要发 Markdown”场景。

特性：

• 自动按字节分包

• 支持 card_v2 与 post 两条发送路由

• post 路由下支持表格降级

示例 1：默认发 schema 2.0 卡片

bot.send_markdown_message_to_chat(
    "oc_xxx",
    md_text="# 发布完成\n\n- API: pass\n- Worker: pass",
    title="发布结果"
)

示例 2：强制走 rich_text 路由，并处理 Markdown 表格

bot.send_markdown_message_to_chat(
    "oc_xxx",
    md_text="""
    # 回归看板

    | 模块 | 状态 |
    | --- | --- |
    | 登录 | pass |
    | 支付 | pass |
    """,
    title="回归结果",
    prefer="post",
    table_fallback="code_block"
)

示例 3：超长日报自动分片

bot.send_markdown_message_to_chat(
    "oc_xxx",
    md_text=very_long_markdown,
    title="长文日报",
    prefer="card_v2",
    max_message_bytes=12000
)

消息后处理

引用回复

reply_message(message_id: str, msg_type: str, content, *, reply_in_thread: bool = False, uuid: str = '')

这是监听场景里最常用的方法。

示例：引用回复文本

bot.reply_message(
    message_id="om_xxx",
    msg_type="text",
    content={"text": "收到文本"}
)

示例：在线程中回复

bot.reply_message(
    message_id="om_xxx",
    msg_type="text",
    content={"text": "这条回复会进入 thread"},
    reply_in_thread=True
)

示例：引用回复卡片

card = CardContentV2(title="处理结果")
card.add_markdown("已收到你的请求，正在执行。")
bot.reply_message("om_xxx", "interactive", card.get_card())

示例：按消息类型动态引用回复

def reply_by_type(message_id: str, kind: str):
    mapping = {
        "text": "收到文本",
        "image": "收到图片",
        "file": "收到文件",
        "audio": "收到音频",
        "post": "收到 rich_text",
        "interactive": "收到 card",
    }
    bot.reply_message(
        message_id,
        "text",
        {"text": mapping.get(kind, f"收到 {kind}")}
    )

转发消息

forward_message(message_id: str, receive_id: str, *, receive_id_type: str = 'chat_id', uuid: str = '')

示例：把群里的告警转发给值班人

bot.forward_message(
    message_id="om_alert_xxx",
    receive_id="ou_duty_xxx",
    receive_id_type="open_id"
)

示例：先引用回复，再把原始消息转发给二线支持

bot.reply_message("om_xxx", "text", {"text": "已转交二线支持"})
bot.forward_message("om_xxx", "ou_level2_xxx", receive_id_type="open_id")

撤回消息

recall_message(message_id: str)

适合“发错消息立即撤回”或“临时状态消息在成功后撤回”。

示例：短暂提示后立即撤回

sent = bot.send_text_to_chat("oc_xxx", "这是一条临时提示")
bot.recall_message(sent["message_id"])

查询消息

• get_message(message_id)

• get_message_list(chat_id, start_time, end_time, sort_type="", page_size=50, page_token="")

示例：拉取某时间范围内的历史消息

history = bot.get_message_list(
    chat_id="oc_xxx",
    start_time="1735603200000",
    end_time="1735689600000",
    sort_type="ByCreateTimeAsc"
)

示例：先查单条消息，再决定是否转发

detail = bot.get_message("om_xxx")
content = detail["body"]["content"]
if "紧急" in content:
    bot.forward_message("om_xxx", "ou_duty_xxx", receive_id_type="open_id")

编辑消息

• edit_text_message(message_id, text)

• edit_post_message(message_id, post_content)

• edit_card_message(message_id, card)

注意：

• edit_text_message / edit_post_message 底层对应飞书 PUT /im/v1/messages/:message_id，只适用于 text / post 消息，也就是文本 / rich_text 消息

• edit_card_message 底层对应飞书卡片更新接口，适用于交互式卡片；做“同一条消息持续刷新”的效果时，通常也是走卡片更新

• 飞书官方限制一条消息最多编辑 20 次；且只能编辑自己发送、未撤回、未超出可编辑时间的消息

• 文本 / 富文本和卡片更新接口是分开的，不能混用；例如不能拿卡片 JSON 去调 edit_text_message / edit_post_message

• 如果你只是想按消息类型做原位编辑，优先使用这 3 个 edit_* 接口，不再需要区分旧的底层方法名

示例：把“处理中”卡片更新成“已完成”

bot.edit_card_message(
    message_id="om_xxx",
    card={
        "type": "template",
        "data": {
            "template_id": "AAqC5c999",
            "template_variable": {"status": "已完成"}
        }
    }
)

示例：先发文本，再更新文本内容

sent = bot.send_text_to_chat("oc_xxx", "初版结论")
bot.edit_text_message(sent["message_id"], "初版结论\n补充说明：影响范围仅限灰度环境")

示例：先发富文本，再更新富文本内容

post = PostContent(title="阶段播报")
post.add_markdown("第一版结果")
sent = bot.send_rich_text_to_chat("oc_xxx", post.get_content())

post2 = PostContent(title="阶段播报")
post2.add_markdown("第一版结果\n\n补充：已完成二次校验")
bot.edit_post_message(sent["message_id"], post2.get_content())

示例：直接编辑已发送的卡片

bot.edit_card_message(
    "om_xxx",
    {
        "type": "template",
        "data": {
            "template_id": "AAqC5c999",
            "template_variable": {"status": "处理中", "detail": "第 2 步 / 共 5 步"}
        }
    }
)

已读与加急

• get_message_read_users(message_id, ...)

• urgent_message(message_id, urgent_type, user_open_ids, ...)

urgent_type 支持：

• app

• phone

• sms

示例：先发消息，再给值班人加急

sent = bot.send_text_to_chat("oc_xxx", "生产告警，请处理")
bot.urgent_message(
    sent["message_id"],
    urgent_type="app",
    user_open_ids=["ou_duty_xxx"]
)

示例：查询哪些人已读，再决定是否继续催办

readers = bot.get_message_read_users("om_xxx")
if not readers.get("items"):
    bot.urgent_message("om_xxx", "sms", ["ou_duty_xxx"])

reaction

• add_reaction(message_id, emoji_type)

• delete_reaction(message_id, reaction_id)

• list_reactions(message_id, ...)

emoji_type 使用飞书 reaction code，不是 Unicode 字符。

常用值包括：

• THUMBSUP

• OK

• HEART

• HAHA

• WITTY

完整列表可通过 PostContent.list_emoji_types() 打开飞书官方表情页。

示例：临时 reaction

reaction = bot.add_reaction("om_xxx", "THUMBSUP")
bot.delete_reaction("om_xxx", reaction["reaction_id"])

示例：统计某条消息有哪些 reaction

from pywayne.tools import wayne_print

data = bot.list_reactions("om_xxx")
wayne_print(data, color="cyan")

示例：把 reaction 当作任务状态灯

reaction = bot.add_reaction("om_xxx", "WITTY")
try:
    bot.reply_message("om_xxx", "text", {"text": "处理中"})
except Exception:
    bot.add_reaction("om_xxx", "HAHA")
    raise
finally:
    bot.delete_reaction("om_xxx", reaction["reaction_id"])

置顶

• pin_message(message_id)

• unpin_message(message_id)

• list_pinned_messages(chat_id, ...)

示例：先回复，再置顶

reply = bot.reply_message("om_xxx", "text", {"text": "这是最终结论"})
bot.pin_message(reply["message_id"])

示例：查询置顶消息并输出摘要

from pywayne.tools import wayne_print

pinned = bot.list_pinned_messages("oc_xxx")
for item in pinned.get("items", []):
    wayne_print(item["message_id"], color="cyan")

流式卡片回复

这一组方法用于“先回复一张卡片，再不断原位刷新同一张卡片内容”，适合接 LLM 流式输出。

• build_streaming_card(md_text, ...)

• reply_streaming_card(message_id, ...)

• update_streaming_card(message_id, md_text, ...)

• recolor_streaming_card(message_id, md_text, ...)

• stream_reply_card(source_message_id, text_stream, ...)

• astream_reply_card(source_message_id, text_stream, ...)

设计约束：

• 走的是“先发 card 卡片，再反复更新该消息”的路径

• 更新时传入的是“当前完整文本”，不是 delta

• 默认 update_interval=0.25，是为了尽量避开单消息高频更新限制

• 卡片默认带 config.update_multi=true

• 颜色不是任意 RGB，而是飞书卡片头部的预设模板色

• final_template 用于流式结束后的最终颜色

• recolor_streaming_card 适合手动切换为成功 / 失败 / 警告状态色

常用头部模板色：

• blue

• wathet

• turquoise

• green

• yellow

• orange

• red

• carmine

• violet

• purple

• indigo

• grey

你也可以直接调用 CardContentV2.list_header_templates() 获取当前内置常用列表。

示例 1：手工启动 + 手工更新

reply = bot.reply_streaming_card(
    "om_xxx",
    title="AI 回复中",
    initial_md="正在思考..."
)

card_message_id = reply["message_id"]
bot.update_streaming_card(card_message_id, "第一段输出", title="AI 回复中")
bot.update_streaming_card(card_message_id, "完整输出内容", title="AI 回复完成", done=True)

示例 2：完成后自动从蓝色切到绿色

def fake_stream():
    yield "第一段输出\\n"
    yield "第二段输出\\n"
    yield "第三段输出\\n"

result = bot.stream_reply_card(
    "om_xxx",
    fake_stream(),
    title="AI 回复中",
    template="blue",
    final_template="green",
    status_text="生成中...",
    final_status_text="已完成"
)

示例 3：失败时改成红色

reply = bot.reply_streaming_card(
    "om_xxx",
    title="任务执行中",
    template="blue",
    initial_md="开始执行..."
)

card_message_id = reply["message_id"]

try:
    bot.update_streaming_card(card_message_id, "步骤 1 成功\\n步骤 2 成功", title="任务执行中")
    raise RuntimeError("第三步失败")
except Exception as exc:
    bot.recolor_streaming_card(
        card_message_id,
        f"步骤 1 成功\\n步骤 2 成功\\n\\n错误信息：{exc}",
        title="任务执行失败",
        template="red",
        status_text="执行失败",
        done=True
    )

示例 4：逐字更新，完成后自动变绿色

import time

def char_stream():
    sentence = "这是一个逐字流式卡片回复示例。"
    for ch in sentence:
        yield ch
        time.sleep(0.3)

bot.stream_reply_card(
    "om_xxx",
    char_stream(),
    title="逐字输出中",
    template="blue",
    final_template="green",
    status_text="生成中...",
    final_status_text="已完成",
    update_interval=0.25
)

示例 5：处理中为蓝色，待人工确认切橙色，最终完成切绿色

reply = bot.reply_streaming_card(
    "om_xxx",
    title="审批任务",
    template="blue",
    initial_md="系统已开始自动分析"
)

card_message_id = reply["message_id"]
bot.update_streaming_card(
    card_message_id,
    "自动分析完成，等待人工确认",
    title="审批任务",
    template="orange",
    done=False,
    status_text="等待人工确认"
)
bot.recolor_streaming_card(
    card_message_id,
    "自动分析完成\n人工确认通过",
    title="审批任务",
    template="green",
    status_text="已完成"
)

示例 6：直接消费同步生成器

def fake_stream():
    yield "你好，"
    yield "这是"
    yield "一段流式输出。"

result = bot.stream_reply_card(
    "om_xxx",
    fake_stream(),
    title="AI 回复中"
)

示例 7：异步生成器版本

async def fake_astream():
    yield "第一段"
    yield "第二段"

result = await bot.astream_reply_card(
    "om_xxx",
    fake_astream(),
    title="AI 回复中",
    template="wathet",
    final_template="green",
    final_status_text="回答完成"
)

群管理

创建 / 删除 / 更新群

• create_chat

• delete_chat

• update_chat

• transfer_chat_owner

示例：创建一个项目群并设置群主

chat = bot.create_chat(
    name="项目 Alpha",
    user_open_ids=["ou_a", "ou_b", "ou_c"],
    description="Alpha 项目协作群",
    owner_open_id="ou_a"
)

示例：修改群资料

bot.update_chat(
    chat_id="oc_xxx",
    name="项目 Alpha - 灰度群",
    description="用于灰度发布值守"
)

示例：转让群主后再更新群描述

bot.transfer_chat_owner("oc_xxx", "ou_new_owner")
bot.update_chat(
    chat_id="oc_xxx",
    description="群主已更新，请按新流程协作"
)

成员管理

• add_members_to_chat

• remove_members_from_chat

• set_chat_admin

示例：加人并设置管理员

bot.add_members_to_chat("oc_xxx", ["ou_dev1", "ou_dev2"])
bot.set_chat_admin("oc_xxx", ["ou_dev1"], is_admin=True)

示例：移除管理员权限

bot.set_chat_admin("oc_xxx", ["ou_dev1"], is_admin=False)

示例：拉人、踢人、再补发说明

bot.add_members_to_chat("oc_xxx", ["ou_new1", "ou_new2"])
bot.remove_members_from_chat("oc_xxx", ["ou_old1"])
bot.send_text_to_chat("oc_xxx", "成员已调整，请关注新的值班安排")

群公告

• get_chat_announcement

• set_chat_announcement

set_chat_announcement 走的是飞书 patch 语义，参数 requests 需要直接传飞书公告 API 的 patch 操作列表。

示例：读取公告

announcement = bot.get_chat_announcement("oc_xxx")

示例：用 patch 语义更新公告

bot.set_chat_announcement(
    "oc_xxx",
    requests=[
        {
            "op": "replace",
            "path": "/content",
            "value": "今晚 23:00 维护，请提前保存工作内容。"
        }
    ]
)

资源与下载

下载单条消息中的资源

• download_message_resource(message_id, resource_type, save_path, file_key=None)

resource_type 常见取值：

• image

• file

• audio

• media

• video

示例：下载图片消息中的原图

bot.download_message_resource(
    message_id="om_xxx",
    resource_type="image",
    save_path="/tmp/msg.png",
    file_key="img_xxx"
)

批量下载消息中的全部资源

• download_message_resources(message_id, message_content, save_dir)

适合“你已经拿到消息正文 JSON，想把里面所有 file_key/image_key 对应的资源一次性落盘”。

示例：下载一条消息中的全部资源后统一归档

import json

detail = bot.get_message("om_xxx")
content = json.loads(detail["body"]["content"])
bot.download_message_resources("om_xxx", content, "/tmp/msg_assets")

用户与群信息

• get_user_info(emails, mobiles)

• get_group_list()

• find_chat_ids_by_name(group_name)

• get_chat_members(chat_id)

• find_member_open_ids_by_name(chat_id, member_name)

• get_chat_and_user_name(chat_id, user_id)

示例：按群名查 ID，再按用户名查成员 open_id

chat_ids = bot.find_chat_ids_by_name("项目 Alpha")
if chat_ids:
    open_ids = bot.find_member_open_ids_by_name(chat_ids[0], "Wayne")

示例：同时查群名和发件人姓名

group_name, user_name = bot.get_chat_and_user_name("oc_xxx", "ou_xxx")

批量发送

batch_send_message(msg_type: str, *, content=None, card=None, user_open_ids=None, department_ids=None, user_ids=None, union_ids=None)

这不是发到群，而是批量发给用户或部门。

特点：

• 底层走 /message/v4/batch_send/

• 支持文本和卡片

• 支持 open_id / user_id / union_id / department_id 目标

• 批量消息不是普通群消息，不能像普通消息那样引用回复

示例：批量发文本通知

bot.batch_send_message(
    "text",
    content="今晚 23:00 系统维护",
    user_open_ids=["ou_a", "ou_b", "ou_c"]
)

示例：批量发卡片

bot.batch_send_message(
    "interactive",
    card={
        "header": {"title": {"content": "运维通知", "tag": "plain_text"}},
        "elements": [{"tag": "markdown", "content": "请及时确认"}]
    },
    department_ids=["od_xxx"]
)

示例：批量发通知后轮询进度

from pywayne.tools import wayne_print

result = bot.batch_send_message(
    "text",
    content="请在今晚 18:00 前完成确认",
    user_open_ids=["ou_a", "ou_b"]
)
task_id = result.get("task_id")
if task_id:
    progress = bot.get_batch_message_progress(task_id)
    wayne_print(progress, color="cyan")

组合场景示例

场景 1：发日报卡片，用户点击后再原位更新

card = CardContentV2(title="日报提交")
card.add_markdown("请点击按钮确认今天已提交日报。")
bot.send_card_to_chat("oc_xxx", card.get_card())

# 按钮点击后的更新逻辑放到 LarkBotListener.card_action_handler 中处理

场景 2：收到用户消息后引用回复，并把回复置顶

reply = bot.reply_message("om_xxx", "text", {"text": "这是最终处理结论"})
bot.pin_message(reply["message_id"])

场景 3：先发“处理中”，完成后更新成最终状态

msg = bot.send_card_to_chat(
    "oc_xxx",
    {
        "header": {"title": {"content": "任务状态", "tag": "plain_text"}},
        "elements": [{"tag": "markdown", "content": "处理中..."}]
    }
)

bot.edit_card_message(
    msg["message_id"],
    {
        "header": {"title": {"content": "任务状态", "tag": "plain_text"}},
        "elements": [{"tag": "markdown", "content": "已完成"}]
    }
)

场景 4：收到告警后，转发给值班人并加急

bot.forward_message("om_alert_xxx", "ou_duty_xxx", receive_id_type="open_id")
bot.urgent_message("om_alert_xxx", "app", ["ou_duty_xxx"])

场景 5：使用 reaction 作为处理中状态

reaction = bot.add_reaction("om_xxx", "WITTY")
try:
    bot.reply_message("om_xxx", "text", {"text": "处理中"})
finally:
    bot.delete_reaction("om_xxx", reaction["reaction_id"])

场景 5.1：把 LLM 流式输出刷到同一张回复卡片上

def llm_stream():
    yield "今天的分析如下：\n\n"
    yield "1. 指标整体稳定\n"
    yield "2. 风险点主要在支付链路\n"
    yield "3. 建议先观察 30 分钟\n"

bot.stream_reply_card(
    "om_xxx",
    llm_stream(),
    title="分析结果生成中",
    final_status_text="已完成"
)

场景 5.2：逐字流式输出，完成态改绿色，失败态改红色

import time

reply = bot.reply_streaming_card(
    "om_xxx",
    title="逐字输出测试",
    template="blue",
    initial_md=""
)

card_message_id = reply["message_id"]
text = "这是一段逐字输出的测试文本。"
current = ""

try:
    for ch in text:
        current += ch
        bot.update_streaming_card(
            card_message_id,
            current,
            title="逐字输出测试",
            template="blue",
            status_text="生成中..."
        )
        time.sleep(0.3)

    bot.recolor_streaming_card(
        card_message_id,
        current,
        title="逐字输出测试",
        template="green",
        status_text="已完成"
    )
except Exception as exc:
    bot.recolor_streaming_card(
        card_message_id,
        current + f"\n\n错误信息：{exc}",
        title="逐字输出测试",
        template="red",
        status_text="失败"
    )

场景 6：创建专项群，拉人，设管理员，发欢迎卡片

group = bot.create_chat(
    name="专项推进群",
    user_open_ids=["ou_a", "ou_b", "ou_c"],
    description="专项推进"
)
chat_id = group["chat_id"]
bot.set_chat_admin(chat_id, ["ou_a"], is_admin=True)

card = CardContentV2(title="欢迎加入")
card.add_markdown("请查看群公告并完成本周任务认领。")
bot.send_card_to_chat(chat_id, card.get_card())

场景 7：长 Markdown 公告自动切片发送

bot.send_markdown_message_to_chat(
    "oc_xxx",
    md_text=huge_release_note,
    title="发布说明",
    prefer="card_v2",
    max_message_bytes=10000
)

场景 8：按名字找目标群和目标人，再定向发送

chat_ids = bot.find_chat_ids_by_name("值班群")
if chat_ids:
    bot.send_text_to_chat(chat_ids[0], "今晚注意观察监控")

open_ids = bot.find_member_open_ids_by_name(chat_ids[0], "Wayne")
if open_ids:
    bot.send_text_to_user(open_ids[0], "请确认值班")

场景 8.1：先按群名查群，再发送长 Markdown 卡片

chat_ids = bot.find_chat_ids_by_name("测试3")
if chat_ids:
    bot.send_markdown_message_to_chat(
        chat_ids[0],
        md_text="# 自动通知\n\n- 功能已发布\n- 请在群内验证",
        title="系统通知"
    )

场景 9：下载消息附件后再二次转发

bot.download_message_resource("om_xxx", "file", "/tmp/input.pdf", "file_xxx")
new_key = bot.upload_file("/tmp/input.pdf", file_type="pdf")
bot.send_file_to_chat("oc_other_xxx", new_key)

场景 10：读取群公告并同步成卡片消息

data = bot.get_chat_announcement("oc_xxx")
card = CardContentV2(title="当前群公告")
card.add_markdown(str(data))
bot.send_card_to_chat("oc_xxx", card.get_card())

场景 11：查消息详情 -> 转发 -> 加急 -> 置顶结论

detail = bot.get_message("om_alert_xxx")
if "严重" in detail["body"]["content"]:
    bot.forward_message("om_alert_xxx", "ou_duty_xxx", receive_id_type="open_id")
    bot.urgent_message("om_alert_xxx", "app", ["ou_duty_xxx"])
    reply = bot.reply_message("om_alert_xxx", "text", {"text": "已转交值班并加急"})
    bot.pin_message(reply["message_id"])

注意事项

1. reply_message 的 content 会被自动 JSON 序列化；文本消息通常传 {"text": "..."}。

2. interactive 这个飞书类型值对应的是 card 卡片；发送与更新时，传入的是卡片 JSON，不需要你手动再做 json.dumps。

3. reaction 的 emoji_type 不是表情符号本身，而是飞书定义的名称。

4. batch_send_message 面向用户 / 部门，不面向群；它和群消息的生命周期不同。

5. set_chat_announcement 目前直接暴露飞书 patch 风格参数，适合需要精确控制公告 patch 的场景。

6. 流式卡片更新传入的是“当前完整文本”，不是本次新增片段；如果你只传 delta，卡片内容会丢前文。

7. 卡片颜色使用的是飞书头部模板色，不是任意 RGB；完成自动变绿需要显式传 final_template="green" 或手动 recolor_streaming_card。