🐛 extension select

resolve #94

Author: RF-Tar-Railt

src/nonebot_plugin_alconna/extension.py

@@ -6,7 +6,6 @@
 import importlib as imp
 from weakref import finalize
 from dataclasses import dataclass
-from typing_extensions import Self
 from abc import ABCMeta, abstractmethod
 from typing import TYPE_CHECKING, Any, Union, Generic, Literal, TypeVar, ClassVar
 
@@ -129,82 +128,9 @@ def id(self) -> str:
 unimsg_origin_cache: LRU[str, UniMessage] = LRU(16)
 
 
-class ExtensionExecutor:
-    globals: ClassVar[list[type[Extension] | Extension]] = [DefaultExtension()]
-    _rule: AlconnaRule
-
-    def __init__(
-        self,
-        rule: AlconnaRule,
-        extensions: list[type[Extension] | Extension] | None = None,
-        excludes: list[str | type[Extension]] | None = None,
-    ):
-        self.extensions: list[Extension] = []
-        for ext in self.globals:
-            if isinstance(ext, type):
-                self.extensions.append(ext())
-            else:
-                self.extensions.append(ext)
-        for ext in extensions or []:
-            if isinstance(ext, type):
-                self.extensions.append(ext())
-            else:
-                self.extensions.append(ext)
-        for exl in excludes or []:
-            if isinstance(exl, str) and exl.startswith("!"):
-                raise ValueError(lang.require("nbp-alc", "error.extension.forbid_exclude"))
-        self._excludes = set(excludes or [])
-        self.extensions = [
-            ext
-            for ext in self.extensions
-            if ext.id not in self._excludes
-            and ext.__class__ not in self._excludes
-            and (not (ns := ext.namespace) or ns == rule._namespace)
-        ]
-        self.context: list[Extension] = []
-        self._rule = rule
-
-        _callbacks.add(self._callback)
-
-        finalize(self, _callbacks.remove, self._callback)
-
-    def _callback(self, *append_global_ext: type[Extension] | Extension):
-        for _ext in append_global_ext:
-            if isinstance(_ext, type):
-                _ext = _ext()
-            if _ext.id in self._excludes or _ext.__class__ in self._excludes:
-                continue
-            if (ns := _ext.namespace) and ns != self._rule._namespace:
-                continue
-            self.extensions.append(_ext)
-            _ext.post_init(self._rule.command())  # type: ignore
-
-    def __enter__(self) -> Self:
-        return self
-
-    def __exit__(self, exc_type, exc_value, traceback):
-        self.clear()
-
-    def select(self, bot: Bot, event: Event) -> Self:
-        self.context = [ext for ext in self.extensions if ext.validate(bot, event)]
-        self.context.sort(key=lambda ext: ext.priority)
-        return self
-
-    def clear(self) -> None:
-        self.context.clear()
-
-    async def output_converter(self, output_type: OutputType, content: str) -> UniMessage:
-        exc = None
-        for ext in self.context:
-            if not ext._overrides["output_converter"]:
-                continue
-            try:
-                return await ext.output_converter(output_type, content)
-            except Exception as e:
-                exc = e
-        if not exc:
-            return UniMessage()
-        raise exc  # type: ignore
+@dataclass
+class SelectedExtensions:
+    context: list[Extension]
 
     async def message_provider(
         self, event: Event, state: T_State, bot: Bot, use_origin: bool = False
@@ -269,13 +195,82 @@ async def parse_wrapper(self, bot: Bot, state: T_State, event: Event, res: Arpar
             *(ext.parse_wrapper(bot, state, event, res) for ext in self.context if ext._overrides["parse_wrapper"])
         )
 
+    async def output_converter(self, output_type: OutputType, content: str) -> UniMessage:
+        exc = None
+        for ext in self.context:
+            if not ext._overrides["output_converter"]:
+                continue
+            try:
+                return await ext.output_converter(output_type, content)
+            except Exception as e:
+                exc = e
+        if not exc:
+            return UniMessage()
+        raise exc  # type: ignore
+
     async def send_wrapper(self, bot: Bot, event: Event, send: TM) -> TM:
         res = send
         for ext in self.context:
             if ext._overrides["send_wrapper"]:
                 res = await ext.send_wrapper(bot, event, res)
         return res
 
+
+class ExtensionExecutor(SelectedExtensions):
+    globals: ClassVar[list[type[Extension] | Extension]] = [DefaultExtension()]
+    _rule: AlconnaRule
+
+    def __init__(
+        self,
+        rule: AlconnaRule,
+        extensions: list[type[Extension] | Extension] | None = None,
+        excludes: list[str | type[Extension]] | None = None,
+    ):
+        self.extensions: list[Extension] = []
+        for ext in self.globals:
+            if isinstance(ext, type):
+                self.extensions.append(ext())
+            else:
+                self.extensions.append(ext)
+        for ext in extensions or []:
+            if isinstance(ext, type):
+                self.extensions.append(ext())
+            else:
+                self.extensions.append(ext)
+        for exl in excludes or []:
+            if isinstance(exl, str) and exl.startswith("!"):
+                raise ValueError(lang.require("nbp-alc", "error.extension.forbid_exclude"))
+        self._excludes = set(excludes or [])
+        self.extensions = [
+            ext
+            for ext in self.extensions
+            if ext.id not in self._excludes
+            and ext.__class__ not in self._excludes
+            and (not (ns := ext.namespace) or ns == rule._namespace)
+        ]
+        self.context = self.extensions
+        self._rule = rule
+
+        _callbacks.add(self._callback)
+
+        finalize(self, _callbacks.remove, self._callback)
+
+    def _callback(self, *append_global_ext: type[Extension] | Extension):
+        for _ext in append_global_ext:
+            if isinstance(_ext, type):
+                _ext = _ext()
+            if _ext.id in self._excludes or _ext.__class__ in self._excludes:
+                continue
+            if (ns := _ext.namespace) and ns != self._rule._namespace:
+                continue
+            self.extensions.append(_ext)
+            _ext.post_init(self._rule.command())  # type: ignore
+
+    def select(self, bot: Bot, event: Event) -> SelectedExtensions:
+        context = [ext for ext in self.extensions if ext.validate(bot, event)]
+        context.sort(key=lambda ext: ext.priority)
+        return SelectedExtensions(context)
+
     def before_catch(self, name: str, annotation: Any, default: Any) -> bool:
         for ext in self.extensions:
             if ext._overrides["catch"]:

src/nonebot_plugin_alconna/matcher.py

@@ -17,7 +17,6 @@
 from tarina.lang.model import LangItem
 from nonebot.permission import Permission
 from nonebot.dependencies import Dependent
-from nonebot.message import run_postprocessor
 from arclet.alconna.tools import AlconnaFormat
 from nonebot.consts import ARG_KEY, RECEIVE_KEY
 from nonebot.internal.params import DefaultParam
@@ -1074,9 +1073,3 @@ def referent(cmd: str | Alconna | None) -> type[AlconnaMatcher] | None:
         except KeyError:
             return None
     return None
-
-
-@run_postprocessor
-@annotation(matcher=AlconnaMatcher)
-def _exit_executor(matcher: AlconnaMatcher):
-    matcher.executor.clear()

src/nonebot_plugin_alconna/params.py

@@ -16,7 +16,7 @@
 
 from .typings import CHECK, MIDDLEWARE
 from .model import T, Match, Query, CommandResult
-from .extension import Extension, ExtensionExecutor
+from .extension import Extension, ExtensionExecutor, SelectedExtensions
 from .consts import ALCONNA_RESULT, ALCONNA_ARG_KEY, ALCONNA_ARG_PATH, ALCONNA_EXTENSION, ALCONNA_EXEC_RESULT
 
 T_Duplication = TypeVar("T_Duplication", bound=Duplication)
@@ -204,8 +204,8 @@ async def _arparma_check(bot: Bot, state: T_State, event: Event, matcher: Matche
 
 def AlconnaExtension(target: type[T_Extension]) -> T_Extension:
     def _alconna_extension(state: T_State):
-        exts = state[ALCONNA_EXTENSION]
-        return next((i for i in exts if isinstance(i, target)), None)  # type: ignore
+        selected: SelectedExtensions = state[ALCONNA_EXTENSION]
+        return next((i for i in selected.context if isinstance(i, target)), None)
 
     return Depends(_alconna_extension, use_cache=False)
 
@@ -261,6 +261,8 @@ def _check_param(cls, param: inspect.Parameter, allow_types: tuple[type[Param],
             return cls(..., type=Alconna)
         if annotation is Duplication:
             return cls(..., type=Duplication)
+        if annotation is SelectedExtensions:
+            return cls(..., type=SelectedExtensions)
         if inspect.isclass(annotation) and issubclass(annotation, Duplication):
             return cls(..., anno=param.annotation, type=Duplication)
         if inspect.isclass(annotation) and issubclass(annotation, Extension):
@@ -288,9 +290,11 @@ async def _solve(self, matcher: Matcher, event: Event, state: T_State, **kwargs:
             if anno := self.extra.get("anno"):
                 return anno(res.result)
             return generate_duplication(res.source)(res.result)
+        if t is SelectedExtensions:
+            return state[ALCONNA_EXTENSION]
         if t is Extension:
             anno = self.extra["anno"]
-            return next((i for i in state[ALCONNA_EXTENSION] if isinstance(i, anno)), None)  # type: ignore
+            return next((i for i in state[ALCONNA_EXTENSION].context if isinstance(i, anno)), None)  # type: ignore
         if t is Match:
             target = res.result.all_matched_args.get(self.extra["name"], Empty)
             return Match(target, target != Empty)
@@ -320,18 +324,6 @@ async def _solve(self, matcher: Matcher, event: Event, state: T_State, **kwargs:
             return result
         return self.default if self.default not in (..., Empty) else PydanticUndefined
 
-    # async def _check(self, state: T_State, **kwargs: Any) -> Any:
-    #     if self.extra["type"] == Any:
-    #         if (
-    #             self.extra["name"] in _alconna_result(state).result.all_matched_args
-    #             or ALCONNA_ARG_KEY.format(key=self.extra["name"]) in state
-    #             or ((path := state.get(ALCONNA_ARG_PATH)) and path.endswith(f".{self.extra['name']}"))
-    #         ):
-    #             return True
-    #         if self.default not in (..., Empty):
-    #             return True
-    #         return False
-
 
 class _Dispatch:
     def __init__(

src/nonebot_plugin_alconna/rule.py

@@ -19,7 +19,7 @@
 from .uniseg import UniMsg, UniMessage
 from .model import CompConfig, CommandResult
 from .uniseg.constraint import UNISEG_MESSAGE
-from .extension import Extension, ExtensionExecutor
+from .extension import Extension, ExtensionExecutor, SelectedExtensions
 from .consts import ALCONNA_RESULT, ALCONNA_EXTENSION, ALCONNA_EXEC_RESULT, log
 
 try:
@@ -212,9 +212,9 @@ def __hash__(self) -> int:
         return hash(self.command.__hash__())
 
     async def handle(
-        self, cmd: Alconna, bot: Bot, event: Event, state: T_State, msg: UniMessage
+        self, selected: SelectedExtensions, cmd: Alconna, bot: Bot, event: Event, state: T_State, msg: UniMessage
     ) -> Union[Arparma, Literal[False]]:
-        ctx = await self.executor.context_provider(event, bot, state)
+        ctx = await selected.context_provider(event, bot, state)
         try:
             session_id = event.get_session_id()
         except ValueError:
@@ -228,7 +228,7 @@ async def handle(
         if res:
             interface.exit()
             return res
-        if not await self.executor.permission_check(bot, event, cmd):
+        if not await selected.permission_check(bot, event, cmd):
             return False
 
         res = Arparma(
@@ -271,12 +271,10 @@ def _checker(_event: Event):
         return res
 
     async def __call__(self, event: Event, state: T_State, bot: Bot) -> bool:
-        self.executor.select(bot, event)
-        if not (msg := await self.executor.message_provider(event, state, bot, self.use_origin)):
-            self.executor.clear()
+        selected = self.executor.select(bot, event)
+        if not (msg := await selected.message_provider(event, state, bot, self.use_origin)):
             return False
         if not self.response_self and check_self_send(bot, event):
-            self.executor.clear()
             return False
         try:
             session_id = event.get_session_id()
@@ -286,38 +284,33 @@ async def __call__(self, event: Event, state: T_State, bot: Bot) -> bool:
             await self._tasks[session_id]
         cmd = self.command()
         if not cmd:
-            self.executor.clear()
             return False
         if command_manager.is_disable(cmd):
-            self.executor.clear()
             return False
-        msg = await self.executor.receive_wrapper(bot, event, cmd, msg)
+        msg = await selected.receive_wrapper(bot, event, cmd, msg)
         Arparma._additional.update(bot=lambda: bot, event=lambda: event, state=lambda: state)
         state[UNISEG_MESSAGE] = msg
 
         with output_manager.capture(cmd.name) as cap:
             output_manager.set_action(lambda x: x, cmd.name)
-            task = asyncio.create_task(self.handle(cmd, bot, event, state, msg))
+            task = asyncio.create_task(self.handle(selected, cmd, bot, event, state, msg))
             if session_id:
                 self._tasks[session_id] = task
                 task.add_done_callback(lambda _: self._tasks.pop(session_id, None))
             try:
                 arp = await task
                 if arp is False:
-                    self.executor.clear()
                     return False
             except Exception as e:
                 arp = Arparma(cmd._hash, msg, False, error_info=e)
             may_help_text: Optional[str] = cap.get("output", None)
         if not arp.head_matched:
-            self.executor.clear()
             return False
         if not arp.matched and not may_help_text and self.skip:
             log(
                 "TRACE",
                 escape_tag(Lang.nbp_alc.log.parse(msg=msg, cmd=self._path, arp=arp)),
             )
-            self.executor.clear()
             return False
         if arp.head_matched:
             log(
@@ -328,18 +321,15 @@ async def __call__(self, event: Event, state: T_State, bot: Bot) -> bool:
             may_help_text = str(arp.error_info)
         if self.auto_send and may_help_text:
             await self.send(may_help_text, bot, event, arp)
-            self.executor.clear()
             return False
         if self.skip and may_help_text:
-            self.executor.clear()
             return False
-        if not await self.executor.permission_check(bot, event, cmd):
-            self.executor.clear()
+        if not await selected.permission_check(bot, event, cmd):
             return False
-        await self.executor.parse_wrapper(bot, state, event, arp)
+        await selected.parse_wrapper(bot, state, event, arp)
         state[ALCONNA_RESULT] = CommandResult(result=arp, output=may_help_text)
         state[ALCONNA_EXEC_RESULT] = cmd.exec_result
-        state[ALCONNA_EXTENSION] = self.executor.context
+        state[ALCONNA_EXTENSION] = selected
         return True
 
     async def send(self, text: str, bot: Bot, event: Event, arp: Arparma) -> Any:

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

@@ -203,18 +203,18 @@ async def reaction(
         if isinstance(context, Target):
             if context.private:
                 if delete:
-                    return await bot.directMessage_deleteReaction(msg_id=msg_id, emoji=emoji.name or emoji.id)
-                return await bot.directMessage_addReaction(msg_id=msg_id, emoji=emoji.name or emoji.id)
+                    return await bot.directMessage_deleteReaction(msg_id=msg_id, emoji=emoji.id)
+                return await bot.directMessage_addReaction(msg_id=msg_id, emoji=emoji.id)
             if delete:
-                return await bot.message_deleteReaction(msg_id=msg_id, emoji=emoji.name or emoji.id)
-            return await bot.message_addReaction(msg_id=msg_id, emoji=emoji.name or emoji.id)
+                return await bot.message_deleteReaction(msg_id=msg_id, emoji=emoji.id)
+            return await bot.message_addReaction(msg_id=msg_id, emoji=emoji.id)
         if isinstance(context, PrivateMessageEvent):
             if delete:
-                return await bot.directMessage_deleteReaction(msg_id=msg_id, emoji=emoji.name or emoji.id)
-            return await bot.directMessage_addReaction(msg_id=msg_id, emoji=emoji.name or emoji.id)
+                return await bot.directMessage_deleteReaction(msg_id=msg_id, emoji=emoji.id)
+            return await bot.directMessage_addReaction(msg_id=msg_id, emoji=emoji.id)
         if delete:
-            return await bot.message_deleteReaction(msg_id=msg_id, emoji=emoji.name or emoji.id)
-        return await bot.message_addReaction(msg_id=msg_id, emoji=emoji.name or emoji.id)
+            return await bot.message_deleteReaction(msg_id=msg_id, emoji=emoji.id)
+        return await bot.message_addReaction(msg_id=msg_id, emoji=emoji.id)
 
     def get_reply(self, mid: Any):
         _mid: MessageCreateReturn = cast(MessageCreateReturn, mid)