From bcebd701064fb3bd1984c89bde9c6e4317c42368 Mon Sep 17 00:00:00 2001 From: Alero Date: Tue, 20 Oct 2026 13:46:07 +0800 Subject: [PATCH] fix: apply command group custom filters to sub-commands A custom filter registered on a command group was only consulted by CommandGroupFilter.filter(), i.e. by the group handler, which the waking check excludes from the activated handlers. Sub-command handlers are evaluated independently and their CommandFilter never looked at the ancestor groups, so @group.custom_filter(...) silently did nothing for sub-commands. CommandFilter now keeps a reference to its parent command group and checks the ancestor groups' custom filters once the command name matches. CommandGroupFilter also checks its ancestors so nested groups are gated consistently. Group filter results are cached on the event, keeping a filter call to once per message even though both the group handler and the matching sub-command consult it. CommandGroupFilter now matches the command name before evaluating the custom filters, so a filter with side effects (for example sending a hint) is not triggered by unrelated messages. The parent group annotation uses a small protocol so the constructor type hints stay resolvable at runtime without a circular import. Adds tests/unit/test_group_custom_filter.py covering the cascade, nested groups, the sub-command's own filter, single evaluation, group-name help behavior and the registration wiring. --- astrbot/core/star/filter/command.py | 43 +++++- astrbot/core/star/filter/command_group.py | 35 ++++- astrbot/core/star/register/star_handler.py | 1 + tests/unit/test_command_group_filter.py | 11 +- tests/unit/test_group_custom_filter.py | 164 +++++++++++++++++++++ 5 files changed, 250 insertions(+), 4 deletions(-) create mode 100644 tests/unit/test_group_custom_filter.py 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)