✨ support adapter HeyBox

Author: RF-Tar-Railt

src/nonebot_plugin_alconna/uniseg/adapters/heybox/__init__.py

@@ -0,0 +1,17 @@
+from nonebot_plugin_alconna.uniseg.loader import BaseLoader
+from nonebot_plugin_alconna.uniseg.constraint import SupportAdapter
+
+
+class Loader(BaseLoader):
+    def get_adapter(self) -> SupportAdapter:
+        return SupportAdapter.heybox
+
+    def get_builder(self):
+        from .builder import HeyboxMessageBuilder
+
+        return HeyboxMessageBuilder()
+
+    def get_exporter(self):
+        from .exporter import HeyboxMessageExporter
+
+        return HeyboxMessageExporter()

src/nonebot_plugin_alconna/uniseg/adapters/heybox/builder.py

@@ -0,0 +1,15 @@
+from nonebot.adapters.heybox.message import MessageSegment
+
+from nonebot_plugin_alconna.uniseg.segment import Text
+from nonebot_plugin_alconna.uniseg.constraint import SupportAdapter
+from nonebot_plugin_alconna.uniseg.builder import MessageBuilder, build
+
+
+class HeyboxMessageBuilder(MessageBuilder):
+    @classmethod
+    def get_adapter(cls) -> SupportAdapter:
+        return SupportAdapter.heybox
+
+    @build("text")
+    def text(self, seg: MessageSegment):
+        return Text(seg.data["text"])

src/nonebot_plugin_alconna/uniseg/adapters/heybox/exporter.py

@@ -0,0 +1,82 @@
+from typing import TYPE_CHECKING, Union
+
+from tarina import lang
+from nonebot.adapters import Bot, Event
+from nonebot.adapters.heybox.bot import Bot as HeyboxBot
+from nonebot.adapters.heybox.event import UserIMMessageEvent
+from nonebot.adapters.heybox.message import Message, MessageSegment
+
+from nonebot_plugin_alconna.uniseg.constraint import SupportScope
+from nonebot_plugin_alconna.uniseg.segment import At, Text, Image, Reply
+from nonebot_plugin_alconna.uniseg.exporter import Target, SupportAdapter, MessageExporter, SerializeFailed, export
+
+
+class HeyboxMessageExporter(MessageExporter[Message]):
+    @classmethod
+    def get_adapter(cls) -> SupportAdapter:
+        return SupportAdapter.heybox
+
+    def get_message_type(self):
+        return Message
+
+    def get_target(self, event: Event, bot: Union[Bot, None] = None) -> Target:
+        if isinstance(event, UserIMMessageEvent):
+            return Target(
+                event.channel_id,
+                parent_id=event.room_id,
+                channel=True,
+                adapter=self.get_adapter(),
+                self_id=bot.self_id if bot else None,
+                scope=SupportScope.heybox,
+            )
+        raise NotImplementedError
+
+    def get_message_id(self, event: Event) -> str:
+        if isinstance(event, UserIMMessageEvent):
+            return str(event.im_seq)
+        raise NotImplementedError
+
+    @export
+    async def text(self, seg: Text, bot: Union[Bot, None]) -> "MessageSegment":
+        return MessageSegment.text(seg.text)
+
+    @export
+    async def at(self, seg: At, bot: Union[Bot, None]) -> "MessageSegment":
+        if seg.flag == "user":
+            return MessageSegment.mention(seg.target)
+        raise SerializeFailed(
+            lang.require("nbp-uniseg", "failed_segment").format(adapter="qq", seg=seg, target="mention")
+        )
+
+    @export
+    async def image(self, seg: Image, bot: Union[Bot, None]) -> "MessageSegment":
+        if seg.url:
+            return MessageSegment.image(url=seg.url, width=seg.width or 0, height=seg.height or 0)
+        if seg.raw:
+            return MessageSegment.local_image(
+                seg.raw_bytes, width=seg.width or 0, height=seg.height or 0, filename=seg.name
+            )
+        raise SerializeFailed(lang.require("nbp-uniseg", "invalid_segment").format(type="image", seg=seg))
+
+    @export
+    async def reply(self, seg: Reply, bot: Union[Bot, None]) -> "MessageSegment":
+        return MessageSegment("$heybox:reply", {"message_id": seg.id})  # type: ignore
+
+    async def send_to(self, target: Union[Target, Event], bot: Bot, message: Message, **kwargs):
+        assert isinstance(bot, HeyboxBot)
+
+        reply_id = None
+        if message.has("$heybox:reply"):
+            reply_id = message["$heybox:reply", 0].data["message_id"]
+            message = message.exclude("$heybox:reply")
+
+        if isinstance(target, Event):
+            if TYPE_CHECKING:
+                assert isinstance(target, UserIMMessageEvent)
+            return await bot.send(event=target, message=message, **kwargs, is_reply=reply_id is not None)
+        return await bot.send_to_channel(
+            target.parent_id,
+            target.id,
+            message,
+            reply_id=reply_id,
+        )

src/nonebot_plugin_alconna/uniseg/adapters/satori/builder.py

@@ -110,7 +110,9 @@ def image(self, seg: ImageSegment):
         if src.startswith("data:"):
             mime, b64 = src[5:].split(";", 1)
             return Image(raw=b64decode(b64[7:]), mimetype=mime)(*self.generate(seg.children))
-        return Image(seg.data["src"])(*self.generate(seg.children))
+        return Image(seg.data["src"], width=seg.data.get("width"), height=seg.data.get("height"))(
+            *self.generate(seg.children)
+        )
 
     @build("audio")
     def audio(self, seg: AudioSegment):
@@ -122,7 +124,7 @@ def audio(self, seg: AudioSegment):
         if src.startswith("data:"):
             mime, b64 = src[5:].split(";", 1)
             return Audio(raw=b64decode(b64[7:]), mimetype=mime)(*self.generate(seg.children))
-        return Audio(seg.data["src"])(*self.generate(seg.children))
+        return Audio(seg.data["src"], duration=seg.data.get("duration"))(*self.generate(seg.children))
 
     @build("video")
     def video(self, seg: VideoSegment):
@@ -134,7 +136,10 @@ def video(self, seg: VideoSegment):
         if src.startswith("data:"):
             mime, b64 = src[5:].split(";", 1)
             return Video(raw=b64decode(b64[7:]), mimetype=mime)(*self.generate(seg.children))
-        return Video(seg.data["src"])(*self.generate(seg.children))
+        thumb = None
+        if poster := seg.data.get("poster"):
+            thumb = Image(url=poster, width=seg.data.get("width"), height=seg.data.get("height"))
+        return Video(seg.data["src"], thumbnail=thumb)(*self.generate(seg.children))
 
     @build("file")
     def file(self, seg: FileSegment):

src/nonebot_plugin_alconna/uniseg/message.py

@@ -203,6 +203,8 @@ def image(
             path: str | Path | None = None,
             raw: bytes | BytesIO | None = None,
             mimetype: str | None = None,
+            width: int | None = None,
+            height: int | None = None,
             name: str = "image.png",
         ) -> UniMessage[TS1 | Image]:
             """创建图片消息
@@ -214,6 +216,8 @@ def image(
                 raw: 图片原始数据
                 mimetype: 图片 MIME 类型
                 name: 图片名称
+                width: 图片宽度
+                height: 图片高度
             返回:
                 构建的消息
             """
@@ -227,6 +231,7 @@ def video(
             path: str | Path | None = None,
             raw: bytes | BytesIO | None = None,
             mimetype: str | None = None,
+            thumbnail: Image | None = None,
             name: str = "video.mp4",
         ) -> UniMessage[TS1 | Video]:
             """创建视频消息
@@ -237,6 +242,7 @@ def video(
                 path: 视频路径
                 raw: 视频原始数据
                 mimetype: 视频 MIME 类型
+                thumbnail: 视频缩略图
                 name: 视频名称
             返回:
                 构建的消息
@@ -251,6 +257,7 @@ def voice(
             path: str | Path | None = None,
             raw: bytes | BytesIO | None = None,
             mimetype: str | None = None,
+            duration: float | None = None,
             name: str = "voice.wav",
         ) -> UniMessage[TS1 | Voice]:
             """创建语音消息
@@ -261,6 +268,7 @@ def voice(
                 path: 语音路径
                 raw: 语音原始数据
                 mimetype: 语音 MIME 类型
+                duration: 语音时长
                 name: 语音名称
             返回:
                 构建的消息
@@ -275,6 +283,7 @@ def audio(
             path: str | Path | None = None,
             raw: bytes | BytesIO | None = None,
             mimetype: str | None = None,
+            duration: float | None = None,
             name: str = "audio.mp3",
         ) -> UniMessage[TS1 | Audio]:
             """创建音频消息
@@ -285,6 +294,7 @@ def audio(
                 path: 音频路径
                 raw: 音频原始数据
                 mimetype: 音频 MIME 类型
+                duration: 音频时长
                 name: 音频名称
             返回:
                 构建的消息
@@ -454,12 +464,14 @@ def image(
             path: str | Path | None = None,
             raw: bytes | BytesIO | None = None,
             mimetype: str | None = None,
+            width: int | None = None,
+            height: int | None = None,
             name: str = "image.png",
         ) -> UniMessage[TS1 | Image]:
             if isinstance(cls_or_self, UniMessage):
-                cls_or_self.append(Image(id, url, path, raw, mimetype, name))
+                cls_or_self.append(Image(id, url, path, raw, mimetype, name, width, height))
                 return cls_or_self
-            return UniMessage(Image(id, url, path, raw, mimetype, name))
+            return UniMessage(Image(id, url, path, raw, mimetype, name, width, height))
 
         @_method
         def video(
@@ -469,12 +481,13 @@ def video(
             path: str | Path | None = None,
             raw: bytes | BytesIO | None = None,
             mimetype: str | None = None,
+            thumbnail: Image | None = None,
             name: str = "video.mp4",
         ) -> UniMessage[TS1 | Video]:
             if isinstance(cls_or_self, UniMessage):
-                cls_or_self.append(Video(id, url, path, raw, mimetype, name))
+                cls_or_self.append(Video(id, url, path, raw, mimetype, name, thumbnail))
                 return cls_or_self
-            return UniMessage(Video(id, url, path, raw, mimetype, name))
+            return UniMessage(Video(id, url, path, raw, mimetype, name, thumbnail))
 
         @_method
         def voice(
@@ -484,12 +497,13 @@ def voice(
             path: str | Path | None = None,
             raw: bytes | BytesIO | None = None,
             mimetype: str | None = None,
+            duration: float | None = None,
             name: str = "voice.wav",
         ) -> UniMessage[TS1 | Voice]:
             if isinstance(cls_or_self, UniMessage):
-                cls_or_self.append(Voice(id, url, path, raw, mimetype, name))
+                cls_or_self.append(Voice(id, url, path, raw, mimetype, name, duration))
                 return cls_or_self
-            return UniMessage(Voice(id, url, path, raw, mimetype, name))
+            return UniMessage(Voice(id, url, path, raw, mimetype, name, duration))
 
         @_method
         def audio(
@@ -499,12 +513,13 @@ def audio(
             path: str | Path | None = None,
             raw: bytes | BytesIO | None = None,
             mimetype: str | None = None,
+            duration: float | None = None,
             name: str = "audio.mp3",
         ) -> UniMessage[TS1 | Audio]:
             if isinstance(cls_or_self, UniMessage):
-                cls_or_self.append(Audio(id, url, path, raw, mimetype, name))
+                cls_or_self.append(Audio(id, url, path, raw, mimetype, name, duration))
                 return cls_or_self
-            return UniMessage(Audio(id, url, path, raw, mimetype, name))
+            return UniMessage(Audio(id, url, path, raw, mimetype, name, duration))
 
         @_method
         def file(

src/nonebot_plugin_alconna/uniseg/segment.py

@@ -652,6 +652,8 @@ def save(self, media_save_dir: Optional[Union[str, Path]] = None) -> Path:
 class Image(Media):
     """Image对象, 表示一类图片元素"""
 
+    width: Optional[int] = field(default=None)
+    height: Optional[int] = field(default=None)
     name: str = field(default="image.png")
 
     __default_name__ = "image.png"
@@ -661,7 +663,7 @@ class Image(Media):
 class Audio(Media):
     """Audio对象, 表示一类音频元素"""
 
-    duration: Optional[int] = field(default=None)
+    duration: Optional[float] = field(default=None)
     name: str = field(default="audio.mp3")
 
     __default_name__ = "audio.mp3"
@@ -671,7 +673,7 @@ class Audio(Media):
 class Voice(Media):
     """Voice对象, 表示一类语音元素"""
 
-    duration: Optional[int] = field(default=None)
+    duration: Optional[float] = field(default=None)
     name: str = field(default="voice.wav")
 
     __default_name__ = "voice.wav"