diff --git a/packages/nonebot-adapter-feishu/nonebot/adapters/feishu/message.py b/packages/nonebot-adapter-feishu/nonebot/adapters/feishu/message.py index d589e554..a0722f30 100644 --- a/packages/nonebot-adapter-feishu/nonebot/adapters/feishu/message.py +++ b/packages/nonebot-adapter-feishu/nonebot/adapters/feishu/message.py @@ -1,11 +1,23 @@ -from typing import Union, Mapping, Iterable +from typing import Type, Union, Mapping, Iterable + from nonebot.adapters import Message as BaseMessage, MessageSegment as BaseMessageSegment +from nonebot.typing import overrides class MessageSegment(BaseMessageSegment): + """ + 飞书 协议 MessageSegment 适配。具体方法参考协议消息段类型或源码。 + """ + + @classmethod + @overrides(BaseMessageSegment) + def get_message_class(cls) -> Type["Message"]: + return Message def __str__(self) -> str: - raise NotImplementedError + if self.type == "text" or self.type == "hongbao": + return str(self.data["text"]) + return "" def __add__(self, other) -> "Message": return Message(self) + other @@ -14,13 +26,124 @@ class MessageSegment(BaseMessageSegment): return Message(other) + self def is_text(self) -> bool: - raise NotImplementedError + return self.type == "text" + + @staticmethod + def text(text: str) -> "MessageSegment": + return MessageSegment("text", {"text": text}) + + @staticmethod + def post(title: str, content: list) -> "MessageSegment": + return MessageSegment("post", {"title": title, "content": content}) + + @staticmethod + def image(image_key: str) -> "MessageSegment": + return MessageSegment("image", {"image_key": image_key}) + + @staticmethod + def file(file_key: str, file_name: str) -> "MessageSegment": + return MessageSegment("file", { + "file_key": file_key, + "file_name": file_name + }) + + @staticmethod + def audio(file_key: str, duration: int) -> "MessageSegment": + return MessageSegment("audio", { + "file_key": file_key, + "duration": duration + }) + + @staticmethod + def media(file_key: str, image_key: str, file_name: str, + duration: int) -> "MessageSegment": + return MessageSegment( + "media", { + "file_key": file_key, + "image_key": image_key, + "file_name": file_name, + "duration": duration + }) + + @staticmethod + def sticker(file_key) -> "MessageSegment": + return MessageSegment("sticker", {"file_key": file_key}) + + @staticmethod + def interactive(title: str, elements: list) -> "MessageSegment": + return MessageSegment("interactive", { + "title": title, + "elements": elements + }) + + @staticmethod + def hongbao(text: str) -> "MessageSegment": + return MessageSegment("hongbao", {"text": text}) + + @staticmethod + def share_calendar_event(summary: str, start_time: str, + end_time: str) -> "MessageSegment": + return MessageSegment("share_calendar_event", { + "summary": summary, + "start_time": start_time, + "end_time": end_time + }) + + @staticmethod + def share_chat(chat_id: str) -> "MessageSegment": + return MessageSegment("share_chat", {"chat_id": chat_id}) + + @staticmethod + def share_user(user_id: str) -> "MessageSegment": + return MessageSegment("share_user", {"user_id": user_id}) + + @staticmethod + def system(template: str, from_user: list, + to_chatters: list) -> "MessageSegment": + return MessageSegment( + "system", { + "template": template, + "from_user": from_user, + "to_chatters": to_chatters + }) + + @staticmethod + def location(name: str, longitude: str, latitude: str) -> "MessageSegment": + return MessageSegment("location", { + "name": name, + "longitude": longitude, + "latitude": latitude + }) + + @staticmethod + def video_chat(topic: str, start_time: str) -> "MessageSegment": + return MessageSegment("video_chat", { + "topic": topic, + "start_time": start_time, + }) class Message(BaseMessage): + """ + 飞书 协议 Message 适配。 + """ + + @classmethod + @overrides(BaseMessage) + def get_segment_class(cls) -> Type[MessageSegment]: + return MessageSegment @staticmethod def _construct( msg: Union[str, Mapping, Iterable[Mapping]]) -> Iterable[MessageSegment]: + if isinstance(msg, Mapping): + yield MessageSegment(msg["type"], msg.get("data") or {}) + elif isinstance(msg, str): + yield MessageSegment.text(msg) + elif isinstance(msg, Iterable): + for seg in msg: + yield MessageSegment(seg["type"], seg.get("data") or {}) + + def _produce(self) -> dict: raise NotImplementedError