feat: 一个指令只允许一个handler处理, 匹配最长指令序列, 同时兼容前缀指令, 增加指令冲突检查 #2746
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
There was a problem hiding this comment.
你好 - 我已经审查了你的更改，它们看起来很棒！
AI 代理的提示
请解决此代码审查中的评论：
## 个人评论
### 评论 1
<location> `astrbot/core/pipeline/waking_check/stage.py:179` </location>
<code_context>
+ for filter_obj in handler.event_filters:
</code_context>
<issue_to_address>
过滤器检查中的异常处理可能会掩盖底层问题。
记录带有堆栈跟踪的异常将有助于识别过滤器实现中的问题。对于关键错误，请考虑重新抛出而不是仅仅通知用户。
</issue_to_address>
### 评论 2
<location> `astrbot/core/pipeline/waking_check/stage.py:104` </location>
<code_context>
+
+ return None
+
+ def _match_command(self, command_filter, message_str, event):
+ """匹配指令
+
</code_context>
<issue_to_address>
考虑提取一个辅助函数来生成所有完整的命令字符串，并重构命令匹配和冲突检测以使用它，从而减少重复。
这里有两个小的重构，可以在不改变行为的情况下，折叠 `_match_command` 和 `_detect_command_conflicts` 中嵌套的循环和重复的逻辑：
1) 提取一个辅助函数，从 `CommandFilter` 构建所有“完整命令”：
```python
def _full_command_list(command_filter):
parts = []
for base in [command_filter.command_name] + list(command_filter.alias):
for parent in command_filter.parent_command_names:
parts.append(f"{parent} {base}" if parent else base)
return parts
```
2) 使用单个正则表达式传递来简化 `_match_command`，而不是三个 `if` 分支：
```python
import re
def _match_command(self, command_filter, message_str, event):
if not command_filter.custom_filter_ok(event, self.ctx.astrbot_config):
return None
commands = _full_command_list(command_filter)
# build: ^(?:cmd1|cmd2)(?:\s+(.*))?$
pat = re.compile(rf"^(?:{'|'.join(map(re.escape, commands))})(?:\s+(.*))?$")
m = pat.match(message_str)
if not m:
return None
raw_args = m.group(1) or ""
params = raw_args.split() if raw_args.strip() else []
try:
parsed = command_filter.validate_and_convert_params(params, command_filter.handler_params)
return (len(m.group(0).split()[0]), parsed)
except ValueError:
return None
```
3) 在 `_detect_command_conflicts` 中重用 `_full_command_list`：
```python
def _detect_command_conflicts(self, command_handlers):
command_map = defaultdict(list)
for handler, cf in command_handlers:
plugin = star_map.get(handler.handler_module_path, {}).name or "unknown"
for full_cmd in _full_command_list(cf):
command_map[full_cmd].append((handler, plugin))
for cmd, lst in command_map.items():
if len(lst) > 1:
logger.warning(f"冲突: '{cmd}' -> " +
", ".join(f"{p}.{h.handler_full_name}" for h, p in lst))
```
这些更改
- 移除了 `_match_command` 中的三个独立分支
- 消除了别名/父级上的重复循环
- 在匹配和冲突检测之间共享命令生成逻辑
—同时保持了相同的行为。
</issue_to_address>请帮助我更有用！请点击每个评论上的 👍 或 👎，我将使用反馈来改进你的评论。
Original comment in English
Hey there - I've reviewed your changes and they look great!
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location> `astrbot/core/pipeline/waking_check/stage.py:179` </location>
<code_context>
+ for filter_obj in handler.event_filters:
</code_context>
<issue_to_address>
Exception handling in filter checks may mask underlying issues.
Logging exceptions with traceback will help identify issues in filter implementations. For critical errors, consider re-raising instead of only notifying the user.
</issue_to_address>
### Comment 2
<location> `astrbot/core/pipeline/waking_check/stage.py:104` </location>
<code_context>
+
+ return None
+
+ def _match_command(self, command_filter, message_str, event):
+ """匹配指令
+
</code_context>
<issue_to_address>
Consider extracting a helper to generate all full command strings and refactoring command matching and conflict detection to use it for reduced duplication.
Here are two small refactorings that can collapse the nested loops and repeated logic in both `_match_command` and `_detect_command_conflicts` without changing behavior:
1) Extract a helper to build all “full commands” from a `CommandFilter`:
```python
def _full_command_list(command_filter):
parts = []
for base in [command_filter.command_name] + list(command_filter.alias):
for parent in command_filter.parent_command_names:
parts.append(f"{parent} {base}" if parent else base)
return parts
```
2) Simplify `_match_command` with a single regex pass instead of three `if`-branches:
```python
import re
def _match_command(self, command_filter, message_str, event):
if not command_filter.custom_filter_ok(event, self.ctx.astrbot_config):
return None
commands = _full_command_list(command_filter)
# build: ^(?:cmd1|cmd2)(?:\s+(.*))?$
pat = re.compile(rf"^(?:{'|'.join(map(re.escape, commands))})(?:\s+(.*))?$")
m = pat.match(message_str)
if not m:
return None
raw_args = m.group(1) or ""
params = raw_args.split() if raw_args.strip() else []
try:
parsed = command_filter.validate_and_convert_params(params, command_filter.handler_params)
return (len(m.group(0).split()[0]), parsed)
except ValueError:
return None
```
3) Reuse `_full_command_list` in `_detect_command_conflicts`:
```python
def _detect_command_conflicts(self, command_handlers):
command_map = defaultdict(list)
for handler, cf in command_handlers:
plugin = star_map.get(handler.handler_module_path, {}).name or "unknown"
for full_cmd in _full_command_list(cf):
command_map[full_cmd].append((handler, plugin))
for cmd, lst in command_map.items():
if len(lst) > 1:
logger.warning(f"冲突: '{cmd}' -> " +
", ".join(f"{p}.{h.handler_full_name}" for h, p in lst))
```
These two changes
- Remove the three separate branches in `_match_command`
- Eliminate duplicate loops over aliases/parents
- Share command-generation logic between matching and conflict detection
—while keeping identical behavior.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Comment on lines
+179
to
+188
| for filter_obj in handler.event_filters: | ||
| try: | ||
| if isinstance(filter_obj, PermissionTypeFilter): | ||
| if not filter_obj.filter(event, self.ctx.astrbot_config): | ||
| permission_not_pass = True | ||
| permission_filter_raise_error = filter_obj.raise_error | ||
| elif hasattr(filter_obj, "command_name"): | ||
| # 不需要再检查 command filter | ||
| continue | ||
| else: |
There was a problem hiding this comment.
suggestion (bug_risk): 过滤器检查中的异常处理可能会掩盖底层问题。
记录带有堆栈跟踪的异常将有助于识别过滤器实现中的问题。对于关键错误，请考虑重新抛出而不是仅仅通知用户。
Original comment in English
suggestion (bug_risk): Exception handling in filter checks may mask underlying issues.
Logging exceptions with traceback will help identify issues in filter implementations. For critical errors, consider re-raising instead of only notifying the user.
|
|
||
| return None | ||
|
|
||
| def _match_command(self, command_filter, message_str, event): |
There was a problem hiding this comment.
issue (complexity): 考虑提取一个辅助函数来生成所有完整的命令字符串，并重构命令匹配和冲突检测以使用它，从而减少重复。
这里有两个小的重构，可以在不改变行为的情况下，折叠 _match_command 和 _detect_command_conflicts 中嵌套的循环和重复的逻辑：
- 提取一个辅助函数，从
CommandFilter构建所有“完整命令”：
def _full_command_list(command_filter):
parts = []
for base in [command_filter.command_name] + list(command_filter.alias):
for parent in command_filter.parent_command_names:
parts.append(f"{parent} {base}" if parent else base)
return parts- 使用单个正则表达式传递来简化
_match_command，而不是三个if分支：
import re
def _match_command(self, command_filter, message_str, event):
if not command_filter.custom_filter_ok(event, self.ctx.astrbot_config):
return None
commands = _full_command_list(command_filter)
# build: ^(?:cmd1|cmd2)(?:\s+(.*))?$
pat = re.compile(rf"^(?:{'|'.join(map(re.escape, commands))})(?:\s+(.*))?$")
m = pat.match(message_str)
if not m:
return None
raw_args = m.group(1) or ""
params = raw_args.split() if raw_args.strip() else []
try:
parsed = command_filter.validate_and_convert_params(params, command_filter.handler_params)
return (len(m.group(0).split()[0]), parsed)
except ValueError:
return None- 在
_detect_command_conflicts中重用_full_command_list：
def _detect_command_conflicts(self, command_handlers):
command_map = defaultdict(list)
for handler, cf in command_handlers:
plugin = star_map.get(handler.handler_module_path, {}).name or "unknown"
for full_cmd in _full_command_list(cf):
command_map[full_cmd].append((handler, plugin))
for cmd, lst in command_map.items():
if len(lst) > 1:
logger.warning(f"冲突: '{cmd}' -> " +
", ".join(f"{p}.{h.handler_full_name}" for h, p in lst))这些更改
- 移除了
_match_command中的三个独立分支 - 消除了别名/父级上的重复循环
- 在匹配和冲突检测之间共享命令生成逻辑
—同时保持了相同的行为。
Original comment in English
issue (complexity): Consider extracting a helper to generate all full command strings and refactoring command matching and conflict detection to use it for reduced duplication.
Here are two small refactorings that can collapse the nested loops and repeated logic in both _match_command and _detect_command_conflicts without changing behavior:
- Extract a helper to build all “full commands” from a
CommandFilter:
def _full_command_list(command_filter):
parts = []
for base in [command_filter.command_name] + list(command_filter.alias):
for parent in command_filter.parent_command_names:
parts.append(f"{parent} {base}" if parent else base)
return parts- Simplify
_match_commandwith a single regex pass instead of threeif-branches:
import re
def _match_command(self, command_filter, message_str, event):
if not command_filter.custom_filter_ok(event, self.ctx.astrbot_config):
return None
commands = _full_command_list(command_filter)
# build: ^(?:cmd1|cmd2)(?:\s+(.*))?$
pat = re.compile(rf"^(?:{'|'.join(map(re.escape, commands))})(?:\s+(.*))?$")
m = pat.match(message_str)
if not m:
return None
raw_args = m.group(1) or ""
params = raw_args.split() if raw_args.strip() else []
try:
parsed = command_filter.validate_and_convert_params(params, command_filter.handler_params)
return (len(m.group(0).split()[0]), parsed)
except ValueError:
return None- Reuse
_full_command_listin_detect_command_conflicts:
def _detect_command_conflicts(self, command_handlers):
command_map = defaultdict(list)
for handler, cf in command_handlers:
plugin = star_map.get(handler.handler_module_path, {}).name or "unknown"
for full_cmd in _full_command_list(cf):
command_map[full_cmd].append((handler, plugin))
for cmd, lst in command_map.items():
if len(lst) > 1:
logger.warning(f"冲突: '{cmd}' -> " +
", ".join(f"{p}.{h.handler_full_name}" for h, p in lst))These two changes
- Remove the three separate branches in
_match_command - Eliminate duplicate loops over aliases/parents
- Share command-generation logic between matching and conflict detection
—while keeping identical behavior.
Comment on lines
+57
to
+60
| for f in handler.event_filters: | ||
| if hasattr(f, "command_name"): | ||
| return f | ||
| return None |
There was a problem hiding this comment.
suggestion (code-quality): 使用内置函数 next 而不是 for 循环 (use-next)
Suggested change
| for f in handler.event_filters: | |
| if hasattr(f, "command_name"): | |
| return f | |
| return None | |
| return next( | |
| (f for f in handler.event_filters if hasattr(f, "command_name")), None | |
| ) |
Original comment in English
suggestion (code-quality): Use the built-in function next instead of a for-loop (use-next)
Suggested change
| for f in handler.event_filters: | |
| if hasattr(f, "command_name"): | |
| return f | |
| return None | |
| return next( | |
| (f for f in handler.event_filters if hasattr(f, "command_name")), None | |
| ) |
Comment on lines
+82
to
+83
| match_result = self._match_command(command_filter, message_str, event) | ||
| if match_result: |
There was a problem hiding this comment.
suggestion (code-quality): 使用命名表达式简化赋值和条件 (use-named-expression)
Suggested change
| match_result = self._match_command(command_filter, message_str, event) | |
| if match_result: | |
| if match_result := self._match_command( | |
| command_filter, message_str, event | |
| ): |
Original comment in English
suggestion (code-quality): Use named expression to simplify assignment and conditional (use-named-expression)
Suggested change
| match_result = self._match_command(command_filter, message_str, event) | |
| if match_result: | |
| if match_result := self._match_command( | |
| command_filter, message_str, event | |
| ): |
Comment on lines
+228
to
+229
| star_metadata = star_map.get(handler.handler_module_path) | ||
| if star_metadata: |
There was a problem hiding this comment.
suggestion (code-quality): 使用命名表达式简化赋值和条件 (use-named-expression)
Suggested change
| star_metadata = star_map.get(handler.handler_module_path) | |
| if star_metadata: | |
| if star_metadata := star_map.get(handler.handler_module_path): |
Original comment in English
suggestion (code-quality): Use named expression to simplify assignment and conditional (use-named-expression)
Suggested change
| star_metadata = star_map.get(handler.handler_module_path) | |
| if star_metadata: | |
| if star_metadata := star_map.get(handler.handler_module_path): |
Comment on lines
+346
to
+348
| # 检查是否为指令 | ||
| command_filter = self._find_command_filter(handler) | ||
| if command_filter: |
There was a problem hiding this comment.
suggestion (code-quality): 我们发现了这些问题：
- 使用命名表达式简化赋值和条件 (
use-named-expression) - 在 WakingCheckStage.process 中发现低代码质量 - 15% (
low-code-quality)
Suggested change
| # 检查是否为指令 | |
| command_filter = self._find_command_filter(handler) | |
| if command_filter: | |
| if command_filter := self._find_command_filter(handler): |
解释
此函数的质量得分低于 25% 的质量阈值。
此得分是方法长度、认知复杂度和工作内存的组合。
您如何解决此问题？
重构此函数以使其更短、更易读可能是有益的。
- 通过将功能块提取到它们自己的函数中来减少函数长度。这是您可以做的最重要的事情——理想情况下，一个函数应该少于 10 行。
- 减少嵌套，或许可以通过引入守卫子句来尽早返回。
- 确保变量作用域紧密，以便使用相关概念的代码在函数中紧密地放在一起，而不是分散开来。
Original comment in English
suggestion (code-quality): We've found these issues:
- Use named expression to simplify assignment and conditional (
use-named-expression) - Low code quality found in WakingCheckStage.process - 15% (
low-code-quality)
Suggested change
| # 检查是否为指令 | |
| command_filter = self._find_command_filter(handler) | |
| if command_filter: | |
| if command_filter := self._find_command_filter(handler): |
Explanation
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines. - Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.
Soulter
changed the title
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Motivation
解决前缀匹配导致的同时激活两个command handler的情况
要求不能注册同名command 并提供冲突日志
同时兼容alias作为同名时解决方案
Modifications
由于单个command filter不应该获取全部commander来检查最长序列, 也不应该处理冲突
同时并发的command filter执行冲突检测难以协调, 故而在wakingstage单独检查command匹配情况
原有的filter逻辑标记为向后兼容, 而用于注册command名和别名的职能仍然不变
兼容别名, 兼容前缀匹配, 兼容前缀匹配时, 和参数之间没有空格的情况(只有一个参数)
Check
requirements.txt和pyproject.toml文件相应位置。测试截图
Sourcery 总结
统一唤醒阶段的命令匹配，以选择单个最长匹配的处理程序，检测命令冲突，并支持别名和灵活的前缀/参数模式。
新功能：
错误修复：
增强功能：
Original summary in English
Summary by Sourcery
Unify command matching in the waking stage to select a single longest-match handler, detect command conflicts, and support aliases and flexible prefix/parameter patterns
New Features:
Bug Fixes:
Enhancements: