🐛 parse received rich text and replied messages

nonebot/adapters/feishu/message.py

@@ -1,4 +1,4 @@
-from collections.abc import Iterable
+from collections.abc import Iterable, Sequence
 from dataclasses import dataclass
 import json
 import re
@@ -8,7 +8,7 @@
 from nonebot.adapters import Message as BaseMessage
 from nonebot.adapters import MessageSegment as BaseMessageSegment
 
-from .models import Mention
+from .models import Mention, ReplyMention
 
 # 仅接收消息时存在的字段，这些信息在上传文件时提供，发送消息时不传
 _RECEIVE_ONLY_FIELDS = {
@@ -34,6 +34,8 @@
 _POST_TAGS = {"text", "a", "at", "img", "media", "emotion", "code_block", "hr", "md"}
 # 富文本中必须独占一个段落的节点
 _STANDALONE_POST_TAGS = {"img", "media"}
+# 视为文本的消息段，链接和代码块取其中的文字
+_TEXT_TYPES = {"text", "post", "a", "code_block"}
 
 
 class MessageSegment(BaseMessageSegment["Message"]):
@@ -48,7 +50,7 @@ def get_message_class(cls) -> type["Message"]:
 
     @override
     def is_text(self) -> bool:
-        return self.type == "text" or self.type == "post"
+        return self.type in _TEXT_TYPES
 
     @override
     def __str__(self) -> str:
@@ -671,6 +673,68 @@ def __str__(self) -> str:
 }
 
 
+# 接收到的富文本节点 -> 消息段，img 与单独发送的图片一样记为 image
+_RECEIVED_POST_NODES: dict[str, tuple[str, type[MessageSegment]]] = {
+    "text": ("text", Text),
+    "at": ("at", At),
+    "img": ("image", Image),
+    "media": ("media", Media),
+}
+
+
+def _mention_id(mention: Mention | ReplyMention) -> str:
+    # 事件中的 id 是包含各类 id 的对象，获取消息接口返回的是按 id_type 给出的单个 id
+    return mention.id.open_id if isinstance(mention, Mention) else mention.id
+
+
+def _post_node_to_segment(node: dict[str, Any], at_key_to_id: dict[str, str]) -> MessageSegment:
+    tag = node.get("tag", "")
+    # 接收到的节点常带有空的 style 列表，去掉后相邻的纯文本才能合并
+    data = {key: value for key, value in node.items() if key != "tag" and value is not None and value != []}
+
+    if tag == "at":
+        # 接收到的 at 节点中 user_id 是 `@_user_1` 这样的 key，与文本消息一样换成 mentions 中的 id
+        user_id: str = data.get("user_id", "")
+        data["user_id"] = at_key_to_id.get(user_id.removeprefix("@"), user_id)
+
+    seg_type, segment_class = _RECEIVED_POST_NODES.get(tag, (tag, MessageSegment))
+    return segment_class(seg_type, data)
+
+
+def _deserialize_post(content: dict[str, Any], at_key_to_id: dict[str, str]) -> "Message":
+    """把接收到的富文本展开为消息段。
+
+    段落之间以换行分隔，标题作为第一行。展开后 @、图片与文字和文本消息中的一致，
+    @ 检测与命令匹配可以直接使用，再次发送时 `Message.serialize` 会重新组合为富文本。
+    """
+    paragraphs: list[list[dict[str, Any]]] = []
+    if title := content.get("title"):
+        paragraphs.append([{"tag": "text", "text": title}])
+    paragraphs.extend(content.get("content") or [])
+
+    segments: list[MessageSegment] = []
+    for index, paragraph in enumerate(paragraphs):
+        if index:
+            segments.append(Text("text", {"text": "\n"}))
+        segments.extend(_post_node_to_segment(node, at_key_to_id) for node in paragraph)
+
+    msg = Message()
+    for seg in segments:
+        # 合并相邻的纯文本，带样式的文本保持独立
+        last = msg[-1] if msg else None
+        if (
+            seg.type == "text"
+            and seg.data.keys() == {"text"}
+            and last is not None
+            and last.type == "text"
+            and last.data.keys() == {"text"}
+        ):
+            msg[-1] = Text("text", {"text": last.data["text"] + seg.data["text"]})
+        else:
+            msg.append(seg)
+    return msg
+
+
 class Message(BaseMessage[MessageSegment]):
     """
     飞书 协议 Message 适配。
@@ -738,12 +802,21 @@ def serialize(self) -> tuple[str, str]:
             raise ValueError("Cannot serialize empty message")
 
     @staticmethod
-    def deserialize(content: str, mentions: list[Mention] | None, message_type: str) -> "Message":
+    def deserialize(
+        content: str,
+        mentions: Sequence[Mention | ReplyMention] | None,
+        message_type: str,
+    ) -> "Message":
+        """解析接收到的消息内容。
+
+        `mentions` 可以是事件中的 `Mention`，也可以是获取消息接口返回的 `ReplyMention`，
+        `@_user_1` 这样的 key 会换成对应的 id。富文本会展开为文字、@、图片等消息段。
+        """
         msg = Message()
         parsed_content = json.loads(content)
         at_key_to_id = {
             # wipeout @ at the start of key
-            mention.key[1:]: mention.id.open_id
+            mention.key[1:]: _mention_id(mention)
             for mention in (mentions or [])
         }
 
@@ -776,7 +849,7 @@ def deserialize(content: str, mentions: list[Mention] | None, message_type: str)
                 msg.append(Text("text", {"text": text[text_begin:]}))
 
         elif message_type == "post":
-            msg.append(Post("post", parsed_content))
+            msg.extend(_deserialize_post(parsed_content, at_key_to_id))
 
         else:
             segment_class = _RECEIVED_SEGMENTS.get(message_type, MessageSegment)

nonebot/adapters/feishu/models/common.py

@@ -1,9 +1,12 @@
-from typing import Any, Literal
+from typing import TYPE_CHECKING, Any, Literal
 
 from pydantic import BaseModel
 
 from nonebot.compat import PYDANTIC_V2, ConfigDict
 
+if TYPE_CHECKING:
+    from ..message import Message
+
 
 class EventHeader(BaseModel):
     event_id: str
@@ -125,6 +128,12 @@ class Reply(BaseModel):
     mentions: list[ReplyMention] = []
     upper_message_id: str | None = None
 
+    def get_message(self) -> "Message":
+        """被回复消息的内容，按消息类型解析，@ 换成 `mentions` 中给出的 id（默认为 open_id）。"""
+        from ..message import Message
+
+        return Message.deserialize(self.body.content, self.mentions, self.msg_type)
+
     if PYDANTIC_V2:
         model_config = ConfigDict(extra="allow")
     else:

tests/test_feishu_bot.py

@@ -133,3 +133,36 @@ async def handle_event(bot: Bot, event: GroupMessageEvent) -> None:
     assert handled == [event]
     assert event.get_message()[0].type == msg_type
     assert event.get_event_description()
+
+
+@pytest.mark.anyio
+async def test_check_tome_in_post():
+    """@ 机器人的富文本与文本消息一样去掉开头的 @，命令可以匹配"""
+    event_data = {k: v for k, v in test_events[1].items() if k != "_model"}
+    content = {
+        "title": "",
+        "content": [
+            [
+                {"tag": "at", "user_id": "@_user_1", "user_name": "bot", "style": []},
+                {"tag": "text", "text": " /say 34567", "style": []},
+            ],
+            [{"tag": "img", "image_key": "img_1"}],
+        ],
+    }
+    message = {**event_data["event"]["message"], "message_type": "post", "content": json.dumps(content)}
+    event = Adapter.json_to_event({**event_data, "event": {**event_data["event"], "message": message}})
+    assert isinstance(event, GroupMessageEvent)
+
+    bot = Bot(
+        nonebot.get_adapter(Adapter),
+        bot_id,
+        bot_config=type_validate_python(BotConfig, test_bot),
+        bot_info=type_validate_python(BotInfo, bots_info["bot"]),
+    )
+    _check_at_me(bot, event)
+
+    assert event.to_me
+    assert event.get_message().extract_plain_text() == "/say 34567\n"
+    assert [seg.type for seg in event.get_message()] == ["text", "image"]
+    assert event.original_message[0].type == "at"
+    assert event.original_message[0].data["user_id"] == bots_info["bot"]["open_id"]

tests/test_feishu_message.py

@@ -20,6 +20,8 @@
     Todo,
     VideoChat,
 )
+from nonebot.adapters.feishu.models import Mention, Reply, UserId
+from nonebot.compat import type_validate_python
 
 TEXT_A = {"tag": "text", "text": "a"}
 TEXT_B = {"tag": "text", "text": "b"}
@@ -415,3 +417,135 @@ def test_serialize_sends_built_card():
 
     assert msg_type == "interactive"
     assert json.loads(content)["elements"] == [{"tag": "hr"}]
+
+
+BOT_MENTION = Mention(
+    key="@_user_1",
+    id=UserId(open_id="ou_bot", user_id="u_bot", union_id="on_bot"),
+    name="bot",
+    tenant_key="tk",
+)
+# 接收到的富文本：没有语言层，at 节点的 user_id 是 mentions 中的 key，节点常带空的 style
+RECEIVED_POST = {
+    "title": "",
+    "content": [
+        [
+            {"tag": "at", "user_id": "@_user_1", "user_name": "bot", "style": []},
+            {"tag": "text", "text": " /explain this", "style": []},
+        ],
+        [{"tag": "img", "image_key": "img_1"}],
+        [
+            {"tag": "text", "text": "see "},
+            {"tag": "a", "href": "https://example.com", "text": "docs", "style": []},
+            {"tag": "emotion", "emoji_type": "SMILE"},
+        ],
+        [{"tag": "code_block", "language": "PYTHON", "text": "print(1)\n"}],
+    ],
+}
+
+
+def _flat(msg: Message) -> list[tuple[str, dict[str, Any]]]:
+    return [(seg.type, seg.data) for seg in msg]
+
+
+def test_deserialize_post_into_segments():
+    msg = Message.deserialize(json.dumps(RECEIVED_POST), [BOT_MENTION], "post")
+
+    assert _flat(msg) == [
+        ("at", {"user_id": "ou_bot", "user_name": "bot"}),
+        ("text", {"text": " /explain this\n"}),
+        ("image", {"image_key": "img_1"}),
+        ("text", {"text": "\nsee "}),
+        ("a", {"href": "https://example.com", "text": "docs"}),
+        ("emotion", {"emoji_type": "SMILE"}),
+        ("text", {"text": "\n"}),
+        ("code_block", {"language": "PYTHON", "text": "print(1)\n"}),
+    ]
+    assert isinstance(msg[2], Image)
+    assert msg.extract_plain_text() == " /explain this\n\nsee docs\nprint(1)\n"
+
+
+def test_deserialize_post_puts_title_on_first_line():
+    received = {"title": "title", "content": [[TEXT_A], [], [TEXT_B]]}
+
+    msg = Message.deserialize(json.dumps(received), None, "post")
+
+    assert _flat(msg) == [("text", {"text": "title\na\n\nb"})]
+
+
+def test_deserialize_post_keeps_styled_text_apart():
+    bold = {"tag": "text", "text": "b", "style": ["bold"]}
+    received = {"title": "", "content": [[TEXT_A, bold, TEXT_A]]}
+
+    msg = Message.deserialize(json.dumps(received), None, "post")
+
+    assert _flat(msg) == [
+        ("text", {"text": "a"}),
+        ("text", {"text": "b", "style": ["bold"]}),
+        ("text", {"text": "a"}),
+    ]
+
+
+def test_deserialize_post_keeps_at_without_mention():
+    received = {"title": "", "content": [[{"tag": "at", "user_id": "@_user_2"}]]}
+
+    msg = Message.deserialize(json.dumps(received), [BOT_MENTION], "post")
+
+    assert _flat(msg) == [("at", {"user_id": "@_user_2"})]
+
+
+def test_deserialize_empty_post():
+    assert Message.deserialize(json.dumps({"title": "", "content": []}), None, "post") == Message()
+
+
+def test_received_post_can_be_sent_again():
+    msg = Message.deserialize(json.dumps(RECEIVED_POST), [BOT_MENTION], "post")
+
+    msg_type, content = msg.serialize()
+
+    assert msg_type == "post"
+    assert json.loads(content)["zh_cn"]["content"] == [
+        [{"tag": "at", "user_id": "ou_bot", "user_name": "bot"}, {"tag": "text", "text": " /explain this\n"}],
+        [{"tag": "img", "image_key": "img_1"}],
+        [
+            {"tag": "text", "text": "\nsee "},
+            {"tag": "a", "href": "https://example.com", "text": "docs"},
+            {"tag": "emotion", "emoji_type": "SMILE"},
+            {"tag": "text", "text": "\n"},
+            {"tag": "code_block", "language": "PYTHON", "text": "print(1)\n"},
+        ],
+    ]
+
+
+def _reply(msg_type: str, content: dict[str, Any], mentions: list[dict[str, str]] | None = None) -> Reply:
+    return type_validate_python(
+        Reply,
+        {
+            "message_id": "om_parent",
+            "msg_type": msg_type,
+            "create_time": "0",
+            "update_time": "0",
+            "deleted": False,
+            "updated": False,
+            "chat_id": "oc_1",
+            "sender": {"id": "ou_sender", "id_type": "open_id", "sender_type": "user", "tenant_key": "tk"},
+            "body": {"content": json.dumps(content)},
+            "mentions": mentions or [],
+        },
+    )
+
+
+def test_reply_get_message_resolves_mentions():
+    # 获取消息接口返回的 mentions 中 id 是单个字符串，不是事件中的 UserId
+    mention = {"key": "@_user_1", "id": "ou_1", "id_type": "open_id", "name": "Tom", "tenant_key": "tk"}
+    reply = _reply("text", {"text": "@_user_1 hi"}, [mention])
+
+    assert _flat(reply.get_message()) == [("at", {"user_id": "ou_1"}), ("text", {"text": " hi"})]
+
+
+def test_reply_get_message_parses_by_type():
+    assert _flat(_reply("image", {"image_key": "img_1"}).get_message()) == [("image", {"image_key": "img_1"})]
+    assert _flat(_reply("post", {"title": "", "content": [[TEXT_A], [IMG]]}).get_message()) == [
+        ("text", {"text": "a\n"}),
+        ("image", {"image_key": "ik"}),
+    ]