Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 42 additions & 1 deletion astrbot/core/star/filter/command.py
Original file line number Diff line number Diff line change
@@ -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
Expand All @@ -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):
"""标记指令完成其他参数接收后的所有剩余文本。"""

Expand All @@ -37,13 +56,15 @@ 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()
self._original_command_name = command_name
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] = []
Expand Down Expand Up @@ -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],
Expand Down Expand Up @@ -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(" ")
# 去除空字符串
Expand Down
35 changes: 33 additions & 2 deletions astrbot/core/star/filter/command_group.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Comment thread
sourcery-ai[bot] marked this conversation as resolved.
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:
Expand All @@ -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
Expand All @@ -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
1 change: 1 addition & 0 deletions astrbot/core/star/register/star_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
11 changes: 10 additions & 1 deletion tests/unit/test_command_group_filter.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,14 +35,23 @@ 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"],
)
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, {})
Expand Down
164 changes: 164 additions & 0 deletions tests/unit/test_group_custom_filter.py
Original file line number Diff line number Diff line change
@@ -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)
Loading