✨ support Mail adapter

Author: RF-Tar-Railt

pdm.lock

@@ -67,6 +67,7 @@ dev = [
     "pytest-mock>=3.14.0",
     "nonebot-plugin-localstore>=0.7.1",
     "pyyaml>=6.0.1",
+    "nonebot-adapter-mail>=1.0.0a3",
 ]
 [tool.pdm.build]
 includes = ["src/nonebot_plugin_alconna"]

pyproject.toml

@@ -21,6 +21,7 @@ def get_adapter(cls) -> SupportAdapter:
     def get_target(self, event: Event, bot: Union[Bot, None] = None) -> Target:
         return Target(
             event.get_user_id(),
+            private=True,
             adapter=self.get_adapter(),
             self_id=bot.self_id if bot else None,
             scope=SupportScope.console,

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

@@ -1,3 +1,4 @@
+from mimetypes import guess_type
 from typing import TYPE_CHECKING
 
 from nonebot.adapters import Bot, Event
@@ -18,7 +19,7 @@
 
 from nonebot_plugin_alconna.uniseg.constraint import SupportAdapter
 from nonebot_plugin_alconna.uniseg.builder import MessageBuilder, build
-from nonebot_plugin_alconna.uniseg.segment import At, AtAll, Image, Other, Reply, Button, Keyboard
+from nonebot_plugin_alconna.uniseg.segment import At, File, AtAll, Audio, Image, Other, Reply, Video, Button, Keyboard
 
 
 class DiscordMessageBuilder(MessageBuilder):
@@ -54,7 +55,17 @@ def sticker(self, seg: StickerSegment):
 
     @build("attachment")
     def attachment(self, seg: AttachmentSegment):
-        return Image(id=seg.data["attachment"].filename)
+        mtype = guess_type(seg.data["attachment"].filename)[0]
+        if mtype and mtype.startswith("image"):
+            return Image(id=seg.data["attachment"].filename, name=seg.data["attachment"].filename)
+        if mtype and mtype.startswith("video"):
+            return Video(id=seg.data["attachment"].filename, name=seg.data["attachment"].filename)
+        if mtype and mtype.startswith("audio"):
+            return Audio(id=seg.data["attachment"].filename, name=seg.data["attachment"].filename)
+        return File(
+            id=seg.data["attachment"].filename,
+            name=seg.data["attachment"].filename,
+        )
 
     @build("reference")
     def reference(self, seg: ReferenceSegment):

src/nonebot_plugin_alconna/uniseg/adapters/discord/builder.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.mail
+
+    def get_builder(self):
+        from .builder import MailMessageBuilder
+
+        return MailMessageBuilder()
+
+    def get_exporter(self):
+        from .exporter import MailMessageExporter
+
+        return MailMessageExporter()

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

@@ -0,0 +1,53 @@
+from typing import TYPE_CHECKING
+
+from nonebot.adapters import Bot, Event
+from nonebot.adapters.mail.event import NewMailMessageEvent
+from nonebot.adapters.mail.message import Html, Attachment, MessageSegment
+
+from nonebot_plugin_alconna.uniseg.constraint import SupportAdapter
+from nonebot_plugin_alconna.uniseg.builder import MessageBuilder, build
+from nonebot_plugin_alconna.uniseg.segment import File, Text, Audio, Image, Reply, Video
+
+
+class MailMessageBuilder(MessageBuilder[MessageSegment]):
+    @classmethod
+    def get_adapter(cls) -> SupportAdapter:
+        return SupportAdapter.mail
+
+    @build("html")
+    def html(self, seg: Html):
+        return Text(seg.data["html"]).mark(0, len(seg.data["html"]), "html")
+
+    @build("attachment")
+    def attachment(self, seg: Attachment):
+        mtype = seg.data["content_type"]
+        if mtype and mtype.startswith("image"):
+            return Image(
+                raw=seg.data["data"],
+                mimetype=mtype,
+                name=seg.data["name"],
+            )
+        if mtype and mtype.startswith("audio"):
+            return Audio(
+                raw=seg.data["data"],
+                mimetype=mtype,
+                name=seg.data["name"],
+            )
+        if mtype and mtype.startswith("video"):
+            return Video(
+                raw=seg.data["data"],
+                mimetype=mtype,
+                name=seg.data["name"],
+            )
+        return File(
+            raw=seg.data["data"],
+            mimetype=seg.data["content_type"],
+            name=seg.data["name"],
+        )
+
+    async def extract_reply(self, event: Event, bot: Bot):
+        if TYPE_CHECKING:
+            assert isinstance(event, NewMailMessageEvent)
+
+        if event.reply:
+            return Reply(event.reply.id, msg=event.reply.message, origin=event.reply)

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

@@ -0,0 +1,98 @@
+from pathlib import Path
+from typing import Union
+
+from tarina import lang
+from nonebot.adapters import Bot, Event
+from nonebot.internal.driver import Request
+from nonebot.adapters.mail import Bot as MailBot
+from nonebot.adapters.mail.event import NewMailMessageEvent
+from nonebot.adapters.mail.message import Message, MessageSegment
+
+from nonebot_plugin_alconna.uniseg.constraint import SupportScope, SerializeFailed
+from nonebot_plugin_alconna.uniseg.exporter import Target, SupportAdapter, MessageExporter, export
+from nonebot_plugin_alconna.uniseg.segment import At, File, Text, Audio, Image, Reply, Video, Voice
+
+
+class MailMessageExporter(MessageExporter[Message]):
+    def get_message_type(self):
+        return Message
+
+    @classmethod
+    def get_adapter(cls) -> SupportAdapter:
+        return SupportAdapter.mail
+
+    def get_target(self, event: Event, bot: Union[Bot, None] = None) -> Target:
+        assert isinstance(event, NewMailMessageEvent)
+        return Target(
+            event.get_user_id(),
+            private=True,
+            adapter=self.get_adapter(),
+            self_id=bot.self_id if bot else None,
+            scope=SupportScope.mail,
+        )
+
+    def get_message_id(self, event: Event) -> str:
+        assert isinstance(event, NewMailMessageEvent)
+        return event.id
+
+    @export
+    async def text(self, seg: Text, bot: Union[Bot, None]) -> "MessageSegment":
+        if not seg.styles:
+            return MessageSegment.text(seg.text)
+        style = seg.extract_most_style()
+        if style == "link":
+            if not getattr(seg, "_children", []):
+                return MessageSegment.html(f'<a href="{seg.text}">{seg.text}</a>')
+            else:
+                return MessageSegment.html(f'<a href="{seg.text}">{seg._children[0].text}</a>')  # type: ignore
+        return MessageSegment.html(str(seg))
+
+    @export
+    async def at(self, seg: At, bot: Union[Bot, None]) -> "MessageSegment":
+        if seg.flag == "user":
+            return MessageSegment.html(f'<a href="mailto:{seg.target}">@{seg.target}</a>')
+        elif seg.flag == "channel":
+            return MessageSegment.html(f" #{seg.target}")
+        else:
+            raise SerializeFailed(lang.require("nbp-uniseg", "invalid_segment").format(type="at", seg=seg))
+
+    @export
+    async def media(self, seg: Union[Image, Voice, Video, Audio, File], bot: Union[Bot, None]) -> "MessageSegment":
+        name = seg.__class__.__name__.lower()
+
+        if seg.raw and (seg.id or seg.name):
+            return MessageSegment.attachment(seg.raw, seg.id or seg.name, seg.mimetype)
+        elif seg.path:
+            path = Path(seg.path)
+            return MessageSegment.attachment(path, path.name)
+        elif bot and seg.url:
+            if name == "image":
+                return MessageSegment.html(f'<img src="{seg.url}" />')
+            elif name == "video":
+                return MessageSegment.html(f'<video src="{seg.url}" controls />')
+            elif name in ["audio", "voice"]:
+                return MessageSegment.html(f'<audio src="{seg.url}" controls />')
+            resp = await bot.adapter.request(Request("GET", seg.url))
+            return MessageSegment.attachment(
+                resp.content,  # type: ignore
+                seg.id or seg.name or seg.url.split("/")[-1],
+            )
+        else:
+            raise SerializeFailed(lang.require("nbp-uniseg", "invalid_segment").format(type=name, seg=seg))
+
+    @export
+    async def reply(self, seg: Reply, bot: Union[Bot, None]) -> "MessageSegment":
+        return MessageSegment("mail:reply", {"message_id": seg.id})  # type: ignore
+
+    async def send_to(self, target: Union[Target, Event], bot: Bot, message: Message, **kwargs):
+        assert isinstance(bot, MailBot)
+
+        in_reply_to = None
+        if message.has("$mail:reply"):
+            reply = message["mail:reply", 0]
+            message = message.exclude("mail:reply")
+            in_reply_to = reply.data["message_id"]
+
+        if isinstance(target, Event):
+            return await bot.send(target, message, in_reply_to=in_reply_to, **kwargs)  # type: ignore
+        return await bot.send_to(recipient=target.id, message=message, in_reply_to=in_reply_to, **kwargs)

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

@@ -0,0 +1,21 @@
+from typing import TYPE_CHECKING, Union
+
+from nonebot.adapters import Bot
+from nonebot.adapters.mail.bot import Bot as MailBot
+
+from nonebot_plugin_alconna.uniseg.target import Target, TargetFetcher
+from nonebot_plugin_alconna.uniseg.constraint import SupportScope, SupportAdapter
+
+
+class MailTargetFetcher(TargetFetcher):
+    @classmethod
+    def get_adapter(cls) -> SupportAdapter:
+        return SupportAdapter.mail
+
+    async def fetch(self, bot: Bot, target: Union[Target, None] = None):
+        if TYPE_CHECKING:
+            assert isinstance(bot, MailBot)
+        if target and not target.private:
+            return
+        for uid in await bot.get_unseen_uids():
+            yield Target(uid, private=True, adapter=self.get_adapter(), self_id=bot.self_id, scope=SupportScope.mail)