Merge branch 'master' into master

src/nonebot_plugin_alconna/matcher.py

@@ -5,13 +5,13 @@
 
 import nepattern.main
 from nonebot.rule import Rule
-from tarina import is_awaitable
 from nonebot.params import Depends
 from nonebot.matcher import Matcher
 from nepattern import AnyOne, AnyString
 from nonebot.permission import Permission
 from nonebot.dependencies import Dependent
 from arclet.alconna.tools import AlconnaFormat
+from tarina import is_awaitable, run_always_await
 from arclet.alconna.tools.construct import FuncMounter
 from arclet.alconna import Arg, Args, Alconna, command_manager
 from nonebot.typing import T_State, T_Handler, T_RuleChecker, T_PermissionChecker
@@ -28,7 +28,7 @@
 from .model import CompConfig
 from .consts import ALCONNA_ARG_KEY
 from .typings import MReturn, TConvert
-from .params import Check, AlcExecResult, assign, _seminal
+from .params import MIDDLEWARE, Check, AlcExecResult, assign, _seminal
 
 
 class ArgsMounter(Protocol):
@@ -110,6 +110,7 @@ def got_path(
         cls,
         path: str,
         prompt: str | Message | MessageSegment | MessageTemplate | None = None,
+        middleware: MIDDLEWARE | None = None,
         parameterless: Iterable[Any] | None = None,
     ) -> Callable[[T_Handler], T_Handler]:
         """装饰一个函数来指示 NoneBot 获取一个路径下的参数 `key`
@@ -128,13 +129,15 @@ def got_path(
         if not (arg := extract_arg(path, cls.command)):
             raise ValueError(f"Path {path} not found in Alconna")
 
-        async def _key_getter(event: Event, matcher: AlconnaMatcher):
+        async def _key_getter(event: Event, bot: Bot, matcher: AlconnaMatcher):
             matcher.set_target(ALCONNA_ARG_KEY.format(key=path))
             if matcher.get_target() == ALCONNA_ARG_KEY.format(key=path):
                 ms = event.get_message()[-1]
                 if (res := _validate(arg, ms)) is None:
                     await matcher.reject(prompt)
                     return
+                if middleware:
+                    res = await run_always_await(middleware, bot, matcher.state, res)
                 matcher.set_path_arg(path, res)
                 return
             if matcher.state.get(ALCONNA_ARG_KEY.format(key=path), ...) is not ...:

src/nonebot_plugin_alconna/params.py

@@ -21,8 +21,7 @@
 
 T_Duplication = TypeVar("T_Duplication", bound=Duplication)
 TS = TypeVar("TS", bound=Union[Segment, str])
-MATCH_MIDDLEWARE: TypeAlias = Callable[[Bot, T_State, Match], Any]
-QUERY_MIDDLEWARE: TypeAlias = Callable[[Bot, T_State, Query], Any]
+MIDDLEWARE: TypeAlias = Callable[[Bot, T_State, Any], Any]
 
 
 def _alconna_result(state: T_State) -> CommandResult:
@@ -49,12 +48,12 @@ def AlconnaMatches() -> Arparma:
     return Depends(_alconna_matches, use_cache=False)
 
 
-def AlconnaMatch(name: str, middleware: Optional[MATCH_MIDDLEWARE] = None) -> Match:
+def AlconnaMatch(name: str, middleware: Optional[MIDDLEWARE] = None) -> Match:
     async def _alconna_match(state: T_State, bot: Bot) -> Match:
         arp = _alconna_result(state).result
         mat = Match(arp.all_matched_args.get(name, Empty), name in arp.all_matched_args)
         if middleware and mat.available:
-            mat.result = await run_always_await(middleware, bot, state, mat)
+            mat.result = await run_always_await(middleware, bot, state, mat.result)
         return mat
 
     return Depends(_alconna_match, use_cache=False)
@@ -63,7 +62,7 @@ async def _alconna_match(state: T_State, bot: Bot) -> Match:
 def AlconnaQuery(
     path: str,
     default: Union[T, Empty] = Empty,
-    middleware: Optional[QUERY_MIDDLEWARE] = None,
+    middleware: Optional[MIDDLEWARE] = None,
 ) -> Query[T]:
     async def _alconna_query(state: T_State, bot: Bot) -> Query:
         arp = _alconna_result(state).result
@@ -75,7 +74,7 @@ async def _alconna_query(state: T_State, bot: Bot) -> Query:
         elif default != Empty:
             q.available = True
         if middleware and q.available:
-            q.result = await run_always_await(middleware, bot, state, q)
+            q.result = await run_always_await(middleware, bot, state, q.result)
         return q
 
     return Depends(_alconna_query, use_cache=False)
@@ -107,6 +106,16 @@ def _alconna_arg(state: T_State) -> Any:
     return Depends(_alconna_arg, use_cache=False)
 
 
+# def AlconnaArg(path: str, middleware: Optional[MIDDLEWARE] = None) -> Any:
+#     async def _alconna_arg(state: T_State, bot: Bot) -> Any:
+#         arg = state[ALCONNA_ARG_KEY.format(key=path)]
+#         if middleware:
+#             return await run_always_await(middleware, bot, state, arg)
+#         return arg
+#
+#     return Depends(_alconna_arg, use_cache=False)
+
+
 def _seg_match_msg(state: T_State) -> Message:
     return state[SEGMATCH_MSG]
 

src/nonebot_plugin_alconna/tools.py

@@ -90,37 +90,35 @@ async def reply_handle(event: Event, bot: Bot):
     return None
 
 
-async def image_fetch(bot: Bot, state: T_State, img: Match[Image]):
-    if not img.available:
-        return None
-    if img.result.url:  # mirai2, qqguild, kook, villa, feishu, minecraft, ding
-        req = Request("GET", img.result.url)
+async def image_fetch(bot: Bot, state: T_State, img: Image):
+    if img.url:  # mirai2, qqguild, kook, villa, feishu, minecraft, ding
+        req = Request("GET", img.url)
         resp = await bot.adapter.request(req)
         return resp.content
-    if not img.result.id:
+    if not img.id:
         return None
     adapter_name = bot.adapter.get_name()
     if adapter_name == "OneBot V11":
-        url = (await bot.get_image(file=img.result.id))["data"]["url"]
+        url = (await bot.get_image(file=img.id))["data"]["url"]
         req = Request("GET", url)
         resp = await bot.adapter.request(req)
         return resp.content
     if adapter_name == "OneBot V12":
-        return (await bot.get_file(type="data", file_id=img.result.id))["data"]
+        return (await bot.get_file(type="data", file_id=img.id))["data"]
     if adapter_name == "mirai2":
         url = (
             f"https://gchat.qpic.cn/gchatpic_new/0/0-0-"
-            f"{img.result.id.replace('-', '').upper()}/0"
+            f"{img.id.replace('-', '').upper()}/0"
         )
         req = Request("GET", url)
         resp = await bot.adapter.request(req)
         return resp.content
     if adapter_name == "Telegram":
         url = (
-                URL(bot.bot_config.api_server)
-                / "file"
-                / f"bot{bot.bot_config.token}"
-                / img.result.id
+            URL(bot.bot_config.api_server)
+            / "file"
+            / f"bot{bot.bot_config.token}"
+            / img.id
         )
         req = Request("GET", url)
         resp = await bot.adapter.request(req)

src/test/plugins/demo.py

@@ -188,9 +188,17 @@ async def bind_handle(reply: Reply = SegMatchResult(Reply)):
     await bind.send(str(reply))
 
 
-wc = on_alconna(Alconna("wc", Args["img", Image]))
+wc = on_alconna(Alconna("wc", Args["img?", Image]))
 
 
 @wc.handle()
-async def _(img: Match[bytes] = AlconnaMatch("img", image_fetch)):
-    print(img.result[:100])
+async def wc_h(
+    matcher: AlconnaMatcher, img: Match[list] = AlconnaMatch("img", image_fetch)
+):
+    if img.available:
+        matcher.set_path_arg("img", img.result)
+
+
+@wc.got_path("img", prompt="请输入图片", middleware=image_fetch)
+async def wc_g(img: list = AlconnaArg("img")):
+    await wc.send(f"img: {img[:10]}")