Feature add fileupload to plugin config #4539
Open
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
dosubot
bot
added
size:XL
![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 个问题，并留下了一些总体反馈：
- 在
FileConfigItem.vue中，部分上传失败时的错误 toast 使用了errors.join('\\n')，这会在界面上渲染为字面量的反斜杠 + n；建议改成errors.join('\n')，这样每条错误会单独显示在一行。 FileConfigItem.vue从未调用后端的delete_plugin_file接口，所以在 UI 中删除文件只会更新本地列表，不会删除已暂存的上传文件；建议在删除「pill」时调用删除 API，这样在保存前就能清理临时文件。
给 AI Agent 的提示
Please address the comments from this code review:
## Overall Comments
- In `FileConfigItem.vue`, the error toast for partial upload failures uses `errors.join('\\n')`, which will render a literal backslash+n in the UI; consider changing this to `errors.join('\n')` so each error appears on its own line.
- The `delete_plugin_file` backend endpoint is never called from `FileConfigItem.vue`, so deleting a file from the UI only updates the local list and won't remove any already-staged uploads; consider invoking the delete API when a pill is removed so temporary files are cleaned up before saving.
## Individual Comments
### Comment 1
<location> `dashboard/src/components/shared/FileConfigItem.vue:212` </location>
<code_context>
+ }
+}
+
+const deleteFile = (filePath) => {
+ fileList.value = fileList.value.filter((item) => item !== filePath)
+ toast.success(tm('fileUpload.deleteSuccess'))
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Deleting a file only updates the local list and never calls the backend delete endpoint, which may leave staged uploads behind.
This handler only updates `fileList` and shows a toast; it never calls `/api/config/plugin/file/delete`. As a result, files uploaded and then removed before saving will remain in the staging area. Please also invoke the delete API for the given `filePath` so backend storage stays consistent with the UI.
Suggested implementation:
```
const deleteFile = async (filePath) => {
try {
const response = await fetch('/api/config/plugin/file/delete', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({ filePath }),
})
const data = await response.json().catch(() => ({}))
if (!response.ok || (data && data.success === false)) {
throw new Error(data?.message || 'File delete failed')
}
fileList.value = fileList.value.filter((item) => item !== filePath)
toast.success(tm('fileUpload.deleteSuccess'))
} catch (error) {
console.error('File delete failed:', error)
toast.error(tm('fileUpload.deleteFailed'))
}
}
```
1. Ensure the backend `/api/config/plugin/file/delete` endpoint expects a JSON body with `{ filePath: string }`. If it uses a different payload shape, update the `body: JSON.stringify(...)` accordingly.
2. Confirm that the i18n key `fileUpload.deleteFailed` exists in your localization files. If not, add it (e.g., "Failed to delete file.").
3. If your project already uses a centralized HTTP client (e.g., Axios instance or a composable like `useApiClient`), you may want to replace the `fetch` call with that client for consistency.
</issue_to_address>
### Comment 2
<location> `astrbot/dashboard/routes/config.py:1129` </location>
<code_context>
+ if isinstance(file_types, list):
+ allowed_exts = [str(ext).lstrip(".").lower() for ext in file_types if str(ext).strip()]
+
+ files = await request.files
+ if not files:
+ return Response().error("No files uploaded").__dict__
</code_context>
<issue_to_address>
**🚨 suggestion (security):** The upload endpoint lacks any explicit limits or validation on file size/count, which can be abused.
`upload_plugin_file` accepts and writes all incoming files without validating per-file size, total payload size, or file count. This allows very large uploads that can exhaust disk or memory. Please add reasonable limits (per-file, per-request, and/or total size) and return a clear error when those limits are exceeded.
Suggested implementation:
```python
files = await request.files
if not files:
return Response().error("No files uploaded").__dict__
# Enforce basic upload limits
MAX_FILE_COUNT = 10
MAX_TOTAL_UPLOAD_SIZE = 50 * 1024 * 1024 # 50 MB
# Limit number of files
try:
file_count = len(files)
except TypeError:
# Fallback if `files` is not directly countable
file_count = sum(1 for _ in files)
if file_count > MAX_FILE_COUNT:
return (
Response()
.error(f"Too many files uploaded. Maximum allowed is {MAX_FILE_COUNT}.")
.__dict__
)
# Limit total upload size via Content-Length header if available
content_length = request.headers.get("content-length") or request.headers.get("Content-Length")
if content_length:
try:
total_size = int(content_length)
except (TypeError, ValueError):
total_size = None
else:
if total_size > MAX_TOTAL_UPLOAD_SIZE:
return (
Response()
.error(
f"Total upload size exceeds limit of {MAX_TOTAL_UPLOAD_SIZE // (1024 * 1024)} MB."
)
.__dict__
)
```
1. If your framework/request abstraction exposes a more accurate way to determine total payload size or per-file size (e.g., `request.body_size`, `file.size`, etc.), you may want to use that instead of relying solely on the `Content-Length` header.
2. If there are other upload endpoints, consider centralizing these limits (e.g., module-level constants or configuration) so they can be reused and adjusted in a single place.
</issue_to_address>
### Comment 3
<location> `astrbot/dashboard/routes/config.py:897` </location>
<code_context>
return Response().error(str(e)).__dict__
+
+ def _get_plugin_metadata_by_name(self, plugin_name: str):
+ for plugin_md in star_registry:
+ if plugin_md.name == plugin_name:
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the new plugin file/config handling into a dedicated helper module so the route class stays lean and only orchestrates HTTP flow.
You can reduce the added complexity substantially by pulling the file/config logic into a focused helper and keeping the route class thin. This keeps all behavior but localizes the low‑level concerns.
### 1. Extract a `PluginFileConfigManager`
Move the helpers and core logic out of `AstrBotConfig` into a dedicated module, e.g. `astrbot/core/config/plugin_file_config.py`:
```python
# astrbot/core/config/plugin_file_config.py
import os
from typing import Iterable
from astrbot.core.utils.astrbot_path import (
get_astrbot_plugin_data_path,
get_astrbot_temp_path,
)
from astrbot.core.utils.io import remove_dir
def sanitize_filename(name: str) -> str:
cleaned = os.path.basename(name).strip()
if not cleaned or cleaned in {".", ".."}:
return ""
for sep in (os.sep, os.altsep):
if sep:
cleaned = cleaned.replace(sep, "_")
return cleaned
def sanitize_path_segment(segment: str) -> str:
cleaned = []
for ch in segment:
if (
("a" <= ch <= "z")
or ("A" <= ch <= "Z")
or ch.isdigit()
or ch in {"-", "_"}
):
cleaned.append(ch)
else:
cleaned.append("_")
result = "".join(cleaned).strip("_")
return result or "_"
def config_key_to_folder(key_path: str) -> str:
parts = [sanitize_path_segment(p) for p in key_path.split(".") if p]
return "/".join(parts) if parts else "_"
def normalize_rel_path(rel_path: str | None) -> str | None:
if not isinstance(rel_path, str):
return None
rel = rel_path.replace("\\", "/").lstrip("/")
if not rel:
return None
parts = [p for p in rel.split("/") if p]
if any(part in {".", ".."} for part in parts):
return None
if rel.startswith("../") or "/../" in rel:
return None
return "/".join(parts)
```
Keep the list / schema / fs logic separately in the same module:
```python
def normalize_file_list(value, key_path: str) -> tuple[list[str], bool]:
if value is None:
return [], False
if not isinstance(value, list):
raise ValueError(f"Invalid file list for {key_path}")
folder = config_key_to_folder(key_path)
expected_prefix = f"files/{folder}/"
results: list[str] = []
changed = False
for item in value:
if not isinstance(item, str):
raise ValueError(f"Invalid file entry for {key_path}")
rel = normalize_rel_path(item)
if not rel or not rel.startswith("files/"):
raise ValueError(f"Invalid file path: {item}")
if rel.startswith(expected_prefix):
results.append(rel)
continue
if rel.count("/") == 1:
filename = rel.split("/", 1)[1]
if not filename:
raise ValueError(f"Invalid file path: {item}")
results.append(f"{expected_prefix}{filename}")
changed = True
continue
raise ValueError(f"Invalid file path: {item}")
return results, changed
def apply_plugin_file_ops(plugin_name: str, md, post_configs: dict) -> None:
schema = getattr(md.config, "schema", None) if md and md.config else None
if not isinstance(schema, dict):
return
# you can also move _collect_file_keys/_get_value_by_path/_set_value_by_path here
# ... existing logic moved verbatim from AstrBotConfig._apply_plugin_file_ops ...
```
You can also move `_collect_file_keys`, `_get_value_by_path`, `_set_value_by_path`, and `_get_schema_item` into this module so all schema/path knowledge is in one place.
### 2. Thin the route class to orchestration only
Then `AstrBotConfig` becomes mostly orchestration, which is easier to read:
```python
# config.py
from astrbot.core.config.plugin_file_config import (
sanitize_filename,
normalize_rel_path,
config_key_to_folder,
apply_plugin_file_ops,
)
async def upload_plugin_file(self):
plugin_name = request.args.get("plugin_name")
key_path = request.args.get("key")
if not plugin_name or not key_path:
return Response().error("Missing plugin_name or key parameter").__dict__
md = self._get_plugin_metadata_by_name(plugin_name)
if not md or not md.config:
return Response().error(
f"Plugin {plugin_name} not found or has no config",
).__dict__
meta = self._get_schema_item(md.config.schema, key_path)
if not meta or meta.get("type") != "file":
return Response().error("Config item not found or not file type").__dict__
file_types = meta.get("file_types")
allowed_exts = []
if isinstance(file_types, list):
allowed_exts = [str(ext).lstrip(".").lower() for ext in file_types if str(ext).strip()]
files = await request.files
if not files:
return Response().error("No files uploaded").__dict__
staging_root = os.path.join(
get_astrbot_temp_path(),
"plugin_file_uploads",
plugin_name,
)
os.makedirs(staging_root, exist_ok=True)
uploaded = []
folder = config_key_to_folder(key_path)
errors = []
for file in files.values():
filename = sanitize_filename(file.filename or "")
# ... unchanged logic using helpers from plugin_file_config ...
# ... unchanged response construction ...
```
And for saving plugin configs:
```python
async def _save_plugin_configs(self, post_configs: dict, plugin_name: str):
# ... plugin_md lookup unchanged ...
try:
errors, post_configs = validate_config(
post_configs, getattr(md.config, "schema", {}), is_core=False
)
if errors:
raise ValueError(f"格式校验未通过: {errors}")
apply_plugin_file_ops(plugin_name, md, post_configs)
md.config.save_config(post_configs)
except Exception as e:
raise e
```
This preserves behavior but:
- Concentrates schema/path/file logic into one module.
- Keeps route methods focused on HTTP concerns and delegating to the helper.
- Makes future changes to file layout or migration logic localized to `plugin_file_config.py`.
</issue_to_address>帮我变得更有用！请在每条评论上点击 👍 或 👎，我会根据你的反馈改进后续评审。
Original comment in English
Hey - I've found 3 issues, and left some high level feedback:
- In
FileConfigItem.vue, the error toast for partial upload failures useserrors.join('\\n'), which will render a literal backslash+n in the UI; consider changing this toerrors.join('\n')so each error appears on its own line. - The
delete_plugin_filebackend endpoint is never called fromFileConfigItem.vue, so deleting a file from the UI only updates the local list and won't remove any already-staged uploads; consider invoking the delete API when a pill is removed so temporary files are cleaned up before saving.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `FileConfigItem.vue`, the error toast for partial upload failures uses `errors.join('\\n')`, which will render a literal backslash+n in the UI; consider changing this to `errors.join('\n')` so each error appears on its own line.
- The `delete_plugin_file` backend endpoint is never called from `FileConfigItem.vue`, so deleting a file from the UI only updates the local list and won't remove any already-staged uploads; consider invoking the delete API when a pill is removed so temporary files are cleaned up before saving.
## Individual Comments
### Comment 1
<location> `dashboard/src/components/shared/FileConfigItem.vue:212` </location>
<code_context>
+ }
+}
+
+const deleteFile = (filePath) => {
+ fileList.value = fileList.value.filter((item) => item !== filePath)
+ toast.success(tm('fileUpload.deleteSuccess'))
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Deleting a file only updates the local list and never calls the backend delete endpoint, which may leave staged uploads behind.
This handler only updates `fileList` and shows a toast; it never calls `/api/config/plugin/file/delete`. As a result, files uploaded and then removed before saving will remain in the staging area. Please also invoke the delete API for the given `filePath` so backend storage stays consistent with the UI.
Suggested implementation:
```
const deleteFile = async (filePath) => {
try {
const response = await fetch('/api/config/plugin/file/delete', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({ filePath }),
})
const data = await response.json().catch(() => ({}))
if (!response.ok || (data && data.success === false)) {
throw new Error(data?.message || 'File delete failed')
}
fileList.value = fileList.value.filter((item) => item !== filePath)
toast.success(tm('fileUpload.deleteSuccess'))
} catch (error) {
console.error('File delete failed:', error)
toast.error(tm('fileUpload.deleteFailed'))
}
}
```
1. Ensure the backend `/api/config/plugin/file/delete` endpoint expects a JSON body with `{ filePath: string }`. If it uses a different payload shape, update the `body: JSON.stringify(...)` accordingly.
2. Confirm that the i18n key `fileUpload.deleteFailed` exists in your localization files. If not, add it (e.g., "Failed to delete file.").
3. If your project already uses a centralized HTTP client (e.g., Axios instance or a composable like `useApiClient`), you may want to replace the `fetch` call with that client for consistency.
</issue_to_address>
### Comment 2
<location> `astrbot/dashboard/routes/config.py:1129` </location>
<code_context>
+ if isinstance(file_types, list):
+ allowed_exts = [str(ext).lstrip(".").lower() for ext in file_types if str(ext).strip()]
+
+ files = await request.files
+ if not files:
+ return Response().error("No files uploaded").__dict__
</code_context>
<issue_to_address>
**🚨 suggestion (security):** The upload endpoint lacks any explicit limits or validation on file size/count, which can be abused.
`upload_plugin_file` accepts and writes all incoming files without validating per-file size, total payload size, or file count. This allows very large uploads that can exhaust disk or memory. Please add reasonable limits (per-file, per-request, and/or total size) and return a clear error when those limits are exceeded.
Suggested implementation:
```python
files = await request.files
if not files:
return Response().error("No files uploaded").__dict__
# Enforce basic upload limits
MAX_FILE_COUNT = 10
MAX_TOTAL_UPLOAD_SIZE = 50 * 1024 * 1024 # 50 MB
# Limit number of files
try:
file_count = len(files)
except TypeError:
# Fallback if `files` is not directly countable
file_count = sum(1 for _ in files)
if file_count > MAX_FILE_COUNT:
return (
Response()
.error(f"Too many files uploaded. Maximum allowed is {MAX_FILE_COUNT}.")
.__dict__
)
# Limit total upload size via Content-Length header if available
content_length = request.headers.get("content-length") or request.headers.get("Content-Length")
if content_length:
try:
total_size = int(content_length)
except (TypeError, ValueError):
total_size = None
else:
if total_size > MAX_TOTAL_UPLOAD_SIZE:
return (
Response()
.error(
f"Total upload size exceeds limit of {MAX_TOTAL_UPLOAD_SIZE // (1024 * 1024)} MB."
)
.__dict__
)
```
1. If your framework/request abstraction exposes a more accurate way to determine total payload size or per-file size (e.g., `request.body_size`, `file.size`, etc.), you may want to use that instead of relying solely on the `Content-Length` header.
2. If there are other upload endpoints, consider centralizing these limits (e.g., module-level constants or configuration) so they can be reused and adjusted in a single place.
</issue_to_address>
### Comment 3
<location> `astrbot/dashboard/routes/config.py:897` </location>
<code_context>
return Response().error(str(e)).__dict__
+
+ def _get_plugin_metadata_by_name(self, plugin_name: str):
+ for plugin_md in star_registry:
+ if plugin_md.name == plugin_name:
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the new plugin file/config handling into a dedicated helper module so the route class stays lean and only orchestrates HTTP flow.
You can reduce the added complexity substantially by pulling the file/config logic into a focused helper and keeping the route class thin. This keeps all behavior but localizes the low‑level concerns.
### 1. Extract a `PluginFileConfigManager`
Move the helpers and core logic out of `AstrBotConfig` into a dedicated module, e.g. `astrbot/core/config/plugin_file_config.py`:
```python
# astrbot/core/config/plugin_file_config.py
import os
from typing import Iterable
from astrbot.core.utils.astrbot_path import (
get_astrbot_plugin_data_path,
get_astrbot_temp_path,
)
from astrbot.core.utils.io import remove_dir
def sanitize_filename(name: str) -> str:
cleaned = os.path.basename(name).strip()
if not cleaned or cleaned in {".", ".."}:
return ""
for sep in (os.sep, os.altsep):
if sep:
cleaned = cleaned.replace(sep, "_")
return cleaned
def sanitize_path_segment(segment: str) -> str:
cleaned = []
for ch in segment:
if (
("a" <= ch <= "z")
or ("A" <= ch <= "Z")
or ch.isdigit()
or ch in {"-", "_"}
):
cleaned.append(ch)
else:
cleaned.append("_")
result = "".join(cleaned).strip("_")
return result or "_"
def config_key_to_folder(key_path: str) -> str:
parts = [sanitize_path_segment(p) for p in key_path.split(".") if p]
return "/".join(parts) if parts else "_"
def normalize_rel_path(rel_path: str | None) -> str | None:
if not isinstance(rel_path, str):
return None
rel = rel_path.replace("\\", "/").lstrip("/")
if not rel:
return None
parts = [p for p in rel.split("/") if p]
if any(part in {".", ".."} for part in parts):
return None
if rel.startswith("../") or "/../" in rel:
return None
return "/".join(parts)
```
Keep the list / schema / fs logic separately in the same module:
```python
def normalize_file_list(value, key_path: str) -> tuple[list[str], bool]:
if value is None:
return [], False
if not isinstance(value, list):
raise ValueError(f"Invalid file list for {key_path}")
folder = config_key_to_folder(key_path)
expected_prefix = f"files/{folder}/"
results: list[str] = []
changed = False
for item in value:
if not isinstance(item, str):
raise ValueError(f"Invalid file entry for {key_path}")
rel = normalize_rel_path(item)
if not rel or not rel.startswith("files/"):
raise ValueError(f"Invalid file path: {item}")
if rel.startswith(expected_prefix):
results.append(rel)
continue
if rel.count("/") == 1:
filename = rel.split("/", 1)[1]
if not filename:
raise ValueError(f"Invalid file path: {item}")
results.append(f"{expected_prefix}{filename}")
changed = True
continue
raise ValueError(f"Invalid file path: {item}")
return results, changed
def apply_plugin_file_ops(plugin_name: str, md, post_configs: dict) -> None:
schema = getattr(md.config, "schema", None) if md and md.config else None
if not isinstance(schema, dict):
return
# you can also move _collect_file_keys/_get_value_by_path/_set_value_by_path here
# ... existing logic moved verbatim from AstrBotConfig._apply_plugin_file_ops ...
```
You can also move `_collect_file_keys`, `_get_value_by_path`, `_set_value_by_path`, and `_get_schema_item` into this module so all schema/path knowledge is in one place.
### 2. Thin the route class to orchestration only
Then `AstrBotConfig` becomes mostly orchestration, which is easier to read:
```python
# config.py
from astrbot.core.config.plugin_file_config import (
sanitize_filename,
normalize_rel_path,
config_key_to_folder,
apply_plugin_file_ops,
)
async def upload_plugin_file(self):
plugin_name = request.args.get("plugin_name")
key_path = request.args.get("key")
if not plugin_name or not key_path:
return Response().error("Missing plugin_name or key parameter").__dict__
md = self._get_plugin_metadata_by_name(plugin_name)
if not md or not md.config:
return Response().error(
f"Plugin {plugin_name} not found or has no config",
).__dict__
meta = self._get_schema_item(md.config.schema, key_path)
if not meta or meta.get("type") != "file":
return Response().error("Config item not found or not file type").__dict__
file_types = meta.get("file_types")
allowed_exts = []
if isinstance(file_types, list):
allowed_exts = [str(ext).lstrip(".").lower() for ext in file_types if str(ext).strip()]
files = await request.files
if not files:
return Response().error("No files uploaded").__dict__
staging_root = os.path.join(
get_astrbot_temp_path(),
"plugin_file_uploads",
plugin_name,
)
os.makedirs(staging_root, exist_ok=True)
uploaded = []
folder = config_key_to_folder(key_path)
errors = []
for file in files.values():
filename = sanitize_filename(file.filename or "")
# ... unchanged logic using helpers from plugin_file_config ...
# ... unchanged response construction ...
```
And for saving plugin configs:
```python
async def _save_plugin_configs(self, post_configs: dict, plugin_name: str):
# ... plugin_md lookup unchanged ...
try:
errors, post_configs = validate_config(
post_configs, getattr(md.config, "schema", {}), is_core=False
)
if errors:
raise ValueError(f"格式校验未通过: {errors}")
apply_plugin_file_ops(plugin_name, md, post_configs)
md.config.save_config(post_configs)
except Exception as e:
raise e
```
This preserves behavior but:
- Concentrates schema/path/file logic into one module.
- Keeps route methods focused on HTTP concerns and delegating to the helper.
- Makes future changes to file layout or migration logic localized to `plugin_file_config.py`.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| return Response().error(str(e)).__dict__ | ||
|
|
||
|
|
||
| def _get_plugin_metadata_by_name(self, plugin_name: str): |
Contributor
There was a problem hiding this comment.
issue (complexity): 建议将新增加的插件文件/配置处理逻辑抽取到独立的辅助模块中，这样路由类可以保持精简，只负责编排 HTTP 流程。
你可以通过把文件/配置相关逻辑集中到一个专门的 helper 中，并让路由类保持“瘦身”，显著降低新增的复杂度。这样既保留所有行为，又把底层细节局部化。
1. 抽取一个 PluginFileConfigManager
将辅助函数和核心逻辑从 AstrBotConfig 中移到一个独立模块，例如 astrbot/core/config/plugin_file_config.py：
# astrbot/core/config/plugin_file_config.py
import os
from typing import Iterable
from astrbot.core.utils.astrbot_path import (
get_astrbot_plugin_data_path,
get_astrbot_temp_path,
)
from astrbot.core.utils.io import remove_dir
def sanitize_filename(name: str) -> str:
cleaned = os.path.basename(name).strip()
if not cleaned or cleaned in {".", ".."}:
return ""
for sep in (os.sep, os.altsep):
if sep:
cleaned = cleaned.replace(sep, "_")
return cleaned
def sanitize_path_segment(segment: str) -> str:
cleaned = []
for ch in segment:
if (
("a" <= ch <= "z")
or ("A" <= ch <= "Z")
or ch.isdigit()
or ch in {"-", "_"}
):
cleaned.append(ch)
else:
cleaned.append("_")
result = "".join(cleaned).strip("_")
return result or "_"
def config_key_to_folder(key_path: str) -> str:
parts = [sanitize_path_segment(p) for p in key_path.split(".") if p]
return "/".join(parts) if parts else "_"
def normalize_rel_path(rel_path: str | None) -> str | None:
if not isinstance(rel_path, str):
return None
rel = rel_path.replace("\\", "/").lstrip("/")
if not rel:
return None
parts = [p for p in rel.split("/") if p]
if any(part in {".", ".."} for part in parts):
return None
if rel.startswith("../") or "/../" in rel:
return None
return "/".join(parts)将列表/Schema/文件系统相关逻辑也放在同一模块中：
def normalize_file_list(value, key_path: str) -> tuple[list[str], bool]:
if value is None:
return [], False
if not isinstance(value, list):
raise ValueError(f"Invalid file list for {key_path}")
folder = config_key_to_folder(key_path)
expected_prefix = f"files/{folder}/"
results: list[str] = []
changed = False
for item in value:
if not isinstance(item, str):
raise ValueError(f"Invalid file entry for {key_path}")
rel = normalize_rel_path(item)
if not rel or not rel.startswith("files/"):
raise ValueError(f"Invalid file path: {item}")
if rel.startswith(expected_prefix):
results.append(rel)
continue
if rel.count("/") == 1:
filename = rel.split("/", 1)[1]
if not filename:
raise ValueError(f"Invalid file path: {item}")
results.append(f"{expected_prefix}{filename}")
changed = True
continue
raise ValueError(f"Invalid file path: {item}")
return results, changed
def apply_plugin_file_ops(plugin_name: str, md, post_configs: dict) -> None:
schema = getattr(md.config, "schema", None) if md and md.config else None
if not isinstance(schema, dict):
return
# you can also move _collect_file_keys/_get_value_by_path/_set_value_by_path here
# ... existing logic moved verbatim from AstrBotConfig._apply_plugin_file_ops ...你也可以将 _collect_file_keys、_get_value_by_path、_set_value_by_path 和 _get_schema_item 挪到这个模块里，这样所有 Schema/路径相关的知识都集中在一处。
2. 让路由类只负责编排
这样 AstrBotConfig 基本上就只负责编排逻辑，更容易阅读：
# config.py
from astrbot.core.config.plugin_file_config import (
sanitize_filename,
normalize_rel_path,
config_key_to_folder,
apply_plugin_file_ops,
)
async def upload_plugin_file(self):
plugin_name = request.args.get("plugin_name")
key_path = request.args.get("key")
if not plugin_name or not key_path:
return Response().error("Missing plugin_name or key parameter").__dict__
md = self._get_plugin_metadata_by_name(plugin_name)
if not md or not md.config:
return Response().error(
f"Plugin {plugin_name} not found or has no config",
).__dict__
meta = self._get_schema_item(md.config.schema, key_path)
if not meta or meta.get("type") != "file":
return Response().error("Config item not found or not file type").__dict__
file_types = meta.get("file_types")
allowed_exts = []
if isinstance(file_types, list):
allowed_exts = [str(ext).lstrip(".").lower() for ext in file_types if str(ext).strip()]
files = await request.files
if not files:
return Response().error("No files uploaded").__dict__
staging_root = os.path.join(
get_astrbot_temp_path(),
"plugin_file_uploads",
plugin_name,
)
os.makedirs(staging_root, exist_ok=True)
uploaded = []
folder = config_key_to_folder(key_path)
errors = []
for file in files.values():
filename = sanitize_filename(file.filename or "")
# ... unchanged logic using helpers from plugin_file_config ...
# ... unchanged response construction ...保存插件配置时：
async def _save_plugin_configs(self, post_configs: dict, plugin_name: str):
# ... plugin_md lookup unchanged ...
try:
errors, post_configs = validate_config(
post_configs, getattr(md.config, "schema", {}), is_core=False
)
if errors:
raise ValueError(f"格式校验未通过: {errors}")
apply_plugin_file_ops(plugin_name, md, post_configs)
md.config.save_config(post_configs)
except Exception as e:
raise e这样既保留了原有行为，又能：
- 将 Schema/路径/文件相关逻辑集中在一个模块中；
- 让路由方法专注于 HTTP 相关处理，并委托给辅助模块；
- 方便未来只在
plugin_file_config.py中调整文件布局或迁移逻辑。
Original comment in English
issue (complexity): Consider extracting the new plugin file/config handling into a dedicated helper module so the route class stays lean and only orchestrates HTTP flow.
You can reduce the added complexity substantially by pulling the file/config logic into a focused helper and keeping the route class thin. This keeps all behavior but localizes the low‑level concerns.
1. Extract a PluginFileConfigManager
Move the helpers and core logic out of AstrBotConfig into a dedicated module, e.g. astrbot/core/config/plugin_file_config.py:
# astrbot/core/config/plugin_file_config.py
import os
from typing import Iterable
from astrbot.core.utils.astrbot_path import (
get_astrbot_plugin_data_path,
get_astrbot_temp_path,
)
from astrbot.core.utils.io import remove_dir
def sanitize_filename(name: str) -> str:
cleaned = os.path.basename(name).strip()
if not cleaned or cleaned in {".", ".."}:
return ""
for sep in (os.sep, os.altsep):
if sep:
cleaned = cleaned.replace(sep, "_")
return cleaned
def sanitize_path_segment(segment: str) -> str:
cleaned = []
for ch in segment:
if (
("a" <= ch <= "z")
or ("A" <= ch <= "Z")
or ch.isdigit()
or ch in {"-", "_"}
):
cleaned.append(ch)
else:
cleaned.append("_")
result = "".join(cleaned).strip("_")
return result or "_"
def config_key_to_folder(key_path: str) -> str:
parts = [sanitize_path_segment(p) for p in key_path.split(".") if p]
return "/".join(parts) if parts else "_"
def normalize_rel_path(rel_path: str | None) -> str | None:
if not isinstance(rel_path, str):
return None
rel = rel_path.replace("\\", "/").lstrip("/")
if not rel:
return None
parts = [p for p in rel.split("/") if p]
if any(part in {".", ".."} for part in parts):
return None
if rel.startswith("../") or "/../" in rel:
return None
return "/".join(parts)Keep the list / schema / fs logic separately in the same module:
def normalize_file_list(value, key_path: str) -> tuple[list[str], bool]:
if value is None:
return [], False
if not isinstance(value, list):
raise ValueError(f"Invalid file list for {key_path}")
folder = config_key_to_folder(key_path)
expected_prefix = f"files/{folder}/"
results: list[str] = []
changed = False
for item in value:
if not isinstance(item, str):
raise ValueError(f"Invalid file entry for {key_path}")
rel = normalize_rel_path(item)
if not rel or not rel.startswith("files/"):
raise ValueError(f"Invalid file path: {item}")
if rel.startswith(expected_prefix):
results.append(rel)
continue
if rel.count("/") == 1:
filename = rel.split("/", 1)[1]
if not filename:
raise ValueError(f"Invalid file path: {item}")
results.append(f"{expected_prefix}{filename}")
changed = True
continue
raise ValueError(f"Invalid file path: {item}")
return results, changed
def apply_plugin_file_ops(plugin_name: str, md, post_configs: dict) -> None:
schema = getattr(md.config, "schema", None) if md and md.config else None
if not isinstance(schema, dict):
return
# you can also move _collect_file_keys/_get_value_by_path/_set_value_by_path here
# ... existing logic moved verbatim from AstrBotConfig._apply_plugin_file_ops ...You can also move _collect_file_keys, _get_value_by_path, _set_value_by_path, and _get_schema_item into this module so all schema/path knowledge is in one place.
2. Thin the route class to orchestration only
Then AstrBotConfig becomes mostly orchestration, which is easier to read:
# config.py
from astrbot.core.config.plugin_file_config import (
sanitize_filename,
normalize_rel_path,
config_key_to_folder,
apply_plugin_file_ops,
)
async def upload_plugin_file(self):
plugin_name = request.args.get("plugin_name")
key_path = request.args.get("key")
if not plugin_name or not key_path:
return Response().error("Missing plugin_name or key parameter").__dict__
md = self._get_plugin_metadata_by_name(plugin_name)
if not md or not md.config:
return Response().error(
f"Plugin {plugin_name} not found or has no config",
).__dict__
meta = self._get_schema_item(md.config.schema, key_path)
if not meta or meta.get("type") != "file":
return Response().error("Config item not found or not file type").__dict__
file_types = meta.get("file_types")
allowed_exts = []
if isinstance(file_types, list):
allowed_exts = [str(ext).lstrip(".").lower() for ext in file_types if str(ext).strip()]
files = await request.files
if not files:
return Response().error("No files uploaded").__dict__
staging_root = os.path.join(
get_astrbot_temp_path(),
"plugin_file_uploads",
plugin_name,
)
os.makedirs(staging_root, exist_ok=True)
uploaded = []
folder = config_key_to_folder(key_path)
errors = []
for file in files.values():
filename = sanitize_filename(file.filename or "")
# ... unchanged logic using helpers from plugin_file_config ...
# ... unchanged response construction ...And for saving plugin configs:
async def _save_plugin_configs(self, post_configs: dict, plugin_name: str):
# ... plugin_md lookup unchanged ...
try:
errors, post_configs = validate_config(
post_configs, getattr(md.config, "schema", {}), is_core=False
)
if errors:
raise ValueError(f"格式校验未通过: {errors}")
apply_plugin_file_ops(plugin_name, md, post_configs)
md.config.save_config(post_configs)
except Exception as e:
raise eThis preserves behavior but:
- Concentrates schema/path/file logic into one module.
- Keeps route methods focused on HTTP concerns and delegating to the helper.
- Makes future changes to file layout or migration logic localized to
plugin_file_config.py.
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.
由于越来越多的插件需要上传文件，故本插件新增插件配置文件上传类型，支持多文件上传、类型限制、文件展示与删除，并采用“先暂存后保存”的流程，同时将文件按配置项分目录存放，提升安全性和原子性。
Modifications / 改动点
本pr是#2734 的重开，由于2734过于早，已经不适用当前版本，故进行了重构。
astrbot/dashboard/routes/config.py：新增 file 类型配置校验、文件上传/删除接口、保存时文件暂存迁移与按配置项分目录存储、旧路径迁移与清理。dashboard/src/components/shared/FileConfigItem.vue：新增文件配置项 UI（上传、拖拽、展示、删除、分页布局、固定大小滚动、完成按钮）。dashboard/src/components/shared/ConfigItemRenderer.vue/dashboard/src/components/shared/AstrBotConfig.vue/dashboard/src/views/ExtensionPage.vue：传递 pluginName 和 configKey 支持上传行为。dashboard/src/i18n/locales/zh-CN/features/config.json、dashboard/src/i18n/locales/en-US/features/config.json：补充 fileUpload i18n 文案。astrbot/core/config/default.py：注册 file 配置类型默认值。用于在插件配置中提供文件上传能力，支持多文件、拖拽上传、类型限制与文件管理。
配置示例
{
"demo_files": {
"type": "file",
"description": "Uploaded files for demo",
"default": [],
"file_types": ["pdf", "docx"]
}
}
参数说明
type: 固定为 file。
default: 默认值，推荐 []。
file_types: 可选。允许的文件扩展名列表（不带点）。如 ["pdf", "docx"]。不填写则不限制类型。
Screenshots or Test Results / 运行截图或测试结果
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.