✨ support adapter WXMP

Author: RF-Tar-Railt

example/bot.py

@@ -11,7 +11,8 @@
 # driver.register_adapter(SatoriAdapter)
 
 # nonebot.require("nonebot_plugin_alconna")
-nonebot.load_plugins("plugins")
+# nonebot.load_plugins("plugins")
+nonebot.load_plugin("plugins.demo1")
 
 
 async def _():

pdm.lock

@@ -14,7 +14,7 @@ dependencies = [
     "nonebot-plugin-waiter>=0.6.0",
 ]
 dynamic = ["version"]
-requires-python = ">=3.9"
+requires-python = ">=3.11"
 readme = "README.md"
 license = {text = "MIT"}
 keywords = [
@@ -69,6 +69,8 @@ dev = [
     "nonebot-plugin-localstore>=0.7.1",
     "pyyaml>=6.0.1",
     "fix-future-annotations>=0.5.0",
+    "nonebot-adapter-heybox>=0.1.1; python_version >= \"3.11\"",
+    "nonebot-adapter-wxmp>=0.1.4; python_version >= \"3.10\"",
 ]
 [tool.pdm.build]
 includes = ["src/nonebot_plugin_alconna"]

pyproject.toml

@@ -356,7 +356,7 @@ async def send_to(self, target: Union[Target, Event], bot: Bot, message: Message
                 message = message.exclude("mention_channel", "mention_user", "mention_everyone", "reference")
             return await bot.send(event=target, message=message, **kwargs)
 
-        if target.extra.get("qq.reply_seq"):
+        if target.extra.get("qq.reply_seq") is not None:
             target.extra["qq.reply_seq"] += 1
 
         if target.channel:

src/nonebot_plugin_alconna/uniseg/adapters/qq/exporter.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.wxmp
+
+    def get_builder(self):
+        from .builder import WXMPMessageBuilder
+
+        return WXMPMessageBuilder()
+
+    def get_exporter(self):
+        from .exporter import WXMPMessageExporter
+
+        return WXMPMessageExporter()

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

@@ -0,0 +1,47 @@
+from nonebot.adapters.wxmp.message import MessageSegment
+from nonebot.adapters.wxmp.message import Text as TextSegment
+from nonebot.adapters.wxmp.message import Emjoy as EmojiSegment
+from nonebot.adapters.wxmp.message import Image as ImageSegment
+from nonebot.adapters.wxmp.message import Video as VideoSegment
+from nonebot.adapters.wxmp.message import Voice as VoiceSegment
+from nonebot.adapters.wxmp.message import Location as LocationSegment
+from nonebot.adapters.wxmp.message import Miniprogrampage as MiniProgramSegment
+
+from nonebot_plugin_alconna.uniseg.constraint import SupportAdapter
+from nonebot_plugin_alconna.uniseg.builder import MessageBuilder, build
+from nonebot_plugin_alconna.uniseg.segment import Text, Audio, Emoji, Hyper, Image, Other, Video
+
+
+class WXMPMessageBuilder(MessageBuilder[MessageSegment]):
+    @classmethod
+    def get_adapter(cls) -> SupportAdapter:
+        return SupportAdapter.wxmp
+
+    @build("text")
+    def build_text(self, seg: TextSegment):
+        return Text(seg.data["text"])
+
+    @build("image")
+    def build_image(self, seg: ImageSegment):
+        return Image(id=seg.data["media_id"], url=str(seg.data["file_url"]))
+
+    @build("miniprogrampage")
+    def build_miniprogrampage(self, seg: MiniProgramSegment):
+        return Hyper("json", content={**seg.data})
+
+    @build("video")
+    def build_video(self, seg: VideoSegment):
+        return Video(id=seg.data["media_id"])
+
+    @build("voice")
+    def build_voice(self, seg: VoiceSegment):
+        return Audio(id=seg.data["media_id"])
+
+    @build("location")
+    def build_location(self, seg: LocationSegment):
+        return Other(seg)
+
+    @build("emjoy")
+    def build_emoji(self, seg: EmojiSegment):
+        t = seg.data["emjoy"]
+        return Emoji(t.value, t.name)

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

@@ -0,0 +1,87 @@
+from pathlib import Path
+from typing import Union
+
+from tarina import lang
+from nonebot.adapters import Bot, Event
+from nonebot.adapters.wxmp import Bot as WXMPBot
+from nonebot.adapters.wxmp.event import MessageEvent
+from nonebot.adapters.wxmp.event import Event as WXMPEvent
+from nonebot.adapters.wxmp.message import Message, EmjoyType, MessageSegment
+
+from nonebot_plugin_alconna.uniseg.constraint import SupportScope, SerializeFailed
+from nonebot_plugin_alconna.uniseg.segment import Text, Audio, Emoji, Hyper, Image, Video, Voice
+from nonebot_plugin_alconna.uniseg.exporter import Target, SupportAdapter, MessageExporter, export
+
+
+class WXMPMessageExporter(MessageExporter[Message]):
+    def get_message_type(self):
+        return Message
+
+    @classmethod
+    def get_adapter(cls) -> SupportAdapter:
+        return SupportAdapter.wxmp
+
+    def get_target(self, event: Event, bot: Union[Bot, None] = None) -> Target:
+        assert isinstance(event, WXMPEvent)
+        return Target(
+            event.get_user_id(),
+            private=True,
+            adapter=self.get_adapter(),
+            self_id=bot.self_id if bot else None,
+            scope=SupportScope.wechat_oap,
+        )
+
+    def get_message_id(self, event: Event) -> str:
+        assert isinstance(event, MessageEvent)
+        return str(event.message_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":
+            title = desc = url = seg.text
+            if getattr(seg, "_children", []):
+                title = desc = seg._children[0].text  # type: ignore
+            return MessageSegment.link(title, desc, url)
+        return MessageSegment.text(seg.text)
+
+    @export
+    async def media(self, seg: Union[Image, Voice, Video, Audio], bot: Union[Bot, None]) -> "MessageSegment":
+        name = seg.__class__.__name__.lower()
+        methods = {
+            "image": MessageSegment.image,
+            "voice": MessageSegment.voice,
+            "video": MessageSegment.video,
+            "audio": MessageSegment.voice,
+        }
+        if seg.id:
+            return methods[name](media_id=seg.id)
+        if seg.raw:
+            if name in ("voice", "audio"):
+                return MessageSegment.voice(file=seg.raw_bytes, format=seg.mimetype)
+            return methods[name](file=seg.raw_bytes)
+        if seg.path:
+            return methods[name](file_path=Path(seg.path))
+        if seg.url and name == "image":
+            return MessageSegment.image(file_url=seg.url)  # type: ignore
+        raise SerializeFailed(lang.require("nbp-uniseg", "invalid_segment").format(type=name, seg=seg))
+
+    @export
+    async def emoji(self, seg: Emoji, bot: Union[Bot, None]) -> "MessageSegment":
+        t = EmjoyType(seg.name)
+        return MessageSegment.emjoy(t)
+
+    @export
+    async def hyper(self, seg: Hyper, bot: Union[Bot, None]) -> "MessageSegment":
+        if isinstance(seg.content, dict):
+            return MessageSegment.miniprogrampage(**seg.content)
+        raise SerializeFailed(lang.require("nbp-uniseg", "invalid_segment").format(type="hyper", seg=seg))
+
+    async def send_to(self, target: Union[Target, Event], bot: Bot, message: Message, **kwargs):
+        assert isinstance(bot, WXMPBot)
+
+        if isinstance(target, Event):
+            return await bot.send(target, message, **kwargs)  # type: ignore
+        return await bot.send_custom_message(user_id=target.id, message=message)

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

@@ -17,6 +17,7 @@ class SupportAdapter(str, Enum):
     dodo = "DoDo"
     feishu = "Feishu"
     github = "GitHub"
+    heybox = "Heybox"
     kritor = "Kritor"
     kook = "Kaiheila"
     mail = "Mail"
@@ -30,6 +31,7 @@ class SupportAdapter(str, Enum):
     satori = "Satori"
     telegram = "Telegram"
     tail_chat = "Tailchat"
+    wxmp = "WXMP"
 
     nonebug = "fake"
 
@@ -47,6 +49,7 @@ class SupportScope(str, Enum):
     discord = "Discord"
     feishu = "Feishu"
     dodo = "DoDo"
+    heybox = "Heybox"
     kook = "Kaiheila"
     mail = "Mail"
     minecraft = "Minecraft"
@@ -96,6 +99,7 @@ def ensure_satori(platform: str):
             "kook": SupportScope.kook,
             "dingtalk": SupportScope.ding,
             "mail": SupportScope.mail,
+            "heybox": SupportScope.heybox,
         }.get(platform, SupportScope.satori_other)
 
 
@@ -108,6 +112,7 @@ class SupportAdapterModule(str, Enum):
     dodo = "nonebot.adapters.dodo"
     feishu = "nonebot.adapters.feishu"
     github = "nonebot.adapters.github"
+    heybox = "nonebot.adapters.heybox"
     kritor = "nonebot.adapters.kritor"
     kook = "nonebot.adapters.kaiheila"
     mail = "nonebot.adapters.mail"
@@ -121,6 +126,7 @@ class SupportAdapterModule(str, Enum):
     satori = "nonebot.adapters.satori"
     telegram = "nonebot.adapters.telegram"
     tail_chat = "nonebot_adapter_tailchat"
+    wxmp = "nonebot.adapters.wxmp"
 
 
 UNISEG_MESSAGE: Literal["_alc_uniseg_message"] = "_alc_uniseg_message"

src/nonebot_plugin_alconna/uniseg/constraint.py

@@ -439,14 +439,14 @@ async def select_wechat(target: "Target", bot: Bot):
         return False
     if bot.adapter.get_name() != SupportAdapter.onebot12:
         return False
-    return not hasattr(bot, "platform") or bot.platform == "wechat"
+    return hasattr(bot, "platform") and bot.platform == "wechat"
 
 
 @_register(SupportScope.wechat_oap)
 async def select_wechat_oap(target: "Target", bot: Bot):
     if target.channel:
         return False
-    if bot.adapter.get_name() != SupportAdapter.satori:
+    if bot.adapter.get_name() not in {SupportAdapter.wxmp, SupportAdapter.satori}:
         return False
     return not hasattr(bot, "platform") or bot.platform == "wechat-official"
 
@@ -457,7 +457,7 @@ async def select_wecom(target: "Target", bot: Bot):
         return False
     if bot.adapter.get_name() != SupportAdapter.satori:
         return False
-    return not hasattr(bot, "platform") or bot.platform == "wecom"
+    return hasattr(bot, "platform") and bot.platform == "wecom"
 
 
 @_register(SupportScope.tail_chat)
@@ -474,3 +474,12 @@ async def select_mail(target: "Target", bot: Bot):
     if bot.adapter.get_name() not in {SupportAdapter.mail, SupportAdapter.satori}:
         return False
     return not hasattr(bot, "platform") or bot.platform == "mail"
+
+
+@_register(SupportScope.heybox)
+async def select_heybox(target: "Target", bot: Bot):
+    if not target.channel:
+        return False
+    if bot.adapter.get_name() not in {SupportAdapter.heybox, SupportAdapter.satori}:
+        return False
+    return not hasattr(bot, "platform") or bot.platform == "heybox"