feat: 新增插件处理器优先级管理API及前端展示 #4716
Open
+258
−2
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
- 新增后端优先级管理API: - GET /api/plugin/priority: 获取所有处理器优先级信息 - POST /api/plugin/priority: 更新处理器优先级 - POST /api/plugin/priority/reset: 重置优先级为原始值 - 通过 shared_preferences 持久化优先级覆盖配置 - 在核心生命周期初始化时加载优先级覆盖 - 为 CommandDescriptor 和指令 API 响应添加 priority 字段 - 在 dashboard CommandTable 组件中添加优先级列展示 - 添加优先级列的 i18n 翻译 (中文/英文)
dosubot
bot
added
the
size:L
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - 我发现了两个问题，并给出了一些总体反馈：
- 在
get_priority中，备用描述字符串'无描述'以及filters_summary的取值（'auto','event_listener'等）是硬编码且未本地化的，这会导致 API 使用方收到混合语言的内容；建议要么保持这些字段与语言无关（例如使用代码/枚举），要么把它们接入现有的 i18n 机制。 get_priority、update_priority和reset_priority中宽泛的except Exception捕获会让调试变得更困难，也可能掩盖编程错误；更好的做法是显式捕获并处理预期的错误类型，让非预期异常继续抛出（或者至少在重新记录日志时加入更多结构化上下文）。
面向 AI Agent 的提示词
Please address the comments from this code review:
## Overall Comments
- In `get_priority`, the fallback description string `'无描述'` and the `filters_summary` values (`'auto'`, `'event_listener'`, etc.) are hardcoded and not localized, which means API consumers will get mixed-language content; consider either keeping these fields language-agnostic (e.g., codes) or wiring them through the existing i18n mechanism.
- The broad `except Exception` blocks in `get_priority`, `update_priority`, and `reset_priority` make debugging harder and can hide programming errors; it would be better to catch and handle expected error types explicitly and let unexpected exceptions surface (or at least re-log them with more structured context).
## Individual Comments
### Comment 1
<location> `astrbot/core/star/command_management.py:292` </location>
<code_context>
is_group=isinstance(filter_ref, CommandGroupFilter),
is_sub_command=is_sub_command,
reserved=plugin_meta.reserved if plugin_meta else False,
+ priority=handler.extras_configs.get("priority", 0),
)
return descriptor
</code_context>
<issue_to_address>
**question:** Consider exposing effective priority (respecting overrides) rather than only the static extras_configs priority.
Since `CommandDescriptor.priority` comes directly from `handler.extras_configs.get("priority", 0)`, it won’t reflect any overrides applied by `StarHandlerRegistry.load_priority_overrides`. If this priority is meant to match actual execution order (e.g., in the command table UI), consider passing the effective priority instead—either via a `StarHandlerRegistry.get_effective_priority(handler)` helper or by injecting the effective value when building the descriptor—so the displayed priority matches the one used for ordering.
</issue_to_address>
### Comment 2
<location> `astrbot/dashboard/routes/plugin.py:421` </location>
<code_context>
+
+ return ", ".join(parts) if parts else "event_listener"
+
+ async def get_priority(self):
+ try:
+ stored = await sp.global_get("handler_priority_overrides", {})
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared storage/priority handling, validation, and filter-summary logic into small helpers to make these endpoints shorter and easier to follow.
You can reduce duplication and make the flow easier to follow by extracting the repeated storage / map handling and the error wrapper into small helpers, and by centralizing some validation. Here are a few focused refactors that keep behavior intact:
### 1. Extract storage + priority map handling
You build `stored` and `priority_map` in all three endpoints. Pull that into helpers:
```python
async def _load_priority_overrides(self) -> tuple[dict, dict[str, int]]:
stored = await sp.global_get("handler_priority_overrides", {})
if not isinstance(stored, dict):
stored = {}
priority_map: dict[str, int] = {}
for handler_full_name, v in stored.items():
if isinstance(v, dict) and "priority" in v:
priority_map[handler_full_name] = int(v.get("priority", 0))
return stored, priority_map
async def _save_priority_overrides(
self,
stored: dict,
) -> dict[str, int]:
await sp.global_put("handler_priority_overrides", stored)
# rebuild priority_map from stored
priority_map = {
k: v.get("priority", 0)
for k, v in stored.items()
if isinstance(v, dict)
}
star_handlers_registry.load_priority_overrides(priority_map)
return priority_map
```
Then your endpoints shrink:
```python
async def get_priority(self):
try:
stored, priority_map = await self._load_priority_overrides()
star_handlers_registry.load_priority_overrides(priority_map)
# existing plugin/handler iteration logic unchanged...
except Exception as e:
logger.error(f"/api/plugin/priority: {traceback.format_exc()}")
return Response().error(str(e)).__dict__
```
```python
async def update_priority(self):
if DEMO_MODE:
return Response().error(
"You are not permitted to do this operation in demo mode"
).__dict__
try:
data = await request.get_json()
updates = data.get("updates", []) if isinstance(data, dict) else []
# validation (see below)...
stored, priority_map = await self._load_priority_overrides()
# apply updates to stored + priority_map...
await self._save_priority_overrides(stored)
return Response().ok(None, "优先级更新成功").__dict__
except Exception as e:
logger.error(f"/api/plugin/priority: {traceback.format_exc()}")
return Response().error(str(e)).__dict__
```
```python
async def reset_priority(self):
if DEMO_MODE:
return Response().error(
"You are not permitted to do this operation in demo mode"
).__dict__
try:
data = await request.get_json()
# validation (see below)...
stored, _ = await self._load_priority_overrides()
if reset_all:
stored = {}
else:
for h in handler_full_names:
if isinstance(h, str):
stored.pop(h, None)
await self._save_priority_overrides(stored)
return Response().ok(None, "重置成功").__dict__
except Exception as e:
logger.error(f"/api/plugin/priority/reset: {traceback.format_exc()}")
return Response().error(str(e)).__dict__
```
This keeps all behavior but makes the main handlers read as “parse/validate → apply → save”.
### 2. Centralize `update_priority` validation
You can pull the inline checks into a helper that returns a normalized, validated list of updates or an error response:
```python
def _validate_priority_updates(self, data: Any) -> tuple[list[dict] | None, dict | None]:
updates = data.get("updates", []) if isinstance(data, dict) else []
if not isinstance(updates, list) or not updates:
return None, Response().error("updates fields must be a non-empty list").__dict__
normalized: list[dict] = []
for item in updates:
if not isinstance(item, dict):
continue
handler_full_name = item.get("handler_full_name")
if not isinstance(handler_full_name, str) or not handler_full_name:
continue
if "priority" not in item:
continue
try:
new_priority = int(item["priority"])
except (TypeError, ValueError):
return None, Response().error(
f"invalid priority for handler: {handler_full_name}"
).__dict__
handler = star_handlers_registry.get_handler_by_full_name(handler_full_name)
if handler is None:
return None, Response().error(
f"handler not found: {handler_full_name}"
).__dict__
normalized.append(
{
"handler": handler,
"handler_full_name": handler_full_name,
"priority": new_priority,
}
)
if not normalized:
return None, Response().error(
"updates fields must contain at least one valid entry"
).__dict__
return normalized, None
```
Use it inside `update_priority`:
```python
async def update_priority(self):
if DEMO_MODE:
return Response().error(
"You are not permitted to do this operation in demo mode"
).__dict__
try:
data = await request.get_json()
updates, error = self._validate_priority_updates(data)
if error:
return error
stored, priority_map = await self._load_priority_overrides()
for item in updates:
handler = item["handler"]
handler_full_name = item["handler_full_name"]
new_priority = item["priority"]
original_priority = handler.extras_configs.get("priority", 0)
stored[handler_full_name] = {
"priority": new_priority,
"original_priority": original_priority,
}
priority_map[handler_full_name] = new_priority
await self._save_priority_overrides(stored)
return Response().ok(None, "优先级更新成功").__dict__
except Exception as e:
logger.error(f"/api/plugin/priority: {traceback.format_exc()}")
return Response().error(str(e)).__dict__
```
This removes most of the scattered inline checks and early returns from the main method.
### 3. Normalize `reset_priority` request parsing
Similarly, you can pull `reset_all` / `handler_full_names` parsing into a helper to avoid nested conditionals:
```python
def _parse_reset_request(self, data: Any) -> tuple[bool, list[str] | None, dict | None]:
if not isinstance(data, dict):
return False, None, Response().error("invalid request body").__dict__
reset_all = data.get("reset_all", False) is True
handler_full_names = data.get("handler_full_names", [])
if reset_all:
return True, [], None
if not isinstance(handler_full_names, list) or not handler_full_names:
return False, None, Response().error(
"handler_full_names must be a non-empty list or reset_all=true"
).__dict__
valid_names = [h for h in handler_full_names if isinstance(h, str)]
if not valid_names:
return False, None, Response().error(
"handler_full_names must contain at least one valid string"
).__dict__
return False, valid_names, None
```
Usage:
```python
async def reset_priority(self):
if DEMO_MODE:
return Response().error(
"You are not permitted to do this operation in demo mode"
).__dict__
try:
data = await request.get_json()
reset_all, handler_full_names, error = self._parse_reset_request(data)
if error:
return error
stored, _ = await self._load_priority_overrides()
if reset_all:
stored = {}
else:
for h in handler_full_names:
stored.pop(h, None)
await self._save_priority_overrides(stored)
return Response().ok(None, "重置成功").__dict__
except Exception as e:
logger.error(f"/api/plugin/priority/reset: {traceback.format_exc()}")
return Response().error(str(e)).__dict__
```
This keeps the logic identical while making it easier to extend with new filter types.
### 4. Optional: Simplify `_build_filters_summary`
You can separate type → text mapping to avoid growing `if/elif` chains:
```python
def _filter_to_summary(self, f: Any) -> str | None:
if isinstance(f, CommandFilter):
base = getattr(f, "parent_command_names", None)
base = base[0] if base else ""
cmd = f"{base} {f.command_name}".strip()
return f"command: {cmd}"
if isinstance(f, CommandGroupFilter):
try:
cmd = f.get_complete_command_names()[0].strip()
except Exception:
cmd = "command_group"
return f"command_group: {cmd}"
if isinstance(f, RegexFilter):
return f"regex: {f.regex_str}"
if isinstance(f, PermissionTypeFilter):
return "permission"
return None
def _build_filters_summary(self, handler: StarHandlerMetadata) -> str:
if handler.event_type != EventType.AdapterMessageEvent:
return "auto"
parts = []
for f in handler.event_filters:
summary = self._filter_to_summary(f)
if summary:
parts.append(summary)
return ", ".join(parts) if parts else "event_listener"
```
This keeps the logic identical while making it easier to extend with new filter types.
</issue_to_address>帮我变得更有用！请对每条评论点 👍 或 👎，我会根据你的反馈改进后续的 Review。
Original comment in English
Hey - I've found 2 issues, and left some high level feedback:
- In
get_priority, the fallback description string'无描述'and thefilters_summaryvalues ('auto','event_listener', etc.) are hardcoded and not localized, which means API consumers will get mixed-language content; consider either keeping these fields language-agnostic (e.g., codes) or wiring them through the existing i18n mechanism. - The broad
except Exceptionblocks inget_priority,update_priority, andreset_prioritymake debugging harder and can hide programming errors; it would be better to catch and handle expected error types explicitly and let unexpected exceptions surface (or at least re-log them with more structured context).
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `get_priority`, the fallback description string `'无描述'` and the `filters_summary` values (`'auto'`, `'event_listener'`, etc.) are hardcoded and not localized, which means API consumers will get mixed-language content; consider either keeping these fields language-agnostic (e.g., codes) or wiring them through the existing i18n mechanism.
- The broad `except Exception` blocks in `get_priority`, `update_priority`, and `reset_priority` make debugging harder and can hide programming errors; it would be better to catch and handle expected error types explicitly and let unexpected exceptions surface (or at least re-log them with more structured context).
## Individual Comments
### Comment 1
<location> `astrbot/core/star/command_management.py:292` </location>
<code_context>
is_group=isinstance(filter_ref, CommandGroupFilter),
is_sub_command=is_sub_command,
reserved=plugin_meta.reserved if plugin_meta else False,
+ priority=handler.extras_configs.get("priority", 0),
)
return descriptor
</code_context>
<issue_to_address>
**question:** Consider exposing effective priority (respecting overrides) rather than only the static extras_configs priority.
Since `CommandDescriptor.priority` comes directly from `handler.extras_configs.get("priority", 0)`, it won’t reflect any overrides applied by `StarHandlerRegistry.load_priority_overrides`. If this priority is meant to match actual execution order (e.g., in the command table UI), consider passing the effective priority instead—either via a `StarHandlerRegistry.get_effective_priority(handler)` helper or by injecting the effective value when building the descriptor—so the displayed priority matches the one used for ordering.
</issue_to_address>
### Comment 2
<location> `astrbot/dashboard/routes/plugin.py:421` </location>
<code_context>
+
+ return ", ".join(parts) if parts else "event_listener"
+
+ async def get_priority(self):
+ try:
+ stored = await sp.global_get("handler_priority_overrides", {})
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared storage/priority handling, validation, and filter-summary logic into small helpers to make these endpoints shorter and easier to follow.
You can reduce duplication and make the flow easier to follow by extracting the repeated storage / map handling and the error wrapper into small helpers, and by centralizing some validation. Here are a few focused refactors that keep behavior intact:
### 1. Extract storage + priority map handling
You build `stored` and `priority_map` in all three endpoints. Pull that into helpers:
```python
async def _load_priority_overrides(self) -> tuple[dict, dict[str, int]]:
stored = await sp.global_get("handler_priority_overrides", {})
if not isinstance(stored, dict):
stored = {}
priority_map: dict[str, int] = {}
for handler_full_name, v in stored.items():
if isinstance(v, dict) and "priority" in v:
priority_map[handler_full_name] = int(v.get("priority", 0))
return stored, priority_map
async def _save_priority_overrides(
self,
stored: dict,
) -> dict[str, int]:
await sp.global_put("handler_priority_overrides", stored)
# rebuild priority_map from stored
priority_map = {
k: v.get("priority", 0)
for k, v in stored.items()
if isinstance(v, dict)
}
star_handlers_registry.load_priority_overrides(priority_map)
return priority_map
```
Then your endpoints shrink:
```python
async def get_priority(self):
try:
stored, priority_map = await self._load_priority_overrides()
star_handlers_registry.load_priority_overrides(priority_map)
# existing plugin/handler iteration logic unchanged...
except Exception as e:
logger.error(f"/api/plugin/priority: {traceback.format_exc()}")
return Response().error(str(e)).__dict__
```
```python
async def update_priority(self):
if DEMO_MODE:
return Response().error(
"You are not permitted to do this operation in demo mode"
).__dict__
try:
data = await request.get_json()
updates = data.get("updates", []) if isinstance(data, dict) else []
# validation (see below)...
stored, priority_map = await self._load_priority_overrides()
# apply updates to stored + priority_map...
await self._save_priority_overrides(stored)
return Response().ok(None, "优先级更新成功").__dict__
except Exception as e:
logger.error(f"/api/plugin/priority: {traceback.format_exc()}")
return Response().error(str(e)).__dict__
```
```python
async def reset_priority(self):
if DEMO_MODE:
return Response().error(
"You are not permitted to do this operation in demo mode"
).__dict__
try:
data = await request.get_json()
# validation (see below)...
stored, _ = await self._load_priority_overrides()
if reset_all:
stored = {}
else:
for h in handler_full_names:
if isinstance(h, str):
stored.pop(h, None)
await self._save_priority_overrides(stored)
return Response().ok(None, "重置成功").__dict__
except Exception as e:
logger.error(f"/api/plugin/priority/reset: {traceback.format_exc()}")
return Response().error(str(e)).__dict__
```
This keeps all behavior but makes the main handlers read as “parse/validate → apply → save”.
### 2. Centralize `update_priority` validation
You can pull the inline checks into a helper that returns a normalized, validated list of updates or an error response:
```python
def _validate_priority_updates(self, data: Any) -> tuple[list[dict] | None, dict | None]:
updates = data.get("updates", []) if isinstance(data, dict) else []
if not isinstance(updates, list) or not updates:
return None, Response().error("updates fields must be a non-empty list").__dict__
normalized: list[dict] = []
for item in updates:
if not isinstance(item, dict):
continue
handler_full_name = item.get("handler_full_name")
if not isinstance(handler_full_name, str) or not handler_full_name:
continue
if "priority" not in item:
continue
try:
new_priority = int(item["priority"])
except (TypeError, ValueError):
return None, Response().error(
f"invalid priority for handler: {handler_full_name}"
).__dict__
handler = star_handlers_registry.get_handler_by_full_name(handler_full_name)
if handler is None:
return None, Response().error(
f"handler not found: {handler_full_name}"
).__dict__
normalized.append(
{
"handler": handler,
"handler_full_name": handler_full_name,
"priority": new_priority,
}
)
if not normalized:
return None, Response().error(
"updates fields must contain at least one valid entry"
).__dict__
return normalized, None
```
Use it inside `update_priority`:
```python
async def update_priority(self):
if DEMO_MODE:
return Response().error(
"You are not permitted to do this operation in demo mode"
).__dict__
try:
data = await request.get_json()
updates, error = self._validate_priority_updates(data)
if error:
return error
stored, priority_map = await self._load_priority_overrides()
for item in updates:
handler = item["handler"]
handler_full_name = item["handler_full_name"]
new_priority = item["priority"]
original_priority = handler.extras_configs.get("priority", 0)
stored[handler_full_name] = {
"priority": new_priority,
"original_priority": original_priority,
}
priority_map[handler_full_name] = new_priority
await self._save_priority_overrides(stored)
return Response().ok(None, "优先级更新成功").__dict__
except Exception as e:
logger.error(f"/api/plugin/priority: {traceback.format_exc()}")
return Response().error(str(e)).__dict__
```
This removes most of the scattered inline checks and early returns from the main method.
### 3. Normalize `reset_priority` request parsing
Similarly, you can pull `reset_all` / `handler_full_names` parsing into a helper to avoid nested conditionals:
```python
def _parse_reset_request(self, data: Any) -> tuple[bool, list[str] | None, dict | None]:
if not isinstance(data, dict):
return False, None, Response().error("invalid request body").__dict__
reset_all = data.get("reset_all", False) is True
handler_full_names = data.get("handler_full_names", [])
if reset_all:
return True, [], None
if not isinstance(handler_full_names, list) or not handler_full_names:
return False, None, Response().error(
"handler_full_names must be a non-empty list or reset_all=true"
).__dict__
valid_names = [h for h in handler_full_names if isinstance(h, str)]
if not valid_names:
return False, None, Response().error(
"handler_full_names must contain at least one valid string"
).__dict__
return False, valid_names, None
```
Usage:
```python
async def reset_priority(self):
if DEMO_MODE:
return Response().error(
"You are not permitted to do this operation in demo mode"
).__dict__
try:
data = await request.get_json()
reset_all, handler_full_names, error = self._parse_reset_request(data)
if error:
return error
stored, _ = await self._load_priority_overrides()
if reset_all:
stored = {}
else:
for h in handler_full_names:
stored.pop(h, None)
await self._save_priority_overrides(stored)
return Response().ok(None, "重置成功").__dict__
except Exception as e:
logger.error(f"/api/plugin/priority/reset: {traceback.format_exc()}")
return Response().error(str(e)).__dict__
```
### 4. Optional: Simplify `_build_filters_summary`
You can separate type → text mapping to avoid growing `if/elif` chains:
```python
def _filter_to_summary(self, f: Any) -> str | None:
if isinstance(f, CommandFilter):
base = getattr(f, "parent_command_names", None)
base = base[0] if base else ""
cmd = f"{base} {f.command_name}".strip()
return f"command: {cmd}"
if isinstance(f, CommandGroupFilter):
try:
cmd = f.get_complete_command_names()[0].strip()
except Exception:
cmd = "command_group"
return f"command_group: {cmd}"
if isinstance(f, RegexFilter):
return f"regex: {f.regex_str}"
if isinstance(f, PermissionTypeFilter):
return "permission"
return None
def _build_filters_summary(self, handler: StarHandlerMetadata) -> str:
if handler.event_type != EventType.AdapterMessageEvent:
return "auto"
parts = []
for f in handler.event_filters:
summary = self._filter_to_summary(f)
if summary:
parts.append(summary)
return ", ".join(parts) if parts else "event_listener"
```
This keeps the logic identical while making it easier to extend with new filter types.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
dosubot
bot
added
area:core
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
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 / 动机
本次改动旨在为后续“插件排序/执行顺序可配置”功能做前置铺垫：先在后端补齐“优先级覆盖”数据层与管理接口，并在前端提供最低限度的优先级展示，便于验证与后续扩展。
具体解决的问题：
Modifications / 改动点
涉及核心文件：
后端：
前端：
This is NOT a breaking change. / 这不是一个破坏性变更。
显示效果
Checklist / 检查清单
requirements.txt和pyproject.toml文件相应位置。/ I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations inrequirements.txtandpyproject.toml.Summary by Sourcery
引入后端插件处理器优先级管理与持久化机制，并通过新的 API 对外暴露，同时在仪表盘 UI 中展示处理器/命令优先级。
新功能：
改进：
Original summary in English
Summary by Sourcery
Introduce backend plugin handler priority management with persistence and expose it via new APIs, and surface handler/command priorities in the dashboard UI.
New Features:
Enhancements: