跳转至

inbound

feishu.im.inbound

飞书入站消息的无状态读取助手。

提供从飞书消息体(im.message.receive_v1 事件中的 message 对象,或 feishu.im.messages.IMNamespace.get 返回的消息数据)中提取信息的纯函数: feishu.im.inbound.message_text 提取可读文本, feishu.im.inbound.is_mentioned 判断机器人是否被提及。

is_mentioned

Python
is_mentioned(message: dict[str, Any], *, open_id: str | None = None, union_id: str | None = None) -> bool

判断消息是否提及(@)了指定用户。

遍历消息的 mentions 数组,若其中任一条目的 id 匹配给定的 open_idunion_id, 则返回 True。不同事件类型与接口版本下,条目的 id 既可能是同时含 open_idunion_id 的字典,也可能是单一字符串,因此对两种形态都进行匹配。

参数:

名称 类型 描述 默认

message

dict[str, Any]

飞书消息体字典,通常含 mentions 数组。

必需

open_id

str | None

待匹配的用户 open ID;为空表示不按 open ID 匹配。

None

union_id

str | None

待匹配的用户 union ID;为空表示不按 union ID 匹配。

None

返回:

类型 描述
bool

消息提及了指定用户时返回 True,否则返回 False

飞书文档

接收消息

示例:

Python Console Session
>>> message = {
...     "content": '{"text":"@_user_1 hi"}',
...     "mentions": [{"key": "@_user_1", "id": {"open_id": "ou_bot", "union_id": "on_bot"}, "name": "Bot"}],
... }
>>> is_mentioned(message, open_id="ou_bot")
True
>>> is_mentioned(message, open_id="ou_other")
False
>>> is_mentioned(message, union_id="on_bot")
True
>>> is_mentioned({"mentions": []}, open_id="ou_bot")
False
源代码位于: feishu/im/inbound.py
Python
def is_mentioned(message: dict[str, Any], *, open_id: str | None = None, union_id: str | None = None) -> bool:
    r"""
    判断消息是否提及(@)了指定用户。

    遍历消息的 `mentions` 数组,若其中任一条目的 `id` 匹配给定的 `open_id` 或 `union_id`,
    则返回 `True`。不同事件类型与接口版本下,条目的 `id` 既可能是同时含 `open_id`、`union_id`
    的字典,也可能是单一字符串,因此对两种形态都进行匹配。

    Args:
        message: 飞书消息体字典,通常含 `mentions` 数组。
        open_id: 待匹配的用户 open ID;为空表示不按 open ID 匹配。
        union_id: 待匹配的用户 union ID;为空表示不按 union ID 匹配。

    Returns:
        消息提及了指定用户时返回 `True`,否则返回 `False`。

    飞书文档:
        [接收消息](https://open.feishu.cn/document/server-docs/im-v1/message/events/receive)

    Examples:
        >>> message = {
        ...     "content": '{"text":"@_user_1 hi"}',
        ...     "mentions": [{"key": "@_user_1", "id": {"open_id": "ou_bot", "union_id": "on_bot"}, "name": "Bot"}],
        ... }
        >>> is_mentioned(message, open_id="ou_bot")
        True
        >>> is_mentioned(message, open_id="ou_other")
        False
        >>> is_mentioned(message, union_id="on_bot")
        True
        >>> is_mentioned({"mentions": []}, open_id="ou_bot")
        False
    """
    return any(
        _mention_matches(mention, open_id=open_id, union_id=union_id) for mention in message.get("mentions") or []
    )

message_content

Python
message_content(message: dict[str, Any]) -> NestedDict

解析飞书消息体中的 content JSON。

参数:

名称 类型 描述 默认

message

dict[str, Any]

飞书消息体字典,含 content

必需

返回:

类型 描述
NestedDict

解析后的 content;缺失、格式错误或非对象时返回空 chanfig.NestedDict

示例:

Python Console Session
1
2
3
4
>>> message_content({"content": '{"text":"hi"}'}).text
'hi'
>>> message_content({"content": "not json"}) == {}
True
源代码位于: feishu/im/inbound.py
Python
def message_content(message: dict[str, Any]) -> NestedDict:
    r"""
    解析飞书消息体中的 `content` JSON。

    Args:
        message: 飞书消息体字典,含 `content`。

    Returns:
        解析后的 `content`;缺失、格式错误或非对象时返回空 [chanfig.NestedDict][]。

    Examples:
        >>> message_content({"content": '{"text":"hi"}'}).text
        'hi'
        >>> message_content({"content": "not json"}) == {}
        True
    """
    raw = message.get("content")
    if raw is None:
        body = message.get("body")
        if isinstance(body, dict):
            raw = body.get("content")
    if not raw:
        return NestedDict()
    try:
        content = json.loads(raw) if isinstance(raw, str) else raw
    except (TypeError, ValueError):
        return NestedDict()
    if isinstance(content, NestedDict):
        return content
    if isinstance(content, dict):
        return NestedDict(content)
    return NestedDict()

message_resource

Python
message_resource(message: dict[str, Any]) -> NestedDict | None

从图片或文件消息中提取可下载资源。

返回值中的 key 可传给 feishu.im.messages.IMNamespace.get_resourcefile_keyresource_type 可作为同名参数传入。

参数:

名称 类型 描述 默认

message

dict[str, Any]

飞书消息体字典。

必需

返回:

类型 描述
NestedDict | None

资源描述;没有图片或文件资源时返回 None

示例:

Python Console Session
1
2
3
4
>>> message_resource({"message_type": "image", "content": '{"image_key":"img_1"}'}).key
'img_1'
>>> message_resource({"message_type": "file", "content": '{"file_key":"file_1","file_name":"a.pdf"}'}).name
'a.pdf'
源代码位于: feishu/im/inbound.py
Python
def message_resource(message: dict[str, Any]) -> NestedDict | None:
    r"""
    从图片或文件消息中提取可下载资源。

    返回值中的 `key` 可传给 [feishu.im.messages.IMNamespace.get_resource][] 的 `file_key`,
    `resource_type` 可作为同名参数传入。

    Args:
        message: 飞书消息体字典。

    Returns:
        资源描述;没有图片或文件资源时返回 `None`。

    Examples:
        >>> message_resource({"message_type": "image", "content": '{"image_key":"img_1"}'}).key
        'img_1'
        >>> message_resource({"message_type": "file", "content": '{"file_key":"file_1","file_name":"a.pdf"}'}).name
        'a.pdf'
    """
    content = message_content(message)
    message_type = str(message.get("message_type") or message.get("msg_type") or "")
    image_key = _string(content.get("image_key"))
    if image_key:
        return NestedDict(
            kind="image",
            key=image_key,
            resource_type="image",
            message_type=message_type,
            name=_string(content.get("file_name")) or _string(content.get("name")),
            mime_type=_string(content.get("mime_type")) or _string(content.get("file_type")),
            size=content.get("size") or content.get("file_size"),
        )

    file_key = _string(content.get("file_key"))
    if file_key:
        return NestedDict(
            kind="file",
            key=file_key,
            resource_type="file",
            message_type=message_type,
            name=_string(content.get("file_name")) or _string(content.get("name")),
            mime_type=_string(content.get("mime_type")) or _string(content.get("file_type")),
            size=content.get("size") or content.get("file_size"),
        )
    return None

message_resources

Python
message_resources(message: dict[str, Any]) -> list[NestedDict]

提取消息中的**全部**可下载资源(图片 / 文件)。

单张图片或文件消息返回 1 个资源(同 feishu.im.inbound.message_resource);富文本 post 消息可内嵌 多张图片,逐个返回(按出现顺序、去重)。每个资源的 key / resource_type 可直接传给 feishu.im.messages.IMNamespace.get_resource。无资源时返回空列表。

参数:

名称 类型 描述 默认

message

dict[str, Any]

飞书消息体字典。

必需

返回:

类型 描述
list[NestedDict]

资源描述列表;无图片 / 文件资源时为空列表。

示例:

Python Console Session
1
2
3
4
5
6
7
>>> message_resources({"message_type": "image", "content": '{"image_key":"img_1"}'})[0].key
'img_1'
>>> post = '{"content":[[{"tag":"img","image_key":"a"}],[{"tag":"img","image_key":"b"}]]}'
>>> [r.key for r in message_resources({"message_type": "post", "content": post})]
['a', 'b']
>>> message_resources({"message_type": "text", "content": '{"text":"hi"}'})
[]
源代码位于: feishu/im/inbound.py
Python
def message_resources(message: dict[str, Any]) -> list[NestedDict]:
    r"""
    提取消息中的**全部**可下载资源(图片 / 文件)。

    单张图片或文件消息返回 1 个资源(同 [feishu.im.inbound.message_resource][]);富文本 `post` 消息可内嵌
    多张图片,逐个返回(按出现顺序、去重)。每个资源的 `key` / `resource_type` 可直接传给
    [feishu.im.messages.IMNamespace.get_resource][]。无资源时返回空列表。

    Args:
        message: 飞书消息体字典。

    Returns:
        资源描述列表;无图片 / 文件资源时为空列表。

    Examples:
        >>> message_resources({"message_type": "image", "content": '{"image_key":"img_1"}'})[0].key
        'img_1'
        >>> post = '{"content":[[{"tag":"img","image_key":"a"}],[{"tag":"img","image_key":"b"}]]}'
        >>> [r.key for r in message_resources({"message_type": "post", "content": post})]
        ['a', 'b']
        >>> message_resources({"message_type": "text", "content": '{"text":"hi"}'})
        []
    """
    single = message_resource(message)
    if single is not None:
        return [single]
    content = message_content(message)
    message_type = str(message.get("message_type") or message.get("msg_type") or "")
    rich = content.get("content") or content.get("elements")
    resources: list[NestedDict] = []
    seen: set[str] = set()
    if isinstance(rich, list):
        for line in rich:
            if not isinstance(line, list):
                continue
            for element in line:
                if not isinstance(element, dict) or element.get("tag") != "img":
                    continue
                image_key = _string(element.get("image_key"))
                if image_key and image_key not in seen:
                    seen.add(image_key)
                    resources.append(
                        NestedDict(
                            kind="image",
                            key=image_key,
                            resource_type="image",
                            message_type=message_type,
                            name=_string(element.get("file_name")) or _string(element.get("name")),
                            mime_type=None,
                            size=None,
                        )
                    )
    return resources

message_text

Python
message_text(message: dict[str, Any]) -> str

从飞书消息体中提取可读文本。

解析消息体内的 content JSON:对 text 类型读取 content['text'];对富文本 post 类型 (contentcontent/elements 二维数组)将各段文本以空行拼接,若含 title 则以 Markdown 二级标题形式前置。随后用消息的 mentions 数组将文本中的 @_user_N 占位符替换为 @<姓名>

参数:

名称 类型 描述 默认

message

dict[str, Any]

飞书消息体字典,含 message_type/msg_typecontent,可选 mentions

必需

返回:

类型 描述
str

提取并解析后的文本;无法解析时返回空字符串。

飞书文档

接收消息

示例:

Python Console Session
>>> text_message = {
...     "message_type": "text",
...     "content": '{"text":"@_user_1 你好"}',
...     "mentions": [{"key": "@_user_1", "name": "小明"}],
... }
>>> message_text(text_message)
'@小明 你好'
>>> post_message = {
...     "message_type": "post",
...     "content": '{"title":"标题","content":[[{"tag":"text","text":"正文"}]]}',
... }
>>> message_text(post_message)
'## 标题\n\n正文'
源代码位于: feishu/im/inbound.py
Python
def message_text(message: dict[str, Any]) -> str:
    r"""
    从飞书消息体中提取可读文本。

    解析消息体内的 `content` JSON:对 `text` 类型读取 `content['text']`;对富文本 `post` 类型
    (`content` 含 `content`/`elements` 二维数组)将各段文本以空行拼接,若含 `title` 则以
    Markdown 二级标题形式前置。随后用消息的 `mentions` 数组将文本中的 `@_user_N` 占位符替换为
    `@<姓名>`。

    Args:
        message: 飞书消息体字典,含 `message_type`/`msg_type`、`content`,可选 `mentions`。

    Returns:
        提取并解析后的文本;无法解析时返回空字符串。

    飞书文档:
        [接收消息](https://open.feishu.cn/document/server-docs/im-v1/message/events/receive)

    Examples:
        >>> text_message = {
        ...     "message_type": "text",
        ...     "content": '{"text":"@_user_1 你好"}',
        ...     "mentions": [{"key": "@_user_1", "name": "小明"}],
        ... }
        >>> message_text(text_message)
        '@小明 你好'
        >>> post_message = {
        ...     "message_type": "post",
        ...     "content": '{"title":"标题","content":[[{"tag":"text","text":"正文"}]]}',
        ... }
        >>> message_text(post_message)
        '## 标题\n\n正文'
    """
    content = message_content(message)
    text = _content_text(content)
    for mention in message.get("mentions") or []:
        key = mention.get("key")
        if key:
            text = text.replace(key, f"@{mention.get('name', '')}")
    return text

message_body_text

Python
message_body_text(message: dict[str, Any]) -> str

安全地从飞书消息体中提取并裁剪可读文本。

feishu.im.inbound.message_text 之上加一层防御:解析失败(缺字段 / 非法 JSON)时返回空串而非抛错, 并去除首尾空白。

参数:

名称 类型 描述 默认

message

dict[str, Any]

飞书消息体(事件中的 message 节点)。

必需

返回:

类型 描述
str

裁剪后的可读文本;无法解析时返回空串。

示例:

Python Console Session
1
2
3
4
5
>>> import json
>>> message_body_text({"message_type": "text", "content": json.dumps({"text": "  hi  "})})
'hi'
>>> message_body_text({"message_type": "image", "content": "not json"})
''
源代码位于: feishu/im/inbound.py
Python
def message_body_text(message: dict[str, Any]) -> str:
    r"""
    安全地从飞书消息体中提取并裁剪可读文本。

    在 [feishu.im.inbound.message_text][] 之上加一层防御:解析失败(缺字段 / 非法 JSON)时返回空串而非抛错,
    并去除首尾空白。

    Args:
        message: 飞书消息体(事件中的 `message` 节点)。

    Returns:
        裁剪后的可读文本;无法解析时返回空串。

    Examples:
        >>> import json
        >>> message_body_text({"message_type": "text", "content": json.dumps({"text": "  hi  "})})
        'hi'
        >>> message_body_text({"message_type": "image", "content": "not json"})
        ''
    """
    try:
        return message_text(message).strip()
    except (TypeError, ValueError):
        return ""

message_sender_label

Python
message_sender_label(message: Mapping[str, Any], *, id_formatter: Callable[[str], str] | None = None, default: str = 'unknown') -> str

返回消息发送者的可读名称;无姓名时回退到其 ID(可经 id_formatter 转换),再不行用 default

参数:

名称 类型 描述 默认

message

Mapping[str, Any]

飞书消息体。

必需

id_formatter

Callable[[str], str] | None

可选的 ID 格式化函数,对回退使用的发送者 ID 应用(如脱敏 / 转中文名)。

None

default

str

既无姓名也无 ID 时返回的占位值。默认为 "unknown"

'unknown'

返回:

类型 描述
str

发送者名称、格式化后的 ID,或 default

示例:

Python Console Session
1
2
3
4
5
6
>>> message_sender_label({"sender": {"name": "张三"}})
'张三'
>>> message_sender_label({"sender": {"open_id": "ou_1"}}, id_formatter=str.upper)
'OU_1'
>>> message_sender_label({})
'unknown'
源代码位于: feishu/im/inbound.py
Python
def message_sender_label(
    message: Mapping[str, Any],
    *,
    id_formatter: Callable[[str], str] | None = None,
    default: str = "unknown",
) -> str:
    r"""
    返回消息发送者的可读名称;无姓名时回退到其 ID(可经 `id_formatter` 转换),再不行用 `default`。

    Args:
        message: 飞书消息体。
        id_formatter: 可选的 ID 格式化函数,对回退使用的发送者 ID 应用(如脱敏 / 转中文名)。
        default: 既无姓名也无 ID 时返回的占位值。默认为 `"unknown"`。

    Returns:
        发送者名称、格式化后的 ID,或 `default`。

    Examples:
        >>> message_sender_label({"sender": {"name": "张三"}})
        '张三'
        >>> message_sender_label({"sender": {"open_id": "ou_1"}}, id_formatter=str.upper)
        'OU_1'
        >>> message_sender_label({})
        'unknown'
    """
    sender = message.get("sender") or message.get("sender_id") or {}
    if isinstance(sender, Mapping):
        value = sender.get("name")
        if isinstance(value, str) and value:
            return value
        for key in ("user_id", "open_id", "union_id", "sender_id"):
            value = sender.get(key)
            if isinstance(value, str) and value:
                return id_formatter(value) if id_formatter is not None else value
    return default

message_transcript

Python
message_transcript(messages: Iterable[dict[str, Any]], *, id_formatter: Callable[[str], str] | None = None) -> str

把一组飞书消息渲染为「发送者: 文本」逐行转录;非文本消息以 [类型] 占位。

参数:

名称 类型 描述 默认

messages

Iterable[dict[str, Any]]

飞书消息体的可迭代集合。

必需

id_formatter

Callable[[str], str] | None

可选的发送者 ID 格式化函数,见 feishu.im.inbound.message_sender_label

None

返回:

类型 描述
str

逐行转录文本(行间以换行分隔)。

示例:

Python Console Session
1
2
3
4
>>> import json
>>> msgs = [{"sender": {"name": "张三"}, "message_type": "text", "content": json.dumps({"text": "hi"})}]
>>> message_transcript(msgs)
'张三: hi'
源代码位于: feishu/im/inbound.py
Python
def message_transcript(
    messages: Iterable[dict[str, Any]],
    *,
    id_formatter: Callable[[str], str] | None = None,
) -> str:
    r"""
    把一组飞书消息渲染为「发送者: 文本」逐行转录;非文本消息以 `[类型]` 占位。

    Args:
        messages: 飞书消息体的可迭代集合。
        id_formatter: 可选的发送者 ID 格式化函数,见 [feishu.im.inbound.message_sender_label][]。

    Returns:
        逐行转录文本(行间以换行分隔)。

    Examples:
        >>> import json
        >>> msgs = [{"sender": {"name": "张三"}, "message_type": "text", "content": json.dumps({"text": "hi"})}]
        >>> message_transcript(msgs)
        '张三: hi'
    """
    lines = []
    for item in messages:
        sender = message_sender_label(item, id_formatter=id_formatter)
        text = message_body_text(item)
        if not text:
            text = f"[{item.get('msg_type') or item.get('message_type') or 'non-text'}]"
        lines.append(f"{sender}: {text}")
    return "\n".join(lines)

interactive_card_text

Python
interactive_card_text(message: dict[str, Any]) -> str

从交互卡片消息体中提取可读文本(先解出 content,再交给 feishu.im.inbound.card_text)。

参数:

名称 类型 描述 默认

message

dict[str, Any]

飞书交互卡片消息体。

必需

返回:

类型 描述
str

卡片中的可读文本;无可提取内容时返回空串。

示例:

Python Console Session
>>> interactive_card_text({"content": '{"elements":[{"tag":"markdown","content":"hi"}]}'})
'hi'
源代码位于: feishu/im/inbound.py
Python
def interactive_card_text(message: dict[str, Any]) -> str:
    r"""
    从交互卡片消息体中提取可读文本(先解出 `content`,再交给 [feishu.im.inbound.card_text][])。

    Args:
        message: 飞书交互卡片消息体。

    Returns:
        卡片中的可读文本;无可提取内容时返回空串。

    Examples:
        >>> interactive_card_text({"content": '{"elements":[{"tag":"markdown","content":"hi"}]}'})
        'hi'
    """
    return card_text(message_content(message))

card_text

Python
card_text(card: Mapping[str, Any]) -> str

从飞书卡片中提取可读的 markdown / 文本内容(递归遍历 body.elements 与顶层 elements)。

参数:

名称 类型 描述 默认

card

Mapping[str, Any]

飞书卡片字典。

必需

返回:

类型 描述
str

卡片中各文本片段以空行拼接的结果;无文本时返回空串。

示例:

Python Console Session
>>> card_text({"elements": [{"tag": "markdown", "content": "**hi**"}]})
'**hi**'
源代码位于: feishu/im/inbound.py
Python
def card_text(card: Mapping[str, Any]) -> str:
    r"""
    从飞书卡片中提取可读的 markdown / 文本内容(递归遍历 `body.elements` 与顶层 `elements`)。

    Args:
        card: 飞书卡片字典。

    Returns:
        卡片中各文本片段以空行拼接的结果;无文本时返回空串。

    Examples:
        >>> card_text({"elements": [{"tag": "markdown", "content": "**hi**"}]})
        '**hi**'
    """
    texts: list[str] = []
    body = card.get("body")
    if isinstance(body, Mapping):
        _collect_card_text(body.get("elements"), texts)
    _collect_card_text(card.get("elements"), texts)
    return "\n\n".join(text.strip() for text in texts if text.strip())

card_title

Python
card_title(card: Mapping[str, Any]) -> str

提取飞书卡片头部(header)的标题文本。

参数:

名称 类型 描述 默认

card

Mapping[str, Any]

飞书卡片字典。

必需

返回:

类型 描述
str

标题文本;无 header / 标题时返回空串。

示例:

Python Console Session
1
2
3
4
>>> card_title({"header": {"title": {"content": "标题"}}})
'标题'
>>> card_title({})
''
源代码位于: feishu/im/inbound.py
Python
def card_title(card: Mapping[str, Any]) -> str:
    r"""
    提取飞书卡片头部(header)的标题文本。

    Args:
        card: 飞书卡片字典。

    Returns:
        标题文本;无 header / 标题时返回空串。

    Examples:
        >>> card_title({"header": {"title": {"content": "标题"}}})
        '标题'
        >>> card_title({})
        ''
    """
    header = card.get("header")
    if not isinstance(header, Mapping):
        return ""
    title = header.get("title")
    if isinstance(title, Mapping):
        content = title.get("content")
        return content if isinstance(content, str) else ""
    return title if isinstance(title, str) else ""