🧑‍💻 extract UniMessage.get_message_id, etc. to module functions

Author: RF-Tar-Railt

docs.md

@@ -656,15 +656,15 @@ async def send(
 
 ```python
 from nonebot import Event, Bot
-from nonebot_plugin_alconna.uniseg import UniMessage, Target
+from nonebot_plugin_alconna.uniseg import Target, get_message_id, get_target
 
 
 matcher = on_xxx(...)
 
 @matcher.handle()
-asycn def _(bot: Bot, event: Event):
-    target: Target = UniMessage.get_target(event, bot)
-    msg_id: str = UniMessage.get_message_id(event, bot)
+async def _(bot: Bot, event: Event):
+    target: Target = get_target(event, bot)
+    msg_id: str = get_message_id(event, bot)
 
 ```
 

src/nonebot_plugin_alconna/__init__.py

@@ -94,6 +94,7 @@
 from .rule import AlconnaRule as AlconnaRule
 from .uniseg import CustomNode as CustomNode
 from .uniseg import UniMessage as UniMessage
+from .uniseg import get_target as get_target
 from .extension import Extension as Extension
 from .extension import Interface as Interface
 from .matcher import on_alconna as on_alconna
@@ -105,6 +106,7 @@
 from .params import AlconnaMatch as AlconnaMatch
 from .params import AlconnaQuery as AlconnaQuery
 from .uniseg import SupportScope as SupportScope
+from .uniseg import message_edit as message_edit
 from .model import CommandResult as CommandResult
 from .pattern import select_first as select_first
 from .params import AlcExecResult as AlcExecResult
@@ -117,6 +119,8 @@
 from .uniseg import SupportAdapter as SupportAdapter
 from .uniseg import apply_filehost as apply_filehost
 from .uniseg import custom_handler as custom_handler
+from .uniseg import get_message_id as get_message_id
+from .uniseg import message_recall as message_recall
 from .matcher import AlconnaMatcher as AlconnaMatcher
 from .consts import ALCONNA_ARG_KEY as ALCONNA_ARG_KEY
 from .uniseg import SerializeFailed as SerializeFailed

src/nonebot_plugin_alconna/builtins/extensions/reply.py

@@ -4,8 +4,8 @@
 from arclet.alconna import Alconna
 from nonebot.internal.adapter import Bot, Event
 
-from nonebot_plugin_alconna.uniseg import reply_fetch
 from nonebot_plugin_alconna import Reply, Extension, UniMessage
+from nonebot_plugin_alconna.uniseg import reply_fetch, get_message_id
 
 
 class ReplyRecordExtension(Extension):
@@ -45,7 +45,7 @@ def get_reply(self, message_id: str) -> Optional[Reply]:
     async def receive_wrapper(self, bot: Bot, event: Event, command: Alconna, receive: UniMessage) -> UniMessage:
         if not (reply := await reply_fetch(event, bot)):
             return receive
-        msg_id = UniMessage.get_message_id(event, bot)
+        msg_id = get_message_id(event, bot)
         self.cache[msg_id] = reply
         return receive
 
@@ -93,7 +93,7 @@ async def message_provider(self, event, state, bot, use_origin: bool = False):
             msg = event.get_message()
         except (NotImplementedError, ValueError):
             return None
-        msg_id = UniMessage.get_message_id(event, bot)
+        msg_id = get_message_id(event, bot)
         if msg_id in self.cache:
             return self.cache[msg_id]
         uni_msg = UniMessage.generate_sync(message=msg, bot=bot)

src/nonebot_plugin_alconna/builtins/plugins/with/extension.py

@@ -5,7 +5,7 @@
 from arclet.alconna import Alconna
 from nonebot.internal.adapter import Bot, Event
 
-from nonebot_plugin_alconna import Target, Extension, UniMessage
+from nonebot_plugin_alconna import Target, Extension, UniMessage, get_target, get_message_id
 
 
 class PrefixAppendExtension(Extension):
@@ -31,10 +31,10 @@ def post_init(self, alc: Alconna) -> None:
         self.sep = alc.separators[0]
 
     async def receive_wrapper(self, bot: Bot, event: Event, command: Alconna, receive: UniMessage) -> UniMessage:
-        msg_id = UniMessage.get_message_id(event, bot)
+        msg_id = get_message_id(event, bot)
         if msg_id in self.cache:
             return self.cache[msg_id]
-        target = UniMessage.get_target(event, bot)
+        target = get_target(event, bot)
         prefix = self.supplier(target)
         if not prefix or not command.header_display.endswith(prefix):
             return receive

src/nonebot_plugin_alconna/extension.py

@@ -16,7 +16,7 @@
 from nonebot.compat import PydanticUndefined
 from nonebot.adapters import Bot, Event, Message
 
-from .uniseg import UniMessage
+from .uniseg import UniMessage, get_message_id
 
 OutputType = Literal["help", "shortcut", "completion", "error"]
 TM = TypeVar("TM", bound=Union[str, Message, UniMessage])
@@ -221,7 +221,7 @@ async def message_provider(
         if exc is not None:
             raise exc
         if event.get_type().startswith("message"):
-            msg_id = UniMessage.get_message_id(event, bot)
+            msg_id = get_message_id(event, bot)
             if use_origin and (uni_msg := unimsg_origin_cache.get(msg_id)) is not None:
                 return uni_msg
             if (uni_msg := unimsg_cache.get(msg_id)) is not None:

src/nonebot_plugin_alconna/matcher.py

@@ -37,11 +37,11 @@
 from .util import annotation
 from .model import CompConfig
 from .rule import AlconnaRule
-from .uniseg import Text, Segment, UniMessage
 from .uniseg.fallback import FallbackStrategy
 from .uniseg.template import UniMessageTemplate
 from .uniseg.message import current_send_wrapper
 from .extension import Extension, ExtensionExecutor
+from .uniseg import Text, Segment, UniMessage, get_target, get_message_id
 from .consts import ALCONNA_RESULT, ALCONNA_ARG_KEY, ALCONNA_ARG_PATH, log
 from .params import CHECK, MIDDLEWARE, Check, AlconnaParam, ExtensionParam, assign, _seminal, _Dispatch, merge_path
 
@@ -401,7 +401,7 @@ async def _receive(event: Event, matcher: Matcher) -> None:
             nonlocal id
             if not id:
                 try:
-                    id = UniMessage.get_message_id(event, current_bot.get())
+                    id = get_message_id(event, current_bot.get())
                 except Exception:
                     pass
             matcher.set_target(RECEIVE_KEY.format(id=id))
@@ -620,9 +620,9 @@ def convert(cls, message: _M) -> Message | UniMessage | str:
         if isinstance(message, MessageTemplate):
             return message.format(**state[ALCONNA_RESULT].result.all_matched_args, **state)
         if isinstance(message, UniMessageTemplate):
-            extra = {"$event": event, "$target": UniMessage.get_target(event, bot)}
+            extra = {"$event": event, "$target": get_target(event, bot)}
             try:
-                msg_id = UniMessage.get_message_id(event, bot)
+                msg_id = get_message_id(event, bot)
             except Exception:
                 pass
             else:
@@ -652,10 +652,10 @@ def i18n(
             **state[ALCONNA_RESULT].result.all_matched_args,
             **state,
             "$event": event,
-            "$target": UniMessage.get_target(event, bot),
+            "$target": get_target(event, bot),
         }
         try:
-            msg_id = UniMessage.get_message_id(event, bot)
+            msg_id = get_message_id(event, bot)
         except Exception:
             pass
         else:

src/nonebot_plugin_alconna/uniseg/__init__.py

@@ -43,9 +43,13 @@
 from .segment import CustomNode as CustomNode
 from .tools import image_fetch as image_fetch
 from .tools import reply_fetch as reply_fetch
+from .functions import get_target as get_target
 from .params import MessageTarget as MessageTarget
+from .functions import message_edit as message_edit
 from .constraint import SupportScope as SupportScope
 from .segment import custom_handler as custom_handler
+from .functions import get_message_id as get_message_id
+from .functions import message_recall as message_recall
 from .segment import custom_register as custom_register
 from .constraint import SupportAdapter as SupportAdapter
 from .fallback import FallbackMessage as FallbackMessage

src/nonebot_plugin_alconna/uniseg/functions.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from tarina import lang
+from nonebot.adapters import Bot, Event
+from nonebot.internal.matcher import current_bot, current_event
+
+from .exporter import SerializeFailed
+from .adapters import alter_get_exporter
+
+if TYPE_CHECKING:
+    from .target import Target
+    from .message import UniMessage
+
+
+async def message_recall(
+    message_id: str | None = None, event: Event | None = None, bot: Bot | None = None, adapter: str | None = None
+):
+    if not event:
+        try:
+            event = current_event.get()
+        except LookupError as e:
+            raise SerializeFailed(lang.require("nbp-uniseg", "event_missing")) from e
+    if not bot:
+        try:
+            bot = current_bot.get()
+        except LookupError as e:
+            raise SerializeFailed(lang.require("nbp-uniseg", "bot_missing")) from e
+    if not adapter:
+        _adapter = bot.adapter
+        adapter = _adapter.get_name()
+    if fn := alter_get_exporter(adapter):
+        return await fn.recall(message_id or fn.get_message_id(event), bot, event)
+    raise SerializeFailed(lang.require("nbp-uniseg", "unsupported").format(adapter=adapter))
+
+
+async def message_edit(
+    msg: UniMessage,
+    message_id: str | None = None,
+    event: Event | None = None,
+    bot: Bot | None = None,
+    adapter: str | None = None,
+):
+    if not event:
+        try:
+            event = current_event.get()
+        except LookupError as e:
+            raise SerializeFailed(lang.require("nbp-uniseg", "event_missing")) from e
+    if not bot:
+        try:
+            bot = current_bot.get()
+        except LookupError as e:
+            raise SerializeFailed(lang.require("nbp-uniseg", "bot_missing")) from e
+    if not adapter:
+        _adapter = bot.adapter
+        adapter = _adapter.get_name()
+    if fn := alter_get_exporter(adapter):
+        return await fn.edit(msg, message_id or fn.get_message_id(event), bot, event)
+    raise SerializeFailed(lang.require("nbp-uniseg", "unsupported").format(adapter=adapter))
+
+
+def get_message_id(event: Event | None = None, bot: Bot | None = None, adapter: str | None = None) -> str:
+    if not event:
+        try:
+            event = current_event.get()
+        except LookupError as e:
+            raise SerializeFailed(lang.require("nbp-uniseg", "event_missing")) from e
+    if hasattr(event, "__uniseg_message_id__"):
+        return event.__uniseg_message_id__  # type: ignore
+    if not adapter:
+        if not bot:
+            try:
+                bot = current_bot.get()
+            except LookupError as e:
+                raise SerializeFailed(lang.require("nbp-uniseg", "bot_missing")) from e
+        _adapter = bot.adapter
+        adapter = _adapter.get_name()
+    if fn := alter_get_exporter(adapter):
+        setattr(event, "__uniseg_message_id__", msg_id := fn.get_message_id(event))
+        return msg_id
+    raise SerializeFailed(lang.require("nbp-uniseg", "unsupported").format(adapter=adapter))
+
+
+def get_target(event: Event | None = None, bot: Bot | None = None, adapter: str | None = None) -> Target:
+    if not event:
+        try:
+            event = current_event.get()
+        except LookupError as e:
+            raise SerializeFailed(lang.require("nbp-uniseg", "event_missing")) from e
+    if not adapter:
+        if not bot:
+            try:
+                bot = current_bot.get()
+            except LookupError as e:
+                raise SerializeFailed(lang.require("nbp-uniseg", "bot_missing")) from e
+        _adapter = bot.adapter
+        adapter = _adapter.get_name()
+    if fn := alter_get_exporter(adapter):
+        return fn.get_target(event, bot)
+    raise SerializeFailed(lang.require("nbp-uniseg", "unsupported").format(adapter=adapter))