✨ support build Command object from json/yaml file

pyproject.toml

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

src/nonebot_plugin_alconna/__init__.py

@@ -125,9 +125,13 @@
 from .uniseg import UniversalMessage as UniversalMessage
 from .uniseg import UniversalSegment as UniversalSegment
 from .params import AlconnaExecResult as AlconnaExecResult
+from .matcher import command_from_json as command_from_json
+from .matcher import command_from_yaml as command_from_yaml
 from .params import AlconnaDuplication as AlconnaDuplication
 from .uniseg import apply_media_to_url as apply_media_to_url
 from .uniseg import patch_matcher_send as patch_matcher_send
+from .matcher import commands_from_json as commands_from_json
+from .matcher import commands_from_yaml as commands_from_yaml
 from .consts import ALCONNA_EXEC_RESULT as ALCONNA_EXEC_RESULT
 from .uniseg import apply_fetch_targets as apply_fetch_targets
 from .uniseg import SupportAdapterModule as SupportAdapterModule

src/nonebot_plugin_alconna/matcher.py

@@ -2,6 +2,7 @@
 
 import random
 import weakref
+from pathlib import Path
 from warnings import warn
 from types import FunctionType
 from contextlib import contextmanager
@@ -27,6 +28,7 @@
 from arclet.alconna.tools import AlconnaFormat, AlconnaString
 from nonebot.plugin.on import store_matcher, get_matcher_source
 from arclet.alconna.tools.construct import FuncMounter, MountConfig
+from nonebot.compat import type_validate_json, type_validate_python
 from arclet.alconna import Arg, Args, Alconna, ShortcutArgs, command_manager
 from nonebot.exception import PausedException, FinishedException, RejectedException
 from nonebot.internal.adapter import Bot, Event, Message, MessageSegment, MessageTemplate
@@ -37,8 +39,8 @@
 from .rule import alconna
 from .typings import MReturn
 from .util import annotation
-from .model import CompConfig
 from .pattern import patterns
+from .model import CompConfig, CommandModel
 from .uniseg import Text, Segment, UniMessage
 from .uniseg.fallback import FallbackStrategy
 from .uniseg.template import UniMessageTemplate
@@ -1147,8 +1149,115 @@ async def handle_actions(results: AlcExecResult):
 
         return matcher
 
+    @classmethod
+    def from_model(cls, model: CommandModel):
+        """从 `CommandModel` 生成 `Command` 对象"""
+        cmd = cls(model.command, model.help)
+        if model.usage:
+            cmd.usage(model.usage)
+        if model.examples:
+            cmd.example("\n".join(model.examples))
+        if model.author:
+            cmd.meta.author = model.author
+        if model.namespace:
+            cmd.namespace(model.namespace)
+        cmd.config(
+            model.fuzzy_match,
+            model.fuzzy_threshold,
+            model.raise_exception,
+            model.hide,
+            model.hide_shortcut,
+            model.keep_crlf,
+            model.compact,
+            model.strict,
+            model.context_style,
+            model.extra,
+        )
+        for opt in model.options:
+            cmd.option(opt.name, opt.opt, opt.default)
+        for sub in model.subcommands:
+            cmd.subcommand(sub.name, sub.default)
+        for alias in model.aliases:
+            cmd.alias(alias)
+        for short in model.shortcuts:
+            cmd.shortcut(
+                short.key,
+                command=short.command,
+                arguments=short.args,
+                fuzzy=short.fuzzy,
+                prefix=short.prefix,
+                humanized=short.humanized,
+            )
+        return cmd
+
 
 @run_postprocessor
 @annotation(matcher=AlconnaMatcher)
 def _exit_executor(matcher: AlconnaMatcher):
     matcher.executor.clear()
+
+
+def command_from_json(file: str | Path) -> Command:
+    """从 JSON 文件中加载 Command 对象"""
+    path = Path(file)
+    if not path.exists():
+        raise FileNotFoundError(path)
+    with path.open("r", encoding="utf-8") as f:
+        model = type_validate_json(CommandModel, f.read())
+    return Command.from_model(model)
+
+
+def command_from_yaml(file: str | Path) -> Command:
+    """从 YAML 文件中加载 Command 对象
+
+    使用该函数前请确保已安装 `pyyaml`
+    """
+    try:
+        from yaml import safe_load
+    except ImportError:
+        raise ImportError("Please install pyyaml first")
+    path = Path(file)
+    if not path.exists():
+        raise FileNotFoundError(path)
+    with path.open("r", encoding="utf-8") as f:
+        data = safe_load(f)
+    model = type_validate_python(CommandModel, data)
+    return Command.from_model(model)
+
+
+def commands_from_json(file: str | Path) -> dict[str, Command]:
+    """从单个 JSON 文件，或 JSON 文件目录中加载 Command 对象"""
+    path = Path(file)
+    if not path.exists():
+        raise FileNotFoundError(path)
+    if path.is_dir():
+        return {(cmd := command_from_json(fl)).buffer["command"]: cmd for fl in path.iterdir()}
+    with path.open("r", encoding="utf-8") as f:
+        models = type_validate_json(list[CommandModel], f.read())
+    return {model.command: Command.from_model(model) for model in models}
+
+
+def commands_from_yaml(file: str | Path) -> dict[str, Command]:
+    """从单个 YAML 文件，或 YAML 文件目录中加载 Command 对象
+
+    使用该函数前请确保已安装 `pyyaml`
+
+    在单个 YAML 文件下，若数据为列表，则直接解析为 CommandModel 列表；
+        若数据为字典，则使用 values 解析为 CommandModel 列表
+    """
+    try:
+        from yaml import safe_load
+    except ImportError:
+        raise ImportError("Please install pyyaml first")
+    path = Path(file)
+    if not path.exists():
+        raise FileNotFoundError(path)
+    if path.is_dir():
+        return {(cmd := command_from_yaml(fl)).buffer["command"]: cmd for fl in path.iterdir()}
+    with path.open("r", encoding="utf-8") as f:
+        data = safe_load(f)
+    if isinstance(data, list):
+        models = type_validate_python(list[CommandModel], data)
+    else:
+        models = type_validate_python(list[CommandModel], list(data.values()))
+    return {model.command: Command.from_model(model) for model in models}

src/nonebot_plugin_alconna/model.py

@@ -1,5 +1,5 @@
 from typing_extensions import NotRequired
-from typing import Union, Generic, Literal, TypeVar, Optional, TypedDict
+from typing import Any, Union, Generic, Literal, TypeVar, Optional, TypedDict
 
 from pydantic import Field, BaseModel
 from arclet.alconna import Empty, Alconna, Arparma
@@ -83,3 +83,48 @@ class CompConfig(TypedDict):
     hides: NotRequired[set[Literal["tab", "enter", "exit"]]]
     disables: NotRequired[set[Literal["tab", "enter", "exit"]]]
     lite: NotRequired[bool]
+
+
+class SubcommandModel(BaseModel):
+    name: str
+    default: Any = None
+
+
+class OptionModel(BaseModel):
+    name: str
+    opt: Optional[str] = None
+    default: Any = None
+
+
+class ShortcutModel(BaseModel):
+    key: str
+    command: Optional[str] = None
+    args: Optional[list[str]] = None
+    fuzzy: bool = True
+    prefix: bool = False
+    humanized: Optional[str] = None
+
+
+class CommandModel(BaseModel):
+    command: str
+    help: Optional[str] = None
+    usage: Optional[str] = None
+    examples: Optional[list[str]] = None
+    author: Optional[str] = None
+    fuzzy_match: bool = False
+    fuzzy_threshold: float = 0.6
+    raise_exception: bool = False
+    hide: bool = False
+    hide_shortcut: bool = False
+    keep_crlf: bool = False
+    compact: bool = False
+    strict: bool = True
+    context_style: Optional[Literal["bracket", "parentheses"]] = None
+    extra: Optional[dict[str, Any]] = None
+
+    namespace: Optional[str] = None
+    aliases: set[str] = Field(default_factory=set)
+
+    options: list[OptionModel] = Field(default_factory=list)
+    subcommands: list[SubcommandModel] = Field(default_factory=list)
+    shortcuts: list[ShortcutModel] = Field(default_factory=list)

src/nonebot_plugin_alconna/uniseg/segment.py

@@ -7,10 +7,10 @@
 import contextlib
 from io import BytesIO
 from pathlib import Path
-from functools import reduce, lru_cache
 from datetime import datetime
 from urllib.parse import urlparse
 from typing_extensions import Self
+from functools import reduce, lru_cache
 from collections.abc import Iterable, Awaitable
 from dataclasses import InitVar, field, asdict, fields, dataclass
 from typing import TYPE_CHECKING, Any, Union, Literal, TypeVar, Callable, ClassVar, Optional, Protocol, overload

tests/test_koishi_command.py

@@ -1,3 +1,5 @@
+from pathlib import Path
+
 import pytest
 from nonebug import App
 from nonebot import get_adapter
@@ -6,10 +8,13 @@
 
 from tests.fake import fake_group_message_event_v11
 
+FILE = Path(__file__).parent / "test_koishi_command.yml"
+FILE1 = Path(__file__).parent / "test_koishi_command1.yml"
+
 
 @pytest.mark.asyncio()
 async def test_command(app: App):
-    from nonebot_plugin_alconna import Command
+    from nonebot_plugin_alconna import Command, command_from_yaml, commands_from_yaml
 
     book = (
         Command("book", "测试")
@@ -30,3 +35,33 @@ async def _(arp: Arparma):
         event = fake_group_message_event_v11(message=Message("book --anonymous"), user_id=123)
         ctx.receive_event(bot, event)
         ctx.should_call_send(event, "{'writer': (value=Ellipsis args={'id': 0})}")
+
+    book1 = command_from_yaml(FILE).build()
+
+    @book1.handle()
+    async def _(arp: Arparma):
+        await book1.send(str(arp.options))
+
+    async with app.test_matcher(book1) as ctx:  # type: ignore
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter)
+        event = fake_group_message_event_v11(message=Message("book1 --anonymous"), user_id=123)
+        ctx.receive_event(bot, event)
+        ctx.should_call_send(event, "{'writer': (value=Ellipsis args={'id': 1})}")
+
+    books = [cmd.build() for cmd in commands_from_yaml(FILE1).values()]
+    for matcher in books:
+
+        @matcher.handle()
+        async def _(arp: Arparma):
+            await matcher.send(str(arp.options))
+
+    async with app.test_matcher(books) as ctx:  # type: ignore
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter)
+        event = fake_group_message_event_v11(message=Message("book2 --anonymous"), user_id=123)
+        ctx.receive_event(bot, event)
+        ctx.should_call_send(event, "{'writer': (value=Ellipsis args={'id': 2})}")
+        event1 = fake_group_message_event_v11(message=Message("book3 --anonymous"), user_id=123)
+        ctx.receive_event(bot, event1)
+        ctx.should_call_send(event1, "{'writer': (value=Ellipsis args={'id': 3})}")

tests/test_koishi_command.yml

@@ -0,0 +1,13 @@
+command: book1
+help: 测试
+options:
+  - name: writer
+    opt: "-w <id:int>"
+  - name: writer
+    opt: "--anonymous"
+    default:
+      id: 1
+usage: book [-w <id:int> | --anonymous]
+shortcuts:
+  - key: 测试
+    args: ["--anonymous"]

tests/test_koishi_command1.yml

@@ -0,0 +1,29 @@
+book2:
+  command: book2
+  help: 测试
+  options:
+    - name: writer
+      opt: "-w <id:int>"
+    - name: writer
+      opt: "--anonymous"
+      default:
+        id: 2
+  usage: book [-w <id:int> | --anonymous]
+  shortcuts:
+    - key: 测试
+      args: ["--anonymous"]
+
+book3:
+  command: book3
+  help: 测试
+  options:
+    - name: writer
+      opt: "-w <id:int>"
+    - name: writer
+      opt: "--anonymous"
+      default:
+        id: 3
+  usage: book [-w <id:int> | --anonymous]
+  shortcuts:
+    - key: 测试
+      args: ["--anonymous"]