feat: add batch operation functionality for session management #4194
+1,094
−4
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}
Contributor
There was a problem hiding this comment.
Hey - 我这边发现了 3 个问题，并补充了一些整体性反馈：
- 新的
list_all_umos_with_status接口会把所有去重后的ConversationV2.user_id和所有规则一次性加载到内存里，然后再在 Python 里做过滤/分页；建议把过滤和分页逻辑尽量下推到数据库查询中（并避免使用page_size=99999调用_get_umo_rules），这样在大规模部署时才能保持可用性。 batch_update_service和batch_update_provider中，从scope和group_id推导umos的逻辑几乎完全一致；可以考虑抽一个公共的辅助函数来解析目标 UMO 列表，以减少重复代码并降低两个接口在行为上逐渐产生偏差的风险。- 在
batch_update_provider中，if not provider_type or not provider_id这段检查目前会阻止用一个 falsy/空的provider_id去清除单会话级的 provider，而前端有时会发送provider_id: this.batchChatProvider || null；如果是希望支持“重置为默认”的行为，可能需要显式支持将null作为“清除覆盖配置”的信号，而不是直接拒绝它。
面向 AI Agent 的提示词
请根据下面这次代码评审中的评论进行修改：
## 整体点评
- 新的 `list_all_umos_with_status` 接口会把所有去重后的 `ConversationV2.user_id` 和所有规则一次性加载到内存里，然后再在 Python 里做过滤/分页；建议把过滤和分页逻辑尽量下推到数据库查询中（并避免使用 `page_size=99999` 调用 `_get_umo_rules`），这样在大规模部署时才能保持可用性。
- `batch_update_service` 和 `batch_update_provider` 中，从 `scope` 和 `group_id` 推导 `umos` 的逻辑几乎完全一致；可以考虑抽一个公共的辅助函数来解析目标 UMO 列表，以减少重复代码并降低两个接口在行为上逐渐产生偏差的风险。
- 在 `batch_update_provider` 中，`if not provider_type or not provider_id` 这段检查目前会阻止用一个 falsy/空的 `provider_id` 去清除单会话级的 provider，而前端有时会发送 `provider_id: this.batchChatProvider || null`；如果是希望支持“重置为默认”的行为，可能需要显式支持将 `null` 作为“清除覆盖配置”的信号，而不是直接拒绝它。
## 单条评论
### Comment 1
<location> `dashboard/src/views/SessionManagementPage.vue:1` </location>
<code_context>
-<template>
+<template>
<div class="session-management-page">
<v-container fluid class="pa-0">
</code_context>
<issue_to_address>
**issue (bug_risk):** `<template>` 前的 BOM 字符可能会导致工具链或运行时问题。
`<template>` 标签前存在一个 UTF‑8 BOM 字符（`+<template>`）。这可能会导致某些 linter、构建工具或模板解析器出错。请移除这个 BOM，使文件直接以 `<template>` 开头。
</issue_to_address>
### Comment 2
<location> `dashboard/src/views/SessionManagementPage.vue:617-626` </location>
<code_context>
+ batchLlmStatus: null,
+ batchTtsStatus: null,
+ batchChatProvider: null,
+ batchTtsProvider: null,
+ batchUpdating: false,
+
</code_context>
<issue_to_address>
**issue (bug_risk):** 批量 TTS provider 更新在逻辑上已经打通，但缺少 UI 控件，实际上无法被触发。
`batchTtsProvider` 已经接入到 state、`canApplyBatch` 和 `applyBatchChanges` 中，TTS 的 API 请求体也会被构造，但目前没有任何 UI 控件（例如 `<v-select>`）可以设置它。结果就是 TTS 的批量更新分支永远不会被真正使用。请考虑要么增加一个 TTS provider 的选择控件（类似 `batchChatProvider`），要么删除这部分未使用的 TTS 批量状态/逻辑，避免出现死代码。
</issue_to_address>
### Comment 3
<location> `astrbot/dashboard/routes/session_management.py:558` </location>
<code_context>
+ logger.error(f"获取会话状态列表失败: {e!s}")
+ return Response().error(f"获取会话状态列表失败: {e!s}").__dict__
+
+ async def batch_update_service(self):
+ """批量更新多个 UMO 的服务状态 (LLM/TTS/Session)
+
</code_context>
<issue_to_address>
**issue (complexity):** 建议抽取共享的辅助函数，用于按 scope 解析 UMO 列表以及解析 UMO 字符串，从而让新的批量接口和列表接口更精简，并避免重复的内联字符串处理逻辑。
你可以在保持现有行为不变的前提下，通过提取几个聚焦的小工具方法来减少重复和嵌套。两个收益比较大的点是：
1. **Scope → UMO 列表的解析**（`batch_update_service` 和 `batch_update_provider` 中存在重复）
2. **UMO 解析 / 类型判断**（多个位置都有类似的字符串处理逻辑）
### 1. 抽取按 scope 解析 UMO 列表的公共方法
从 `scope`（包括 `custom_group`）解析 `umos` 的逻辑，在 `batch_update_service` 和 `batch_update_provider` 中几乎一模一样。可以集中到一个方法里：
```python
async def _resolve_target_umos(
self,
scope: str | None,
group_id: str | None,
explicit_umos: list[str] | None,
) -> list[str]:
"""根据 scope / group / 显式 umos 获取目标会话列表."""
if explicit_umos:
return explicit_umos
if not scope:
return []
if scope == "custom_group":
if not group_id:
raise ValueError("请指定分组 ID")
groups = self._get_groups()
if group_id not in groups:
raise KeyError(f"分组 '{group_id}' 不存在")
return groups[group_id].get("umos", [])
async with self.db_helper.get_db() as session:
session: AsyncSession
result = await session.execute(select(ConversationV2.user_id).distinct())
all_umos = [row[0] for row in result.fetchall()]
if scope == "group":
return [u for u in all_umos if self._is_group_umo(u)]
if scope == "private":
return [u for u in all_umos if self._is_private_umo(u)]
if scope == "all":
return all_umos
return []
```
这样 `batch_update_service` 中“解析 umos”的部分就可以大幅简化：
```python
async def batch_update_service(self):
try:
data = await request.get_json()
llm_enabled = data.get("llm_enabled")
tts_enabled = data.get("tts_enabled")
session_enabled = data.get("session_enabled")
if llm_enabled is None and tts_enabled is None and session_enabled is None:
return Response().error("至少需要指定一个要修改的状态").__dict__
try:
umos = await self._resolve_target_umos(
scope=data.get("scope"),
group_id=data.get("group_id"),
explicit_umos=data.get("umos") or [],
)
except ValueError as e:
return Response().error(str(e)).__dict__
except KeyError as e:
return Response().error(str(e)).__dict__
if not umos:
return Response().error("没有找到符合条件的会话").__dict__
# ... 保持现有循环和响应构造逻辑不变 ...
```
而 `batch_update_provider` 也可以直接复用这个 helper：
```python
async def batch_update_provider(self):
try:
data = await request.get_json()
provider_type = data.get("provider_type")
provider_id = data.get("provider_id")
# ... 复用现有的 provider_type_map 逻辑 ...
try:
umos = await self._resolve_target_umos(
scope=data.get("scope"),
group_id=data.get("group_id"),
explicit_umos=data.get("umos") or [],
)
except ValueError as e:
return Response().error(str(e)).__dict__
except KeyError as e:
return Response().error(str(e)).__dict__
if not umos:
return Response().error("没有找到符合条件的会话").__dict__
# ... 保持调用 provider_manager.set_provider 的循环逻辑不变 ...
```
这样可以消除两处端点中重复的数据库访问、`custom_group` 处理以及字符串类型过滤逻辑。
### 2. 统一封装 UMO 解析 / 类型判断
你也可以把 `list_all_umos_with_status` 中，以及上面 scope 过滤中用到的字符串拆分和类型判断逻辑封装起来：
```python
def _parse_umo(self, umo: str) -> dict:
parts = umo.split(":")
return {
"umo": umo,
"platform": parts[0] if len(parts) >= 1 else "unknown",
"message_type": parts[1] if len(parts) >= 2 else "unknown",
"session_id": parts[2] if len(parts) >= 3 else umo,
}
def _is_group_umo(self, umo: str) -> bool:
mt = self._parse_umo(umo)["message_type"].lower()
return mt in {"group", "groupmessage"}
def _is_private_umo(self, umo: str) -> bool:
mt = self._parse_umo(umo)["message_type"].lower()
return mt in {"private", "friend", "friendmessage"}
```
在 `list_all_umos_with_status` 中就可以这样使用：
```python
for umo in all_umos:
parsed = self._parse_umo(umo)
umo_platform = parsed["platform"]
umo_message_type = parsed["message_type"]
umo_session_id = parsed["session_id"]
if message_type == "group" and not self._is_group_umo(umo):
continue
if message_type == "private" and not self._is_private_umo(umo):
continue
if platform and umo_platform != platform:
continue
# ... 保持现有规则 / svc_config / 搜索 / provider 逻辑不变 ...
```
同时 `_resolve_target_umos` 内部的 scope 过滤也可以改用 `_is_group_umo` / `_is_private_umo`，避免重复写字符串模式判断。
这两处小的抽取既能保持所有功能不变，又能让路由方法更短、更易读，减少重复代码，并把 UMO 的语义集中在一个地方管理，方便之后如果格式或支持的消息类型发生变化时进行维护。
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进后续的评审质量。
Original comment in English
Hey - I've found 3 issues, and left some high level feedback:
- The new
list_all_umos_with_statusendpoint loads all distinctConversationV2.user_idand all rules into memory, then filters/paginates in Python; consider pushing filtering/pagination into the DB query (and avoiding thepage_size=99999_get_umo_rulescall) to keep this usable on large installations. - The logic that derives
umosfromscopeandgroup_idinbatch_update_serviceandbatch_update_provideris nearly identical; extracting a shared helper to resolve the target UMO list would reduce duplication and the risk of the two endpoints diverging in behavior. - In
batch_update_provider, the checkif not provider_type or not provider_idcurrently prevents using a falsy/emptyprovider_idto clear a per-session provider, while the frontend sometimes sendsprovider_id: this.batchChatProvider || null; if reset-to-default is intended, you may want to explicitly supportnullas "clear override" instead of rejecting it.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new `list_all_umos_with_status` endpoint loads *all* distinct `ConversationV2.user_id` and all rules into memory, then filters/paginates in Python; consider pushing filtering/pagination into the DB query (and avoiding the `page_size=99999` `_get_umo_rules` call) to keep this usable on large installations.
- The logic that derives `umos` from `scope` and `group_id` in `batch_update_service` and `batch_update_provider` is nearly identical; extracting a shared helper to resolve the target UMO list would reduce duplication and the risk of the two endpoints diverging in behavior.
- In `batch_update_provider`, the check `if not provider_type or not provider_id` currently prevents using a falsy/empty `provider_id` to clear a per-session provider, while the frontend sometimes sends `provider_id: this.batchChatProvider || null`; if reset-to-default is intended, you may want to explicitly support `null` as "clear override" instead of rejecting it.
## Individual Comments
### Comment 1
<location> `dashboard/src/views/SessionManagementPage.vue:1` </location>
<code_context>
-<template>
+<template>
<div class="session-management-page">
<v-container fluid class="pa-0">
</code_context>
<issue_to_address>
**issue (bug_risk):** The BOM character before `<template>` may cause tooling or runtime issues.
There’s a UTF‑8 BOM character before the `<template>` tag (`+<template>`). This can break certain linters, build tools, or template parsers. Please remove the BOM so the file begins directly with `<template>`.
</issue_to_address>
### Comment 2
<location> `dashboard/src/views/SessionManagementPage.vue:617-626` </location>
<code_context>
+ batchLlmStatus: null,
+ batchTtsStatus: null,
+ batchChatProvider: null,
+ batchTtsProvider: null,
+ batchUpdating: false,
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Batch TTS provider update is wired in logic but has no UI control, making it effectively unreachable.
`batchTtsProvider` is wired through state, `canApplyBatch`, and `applyBatchChanges`, and the TTS API payload is constructed, but there’s no UI control (e.g., `<v-select>`) to set it. As a result, the TTS batch path is never exercised. Please either add a TTS provider selector (mirroring `batchChatProvider`) or remove the unused TTS batch state/logic to avoid dead code.
</issue_to_address>
### Comment 3
<location> `astrbot/dashboard/routes/session_management.py:558` </location>
<code_context>
+ logger.error(f"获取会话状态列表失败: {e!s}")
+ return Response().error(f"获取会话状态列表失败: {e!s}").__dict__
+
+ async def batch_update_service(self):
+ """批量更新多个 UMO 的服务状态 (LLM/TTS/Session)
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting shared helpers for resolving UMOs by scope and parsing UMO strings so the new batch endpoints and listing logic stay shorter and avoid duplicated inline string handling.
You can keep all behavior but reduce duplication and nesting by extracting a couple of focused helpers. Two places that give you a lot of leverage are:
1. **Scope → UMOs resolution** (duplicated in `batch_update_service` / `batch_update_provider`)
2. **UMO parsing / type checks** (inline string logic in multiple places)
### 1. Extract shared UMO resolution by scope
The logic to resolve `umos` from `scope` (including `custom_group`) is nearly identical in `batch_update_service` and `batch_update_provider`. You can centralize it:
```python
async def _resolve_target_umos(
self,
scope: str | None,
group_id: str | None,
explicit_umos: list[str] | None,
) -> list[str]:
"""根据 scope / group / 显式 umos 获取目标会话列表."""
if explicit_umos:
return explicit_umos
if not scope:
return []
if scope == "custom_group":
if not group_id:
raise ValueError("请指定分组 ID")
groups = self._get_groups()
if group_id not in groups:
raise KeyError(f"分组 '{group_id}' 不存在")
return groups[group_id].get("umos", [])
async with self.db_helper.get_db() as session:
session: AsyncSession
result = await session.execute(select(ConversationV2.user_id).distinct())
all_umos = [row[0] for row in result.fetchall()]
if scope == "group":
return [u for u in all_umos if self._is_group_umo(u)]
if scope == "private":
return [u for u in all_umos if self._is_private_umo(u)]
if scope == "all":
return all_umos
return []
```
Then `batch_update_service` becomes much simpler in the “resolve umos” part:
```python
async def batch_update_service(self):
try:
data = await request.get_json()
llm_enabled = data.get("llm_enabled")
tts_enabled = data.get("tts_enabled")
session_enabled = data.get("session_enabled")
if llm_enabled is None and tts_enabled is None and session_enabled is None:
return Response().error("至少需要指定一个要修改的状态").__dict__
try:
umos = await self._resolve_target_umos(
scope=data.get("scope"),
group_id=data.get("group_id"),
explicit_umos=data.get("umos") or [],
)
except ValueError as e:
return Response().error(str(e)).__dict__
except KeyError as e:
return Response().error(str(e)).__dict__
if not umos:
return Response().error("没有找到符合条件的会话").__dict__
# ... keep the existing loop & response-building as-is ...
```
And `batch_update_provider` just reuses the same helper:
```python
async def batch_update_provider(self):
try:
data = await request.get_json()
provider_type = data.get("provider_type")
provider_id = data.get("provider_id")
# ... existing provider_type_map logic ...
try:
umos = await self._resolve_target_umos(
scope=data.get("scope"),
group_id=data.get("group_id"),
explicit_umos=data.get("umos") or [],
)
except ValueError as e:
return Response().error(str(e)).__dict__
except KeyError as e:
return Response().error(str(e)).__dict__
if not umos:
return Response().error("没有找到符合条件的会话").__dict__
# ... existing loop that calls provider_manager.set_provider ...
```
This removes duplicated DB access, `custom_group` handling, and string-based type filters in two endpoints.
### 2. Centralize UMO parsing / type predicates
You can also encapsulate the string splitting and type checks used in `list_all_umos_with_status` and the scope filters above:
```python
def _parse_umo(self, umo: str) -> dict:
parts = umo.split(":")
return {
"umo": umo,
"platform": parts[0] if len(parts) >= 1 else "unknown",
"message_type": parts[1] if len(parts) >= 2 else "unknown",
"session_id": parts[2] if len(parts) >= 3 else umo,
}
def _is_group_umo(self, umo: str) -> bool:
mt = self._parse_umo(umo)["message_type"].lower()
return mt in {"group", "groupmessage"}
def _is_private_umo(self, umo: str) -> bool:
mt = self._parse_umo(umo)["message_type"].lower()
return mt in {"private", "friend", "friendmessage"}
```
Then in `list_all_umos_with_status`:
```python
for umo in all_umos:
parsed = self._parse_umo(umo)
umo_platform = parsed["platform"]
umo_message_type = parsed["message_type"]
umo_session_id = parsed["session_id"]
if message_type == "group" and not self._is_group_umo(umo):
continue
if message_type == "private" and not self._is_private_umo(umo):
continue
if platform and umo_platform != platform:
continue
# ... existing rules / svc_config / search / provider logic ...
```
And the scope filtering inside `_resolve_target_umos` uses `_is_group_umo` / `_is_private_umo` instead of repeating string patterns.
These two small extractions keep all functionality, but make the route methods shorter, reduce duplication, and centralize UMO semantics in one place, which will help if the format or supported message types change later.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| @@ -1,4 +1,4 @@ | |||
| <template> | |||
| <template> | |||
Contributor
There was a problem hiding this comment.
issue (bug_risk): <template> 前的 BOM 字符可能会导致工具链或运行时问题。
<template> 标签前存在一个 UTF‑8 BOM 字符（+<template>）。这可能会导致某些 linter、构建工具或模板解析器出错。请移除这个 BOM，使文件直接以 <template> 开头。
Original comment in English
issue (bug_risk): The BOM character before <template> may cause tooling or runtime issues.
There’s a UTF‑8 BOM character before the <template> tag (+<template>). This can break certain linters, build tools, or template parsers. Please remove the BOM so the file begins directly with <template>.
Comment on lines
+617
to
+626
| batchTtsProvider: null, | ||
| batchUpdating: false, | ||
| // 分组管理 | ||
| groups: [], | ||
| groupsLoading: false, | ||
| groupDialog: false, | ||
| groupDialogMode: 'create', | ||
| editingGroup: { | ||
| id: null, |
Contributor
There was a problem hiding this comment.
issue (bug_risk): 批量 TTS provider 更新在逻辑上已经打通，但缺少 UI 控件，实际上无法被触发。
batchTtsProvider 已经接入到 state、canApplyBatch 和 applyBatchChanges 中，TTS 的 API 请求体也会被构造，但目前没有任何 UI 控件（例如 <v-select>）可以设置它。结果就是 TTS 的批量更新分支永远不会被真正使用。请考虑要么增加一个 TTS provider 的选择控件（类似 batchChatProvider），要么删除这部分未使用的 TTS 批量状态/逻辑，避免出现死代码。
Original comment in English
issue (bug_risk): Batch TTS provider update is wired in logic but has no UI control, making it effectively unreachable.
batchTtsProvider is wired through state, canApplyBatch, and applyBatchChanges, and the TTS API payload is constructed, but there’s no UI control (e.g., <v-select>) to set it. As a result, the TTS batch path is never exercised. Please either add a TTS provider selector (mirroring batchChatProvider) or remove the unused TTS batch state/logic to avoid dead code.
| logger.error(f"获取会话状态列表失败: {e!s}") | ||
| return Response().error(f"获取会话状态列表失败: {e!s}").__dict__ | ||
|
|
||
| async def batch_update_service(self): |
Contributor
There was a problem hiding this comment.
issue (complexity): 建议抽取共享的辅助函数，用于按 scope 解析 UMO 列表以及解析 UMO 字符串，从而让新的批量接口和列表接口更精简，并避免重复的内联字符串处理逻辑。
你可以在保持现有行为不变的前提下，通过提取几个聚焦的小工具方法来减少重复和嵌套。两个收益比较大的点是：
- Scope → UMO 列表的解析（
batch_update_service和batch_update_provider中存在重复） - UMO 解析 / 类型判断（多个位置都有类似的字符串处理逻辑）
1. 抽取按 scope 解析 UMO 列表的公共方法
从 scope（包括 custom_group）解析 umos 的逻辑，在 batch_update_service 和 batch_update_provider 中几乎一模一样。可以集中到一个方法里：
async def _resolve_target_umos(
self,
scope: str | None,
group_id: str | None,
explicit_umos: list[str] | None,
) -> list[str]:
"""根据 scope / group / 显式 umos 获取目标会话列表."""
if explicit_umos:
return explicit_umos
if not scope:
return []
if scope == "custom_group":
if not group_id:
raise ValueError("请指定分组 ID")
groups = self._get_groups()
if group_id not in groups:
raise KeyError(f"分组 '{group_id}' 不存在")
return groups[group_id].get("umos", [])
async with self.db_helper.get_db() as session:
session: AsyncSession
result = await session.execute(select(ConversationV2.user_id).distinct())
all_umos = [row[0] for row in result.fetchall()]
if scope == "group":
return [u for u in all_umos if self._is_group_umo(u)]
if scope == "private":
return [u for u in all_umos if self._is_private_umo(u)]
if scope == "all":
return all_umos
return []这样 batch_update_service 中“解析 umos”的部分就可以大幅简化：
async def batch_update_service(self):
try:
data = await request.get_json()
llm_enabled = data.get("llm_enabled")
tts_enabled = data.get("tts_enabled")
session_enabled = data.get("session_enabled")
if llm_enabled is None and tts_enabled is None and session_enabled is None:
return Response().error("至少需要指定一个要修改的状态").__dict__
try:
umos = await self._resolve_target_umos(
scope=data.get("scope"),
group_id=data.get("group_id"),
explicit_umos=data.get("umos") or [],
)
except ValueError as e:
return Response().error(str(e)).__dict__
except KeyError as e:
return Response().error(str(e)).__dict__
if not umos:
return Response().error("没有找到符合条件的会话").__dict__
# ... 保持现有循环和响应构造逻辑不变 ...而 batch_update_provider 也可以直接复用这个 helper：
async def batch_update_provider(self):
try:
data = await request.get_json()
provider_type = data.get("provider_type")
provider_id = data.get("provider_id")
# ... 复用现有的 provider_type_map 逻辑 ...
try:
umos = await self._resolve_target_umos(
scope=data.get("scope"),
group_id=data.get("group_id"),
explicit_umos=data.get("umos") or [],
)
except ValueError as e:
return Response().error(str(e)).__dict__
except KeyError as e:
return Response().error(str(e)).__dict__
if not umos:
return Response().error("没有找到符合条件的会话").__dict__
# ... 保持调用 provider_manager.set_provider 的循环逻辑不变 ...这样可以消除两处端点中重复的数据库访问、custom_group 处理以及字符串类型过滤逻辑。
2. 统一封装 UMO 解析 / 类型判断
你也可以把 list_all_umos_with_status 中，以及上面 scope 过滤中用到的字符串拆分和类型判断逻辑封装起来：
def _parse_umo(self, umo: str) -> dict:
parts = umo.split(":")
return {
"umo": umo,
"platform": parts[0] if len(parts) >= 1 else "unknown",
"message_type": parts[1] if len(parts) >= 2 else "unknown",
"session_id": parts[2] if len(parts) >= 3 else umo,
}
def _is_group_umo(self, umo: str) -> bool:
mt = self._parse_umo(umo)["message_type"].lower()
return mt in {"group", "groupmessage"}
def _is_private_umo(self, umo: str) -> bool:
mt = self._parse_umo(umo)["message_type"].lower()
return mt in {"private", "friend", "friendmessage"}在 list_all_umos_with_status 中就可以这样使用：
for umo in all_umos:
parsed = self._parse_umo(umo)
umo_platform = parsed["platform"]
umo_message_type = parsed["message_type"]
umo_session_id = parsed["session_id"]
if message_type == "group" and not self._is_group_umo(umo):
continue
if message_type == "private" and not self._is_private_umo(umo):
continue
if platform and umo_platform != platform:
continue
# ... 保持现有规则 / svc_config / 搜索 / provider 逻辑不变 ...同时 _resolve_target_umos 内部的 scope 过滤也可以改用 _is_group_umo / _is_private_umo，避免重复写字符串模式判断。
这两处小的抽取既能保持所有功能不变，又能让路由方法更短、更易读，减少重复代码，并把 UMO 的语义集中在一个地方管理，方便之后如果格式或支持的消息类型发生变化时进行维护。
Original comment in English
issue (complexity): Consider extracting shared helpers for resolving UMOs by scope and parsing UMO strings so the new batch endpoints and listing logic stay shorter and avoid duplicated inline string handling.
You can keep all behavior but reduce duplication and nesting by extracting a couple of focused helpers. Two places that give you a lot of leverage are:
- Scope → UMOs resolution (duplicated in
batch_update_service/batch_update_provider) - UMO parsing / type checks (inline string logic in multiple places)
1. Extract shared UMO resolution by scope
The logic to resolve umos from scope (including custom_group) is nearly identical in batch_update_service and batch_update_provider. You can centralize it:
async def _resolve_target_umos(
self,
scope: str | None,
group_id: str | None,
explicit_umos: list[str] | None,
) -> list[str]:
"""根据 scope / group / 显式 umos 获取目标会话列表."""
if explicit_umos:
return explicit_umos
if not scope:
return []
if scope == "custom_group":
if not group_id:
raise ValueError("请指定分组 ID")
groups = self._get_groups()
if group_id not in groups:
raise KeyError(f"分组 '{group_id}' 不存在")
return groups[group_id].get("umos", [])
async with self.db_helper.get_db() as session:
session: AsyncSession
result = await session.execute(select(ConversationV2.user_id).distinct())
all_umos = [row[0] for row in result.fetchall()]
if scope == "group":
return [u for u in all_umos if self._is_group_umo(u)]
if scope == "private":
return [u for u in all_umos if self._is_private_umo(u)]
if scope == "all":
return all_umos
return []Then batch_update_service becomes much simpler in the “resolve umos” part:
async def batch_update_service(self):
try:
data = await request.get_json()
llm_enabled = data.get("llm_enabled")
tts_enabled = data.get("tts_enabled")
session_enabled = data.get("session_enabled")
if llm_enabled is None and tts_enabled is None and session_enabled is None:
return Response().error("至少需要指定一个要修改的状态").__dict__
try:
umos = await self._resolve_target_umos(
scope=data.get("scope"),
group_id=data.get("group_id"),
explicit_umos=data.get("umos") or [],
)
except ValueError as e:
return Response().error(str(e)).__dict__
except KeyError as e:
return Response().error(str(e)).__dict__
if not umos:
return Response().error("没有找到符合条件的会话").__dict__
# ... keep the existing loop & response-building as-is ...And batch_update_provider just reuses the same helper:
async def batch_update_provider(self):
try:
data = await request.get_json()
provider_type = data.get("provider_type")
provider_id = data.get("provider_id")
# ... existing provider_type_map logic ...
try:
umos = await self._resolve_target_umos(
scope=data.get("scope"),
group_id=data.get("group_id"),
explicit_umos=data.get("umos") or [],
)
except ValueError as e:
return Response().error(str(e)).__dict__
except KeyError as e:
return Response().error(str(e)).__dict__
if not umos:
return Response().error("没有找到符合条件的会话").__dict__
# ... existing loop that calls provider_manager.set_provider ...This removes duplicated DB access, custom_group handling, and string-based type filters in two endpoints.
2. Centralize UMO parsing / type predicates
You can also encapsulate the string splitting and type checks used in list_all_umos_with_status and the scope filters above:
def _parse_umo(self, umo: str) -> dict:
parts = umo.split(":")
return {
"umo": umo,
"platform": parts[0] if len(parts) >= 1 else "unknown",
"message_type": parts[1] if len(parts) >= 2 else "unknown",
"session_id": parts[2] if len(parts) >= 3 else umo,
}
def _is_group_umo(self, umo: str) -> bool:
mt = self._parse_umo(umo)["message_type"].lower()
return mt in {"group", "groupmessage"}
def _is_private_umo(self, umo: str) -> bool:
mt = self._parse_umo(umo)["message_type"].lower()
return mt in {"private", "friend", "friendmessage"}Then in list_all_umos_with_status:
for umo in all_umos:
parsed = self._parse_umo(umo)
umo_platform = parsed["platform"]
umo_message_type = parsed["message_type"]
umo_session_id = parsed["session_id"]
if message_type == "group" and not self._is_group_umo(umo):
continue
if message_type == "private" and not self._is_private_umo(umo):
continue
if platform and umo_platform != platform:
continue
# ... existing rules / svc_config / search / provider logic ...And the scope filtering inside _resolve_target_umos uses _is_group_umo / _is_private_umo instead of repeating string patterns.
These two small extractions keep all functionality, but make the route methods shorter, reduce duplication, and centralize UMO semantics in one place, which will help if the format or supported message types change later.
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.
Fixes #4185
解决的问题：
添加的功能：
📝 改动点 / Modifications
核心文件修改：
后端
astrbot/dashboard/routes/session_management.py：list_all_umos_with_statusAPI：获取所有会话及其当前状态（支持分页、搜索、筛选）batch_update_serviceAPI：批量更新 LLM/TTS/Session 开关状态batch_update_providerAPI：批量更新 Provider 配置list_groups、create_group、update_group、delete_group前端
dashboard/src/views/SessionManagementPage.vue：国际化：
en-US/features/session-management.json：添加英文翻译zh-CN/features/session-management.json：添加中文翻译实现的功能：
✅ 批量开关 LLM：一键开启/关闭所有会话的 LLM 功能
✅ 批量开关 TTS：一键开启/关闭所有会话的 TTS 功能
✅ 批量切换模型：统一切换所有会话使用的聊天/TTS/STT 模型
✅ 范围筛选：支持按群聊/私聊/平台/自定义分组进行筛选
✅ 分组管理：支持将会话归类到自定义分组，方便批量管理
This is NOT a breaking change. / 这不是一个破坏性变更。
🖼️ 测试结果 / Screenshots or Test Results
测试场景：
✅ 检查清单 / Checklist
✦ 这个 PR 恢复并增强了 v4.6 的批量操作功能，大幅提升多会话管理效率！🎉
Summary by Sourcery
添加批量会话管理功能，支持批量状态/服务商更新以及自定义会话分组。
新功能：
改进：
Original summary in English
Summary by Sourcery
Add batch session management capabilities with support for bulk status/provider updates and custom session groups.
New Features:
Enhancements: