diff --git a/astrbot/core/star/filter/command.py b/astrbot/core/star/filter/command.py index abf339fda7f..8c1efd387d0 100755 --- a/astrbot/core/star/filter/command.py +++ b/astrbot/core/star/filter/command.py @@ -1,8 +1,10 @@ +from __future__ import annotations + import inspect import re import types import typing -from typing import Any +from typing import Any, Protocol from astrbot.core.config import AstrBotConfig from astrbot.core.platform.astr_message_event import AstrMessageEvent @@ -12,6 +14,23 @@ from .custom_filter import CustomFilter +class CommandGroupLike(Protocol): + """Structural type of ``CommandGroupFilter``. + + ``command_group`` imports ``CommandFilter`` at runtime, so it cannot be + imported back here directly. A protocol keeps the constructor annotations + resolvable for ``typing.get_type_hints`` without a circular import. + """ + + parent_group: CommandGroupLike | None + + def custom_filter_ok(self, event: AstrMessageEvent, cfg: AstrBotConfig) -> bool: ... + + def ancestor_custom_filter_ok( + self, event: AstrMessageEvent, cfg: AstrBotConfig + ) -> bool: ... + + class GreedyStr(str): """标记指令完成其他参数接收后的所有剩余文本。""" @@ -37,6 +56,7 @@ def __init__( alias: set | None = None, handler_md: StarHandlerMetadata | None = None, parent_command_names: list[str] | None = None, + parent_group: CommandGroupLike | None = None, ) -> None: self.command_name = command_name self.alias = alias if alias else set() @@ -44,6 +64,7 @@ def __init__( self.parent_command_names = ( parent_command_names if parent_command_names is not None else [""] ) + self.parent_group = parent_group if handler_md: self.init_handler_md(handler_md) self.custom_filter_list: list[CustomFilter] = [] @@ -90,6 +111,23 @@ def custom_filter_ok(self, event: AstrMessageEvent, cfg: AstrBotConfig) -> bool: return False return True + def parent_custom_filter_ok( + self, event: AstrMessageEvent, cfg: AstrBotConfig + ) -> bool: + """Check the custom filters of every ancestor command group. + + A filter registered on a command group is meant to constrain the whole + group, including its sub-commands. Sub-command handlers are evaluated + independently from the group handler, so the ancestor filters have to + be checked here as well. + """ + parent_group = self.parent_group + if parent_group is None: + return True + return parent_group.custom_filter_ok( + event, cfg + ) and parent_group.ancestor_custom_filter_ok(event, cfg) + def validate_and_convert_params( self, params: list[Any], @@ -205,6 +243,9 @@ def filter(self, event: AstrMessageEvent, cfg: AstrBotConfig) -> bool: if not ok: return False + if not self.parent_custom_filter_ok(event, cfg): + return False + # 分割为列表 ls = message_str.split(" ") # 去除空字符串 diff --git a/astrbot/core/star/filter/command_group.py b/astrbot/core/star/filter/command_group.py index 2e0641362e7..63ae650f795 100755 --- a/astrbot/core/star/filter/command_group.py +++ b/astrbot/core/star/filter/command_group.py @@ -105,9 +105,31 @@ def print_cmd_tree( return "".join(parts) def custom_filter_ok(self, event: AstrMessageEvent, cfg: AstrBotConfig) -> bool: + # The same group filter may be evaluated several times for one message + # (by the group handler and by each matching sub-command), so cache the + # result on the event to keep side effects to a single call. + cache_key = f"custom_filter_ok:{id(self)}" + cached = event.get_extra(cache_key) + if cached is not None: + return cached + + result = True for custom_filter in self.custom_filter_list: if not custom_filter.filter(event, cfg): + result = False + break + event.set_extra(cache_key, result) + return result + + def ancestor_custom_filter_ok( + self, event: AstrMessageEvent, cfg: AstrBotConfig + ) -> bool: + """Check the custom filters of every ancestor command group.""" + parent_group = self.parent_group + while parent_group is not None: + if not parent_group.custom_filter_ok(event, cfg): return False + parent_group = parent_group.parent_group return True def startswith(self, message_str: str) -> bool: @@ -127,10 +149,19 @@ def filter(self, event: AstrMessageEvent, cfg: AstrBotConfig) -> bool: if not event.is_at_or_wake_command: return False - # 判断当前指令组的自定义过滤器 + # Only evaluate the custom filters for messages that target this group, + # so a filter with side effects (for example sending a hint) is not + # triggered by unrelated messages. + if not self.startswith(event.message_str): + return False + + # 判断当前指令组以及上层指令组的自定义过滤器 if not self.custom_filter_ok(event, cfg): return False + if not self.ancestor_custom_filter_ok(event, cfg): + return False + if self.equals(event.message_str.strip()): tree = ( self.group_name @@ -141,4 +172,4 @@ def filter(self, event: AstrMessageEvent, cfg: AstrBotConfig) -> bool: f"参数不足。{self.group_name} 指令组下有如下指令，请参考：\n" + tree, ) - return self.startswith(event.message_str) + return True diff --git a/astrbot/core/star/register/star_handler.py b/astrbot/core/star/register/star_handler.py index 45d91bf15e3..97d5e805899 100644 --- a/astrbot/core/star/register/star_handler.py +++ b/astrbot/core/star/register/star_handler.py @@ -92,6 +92,7 @@ def register_command( alias, None, parent_command_names=parent_command_names, + parent_group=command_name.parent_group, ) command_name.parent_group.add_sub_command_filter(new_command) else: diff --git a/tests/unit/test_command_group_filter.py b/tests/unit/test_command_group_filter.py index 01d3cc886ef..70cead4cfe4 100644 --- a/tests/unit/test_command_group_filter.py +++ b/tests/unit/test_command_group_filter.py @@ -35,6 +35,15 @@ def test_nested_group_matches_whole_words_only(): assert not child.startswith("tool configure") +def make_event(message: str) -> SimpleNamespace: + """Build a minimal event double, including the extras API the filters use.""" + event = SimpleNamespace(is_at_or_wake_command=True, message_str=message) + event._extras = {} + event.get_extra = lambda key=None, default=None: event._extras.get(key, default) + event.set_extra = lambda key, value: event._extras.__setitem__(key, value) + return event + + @pytest.mark.parametrize( "message", ["tool config", "tool config", "tool\tconfig", "tool\nconfig"], @@ -42,7 +51,7 @@ def test_nested_group_matches_whole_words_only(): def test_nested_group_without_subcommand_shows_help(message): parent = CommandGroupFilter("tool") group = CommandGroupFilter("config", parent_group=parent) - event = SimpleNamespace(is_at_or_wake_command=True, message_str=message) + event = make_event(message) with pytest.raises(ValueError, match="参数不足"): group.filter(event, {}) diff --git a/tests/unit/test_group_custom_filter.py b/tests/unit/test_group_custom_filter.py new file mode 100644 index 00000000000..6deb418a4d0 --- /dev/null +++ b/tests/unit/test_group_custom_filter.py @@ -0,0 +1,164 @@ +"""Tests for the cascade of command-group custom filters to sub-commands.""" + +from typing import Any + +import pytest + + +class FakeEvent: + """Minimal AstrMessageEvent stand-in for filter unit tests.""" + + def __init__(self, message_str: str) -> None: + self.message_str = message_str + self.is_at_or_wake_command = True + self._extras: dict[str, Any] = {} + + def get_message_str(self) -> str: + return self.message_str + + def get_extra(self, key: str | None = None, default: Any = None) -> Any: + if key is None: + return default + return self._extras.get(key, default) + + def set_extra(self, key: str, value: Any) -> None: + self._extras[key] = value + + +def make_custom_filter(allowed: bool, calls: list[int] | None = None): + """Build a CustomFilter that returns ``allowed`` and counts invocations.""" + from astrbot.core.star.filter.custom_filter import CustomFilter + + class _Filter(CustomFilter): + def filter(self, event, cfg) -> bool: + if calls is not None: + calls.append(1) + return allowed + + return _Filter() + + +def make_group(name: str, parent_group=None): + from astrbot.core.star.filter.command_group import CommandGroupFilter + + return CommandGroupFilter(name, parent_group=parent_group) + + +def make_sub_command(name: str, group): + from astrbot.core.star.filter.command import CommandFilter + + sub_command = CommandFilter( + name, + parent_command_names=group.get_complete_command_names(), + parent_group=group, + ) + # Normally filled by init_handler_md() while decorating the handler. + sub_command.handler_params = {} + return sub_command + + +def test_group_filter_blocks_sub_command(): + group = make_group("math") + group.add_custom_filter(make_custom_filter(False)) + sub_command = make_sub_command("add", group) + + assert sub_command.filter(FakeEvent("math add"), {}) is False + + +def test_group_filter_allows_sub_command(): + group = make_group("math") + group.add_custom_filter(make_custom_filter(True)) + sub_command = make_sub_command("add", group) + + assert sub_command.filter(FakeEvent("math add"), {}) is True + + +def test_nested_group_filter_blocks_sub_command(): + outer = make_group("math") + outer.add_custom_filter(make_custom_filter(False)) + inner = make_group("calc", parent_group=outer) + inner.add_custom_filter(make_custom_filter(True)) + sub_command = make_sub_command("add", inner) + + assert sub_command.filter(FakeEvent("math calc add"), {}) is False + + +def test_sub_command_filter_still_applies(): + group = make_group("math") + group.add_custom_filter(make_custom_filter(True)) + sub_command = make_sub_command("add", group) + sub_command.add_custom_filter(make_custom_filter(False)) + + assert sub_command.filter(FakeEvent("math add"), {}) is False + + +def test_group_filter_is_evaluated_once_per_message(): + calls: list[int] = [] + group = make_group("math") + group.add_custom_filter(make_custom_filter(True, calls)) + sub_command = make_sub_command("add", group) + event = FakeEvent("math add") + + assert group.filter(event, {}) is True + assert sub_command.filter(event, {}) is True + assert len(calls) == 1 + + +def test_group_filter_not_checked_for_unrelated_message(): + calls: list[int] = [] + group = make_group("math") + group.add_custom_filter(make_custom_filter(False, calls)) + sub_command = make_sub_command("add", group) + + assert sub_command.filter(FakeEvent("other add"), {}) is False + assert calls == [] + + +def test_group_handler_skips_filters_for_unrelated_message(): + calls: list[int] = [] + group = make_group("math") + group.add_custom_filter(make_custom_filter(True, calls)) + + assert group.filter(FakeEvent("weather info"), {}) is False + assert calls == [] + + +def test_group_handler_shows_help_for_exact_group_name(): + group = make_group("math") + group.add_custom_filter(make_custom_filter(True)) + + with pytest.raises(ValueError, match="参数不足"): + group.filter(FakeEvent("math"), {}) + + +def test_group_handler_hides_help_when_filter_denies(): + group = make_group("math") + group.add_custom_filter(make_custom_filter(False)) + + assert group.filter(FakeEvent("math"), {}) is False + + +def test_register_command_attaches_parent_group(): + from astrbot.core.star.filter.command_group import CommandGroupFilter + from astrbot.core.star.register.star_handler import ( + RegisteringCommandable, + register_command, + ) + from astrbot.core.star.star_handler import star_handlers_registry + + group = CommandGroupFilter("_scope_test_math") + + async def handler(event): + pass + + md = None + try: + register_command(RegisteringCommandable(group), "_scope_test_add")(handler) + md = star_handlers_registry.get_handler_by_full_name( + f"{handler.__module__}_{handler.__name__}" + ) + assert md is not None + assert md.event_filters[0].parent_group is group + finally: + if md is not None: + star_handlers_registry.remove(md)