fix: harden plugin dependency loading in frozen app runtime #5015
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
the
size:L
|
Related Documentation Checked 1 published document(s) in 1 knowledge base(s). No updates required. How did I do? Any feedback? |
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey - 我发现了 4 个问题，并留下了一些总体反馈：
- 在
_try_register_distlib_finder中，调用从register_finder(loader_type, resource_finder)改成了register_finder(loader, resource_finder)，但register_finder期望的是 loader 的类/类型而不是实例，这很可能会破坏原本补丁的意图，应该改回使用类型。 - 自定义 loader
_prefer_module_from_site_packages会手动修改sys.modules并从文件路径导入模块；如果导入可能并发发生（例如在异步任务或多线程中），最好用锁来保护这段逻辑，或以其他方式确保不会与其他同样在操作这些模块名的导入之间产生竞态条件。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_try_register_distlib_finder` the call was changed from `register_finder(loader_type, resource_finder)` to `register_finder(loader, resource_finder)`, but `register_finder` expects a loader class/type rather than an instance, so this likely breaks the intended patch and should be switched back to using the type.
- The custom loader `_prefer_module_from_site_packages` manually mutates `sys.modules` and imports modules from file paths; if imports can happen concurrently (e.g., in async tasks or threads) it would be safer to guard this with a lock or otherwise ensure no race conditions with other imports manipulating the same module names.
## Individual Comments
### Comment 1
<location> `astrbot/core/utils/pip_installer.py:79-88` </location>
<code_context>
+ @staticmethod
+ def _purge_conflicting_third_party_modules(prefixes: tuple[str, ...]) -> None:
+ for key in list(sys.modules.keys()):
+ if any(
+ key == prefix or key.startswith(f"{prefix}.") for prefix in prefixes
</code_context>
<issue_to_address>
**issue (bug_risk):** Avoid permanently clearing existing modules in failure paths when preferring site-packages versions.
In `_prefer_module_from_site_packages`, `sys.modules` entries for `module_name` (and submodules) are removed before confirming that the replacement can be loaded. If `spec_from_file_location` is `None`, lacks a loader, or `exec_module` fails, you return/fail with the original modules already purged, which can later break imports (e.g. `starlette`/`fastapi`). Please either defer the purge until after a successful load, or snapshot and restore the previous `sys.modules` entries if loading fails.
</issue_to_address>
### Comment 2
<location> `dashboard/src/views/ExtensionPage.vue:265-274` </location>
<code_context>
+ return 0;
+};
+
+const comparePluginVersions = (leftVersion, rightVersion) => {
+ const left = parsePluginVersion(leftVersion);
+ const right = parsePluginVersion(rightVersion);
+ if (!left || !right) return null;
+
+ const maxLength = Math.max(left.numeric.length, right.numeric.length);
+ for (let i = 0; i < maxLength; i++) {
+ const leftPart = left.numeric[i] ?? 0;
+ const rightPart = right.numeric[i] ?? 0;
+ if (leftPart === rightPart) continue;
+ return leftPart > rightPart ? 1 : -1;
+ }
+
+ return comparePrerelease(left.prerelease, right.prerelease);
+};
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Handling unparsable versions by returning `null` may regress update detection for non-semver versions.
Because `comparePluginVersions` now returns `null` when either version is unparsable, `has_update` becomes `false` where it previously became `true` whenever the versions differed and the online version wasn’t the "unknown" placeholder. This affects cases like a local non‑semver version ("dev", "local", git hash) vs a valid remote semver version: the UI used to show an update, but now won’t. If that’s not intended, consider falling back to a simpler comparison (e.g., string inequality or a heuristic) when parsing fails, or treating an unparsable local version as older than a valid remote version.
</issue_to_address>
### Comment 3
<location> `astrbot/core/utils/pip_installer.py:105-106` </location>
<code_context>
+ return True
+
+
+def _prefer_fastapi_related_modules(site_packages_path: str) -> None:
+ for module_name in ("starlette", "starlette.status"):
+ try:
+ loaded = _prefer_module_from_site_packages(module_name, site_packages_path)
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Consider aligning the set of preferred modules with the ones being purged in `StarManager._purge_conflicting_third_party_modules`.
This prefers only `"starlette"` and `"starlette.status"` from the plugin environment, but `_purge_conflicting_third_party_modules` removes `("starlette", "fastapi", "pydantic", "pydantic_core")` before re-importing. As a result, `fastapi`/`pydantic` may still come from the main environment while `starlette` is forced to the plugin environment. If version consistency across these packages is intended, consider including the other prefixes here as well, or clarify why only `starlette` is preferred.
```suggestion
def _prefer_fastapi_related_modules(site_packages_path: str) -> None:
for module_name in ("starlette", "starlette.status", "fastapi", "pydantic", "pydantic_core"):
```
</issue_to_address>
### Comment 4
<location> `astrbot/core/utils/pip_installer.py:53` </location>
<code_context>
handler.close()
+def _prepend_sys_path(path: str) -> None:
+ normalized_target = os.path.realpath(path)
+ sys.path[:] = [
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the sys.path handling and module preference logic by relying more on Python’s standard import/reload mechanisms instead of custom spec and sys.modules manipulation.
You can keep the “prefer plugin copy” behavior while leaning much more on Python’s import system and simplifying the helpers.
### 1. Simplify `_prepend_sys_path`
You don’t need to rebuild the entire `sys.path` on each call. A tighter version that removes only the first matching occurrence (after normalization) is easier to reason about and cheaper:
```python
def _prepend_sys_path(path: str) -> None:
normalized_target = os.path.realpath(path)
# Remove first existing occurrence only
for i, item in enumerate(sys.path):
if os.path.realpath(item) == normalized_target:
del sys.path[i]
break
sys.path.insert(0, normalized_target)
```
This keeps the de‑duplication behavior but avoids scanning+rebuilding the full list.
### 2. Replace custom import plumbing with `reload`
The current `_prefer_module_from_site_packages` re-implements a large part of the import mechanism (path building, `spec_from_file_location`, manual `sys.modules` sweeping, parent/child wiring). For the use case “prefer the version in this site-packages directory”, you can:
1. Ensure the plugin site-packages is at the front of `sys.path`.
2. Invalidate caches.
3. `reload()` the modules you care about.
That lets the standard import mechanism resolve from the new path ordering and avoids manual module wiring.
A replacement version could look like this:
```python
def _prefer_module_from_site_packages(module_name: str, site_packages_path: str) -> bool:
# Make sure plugin site-packages is first
_prepend_sys_path(site_packages_path)
importlib.invalidate_caches()
try:
# Import (or re-import) using normal import machinery
module = sys.modules.get(module_name)
if module is None:
module = importlib.import_module(module_name)
else:
module = importlib.reload(module)
except ModuleNotFoundError:
return False
logger.info(
"Using %s from plugin site-packages (sys.path[0]=%s)",
module_name,
sys.path[0],
)
return True
```
This preserves the “prefer plugin copy” semantics as long as the plugin site-packages is in front of any other locations that provide the same module.
### 3. Keep `_prefer_fastapi_related_modules` thin
With the simplified `_prefer_module_from_site_packages`, `_prefer_fastapi_related_modules` becomes a very small wrapper, without manual spec handling or `sys.modules` surgery:
```python
def _prefer_fastapi_related_modules(site_packages_path: str) -> None:
for module_name in ("starlette", "starlette.status"):
try:
if not _prefer_module_from_site_packages(module_name, site_packages_path):
logger.warning(
"Module %s not found in plugin site-packages: %s",
module_name,
site_packages_path,
)
except Exception as exc:
logger.warning(
"Failed to prefer module %s from plugin site-packages: %s",
module_name,
exc,
)
```
### 4. Reduce cognitive load in `install`
Once `_prefer_module_from_site_packages` uses normal imports, the `install` flow is easier to reason about:
- Prepend `target_site_packages` once before pip invocation (if necessary for pip).
- After installation, _optionally_ prepend again (with the cheaper `_prepend_sys_path`) and prefer modules, then invalidate caches.
Because `_prepend_sys_path` is idempotent and simple, the double call is cheap and obvious, and the “preference” step is a straightforward reload rather than a custom import path.
</issue_to_address>帮我变得更有用！请对每条评论点 👍 或 👎，我会根据你的反馈改进评审质量。
Original comment in English
Hey - I've found 4 issues, and left some high level feedback:
- In
_try_register_distlib_finderthe call was changed fromregister_finder(loader_type, resource_finder)toregister_finder(loader, resource_finder), butregister_finderexpects a loader class/type rather than an instance, so this likely breaks the intended patch and should be switched back to using the type. - The custom loader
_prefer_module_from_site_packagesmanually mutatessys.modulesand imports modules from file paths; if imports can happen concurrently (e.g., in async tasks or threads) it would be safer to guard this with a lock or otherwise ensure no race conditions with other imports manipulating the same module names.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_try_register_distlib_finder` the call was changed from `register_finder(loader_type, resource_finder)` to `register_finder(loader, resource_finder)`, but `register_finder` expects a loader class/type rather than an instance, so this likely breaks the intended patch and should be switched back to using the type.
- The custom loader `_prefer_module_from_site_packages` manually mutates `sys.modules` and imports modules from file paths; if imports can happen concurrently (e.g., in async tasks or threads) it would be safer to guard this with a lock or otherwise ensure no race conditions with other imports manipulating the same module names.
## Individual Comments
### Comment 1
<location> `astrbot/core/utils/pip_installer.py:79-88` </location>
<code_context>
+ @staticmethod
+ def _purge_conflicting_third_party_modules(prefixes: tuple[str, ...]) -> None:
+ for key in list(sys.modules.keys()):
+ if any(
+ key == prefix or key.startswith(f"{prefix}.") for prefix in prefixes
</code_context>
<issue_to_address>
**issue (bug_risk):** Avoid permanently clearing existing modules in failure paths when preferring site-packages versions.
In `_prefer_module_from_site_packages`, `sys.modules` entries for `module_name` (and submodules) are removed before confirming that the replacement can be loaded. If `spec_from_file_location` is `None`, lacks a loader, or `exec_module` fails, you return/fail with the original modules already purged, which can later break imports (e.g. `starlette`/`fastapi`). Please either defer the purge until after a successful load, or snapshot and restore the previous `sys.modules` entries if loading fails.
</issue_to_address>
### Comment 2
<location> `dashboard/src/views/ExtensionPage.vue:265-274` </location>
<code_context>
+ return 0;
+};
+
+const comparePluginVersions = (leftVersion, rightVersion) => {
+ const left = parsePluginVersion(leftVersion);
+ const right = parsePluginVersion(rightVersion);
+ if (!left || !right) return null;
+
+ const maxLength = Math.max(left.numeric.length, right.numeric.length);
+ for (let i = 0; i < maxLength; i++) {
+ const leftPart = left.numeric[i] ?? 0;
+ const rightPart = right.numeric[i] ?? 0;
+ if (leftPart === rightPart) continue;
+ return leftPart > rightPart ? 1 : -1;
+ }
+
+ return comparePrerelease(left.prerelease, right.prerelease);
+};
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Handling unparsable versions by returning `null` may regress update detection for non-semver versions.
Because `comparePluginVersions` now returns `null` when either version is unparsable, `has_update` becomes `false` where it previously became `true` whenever the versions differed and the online version wasn’t the "unknown" placeholder. This affects cases like a local non‑semver version ("dev", "local", git hash) vs a valid remote semver version: the UI used to show an update, but now won’t. If that’s not intended, consider falling back to a simpler comparison (e.g., string inequality or a heuristic) when parsing fails, or treating an unparsable local version as older than a valid remote version.
</issue_to_address>
### Comment 3
<location> `astrbot/core/utils/pip_installer.py:105-106` </location>
<code_context>
+ return True
+
+
+def _prefer_fastapi_related_modules(site_packages_path: str) -> None:
+ for module_name in ("starlette", "starlette.status"):
+ try:
+ loaded = _prefer_module_from_site_packages(module_name, site_packages_path)
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Consider aligning the set of preferred modules with the ones being purged in `StarManager._purge_conflicting_third_party_modules`.
This prefers only `"starlette"` and `"starlette.status"` from the plugin environment, but `_purge_conflicting_third_party_modules` removes `("starlette", "fastapi", "pydantic", "pydantic_core")` before re-importing. As a result, `fastapi`/`pydantic` may still come from the main environment while `starlette` is forced to the plugin environment. If version consistency across these packages is intended, consider including the other prefixes here as well, or clarify why only `starlette` is preferred.
```suggestion
def _prefer_fastapi_related_modules(site_packages_path: str) -> None:
for module_name in ("starlette", "starlette.status", "fastapi", "pydantic", "pydantic_core"):
```
</issue_to_address>
### Comment 4
<location> `astrbot/core/utils/pip_installer.py:53` </location>
<code_context>
handler.close()
+def _prepend_sys_path(path: str) -> None:
+ normalized_target = os.path.realpath(path)
+ sys.path[:] = [
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the sys.path handling and module preference logic by relying more on Python’s standard import/reload mechanisms instead of custom spec and sys.modules manipulation.
You can keep the “prefer plugin copy” behavior while leaning much more on Python’s import system and simplifying the helpers.
### 1. Simplify `_prepend_sys_path`
You don’t need to rebuild the entire `sys.path` on each call. A tighter version that removes only the first matching occurrence (after normalization) is easier to reason about and cheaper:
```python
def _prepend_sys_path(path: str) -> None:
normalized_target = os.path.realpath(path)
# Remove first existing occurrence only
for i, item in enumerate(sys.path):
if os.path.realpath(item) == normalized_target:
del sys.path[i]
break
sys.path.insert(0, normalized_target)
```
This keeps the de‑duplication behavior but avoids scanning+rebuilding the full list.
### 2. Replace custom import plumbing with `reload`
The current `_prefer_module_from_site_packages` re-implements a large part of the import mechanism (path building, `spec_from_file_location`, manual `sys.modules` sweeping, parent/child wiring). For the use case “prefer the version in this site-packages directory”, you can:
1. Ensure the plugin site-packages is at the front of `sys.path`.
2. Invalidate caches.
3. `reload()` the modules you care about.
That lets the standard import mechanism resolve from the new path ordering and avoids manual module wiring.
A replacement version could look like this:
```python
def _prefer_module_from_site_packages(module_name: str, site_packages_path: str) -> bool:
# Make sure plugin site-packages is first
_prepend_sys_path(site_packages_path)
importlib.invalidate_caches()
try:
# Import (or re-import) using normal import machinery
module = sys.modules.get(module_name)
if module is None:
module = importlib.import_module(module_name)
else:
module = importlib.reload(module)
except ModuleNotFoundError:
return False
logger.info(
"Using %s from plugin site-packages (sys.path[0]=%s)",
module_name,
sys.path[0],
)
return True
```
This preserves the “prefer plugin copy” semantics as long as the plugin site-packages is in front of any other locations that provide the same module.
### 3. Keep `_prefer_fastapi_related_modules` thin
With the simplified `_prefer_module_from_site_packages`, `_prefer_fastapi_related_modules` becomes a very small wrapper, without manual spec handling or `sys.modules` surgery:
```python
def _prefer_fastapi_related_modules(site_packages_path: str) -> None:
for module_name in ("starlette", "starlette.status"):
try:
if not _prefer_module_from_site_packages(module_name, site_packages_path):
logger.warning(
"Module %s not found in plugin site-packages: %s",
module_name,
site_packages_path,
)
except Exception as exc:
logger.warning(
"Failed to prefer module %s from plugin site-packages: %s",
module_name,
exc,
)
```
### 4. Reduce cognitive load in `install`
Once `_prefer_module_from_site_packages` uses normal imports, the `install` flow is easier to reason about:
- Prepend `target_site_packages` once before pip invocation (if necessary for pip).
- After installation, _optionally_ prepend again (with the cheaper `_prepend_sys_path`) and prefer modules, then invalidate caches.
Because `_prepend_sys_path` is idempotent and simple, the double call is cheap and obvious, and the “preference” step is a straightforward reload rather than a custom import path.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| handler.close() | ||
|
|
||
|
|
||
| def _prepend_sys_path(path: str) -> None: |
Contributor
There was a problem hiding this comment.
issue (complexity): 建议尽量依赖 Python 标准的导入/重载机制，而不是自定义 spec 和 sys.modules 操作，从而简化 sys.path 处理和模块优先级逻辑。
你可以在保持“优先使用插件副本”语义的同时，更多地利用 Python 的导入系统，从而简化这些辅助函数。
1. 简化 _prepend_sys_path
你不需要在每次调用时重建整个 sys.path。一个更紧凑的版本只移除第一次匹配到的（归一化后）条目，更好理解且开销更小：
def _prepend_sys_path(path: str) -> None:
normalized_target = os.path.realpath(path)
# 仅移除第一次出现的已有条目
for i, item in enumerate(sys.path):
if os.path.realpath(item) == normalized_target:
del sys.path[i]
break
sys.path.insert(0, normalized_target)这样既保持了去重行为，又避免了扫描+重建整个列表。
2. 用 reload 替换自定义导入管线
当前的 _prefer_module_from_site_packages 重新实现了导入机制中的大量逻辑（路径构造、spec_from_file_location、手动清理 sys.modules、父/子模块关系维护）。对于“优先使用这个 site-packages 目录中的版本”这一需求，你可以：
- 确保插件的 site-packages 在
sys.path的最前面； - 失效导入缓存；
- 对关心的模块调用
reload()。
这样就可以让标准导入机制根据新的路径顺序解析模块，避免手动维护模块关系。
一个替代实现可以是：
def _prefer_module_from_site_packages(module_name: str, site_packages_path: str) -> bool:
# 确保插件的 site-packages 在最前
_prepend_sys_path(site_packages_path)
importlib.invalidate_caches()
try:
# 使用正常导入机制导入（或重新导入）
module = sys.modules.get(module_name)
if module is None:
module = importlib.import_module(module_name)
else:
module = importlib.reload(module)
except ModuleNotFoundError:
return False
logger.info(
"Using %s from plugin site-packages (sys.path[0]=%s)",
module_name,
sys.path[0],
)
return True只要插件的 site-packages 排在其他同样提供该模块的位置之前，就能保持“优先使用插件副本”的语义。
3. 让 _prefer_fastapi_related_modules 更薄
在简化后的 _prefer_module_from_site_packages 基础上，_prefer_fastapi_related_modules 可以变成一个非常小的包装函数，而不需要手动处理 spec 或 sys.modules：
def _prefer_fastapi_related_modules(site_packages_path: str) -> None:
for module_name in ("starlette", "starlette.status"):
try:
if not _prefer_module_from_site_packages(module_name, site_packages_path):
logger.warning(
"Module %s not found in plugin site-packages: %s",
module_name,
site_packages_path,
)
except Exception as exc:
logger.warning(
"Failed to prefer module %s from plugin site-packages: %s",
module_name,
exc,
)4. 降低 install 的理解负担
一旦 _prefer_module_from_site_packages 改为使用正常导入机制，install 流程就更容易理解：
- 在调用 pip 之前（如果 pip 需要）先将
target_site_packages追加到前面； - 安装完成后，可以（使用更廉价的
_prepend_sys_path）再追加一次并执行“优先模块”逻辑，然后失效缓存。
由于 _prepend_sys_path 是幂等且简单的，因此重复调用成本很低且语义明确，“优先使用”步骤只是一次直观的 reload，而不是自定义导入路径。
Original comment in English
issue (complexity): Consider simplifying the sys.path handling and module preference logic by relying more on Python’s standard import/reload mechanisms instead of custom spec and sys.modules manipulation.
You can keep the “prefer plugin copy” behavior while leaning much more on Python’s import system and simplifying the helpers.
1. Simplify _prepend_sys_path
You don’t need to rebuild the entire sys.path on each call. A tighter version that removes only the first matching occurrence (after normalization) is easier to reason about and cheaper:
def _prepend_sys_path(path: str) -> None:
normalized_target = os.path.realpath(path)
# Remove first existing occurrence only
for i, item in enumerate(sys.path):
if os.path.realpath(item) == normalized_target:
del sys.path[i]
break
sys.path.insert(0, normalized_target)This keeps the de‑duplication behavior but avoids scanning+rebuilding the full list.
2. Replace custom import plumbing with reload
The current _prefer_module_from_site_packages re-implements a large part of the import mechanism (path building, spec_from_file_location, manual sys.modules sweeping, parent/child wiring). For the use case “prefer the version in this site-packages directory”, you can:
- Ensure the plugin site-packages is at the front of
sys.path. - Invalidate caches.
reload()the modules you care about.
That lets the standard import mechanism resolve from the new path ordering and avoids manual module wiring.
A replacement version could look like this:
def _prefer_module_from_site_packages(module_name: str, site_packages_path: str) -> bool:
# Make sure plugin site-packages is first
_prepend_sys_path(site_packages_path)
importlib.invalidate_caches()
try:
# Import (or re-import) using normal import machinery
module = sys.modules.get(module_name)
if module is None:
module = importlib.import_module(module_name)
else:
module = importlib.reload(module)
except ModuleNotFoundError:
return False
logger.info(
"Using %s from plugin site-packages (sys.path[0]=%s)",
module_name,
sys.path[0],
)
return TrueThis preserves the “prefer plugin copy” semantics as long as the plugin site-packages is in front of any other locations that provide the same module.
3. Keep _prefer_fastapi_related_modules thin
With the simplified _prefer_module_from_site_packages, _prefer_fastapi_related_modules becomes a very small wrapper, without manual spec handling or sys.modules surgery:
def _prefer_fastapi_related_modules(site_packages_path: str) -> None:
for module_name in ("starlette", "starlette.status"):
try:
if not _prefer_module_from_site_packages(module_name, site_packages_path):
logger.warning(
"Module %s not found in plugin site-packages: %s",
module_name,
site_packages_path,
)
except Exception as exc:
logger.warning(
"Failed to prefer module %s from plugin site-packages: %s",
module_name,
exc,
)4. Reduce cognitive load in install
Once _prefer_module_from_site_packages uses normal imports, the install flow is easier to reason about:
- Prepend
target_site_packagesonce before pip invocation (if necessary for pip). - After installation, optionally prepend again (with the cheaper
_prepend_sys_path) and prefer modules, then invalidate caches.
Because _prepend_sys_path is idempotent and simple, the double call is cheap and obvious, and the “preference” step is a straightforward reload rather than a custom import path.
dosubot
bot
added
the
area:core
zouyonghe
changed the title
Collaborator
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - 我发现了 1 个问题，并给出了一些整体性的反馈：
- 在
_try_register_distlib_finder中，现在是用loader而不是loader_type调用register_finder；但distlib.resources.register_finder期望传入的是 loader 的类型（类），因此建议传递loader_type/type(loader)，以避免只针对单个实例注册 finder。 - 新增的版本解析/比较辅助函数（
parsePluginVersion、comparePluginVersions等）给ExtensionPage.vue增加了不少逻辑；建议将这些逻辑抽取到一个共享的工具模块中，这样视图组件可以更专注于 UI 相关的逻辑，同时这些逻辑也便于复用或在隔离环境中进行单元测试。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_try_register_distlib_finder`, `register_finder` is now called with `loader` instead of `loader_type`; `distlib.resources.register_finder` expects a loader *type* (class), so consider passing `loader_type`/`type(loader)` to avoid registering the finder against a single instance.
- The new version parsing/comparison helpers (`parsePluginVersion`, `comparePluginVersions`, etc.) add quite a bit of logic to `ExtensionPage.vue`; consider extracting them into a shared utility module so the view stays focused on UI concerns and the logic can be reused or unit-tested in isolation.
## Individual Comments
### Comment 1
<location> `astrbot/core/utils/pip_installer.py:173` </location>
<code_context>
+ return dist_infos
+
+
+def _collect_candidate_modules(
+ requirement_names: set[str],
+ site_packages_path: str,
</code_context>
<issue_to_address>
**issue (complexity):** 考虑将新的依赖解析和模块优先级相关的辅助函数收敛为少数几个更高层级的函数，这些函数更多依赖标准库的元数据/导入机制，从而减少自定义逻辑并降低理解成本。
你可以保持当前的行为不变，但通过把零散的辅助函数合并为几个更高层的操作，并更多依赖标准库的导入机制，来减少复杂度。
### 1. 使用 `importlib.metadata` 收敛“依赖 → 模块”的解析流程
`_find_dist_info_paths`、`_extract_top_level_modules` 以及 `_collect_candidate_modules` 的部分逻辑可以用一层简单封装的 `importlib.metadata.distributions` 来替代：
```python
import importlib.metadata as im
def _collect_candidate_modules(
requirement_names: set[str],
site_packages_path: str,
) -> set[str]:
candidates: set[str] = set()
dists = list(im.distributions(path=[site_packages_path]))
# build index once instead of scanning filesystem per requirement
by_name: dict[str, list[im.Distribution]] = {}
for dist in dists:
name = _canonicalize_distribution_name(dist.metadata["Name"])
by_name.setdefault(name, []).append(dist)
for req in requirement_names:
normalized = _canonicalize_distribution_name(req)
for dist in by_name.get(normalized, []):
try:
text = dist.read_text("top_level.txt") or ""
except FileNotFoundError:
text = ""
for line in text.splitlines():
mod = line.strip()
if mod and not mod.startswith("#"):
candidates.add(mod)
# fallback if no matching dist or no top_level.txt
if not by_name.get(normalized):
candidates.add(normalized.replace("-", "_"))
return candidates
```
这允许你删除：
- `_find_dist_info_paths`
- `_extract_top_level_modules`
- 原始 `_collect_candidate_modules` 中与 dist-info 扫描相关的部分
并且去掉手动的 `os.listdir` / 路径拼接逻辑。
### 2. 将模块“优先加载”简化为 `sys.path` + 清理 `sys.modules` + `import_module`
你可以通过使用正常的导入语义来避免在 `_prefer_module_from_site_packages` / `_prefer_modules_from_site_packages` 中使用 `spec_from_file_location`、手动创建模块、手动设置父模块以及 RLock：
```python
def _prefer_modules_from_site_packages(
module_names: set[str],
site_packages_path: str,
) -> dict[str, bool]:
_prepend_sys_path(site_packages_path)
results: dict[str, bool] = {}
for name in sorted(module_names):
base = name.split(".", 1)[0]
# Clear potentially conflicting modules (including submodules)
for key in list(sys.modules.keys()):
if key == base or key.startswith(f"{base}."):
sys.modules.pop(key, None)
try:
importlib.import_module(name)
results[name] = True
except Exception as exc:
logger.warning(
"Failed to import preferred module %s from %s: %s",
name,
site_packages_path,
exc,
)
results[name] = False
return results
```
这样可以保留优先从插件 `site-packages` 加载的行为（因为你已经把它加到了 `sys.path` 前面），但可以移除：
- `_SITE_PACKAGES_IMPORT_LOCK`
- 手动 spec / loader 处理
- 显式的父模块属性绑定
如果你在其它地方已经有通用的 `_purge_modules_by_prefix` / `_purge_conflicting_third_party_modules`，可以在这里复用，而不是内联 `for key in list(sys.modules.keys())` 这一段逻辑。
### 3. 将校验折叠进优先加载步骤
与其在 `_validate_preferred_module_sources` 中做第二次遍历，你可以根据模块实际是从哪里加载的进行即时校验：
```python
def _ensure_preferred_modules(
module_names: set[str],
site_packages_path: str,
) -> None:
_prepend_sys_path(site_packages_path)
unresolved: list[str] = []
for name in sorted(module_names):
# purge and import as above
for key in list(sys.modules.keys()):
if key == name or key.startswith(f"{name}."):
sys.modules.pop(key, None)
try:
mod = importlib.import_module(name)
except Exception as exc:
logger.warning(
"Failed to import preferred module %s from %s: %s",
name,
site_packages_path,
exc,
)
continue
module_file = getattr(mod, "__file__", None)
if not module_file:
continue
module_path = os.path.realpath(module_file)
site_root = os.path.realpath(site_packages_path)
try:
if os.path.commonpath([module_path, site_root]) != site_root:
unresolved.append(f"{name} -> {module_file}")
except ValueError:
unresolved.append(f"{name} -> {module_file}")
if unresolved:
raise RuntimeError(
"检测到插件依赖与当前运行时发生冲突，无法安全加载该插件。"
f"冲突模块: {', '.join(unresolved)}"
)
```
这样，`_prefer_modules_from_site_packages` 和 `_validate_preferred_module_sources` 就可以合并为一个内聚的操作（`_ensure_preferred_modules`）。
### 4. 将安装后的处理封装为单一辅助函数
通过把“插件 site-packages 优先加载 + 冲突检测”封装到一个命名清晰的函数中，可以让 `install` 方法尾部的逻辑更易读：
```python
def _ensure_plugin_dependencies_preferred(
target_site_packages: str,
requested_requirements: set[str],
) -> None:
if not requested_requirements:
return
candidate_modules = _collect_candidate_modules(
requested_requirements,
target_site_packages,
)
if not candidate_modules:
return
_ensure_preferred_modules(candidate_modules, target_site_packages)
```
然后 `install` 就能保持简洁，且其保证的语义也更清晰：
```python
if target_site_packages:
_prepend_sys_path(target_site_packages)
_ensure_plugin_dependencies_preferred(
target_site_packages,
requested_requirements,
)
importlib.invalidate_caches()
```
这样调用方就不需要分别了解“候选模块、优先加载和校验”这些独立概念，从而降低理解成本，同时保持新的冲突预防行为不变。
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进评审质量。
Original comment in English
Hey - I've found 1 issue, and left some high level feedback:
- In
_try_register_distlib_finder,register_finderis now called withloaderinstead ofloader_type;distlib.resources.register_finderexpects a loader type (class), so consider passingloader_type/type(loader)to avoid registering the finder against a single instance. - The new version parsing/comparison helpers (
parsePluginVersion,comparePluginVersions, etc.) add quite a bit of logic toExtensionPage.vue; consider extracting them into a shared utility module so the view stays focused on UI concerns and the logic can be reused or unit-tested in isolation.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_try_register_distlib_finder`, `register_finder` is now called with `loader` instead of `loader_type`; `distlib.resources.register_finder` expects a loader *type* (class), so consider passing `loader_type`/`type(loader)` to avoid registering the finder against a single instance.
- The new version parsing/comparison helpers (`parsePluginVersion`, `comparePluginVersions`, etc.) add quite a bit of logic to `ExtensionPage.vue`; consider extracting them into a shared utility module so the view stays focused on UI concerns and the logic can be reused or unit-tested in isolation.
## Individual Comments
### Comment 1
<location> `astrbot/core/utils/pip_installer.py:173` </location>
<code_context>
+ return dist_infos
+
+
+def _collect_candidate_modules(
+ requirement_names: set[str],
+ site_packages_path: str,
</code_context>
<issue_to_address>
**issue (complexity):** Consider consolidating the new dependency-resolution and module-preference helpers into a few higher-level functions that rely more on stdlib metadata/import mechanisms to reduce custom logic and cognitive load.
You can keep the new behavior but reduce complexity by collapsing the fragmented helpers into a few higher‑level operations and leaning on stdlib import machinery.
### 1. Collapse requirement→module resolution using `importlib.metadata`
`_find_dist_info_paths`, `_extract_top_level_modules`, and part of `_collect_candidate_modules` can be replaced with a small wrapper around `importlib.metadata.distributions`:
```python
import importlib.metadata as im
def _collect_candidate_modules(
requirement_names: set[str],
site_packages_path: str,
) -> set[str]:
candidates: set[str] = set()
dists = list(im.distributions(path=[site_packages_path]))
# build index once instead of scanning filesystem per requirement
by_name: dict[str, list[im.Distribution]] = {}
for dist in dists:
name = _canonicalize_distribution_name(dist.metadata["Name"])
by_name.setdefault(name, []).append(dist)
for req in requirement_names:
normalized = _canonicalize_distribution_name(req)
for dist in by_name.get(normalized, []):
try:
text = dist.read_text("top_level.txt") or ""
except FileNotFoundError:
text = ""
for line in text.splitlines():
mod = line.strip()
if mod and not mod.startswith("#"):
candidates.add(mod)
# fallback if no matching dist or no top_level.txt
if not by_name.get(normalized):
candidates.add(normalized.replace("-", "_"))
return candidates
```
This lets you delete:
- `_find_dist_info_paths`
- `_extract_top_level_modules`
- the dist‑info scanning portion of the original `_collect_candidate_modules`
and removes manual `os.listdir` / path logic.
### 2. Simplify module “preference” to `sys.path` + `sys.modules` purge + `import_module`
You can avoid `spec_from_file_location`, manual module creation, parent wiring, and the RLock in `_prefer_module_from_site_packages` / `_prefer_modules_from_site_packages` by using normal import semantics:
```python
def _prefer_modules_from_site_packages(
module_names: set[str],
site_packages_path: str,
) -> dict[str, bool]:
_prepend_sys_path(site_packages_path)
results: dict[str, bool] = {}
for name in sorted(module_names):
base = name.split(".", 1)[0]
# Clear potentially conflicting modules (including submodules)
for key in list(sys.modules.keys()):
if key == base or key.startswith(f"{base}."):
sys.modules.pop(key, None)
try:
importlib.import_module(name)
results[name] = True
except Exception as exc:
logger.warning(
"Failed to import preferred module %s from %s: %s",
name,
site_packages_path,
exc,
)
results[name] = False
return results
```
This preserves the behavior of preferring plugin `site-packages` (because you already prepended it to `sys.path`), but removes:
- `_SITE_PACKAGES_IMPORT_LOCK`
- manual spec / loader handling
- explicit parent attribute wiring
If you already have a generic `_purge_modules_by_prefix` / `_purge_conflicting_third_party_modules` elsewhere, reuse it here instead of inlining the `for key in list(sys.modules.keys())` logic.
### 3. Fold validation into the preference step
Instead of a second pass in `_validate_preferred_module_sources`, you can validate immediately based on where the module actually loaded from:
```python
def _ensure_preferred_modules(
module_names: set[str],
site_packages_path: str,
) -> None:
_prepend_sys_path(site_packages_path)
unresolved: list[str] = []
for name in sorted(module_names):
# purge and import as above
for key in list(sys.modules.keys()):
if key == name or key.startswith(f"{name}."):
sys.modules.pop(key, None)
try:
mod = importlib.import_module(name)
except Exception as exc:
logger.warning(
"Failed to import preferred module %s from %s: %s",
name,
site_packages_path,
exc,
)
continue
module_file = getattr(mod, "__file__", None)
if not module_file:
continue
module_path = os.path.realpath(module_file)
site_root = os.path.realpath(site_packages_path)
try:
if os.path.commonpath([module_path, site_root]) != site_root:
unresolved.append(f"{name} -> {module_file}")
except ValueError:
unresolved.append(f"{name} -> {module_file}")
if unresolved:
raise RuntimeError(
"检测到插件依赖与当前运行时发生冲突，无法安全加载该插件。"
f"冲突模块: {', '.join(unresolved)}"
)
```
With this, `_prefer_modules_from_site_packages` and `_validate_preferred_module_sources` can be merged into one cohesive operation (`_ensure_preferred_modules`).
### 4. Encapsulate post‑install handling into a single helper
The `install` method’s tail can be made more readable by wrapping all the “plugin site‑packages preference + conflict detection” into one clearly named function:
```python
def _ensure_plugin_dependencies_preferred(
target_site_packages: str,
requested_requirements: set[str],
) -> None:
if not requested_requirements:
return
candidate_modules = _collect_candidate_modules(
requested_requirements,
target_site_packages,
)
if not candidate_modules:
return
_ensure_preferred_modules(candidate_modules, target_site_packages)
```
Then `install` remains simple and the guarantees are obvious:
```python
if target_site_packages:
_prepend_sys_path(target_site_packages)
_ensure_plugin_dependencies_preferred(
target_site_packages,
requested_requirements,
)
importlib.invalidate_caches()
```
This removes the need for the call‑site to know about candidates, preference, and validation as separate concepts and lowers cognitive load while keeping the new conflict‑prevention behavior intact.
</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
size:XL
Collaborator
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
嗨，我发现了 2 个问题，并留下了一些整体性的反馈：
- 在
pip_installer.py中，_extract_requirement_names会完全跳过-r/--requirement的 include，这意味着依赖优先级/冲突检测会漏掉 include 文件中定义的传递性依赖；建议跟踪这些 include（加上递归保护），这样_ensure_plugin_dependencies_preferred的逻辑才能获得完整的依赖视图。 pip_installer.py中新增的模块优先级辅助函数（_collect_candidate_modules/_prefer_modules_from_site_packages/_ensure_preferred_modules）在单个工具文件里引入了大量逻辑；将这部分抽到一个独立模块（例如plugin_site_packages.py）会让 pip 包装器本身更易理解，并在推理进程内 pip 与 import 行为时降低认知负担。
给 AI Agent 的提示
Please address the comments from this code review:
## Overall Comments
- In `pip_installer.py`, `_extract_requirement_names` skips `-r/--requirement` includes entirely, which means dependency-preference/conflict detection will miss transitive requirements defined in included files; consider following those includes (with a recursion guard) so your `_ensure_plugin_dependencies_preferred` logic has a complete view of dependencies.
- The new module-preference helpers in `pip_installer.py` (_collect_candidate_modules / _prefer_modules_from_site_packages / _ensure_preferred_modules) introduce a lot of logic into a single utility file; extracting them into a dedicated module (e.g. `plugin_site_packages.py`) would make the pip wrapper itself easier to follow and reduce the cognitive load when reasoning about in-process pip vs. import behavior.
## Individual Comments
### Comment 1
<location> `astrbot/core/utils/pip_installer.py:233` </location>
<code_context>
+ raise RuntimeError(conflict_message)
+
+
+def _prefer_module_from_site_packages(
+ module_name: str, site_packages_path: str
+) -> bool:
</code_context>
<issue_to_address>
**issue (complexity):** 建议通过复用正常的导入系统、限制依赖“恢复”的范围，并避免做完整的传递依赖展开，来简化新的插件导入和依赖处理逻辑。
你可以在保留当前行为（优先使用插件的 site-packages、检测冲突）的同时，大幅削减复杂度：更多依赖标准导入系统，并收窄“恢复”逻辑的作用范围。
### 1. 简化 `_prefer_module_from_site_packages`（不再手动构建 spec）
你已经通过 `_prepend_sys_path` 确保插件的 `site-packages` 优先出现在 `sys.path` 中。这意味着你可以避免自定义 spec 构造以及大部分对 `sys.modules` 的复杂操作：
```python
def _prefer_module_from_site_packages(
module_name: str,
site_packages_path: str,
) -> bool:
with _SITE_PACKAGES_IMPORT_LOCK:
if not _module_exists_in_site_packages(module_name, site_packages_path):
return False
# Snapshot original modules for rollback
matched_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
original_modules = {key: sys.modules[key] for key in matched_keys}
try:
# Drop existing module(s), then import normally (sys.path is ordered)
for key in matched_keys:
sys.modules.pop(key, None)
module = importlib.import_module(module_name)
# Ensure parent attribute is set, if applicable
if "." in module_name:
parent_name, child_name = module_name.rsplit(".", 1)
parent = sys.modules.get(parent_name)
if parent is not None:
setattr(parent, child_name, module)
logger.info(
"Loaded %s via normal import from plugin site-packages", module_name
)
return True
except Exception:
# Roll back on failure
for key in list(sys.modules.keys()):
if key == module_name or key.startswith(f"{module_name}."):
sys.modules.pop(key, None)
sys.modules.update(original_modules)
raise
```
这样仍然可以得到：
- 从插件路径确定性地重新导入（因为你提前将 `target_site_packages` 放在 `sys.path` 前面），
- 导入失败时可以回滚，
- 不再需要自定义 `spec_from_file_location` 或手动执行 loader。
### 2. 收窄依赖“恢复”和重试逻辑
多轮重试（`max_rounds`、`pending_modules`、`round_progress`）以及基于正则的 `_extract_conflicting_module_name` 会引入大量状态，并且依赖错误信息的具体格式。
你可以将其简化为：
- 只在 `ModuleNotFoundError` 上做恢复，并使用其 `.name` 属性，
- 不再使用全局多轮调度，只对每个依赖做一次本地尝试。
例如：
```python
def _extract_conflicting_module_name(exc: Exception) -> str | None:
if isinstance(exc, ModuleNotFoundError):
name = getattr(exc, "name", None)
if name:
return name.split(".", 1)[0]
return None
```
然后：
```python
def _prefer_module_with_dependency_recovery(
module_name: str,
site_packages_path: str,
) -> bool:
try:
return _prefer_module_from_site_packages(module_name, site_packages_path)
except Exception as exc:
dependency_name = _extract_conflicting_module_name(exc)
if not dependency_name or dependency_name == module_name:
raise
# Try once to prefer the missing dependency, then retry module
if not _prefer_module_from_site_packages(dependency_name, site_packages_path):
raise
logger.info(
"Recovered dependency %s while preferring %s from plugin site-packages.",
dependency_name,
module_name,
)
return _prefer_module_from_site_packages(module_name, site_packages_path)
```
这样 `_prefer_modules_from_site_packages` 就可以简化为单次遍历（不再需要多轮和 pending 列表）：
```python
def _prefer_modules_from_site_packages(
module_names: set[str],
site_packages_path: str,
) -> dict[str, str]:
unresolved: dict[str, str] = {}
for module_name in sorted(module_names):
try:
loaded = _prefer_module_with_dependency_recovery(
module_name,
site_packages_path,
)
if not loaded:
logger.debug(
"Module %s not found in plugin site-packages: %s",
module_name,
site_packages_path,
)
except Exception as exc:
unresolved[module_name] = str(exc)
for module_name, reason in unresolved.items():
logger.warning(
"Failed to prefer module %s from plugin site-packages: %s",
module_name,
reason,
)
return unresolved
```
这样可以保留：
- 对简单缺失导入场景的依赖“恢复”，
- 面向每个模块的错误上报，
但去掉多轮调度和基于错误信息的正则解析。
### 3. 将依赖图扩展限制为直接依赖
当前 `_collect_candidate_modules` 会在 `importlib_metadata.distributions` 上构建完整的传递闭包。对于大多数插件场景，只优先处理*直接*依赖通常已经足够，并且更容易理解。
你可以保留基于元数据的解析逻辑，但提供一个更简单的路径来跳过对 `requires` 的 BFS：
```python
def _collect_candidate_modules(
requirement_names: set[str],
site_packages_path: str,
*,
include_transitive: bool = False,
) -> set[str]:
by_name: dict[str, list[importlib_metadata.Distribution]] = {}
try:
for dist in importlib_metadata.distributions(path=[site_packages_path]):
name = dist.metadata.get("Name")
if not name:
continue
canonical = _canonicalize_distribution_name(name)
by_name.setdefault(canonical, []).append(dist)
except Exception as exc:
logger.warning("读取 site-packages 元数据失败，使用回退模块名: %s", exc)
if include_transitive:
expanded = set()
pending = deque(requirement_names)
while pending:
req = pending.popleft()
if req in expanded:
continue
expanded.add(req)
for dist in by_name.get(req, []):
for dep_line in dist.requires or []:
dep_name = _extract_requirement_name(dep_line)
if dep_name and dep_name not in expanded:
pending.append(dep_name)
requirement_names = expanded
candidates: set[str] = set()
for req in requirement_names:
modules_for_req: set[str] = set()
for dist in by_name.get(req, []):
modules_for_req.update(_extract_top_level_modules(dist))
if modules_for_req:
candidates.update(modules_for_req)
else:
candidates.add(req.replace("-", "_"))
return candidates
```
然后 `_ensure_plugin_dependencies_preferred` 可以通过 `include_transitive=False` 调用它（或将此设为默认）：
```python
candidate_modules = _collect_candidate_modules(
requested_requirements,
target_site_packages,
include_transitive=False,
)
```
这样既保留了未来扩展为完整闭包的能力，又让默认行为更简单、更可预测。
---
这些聚焦的改动：
- 依然保证优先使用由插件安装的依赖，
- 依然能检测未解决的冲突，
- 但去掉了大量底层 import 机制细节和多轮状态处理，使整体行为更容易理解和维护。
</issue_to_address>
### Comment 2
<location> `astrbot/core/star/star_manager.py:691` </location>
<code_context>
+ f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}",
+ )
+
async def install_plugin(self, repo_url: str, proxy=""):
"""从仓库 URL 安装插件
</code_context>
<issue_to_address>
**issue (complexity):** 建议为插件安装抽取一个共享的收尾（finalization）辅助函数，并拆分清理辅助函数，使每个函数只承担单一且更清晰的职责，同时减少重复逻辑。
你可以保留新的行为，但通过抽取共享的“插件安装收尾”流程，并将基于标志位的清理封装到一个辅助函数/类似上下文管理器的结构中，从而降低复杂度。
### 1. 抽取共享的收尾/安装逻辑
`install_plugin` 和 `install_plugin_from_file` 在插件目录创建之后，都遵循相同的模式：
- 定位插件
- 读取 README
- 构建 `plugin_info`
- 调度指标上报（文件安装场景）
可以把这部分提取到一个可复用的辅助函数中，使每个安装函数更线性、更短：
```python
async def _finalize_plugin_install(
self,
dir_name: str,
plugin_path: str,
metric_event: str | None = None,
) -> dict | None:
plugin = self.context.get_registered_star(dir_name)
if not plugin:
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
plugin = star
break
readme_content = None
readme_path = os.path.join(plugin_path, "README.md")
if not os.path.exists(readme_path):
readme_path = os.path.join(plugin_path, "readme.md")
if os.path.exists(readme_path):
try:
with open(readme_path, encoding="utf-8") as f:
readme_content = f.read()
except Exception as e:
logger.warning(f"读取插件 {dir_name} 的 README.md 文件失败: {e!s}")
if not plugin:
return None
plugin_info = {
"repo": plugin.repo,
"readme": readme_content,
"name": plugin.name,
}
if metric_event and plugin.repo:
asyncio.create_task(
Metric.upload(
et=metric_event,
repo=plugin.repo,
),
)
return plugin_info
```
在 `install_plugin` 中使用：
```python
async with self._pm_lock:
plugin_path = ""
dir_name = ""
cleanup_required = False
try:
plugin_path = await self.updator.install(repo_url, proxy)
cleanup_required = True
dir_name = os.path.basename(plugin_path)
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return await self._finalize_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
metric_event=None, # metric for repo_url can stay where it is now
)
except Exception:
if cleanup_required and dir_name and plugin_path:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
)
raise
```
在 `install_plugin_from_file` 中使用：
```python
try:
self.updator.unzip_file(zip_file_path, desti_dir)
cleanup_required = True
# ... metadata conflict check ...
try:
os.remove(zip_file_path)
except BaseException as e:
logger.warning(f"删除插件压缩包失败: {e!s}")
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return await self._finalize_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
metric_event="install_star_f",
)
except Exception:
if cleanup_required:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
)
raise
```
这样可以消除重复的分支逻辑，并且让“收尾”逻辑的职责更加清晰。
### 2. 降低 `_cleanup_failed_plugin_install` 的职责
如果你希望在不改变行为的前提下进一步简化清理逻辑，可以将 `_cleanup_failed_plugin_install` 拆分成更小的辅助函数，让主辅助函数只做调度：
```python
async def _terminate_and_unbind_by_dir_name(self, dir_name: str) -> None:
plugin = None
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
plugin = star
break
if plugin and plugin.name and plugin.module_path:
try:
await self._terminate_plugin(plugin)
except Exception:
logger.warning(traceback.format_exc())
try:
await self._unbind_plugin(plugin.name, plugin.module_path)
except Exception:
logger.warning(traceback.format_exc())
def _remove_failed_plugin_files(self, dir_name: str, plugin_path: str) -> None:
if os.path.exists(plugin_path):
try:
remove_dir(plugin_path)
logger.warning(f"已清理安装失败的插件目录: {plugin_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件目录失败: {plugin_path}，原因: {e!s}",
)
plugin_config_path = os.path.join(
self.plugin_config_path,
f"{dir_name}_config.json",
)
if os.path.exists(plugin_config_path):
try:
os.remove(plugin_config_path)
logger.warning(f"已清理安装失败插件配置: {plugin_config_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}",
)
async def _cleanup_failed_plugin_install(
self,
dir_name: str,
plugin_path: str,
) -> None:
await self._terminate_and_unbind_by_dir_name(dir_name)
self._remove_failed_plugin_files(dir_name, plugin_path)
```
这样可以保持当前行为不变，但每个函数现在只承担一个更容易理解的职责，而主要的安装流程则变得更短、更清晰。
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据你的反馈改进后续评审。
Original comment in English
Hey - I've found 2 issues, and left some high level feedback:
- In
pip_installer.py,_extract_requirement_namesskips-r/--requirementincludes entirely, which means dependency-preference/conflict detection will miss transitive requirements defined in included files; consider following those includes (with a recursion guard) so your_ensure_plugin_dependencies_preferredlogic has a complete view of dependencies. - The new module-preference helpers in
pip_installer.py(_collect_candidate_modules / _prefer_modules_from_site_packages / _ensure_preferred_modules) introduce a lot of logic into a single utility file; extracting them into a dedicated module (e.g.plugin_site_packages.py) would make the pip wrapper itself easier to follow and reduce the cognitive load when reasoning about in-process pip vs. import behavior.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `pip_installer.py`, `_extract_requirement_names` skips `-r/--requirement` includes entirely, which means dependency-preference/conflict detection will miss transitive requirements defined in included files; consider following those includes (with a recursion guard) so your `_ensure_plugin_dependencies_preferred` logic has a complete view of dependencies.
- The new module-preference helpers in `pip_installer.py` (_collect_candidate_modules / _prefer_modules_from_site_packages / _ensure_preferred_modules) introduce a lot of logic into a single utility file; extracting them into a dedicated module (e.g. `plugin_site_packages.py`) would make the pip wrapper itself easier to follow and reduce the cognitive load when reasoning about in-process pip vs. import behavior.
## Individual Comments
### Comment 1
<location> `astrbot/core/utils/pip_installer.py:233` </location>
<code_context>
+ raise RuntimeError(conflict_message)
+
+
+def _prefer_module_from_site_packages(
+ module_name: str, site_packages_path: str
+) -> bool:
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the new plugin import and dependency-handling logic by reusing the normal import system, limiting dependency recovery, and avoiding full transitive requirement expansion.
You can keep the new behavior (prefer plugin site-packages, check for conflicts) while cutting a lot of complexity by leaning more on the standard import system and narrowing the scope of “recovery”.
### 1. Simplify `_prefer_module_from_site_packages` (no manual spec)
You already ensure the plugin `site-packages` is first via `_prepend_sys_path`. That means you can avoid custom spec construction and most of the `sys.modules` surgery:
```python
def _prefer_module_from_site_packages(
module_name: str,
site_packages_path: str,
) -> bool:
with _SITE_PACKAGES_IMPORT_LOCK:
if not _module_exists_in_site_packages(module_name, site_packages_path):
return False
# Snapshot original modules for rollback
matched_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
original_modules = {key: sys.modules[key] for key in matched_keys}
try:
# Drop existing module(s), then import normally (sys.path is ordered)
for key in matched_keys:
sys.modules.pop(key, None)
module = importlib.import_module(module_name)
# Ensure parent attribute is set, if applicable
if "." in module_name:
parent_name, child_name = module_name.rsplit(".", 1)
parent = sys.modules.get(parent_name)
if parent is not None:
setattr(parent, child_name, module)
logger.info(
"Loaded %s via normal import from plugin site-packages", module_name
)
return True
except Exception:
# Roll back on failure
for key in list(sys.modules.keys()):
if key == module_name or key.startswith(f"{module_name}."):
sys.modules.pop(key, None)
sys.modules.update(original_modules)
raise
```
This still gives you:
- deterministic re-import from the plugin path (because you prepend `target_site_packages`),
- rollback on failure,
- no custom `spec_from_file_location` or manual loader execution.
### 2. Narrow dependency “recovery” and retry logic
The multi-round retry (`max_rounds`, `pending_modules`, `round_progress`) and regex-based `_extract_conflicting_module_name` add a lot of state and rely on error-message formats.
You could reduce this to:
- only recover for `ModuleNotFoundError` using the `.name` attribute,
- no multi-round global scheduler; just a single local attempt per dependency.
For example:
```python
def _extract_conflicting_module_name(exc: Exception) -> str | None:
if isinstance(exc, ModuleNotFoundError):
name = getattr(exc, "name", None)
if name:
return name.split(".", 1)[0]
return None
```
And then:
```python
def _prefer_module_with_dependency_recovery(
module_name: str,
site_packages_path: str,
) -> bool:
try:
return _prefer_module_from_site_packages(module_name, site_packages_path)
except Exception as exc:
dependency_name = _extract_conflicting_module_name(exc)
if not dependency_name or dependency_name == module_name:
raise
# Try once to prefer the missing dependency, then retry module
if not _prefer_module_from_site_packages(dependency_name, site_packages_path):
raise
logger.info(
"Recovered dependency %s while preferring %s from plugin site-packages.",
dependency_name,
module_name,
)
return _prefer_module_from_site_packages(module_name, site_packages_path)
```
Then `_prefer_modules_from_site_packages` can be simplified to a single pass (no rounds, no pending list):
```python
def _prefer_modules_from_site_packages(
module_names: set[str],
site_packages_path: str,
) -> dict[str, str]:
unresolved: dict[str, str] = {}
for module_name in sorted(module_names):
try:
loaded = _prefer_module_with_dependency_recovery(
module_name,
site_packages_path,
)
if not loaded:
logger.debug(
"Module %s not found in plugin site-packages: %s",
module_name,
site_packages_path,
)
except Exception as exc:
unresolved[module_name] = str(exc)
for module_name, reason in unresolved.items():
logger.warning(
"Failed to prefer module %s from plugin site-packages: %s",
module_name,
reason,
)
return unresolved
```
This keeps:
- dependency “recovery” for straightforward missing imports,
- per-module error reporting,
but removes the multi-round scheduler and error-message regex parsing.
### 3. Limit dependency graph expansion to direct requirements
`_collect_candidate_modules` currently builds a full transitive closure over `importlib_metadata.distributions`. For most plugin scenarios, preferring only *direct* requirements is usually enough, and much easier to reason about.
You can keep your metadata-based resolution but allow a simpler path that skips BFS over `requires`:
```python
def _collect_candidate_modules(
requirement_names: set[str],
site_packages_path: str,
*,
include_transitive: bool = False,
) -> set[str]:
by_name: dict[str, list[importlib_metadata.Distribution]] = {}
try:
for dist in importlib_metadata.distributions(path=[site_packages_path]):
name = dist.metadata.get("Name")
if not name:
continue
canonical = _canonicalize_distribution_name(name)
by_name.setdefault(canonical, []).append(dist)
except Exception as exc:
logger.warning("读取 site-packages 元数据失败，使用回退模块名: %s", exc)
if include_transitive:
expanded = set()
pending = deque(requirement_names)
while pending:
req = pending.popleft()
if req in expanded:
continue
expanded.add(req)
for dist in by_name.get(req, []):
for dep_line in dist.requires or []:
dep_name = _extract_requirement_name(dep_line)
if dep_name and dep_name not in expanded:
pending.append(dep_name)
requirement_names = expanded
candidates: set[str] = set()
for req in requirement_names:
modules_for_req: set[str] = set()
for dist in by_name.get(req, []):
modules_for_req.update(_extract_top_level_modules(dist))
if modules_for_req:
candidates.update(modules_for_req)
else:
candidates.add(req.replace("-", "_"))
return candidates
```
Then `_ensure_plugin_dependencies_preferred` can call it with `include_transitive=False` (or make that the default):
```python
candidate_modules = _collect_candidate_modules(
requested_requirements,
target_site_packages,
include_transitive=False,
)
```
This keeps the ability to extend to full closure later but keeps the default behavior simpler and more predictable.
---
These focused changes:
- still enforce that plugin-installed dependencies are preferred,
- still detect unresolved conflicts,
- but remove a lot of low-level import machinery and multi-round state handling, making the behavior significantly easier to reason about and maintain.
</issue_to_address>
### Comment 2
<location> `astrbot/core/star/star_manager.py:691` </location>
<code_context>
+ f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}",
+ )
+
async def install_plugin(self, repo_url: str, proxy=""):
"""从仓库 URL 安装插件
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting a shared finalization helper for plugin installs and decomposing the cleanup helper so each function has a single, clearer responsibility and less duplicated logic.
You can keep the new behavior but reduce complexity by extracting the shared “finalize plugin install” flow and hiding the flag-based cleanup behind a helper/context-like function.
### 1. Extract shared finalize/install logic
Both `install_plugin` and `install_plugin_from_file` have the same pattern after the plugin directory exists:
- locate the plugin
- read README
- build `plugin_info`
- schedule metric upload (for file-based install)
Pull that into a reusable helper so each install function is linear and smaller:
```python
async def _finalize_plugin_install(
self,
dir_name: str,
plugin_path: str,
metric_event: str | None = None,
) -> dict | None:
plugin = self.context.get_registered_star(dir_name)
if not plugin:
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
plugin = star
break
readme_content = None
readme_path = os.path.join(plugin_path, "README.md")
if not os.path.exists(readme_path):
readme_path = os.path.join(plugin_path, "readme.md")
if os.path.exists(readme_path):
try:
with open(readme_path, encoding="utf-8") as f:
readme_content = f.read()
except Exception as e:
logger.warning(f"读取插件 {dir_name} 的 README.md 文件失败: {e!s}")
if not plugin:
return None
plugin_info = {
"repo": plugin.repo,
"readme": readme_content,
"name": plugin.name,
}
if metric_event and plugin.repo:
asyncio.create_task(
Metric.upload(
et=metric_event,
repo=plugin.repo,
),
)
return plugin_info
```
Usage in `install_plugin`:
```python
async with self._pm_lock:
plugin_path = ""
dir_name = ""
cleanup_required = False
try:
plugin_path = await self.updator.install(repo_url, proxy)
cleanup_required = True
dir_name = os.path.basename(plugin_path)
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return await self._finalize_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
metric_event=None, # metric for repo_url can stay where it is now
)
except Exception:
if cleanup_required and dir_name and plugin_path:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
)
raise
```
Usage in `install_plugin_from_file`:
```python
try:
self.updator.unzip_file(zip_file_path, desti_dir)
cleanup_required = True
# ... metadata conflict check ...
try:
os.remove(zip_file_path)
except BaseException as e:
logger.warning(f"删除插件压缩包失败: {e!s}")
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return await self._finalize_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
metric_event="install_star_f",
)
except Exception:
if cleanup_required:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
)
raise
```
This removes duplicated branching and makes it clear what “finalization” does.
### 2. Reduce responsibility of `_cleanup_failed_plugin_install`
If you want to further simplify the cleanup logic without changing behavior, split `_cleanup_failed_plugin_install` into smaller helpers and make the main helper just orchestrate:
```python
async def _terminate_and_unbind_by_dir_name(self, dir_name: str) -> None:
plugin = None
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
plugin = star
break
if plugin and plugin.name and plugin.module_path:
try:
await self._terminate_plugin(plugin)
except Exception:
logger.warning(traceback.format_exc())
try:
await self._unbind_plugin(plugin.name, plugin.module_path)
except Exception:
logger.warning(traceback.format_exc())
def _remove_failed_plugin_files(self, dir_name: str, plugin_path: str) -> None:
if os.path.exists(plugin_path):
try:
remove_dir(plugin_path)
logger.warning(f"已清理安装失败的插件目录: {plugin_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件目录失败: {plugin_path}，原因: {e!s}",
)
plugin_config_path = os.path.join(
self.plugin_config_path,
f"{dir_name}_config.json",
)
if os.path.exists(plugin_config_path):
try:
os.remove(plugin_config_path)
logger.warning(f"已清理安装失败插件配置: {plugin_config_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}",
)
async def _cleanup_failed_plugin_install(
self,
dir_name: str,
plugin_path: str,
) -> None:
await self._terminate_and_unbind_by_dir_name(dir_name)
self._remove_failed_plugin_files(dir_name, plugin_path)
```
This keeps behavior intact, but each function now has a single, easier-to-reason-about responsibility and the main install flows become shorter and clearer.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| raise RuntimeError(conflict_message) | ||
|
|
||
|
|
||
| def _prefer_module_from_site_packages( |
Contributor
There was a problem hiding this comment.
issue (complexity): 建议通过复用正常的导入系统、限制依赖“恢复”的范围，并避免做完整的传递依赖展开，来简化新的插件导入和依赖处理逻辑。
你可以在保留当前行为（优先使用插件的 site-packages、检测冲突）的同时，大幅削减复杂度：更多依赖标准导入系统，并收窄“恢复”逻辑的作用范围。
1. 简化 _prefer_module_from_site_packages（不再手动构建 spec）
你已经通过 _prepend_sys_path 确保插件的 site-packages 优先出现在 sys.path 中。这意味着你可以避免自定义 spec 构造以及大部分对 sys.modules 的复杂操作：
def _prefer_module_from_site_packages(
module_name: str,
site_packages_path: str,
) -> bool:
with _SITE_PACKAGES_IMPORT_LOCK:
if not _module_exists_in_site_packages(module_name, site_packages_path):
return False
# Snapshot original modules for rollback
matched_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
original_modules = {key: sys.modules[key] for key in matched_keys}
try:
# Drop existing module(s), then import normally (sys.path is ordered)
for key in matched_keys:
sys.modules.pop(key, None)
module = importlib.import_module(module_name)
# Ensure parent attribute is set, if applicable
if "." in module_name:
parent_name, child_name = module_name.rsplit(".", 1)
parent = sys.modules.get(parent_name)
if parent is not None:
setattr(parent, child_name, module)
logger.info(
"Loaded %s via normal import from plugin site-packages", module_name
)
return True
except Exception:
# Roll back on failure
for key in list(sys.modules.keys()):
if key == module_name or key.startswith(f"{module_name}."):
sys.modules.pop(key, None)
sys.modules.update(original_modules)
raise这样仍然可以得到：
- 从插件路径确定性地重新导入（因为你提前将
target_site_packages放在sys.path前面）， - 导入失败时可以回滚，
- 不再需要自定义
spec_from_file_location或手动执行 loader。
2. 收窄依赖“恢复”和重试逻辑
多轮重试（max_rounds、pending_modules、round_progress）以及基于正则的 _extract_conflicting_module_name 会引入大量状态，并且依赖错误信息的具体格式。
你可以将其简化为：
- 只在
ModuleNotFoundError上做恢复，并使用其.name属性， - 不再使用全局多轮调度，只对每个依赖做一次本地尝试。
例如：
def _extract_conflicting_module_name(exc: Exception) -> str | None:
if isinstance(exc, ModuleNotFoundError):
name = getattr(exc, "name", None)
if name:
return name.split(".", 1)[0]
return None然后：
def _prefer_module_with_dependency_recovery(
module_name: str,
site_packages_path: str,
) -> bool:
try:
return _prefer_module_from_site_packages(module_name, site_packages_path)
except Exception as exc:
dependency_name = _extract_conflicting_module_name(exc)
if not dependency_name or dependency_name == module_name:
raise
# Try once to prefer the missing dependency, then retry module
if not _prefer_module_from_site_packages(dependency_name, site_packages_path):
raise
logger.info(
"Recovered dependency %s while preferring %s from plugin site-packages.",
dependency_name,
module_name,
)
return _prefer_module_from_site_packages(module_name, site_packages_path)这样 _prefer_modules_from_site_packages 就可以简化为单次遍历（不再需要多轮和 pending 列表）：
def _prefer_modules_from_site_packages(
module_names: set[str],
site_packages_path: str,
) -> dict[str, str]:
unresolved: dict[str, str] = {}
for module_name in sorted(module_names):
try:
loaded = _prefer_module_with_dependency_recovery(
module_name,
site_packages_path,
)
if not loaded:
logger.debug(
"Module %s not found in plugin site-packages: %s",
module_name,
site_packages_path,
)
except Exception as exc:
unresolved[module_name] = str(exc)
for module_name, reason in unresolved.items():
logger.warning(
"Failed to prefer module %s from plugin site-packages: %s",
module_name,
reason,
)
return unresolved这样可以保留：
- 对简单缺失导入场景的依赖“恢复”，
- 面向每个模块的错误上报，
但去掉多轮调度和基于错误信息的正则解析。
3. 将依赖图扩展限制为直接依赖
当前 _collect_candidate_modules 会在 importlib_metadata.distributions 上构建完整的传递闭包。对于大多数插件场景，只优先处理直接依赖通常已经足够，并且更容易理解。
你可以保留基于元数据的解析逻辑，但提供一个更简单的路径来跳过对 requires 的 BFS：
def _collect_candidate_modules(
requirement_names: set[str],
site_packages_path: str,
*,
include_transitive: bool = False,
) -> set[str]:
by_name: dict[str, list[importlib_metadata.Distribution]] = {}
try:
for dist in importlib_metadata.distributions(path=[site_packages_path]):
name = dist.metadata.get("Name")
if not name:
continue
canonical = _canonicalize_distribution_name(name)
by_name.setdefault(canonical, []).append(dist)
except Exception as exc:
logger.warning("读取 site-packages 元数据失败，使用回退模块名: %s", exc)
if include_transitive:
expanded = set()
pending = deque(requirement_names)
while pending:
req = pending.popleft()
if req in expanded:
continue
expanded.add(req)
for dist in by_name.get(req, []):
for dep_line in dist.requires or []:
dep_name = _extract_requirement_name(dep_line)
if dep_name and dep_name not in expanded:
pending.append(dep_name)
requirement_names = expanded
candidates: set[str] = set()
for req in requirement_names:
modules_for_req: set[str] = set()
for dist in by_name.get(req, []):
modules_for_req.update(_extract_top_level_modules(dist))
if modules_for_req:
candidates.update(modules_for_req)
else:
candidates.add(req.replace("-", "_"))
return candidates然后 _ensure_plugin_dependencies_preferred 可以通过 include_transitive=False 调用它（或将此设为默认）：
candidate_modules = _collect_candidate_modules(
requested_requirements,
target_site_packages,
include_transitive=False,
)这样既保留了未来扩展为完整闭包的能力，又让默认行为更简单、更可预测。
这些聚焦的改动：
- 依然保证优先使用由插件安装的依赖，
- 依然能检测未解决的冲突，
- 但去掉了大量底层 import 机制细节和多轮状态处理，使整体行为更容易理解和维护。
Original comment in English
issue (complexity): Consider simplifying the new plugin import and dependency-handling logic by reusing the normal import system, limiting dependency recovery, and avoiding full transitive requirement expansion.
You can keep the new behavior (prefer plugin site-packages, check for conflicts) while cutting a lot of complexity by leaning more on the standard import system and narrowing the scope of “recovery”.
1. Simplify _prefer_module_from_site_packages (no manual spec)
You already ensure the plugin site-packages is first via _prepend_sys_path. That means you can avoid custom spec construction and most of the sys.modules surgery:
def _prefer_module_from_site_packages(
module_name: str,
site_packages_path: str,
) -> bool:
with _SITE_PACKAGES_IMPORT_LOCK:
if not _module_exists_in_site_packages(module_name, site_packages_path):
return False
# Snapshot original modules for rollback
matched_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
original_modules = {key: sys.modules[key] for key in matched_keys}
try:
# Drop existing module(s), then import normally (sys.path is ordered)
for key in matched_keys:
sys.modules.pop(key, None)
module = importlib.import_module(module_name)
# Ensure parent attribute is set, if applicable
if "." in module_name:
parent_name, child_name = module_name.rsplit(".", 1)
parent = sys.modules.get(parent_name)
if parent is not None:
setattr(parent, child_name, module)
logger.info(
"Loaded %s via normal import from plugin site-packages", module_name
)
return True
except Exception:
# Roll back on failure
for key in list(sys.modules.keys()):
if key == module_name or key.startswith(f"{module_name}."):
sys.modules.pop(key, None)
sys.modules.update(original_modules)
raiseThis still gives you:
- deterministic re-import from the plugin path (because you prepend
target_site_packages), - rollback on failure,
- no custom
spec_from_file_locationor manual loader execution.
2. Narrow dependency “recovery” and retry logic
The multi-round retry (max_rounds, pending_modules, round_progress) and regex-based _extract_conflicting_module_name add a lot of state and rely on error-message formats.
You could reduce this to:
- only recover for
ModuleNotFoundErrorusing the.nameattribute, - no multi-round global scheduler; just a single local attempt per dependency.
For example:
def _extract_conflicting_module_name(exc: Exception) -> str | None:
if isinstance(exc, ModuleNotFoundError):
name = getattr(exc, "name", None)
if name:
return name.split(".", 1)[0]
return NoneAnd then:
def _prefer_module_with_dependency_recovery(
module_name: str,
site_packages_path: str,
) -> bool:
try:
return _prefer_module_from_site_packages(module_name, site_packages_path)
except Exception as exc:
dependency_name = _extract_conflicting_module_name(exc)
if not dependency_name or dependency_name == module_name:
raise
# Try once to prefer the missing dependency, then retry module
if not _prefer_module_from_site_packages(dependency_name, site_packages_path):
raise
logger.info(
"Recovered dependency %s while preferring %s from plugin site-packages.",
dependency_name,
module_name,
)
return _prefer_module_from_site_packages(module_name, site_packages_path)Then _prefer_modules_from_site_packages can be simplified to a single pass (no rounds, no pending list):
def _prefer_modules_from_site_packages(
module_names: set[str],
site_packages_path: str,
) -> dict[str, str]:
unresolved: dict[str, str] = {}
for module_name in sorted(module_names):
try:
loaded = _prefer_module_with_dependency_recovery(
module_name,
site_packages_path,
)
if not loaded:
logger.debug(
"Module %s not found in plugin site-packages: %s",
module_name,
site_packages_path,
)
except Exception as exc:
unresolved[module_name] = str(exc)
for module_name, reason in unresolved.items():
logger.warning(
"Failed to prefer module %s from plugin site-packages: %s",
module_name,
reason,
)
return unresolvedThis keeps:
- dependency “recovery” for straightforward missing imports,
- per-module error reporting,
but removes the multi-round scheduler and error-message regex parsing.
3. Limit dependency graph expansion to direct requirements
_collect_candidate_modules currently builds a full transitive closure over importlib_metadata.distributions. For most plugin scenarios, preferring only direct requirements is usually enough, and much easier to reason about.
You can keep your metadata-based resolution but allow a simpler path that skips BFS over requires:
def _collect_candidate_modules(
requirement_names: set[str],
site_packages_path: str,
*,
include_transitive: bool = False,
) -> set[str]:
by_name: dict[str, list[importlib_metadata.Distribution]] = {}
try:
for dist in importlib_metadata.distributions(path=[site_packages_path]):
name = dist.metadata.get("Name")
if not name:
continue
canonical = _canonicalize_distribution_name(name)
by_name.setdefault(canonical, []).append(dist)
except Exception as exc:
logger.warning("读取 site-packages 元数据失败，使用回退模块名: %s", exc)
if include_transitive:
expanded = set()
pending = deque(requirement_names)
while pending:
req = pending.popleft()
if req in expanded:
continue
expanded.add(req)
for dist in by_name.get(req, []):
for dep_line in dist.requires or []:
dep_name = _extract_requirement_name(dep_line)
if dep_name and dep_name not in expanded:
pending.append(dep_name)
requirement_names = expanded
candidates: set[str] = set()
for req in requirement_names:
modules_for_req: set[str] = set()
for dist in by_name.get(req, []):
modules_for_req.update(_extract_top_level_modules(dist))
if modules_for_req:
candidates.update(modules_for_req)
else:
candidates.add(req.replace("-", "_"))
return candidatesThen _ensure_plugin_dependencies_preferred can call it with include_transitive=False (or make that the default):
candidate_modules = _collect_candidate_modules(
requested_requirements,
target_site_packages,
include_transitive=False,
)This keeps the ability to extend to full closure later but keeps the default behavior simpler and more predictable.
These focused changes:
- still enforce that plugin-installed dependencies are preferred,
- still detect unresolved conflicts,
- but remove a lot of low-level import machinery and multi-round state handling, making the behavior significantly easier to reason about and maintain.
| f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}", | ||
| ) | ||
|
|
||
| async def install_plugin(self, repo_url: str, proxy=""): |
Contributor
There was a problem hiding this comment.
issue (complexity): 建议为插件安装抽取一个共享的收尾（finalization）辅助函数，并拆分清理辅助函数，使每个函数只承担单一且更清晰的职责，同时减少重复逻辑。
你可以保留新的行为，但通过抽取共享的“插件安装收尾”流程，并将基于标志位的清理封装到一个辅助函数/类似上下文管理器的结构中，从而降低复杂度。
1. 抽取共享的收尾/安装逻辑
install_plugin 和 install_plugin_from_file 在插件目录创建之后，都遵循相同的模式：
- 定位插件
- 读取 README
- 构建
plugin_info - 调度指标上报（文件安装场景）
可以把这部分提取到一个可复用的辅助函数中，使每个安装函数更线性、更短：
async def _finalize_plugin_install(
self,
dir_name: str,
plugin_path: str,
metric_event: str | None = None,
) -> dict | None:
plugin = self.context.get_registered_star(dir_name)
if not plugin:
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
plugin = star
break
readme_content = None
readme_path = os.path.join(plugin_path, "README.md")
if not os.path.exists(readme_path):
readme_path = os.path.join(plugin_path, "readme.md")
if os.path.exists(readme_path):
try:
with open(readme_path, encoding="utf-8") as f:
readme_content = f.read()
except Exception as e:
logger.warning(f"读取插件 {dir_name} 的 README.md 文件失败: {e!s}")
if not plugin:
return None
plugin_info = {
"repo": plugin.repo,
"readme": readme_content,
"name": plugin.name,
}
if metric_event and plugin.repo:
asyncio.create_task(
Metric.upload(
et=metric_event,
repo=plugin.repo,
),
)
return plugin_info在 install_plugin 中使用：
async with self._pm_lock:
plugin_path = ""
dir_name = ""
cleanup_required = False
try:
plugin_path = await self.updator.install(repo_url, proxy)
cleanup_required = True
dir_name = os.path.basename(plugin_path)
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return await self._finalize_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
metric_event=None, # metric for repo_url can stay where it is now
)
except Exception:
if cleanup_required and dir_name and plugin_path:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
)
raise在 install_plugin_from_file 中使用：
try:
self.updator.unzip_file(zip_file_path, desti_dir)
cleanup_required = True
# ... metadata conflict check ...
try:
os.remove(zip_file_path)
except BaseException as e:
logger.warning(f"删除插件压缩包失败: {e!s}")
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return await self._finalize_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
metric_event="install_star_f",
)
except Exception:
if cleanup_required:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
)
raise这样可以消除重复的分支逻辑，并且让“收尾”逻辑的职责更加清晰。
2. 降低 _cleanup_failed_plugin_install 的职责
如果你希望在不改变行为的前提下进一步简化清理逻辑，可以将 _cleanup_failed_plugin_install 拆分成更小的辅助函数，让主辅助函数只做调度：
async def _terminate_and_unbind_by_dir_name(self, dir_name: str) -> None:
plugin = None
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
plugin = star
break
if plugin and plugin.name and plugin.module_path:
try:
await self._terminate_plugin(plugin)
except Exception:
logger.warning(traceback.format_exc())
try:
await self._unbind_plugin(plugin.name, plugin.module_path)
except Exception:
logger.warning(traceback.format_exc())
def _remove_failed_plugin_files(self, dir_name: str, plugin_path: str) -> None:
if os.path.exists(plugin_path):
try:
remove_dir(plugin_path)
logger.warning(f"已清理安装失败的插件目录: {plugin_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件目录失败: {plugin_path}，原因: {e!s}",
)
plugin_config_path = os.path.join(
self.plugin_config_path,
f"{dir_name}_config.json",
)
if os.path.exists(plugin_config_path):
try:
os.remove(plugin_config_path)
logger.warning(f"已清理安装失败插件配置: {plugin_config_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}",
)
async def _cleanup_failed_plugin_install(
self,
dir_name: str,
plugin_path: str,
) -> None:
await self._terminate_and_unbind_by_dir_name(dir_name)
self._remove_failed_plugin_files(dir_name, plugin_path)这样可以保持当前行为不变，但每个函数现在只承担一个更容易理解的职责，而主要的安装流程则变得更短、更清晰。
Original comment in English
issue (complexity): Consider extracting a shared finalization helper for plugin installs and decomposing the cleanup helper so each function has a single, clearer responsibility and less duplicated logic.
You can keep the new behavior but reduce complexity by extracting the shared “finalize plugin install” flow and hiding the flag-based cleanup behind a helper/context-like function.
1. Extract shared finalize/install logic
Both install_plugin and install_plugin_from_file have the same pattern after the plugin directory exists:
- locate the plugin
- read README
- build
plugin_info - schedule metric upload (for file-based install)
Pull that into a reusable helper so each install function is linear and smaller:
async def _finalize_plugin_install(
self,
dir_name: str,
plugin_path: str,
metric_event: str | None = None,
) -> dict | None:
plugin = self.context.get_registered_star(dir_name)
if not plugin:
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
plugin = star
break
readme_content = None
readme_path = os.path.join(plugin_path, "README.md")
if not os.path.exists(readme_path):
readme_path = os.path.join(plugin_path, "readme.md")
if os.path.exists(readme_path):
try:
with open(readme_path, encoding="utf-8") as f:
readme_content = f.read()
except Exception as e:
logger.warning(f"读取插件 {dir_name} 的 README.md 文件失败: {e!s}")
if not plugin:
return None
plugin_info = {
"repo": plugin.repo,
"readme": readme_content,
"name": plugin.name,
}
if metric_event and plugin.repo:
asyncio.create_task(
Metric.upload(
et=metric_event,
repo=plugin.repo,
),
)
return plugin_infoUsage in install_plugin:
async with self._pm_lock:
plugin_path = ""
dir_name = ""
cleanup_required = False
try:
plugin_path = await self.updator.install(repo_url, proxy)
cleanup_required = True
dir_name = os.path.basename(plugin_path)
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return await self._finalize_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
metric_event=None, # metric for repo_url can stay where it is now
)
except Exception:
if cleanup_required and dir_name and plugin_path:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
)
raiseUsage in install_plugin_from_file:
try:
self.updator.unzip_file(zip_file_path, desti_dir)
cleanup_required = True
# ... metadata conflict check ...
try:
os.remove(zip_file_path)
except BaseException as e:
logger.warning(f"删除插件压缩包失败: {e!s}")
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return await self._finalize_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
metric_event="install_star_f",
)
except Exception:
if cleanup_required:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
)
raiseThis removes duplicated branching and makes it clear what “finalization” does.
2. Reduce responsibility of _cleanup_failed_plugin_install
If you want to further simplify the cleanup logic without changing behavior, split _cleanup_failed_plugin_install into smaller helpers and make the main helper just orchestrate:
async def _terminate_and_unbind_by_dir_name(self, dir_name: str) -> None:
plugin = None
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
plugin = star
break
if plugin and plugin.name and plugin.module_path:
try:
await self._terminate_plugin(plugin)
except Exception:
logger.warning(traceback.format_exc())
try:
await self._unbind_plugin(plugin.name, plugin.module_path)
except Exception:
logger.warning(traceback.format_exc())
def _remove_failed_plugin_files(self, dir_name: str, plugin_path: str) -> None:
if os.path.exists(plugin_path):
try:
remove_dir(plugin_path)
logger.warning(f"已清理安装失败的插件目录: {plugin_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件目录失败: {plugin_path}，原因: {e!s}",
)
plugin_config_path = os.path.join(
self.plugin_config_path,
f"{dir_name}_config.json",
)
if os.path.exists(plugin_config_path):
try:
os.remove(plugin_config_path)
logger.warning(f"已清理安装失败插件配置: {plugin_config_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}",
)
async def _cleanup_failed_plugin_install(
self,
dir_name: str,
plugin_path: str,
) -> None:
await self._terminate_and_unbind_by_dir_name(dir_name)
self._remove_failed_plugin_files(dir_name, plugin_path)This keeps behavior intact, but each function now has a single, easier-to-reason-about responsibility and the main install flows become shorter and clearer.
zouyonghe
changed the title
Collaborator
Author
|
@sourcery-ai review |
Contributor
There was a problem hiding this comment.
Hey - 我发现了两个问题，并留下了一些总体反馈：
pip_installer.py中新增的依赖优先级辅助函数（_collect_candidate_modules/_ensure_plugin_dependencies_preferred/_prefer_modules_from_site_packages等）体积较大且彼此高度耦合；可以考虑按职责拆分为更小、更聚焦的工具函数（例如：依赖解析、元数据展开、导入覆盖等），以便后续更容易理解和调试。- 两条插件安装路径（
install_plugin和install_plugin_from_file）现在在_cleanup_failed_plugin_install附近都包含类似的成功/清理和 README 提取逻辑；可以考虑抽取一个共享的辅助函数，减少重复逻辑，并在单一位置保持插件生命周期行为的一致性。 _check_plugin_dept_update现在在单个依赖安装失败时也会无条件返回True（只记录日志）；如果调用方会将返回值用于控制流程，可以考虑是否需要在失败时传递失败标识，而不是始终表示成功。
给 AI Agent 的提示
请根据以下代码评审评论进行修改：
## 总体评论
- `pip_installer.py` 中新增的依赖优先级辅助函数（`_collect_candidate_modules` / `_ensure_plugin_dependencies_preferred` / `_prefer_modules_from_site_packages` 等）体积较大且彼此高度耦合；可以考虑按职责拆分为更小、更聚焦的工具函数（例如：依赖解析、元数据展开、导入覆盖等），以便后续更容易理解和调试。
- 两条插件安装路径（`install_plugin` 和 `install_plugin_from_file`）现在在 `_cleanup_failed_plugin_install` 附近都包含类似的成功/清理和 README 提取逻辑；可以考虑抽取一个共享的辅助函数，减少重复逻辑，并在单一位置保持插件生命周期行为的一致性。
- `_check_plugin_dept_update` 现在在单个依赖安装失败时也会无条件返回 `True`（只记录日志）；如果调用方会将返回值用于控制流程，可以考虑是否需要在失败时传递失败标识，而不是始终表示成功。
## 逐条评论
### 评论 1
<location> `astrbot/core/utils/pip_installer.py:412` </location>
<code_context>
+ return final_unresolved
+
+
+def _ensure_plugin_dependencies_preferred(
+ target_site_packages: str,
+ requested_requirements: set[str],
</code_context>
<issue_to_address>
**issue (complexity):** 请考虑将依赖优先级的编排逻辑封装到一个管理类中，并把底层的导入操作拆分成更小的辅助函数，以提供更简单、更清晰的对外流程。
这里新增的能力很强大，但现在被分散在多个紧密耦合的辅助函数中。在不改变行为的前提下，可以通过以下方式降低理解成本：
1. **将编排逻辑封装到一个小对象中**
2. **把底层的导入操作拆分为更小、更聚焦的辅助函数**
这样既保留所有特性（依赖展开、恢复、重试），又能对外暴露更小的接口面，流程更加清晰。
---
### 1. 将编排逻辑封装进 manager
目前 `_ensure_plugin_dependencies_preferred` + `_collect_candidate_modules` + `_ensure_preferred_modules` + `_prefer_modules_from_site_packages` + `_prefer_module_with_dependency_recovery` 实际上组成了一个状态机。
可以考虑把这些组合为一个小的管理类，并从模块中只暴露一个“入口”方法。这样可以保留所有现有逻辑，但把大部分细节隐藏在实例内部：
```python
class _PluginEnvManager:
def __init__(self, site_packages_path: str):
self.site_packages_path = site_packages_path
def ensure_dependencies_preferred(self, requested_requirements: set[str]) -> None:
if not requested_requirements:
return
candidate_modules = self._collect_candidate_modules(requested_requirements)
if not candidate_modules:
return
self._ensure_preferred_modules(candidate_modules)
# move existing helpers as methods, e.g.:
def _collect_candidate_modules(self, requirement_names: set[str]) -> set[str]:
return _collect_candidate_modules(requirement_names, self.site_packages_path)
def _ensure_preferred_modules(self, module_names: set[str]) -> None:
_ensure_preferred_modules(module_names, self.site_packages_path)
```
这样外部调用点会更简单，模块级的对外接口也会缩小：
```python
if target_site_packages:
_prepend_sys_path(target_site_packages)
manager = _PluginEnvManager(target_site_packages)
manager.ensure_dependencies_preferred(requested_requirements)
```
后续步骤（可选但有帮助）：
- 逐步将 `_collect_candidate_modules`、`_ensure_preferred_modules`、`_prefer_modules_from_site_packages`、`_prefer_module_with_dependency_recovery` 内联/迁移为该类的私有方法，并在所有调用点都使用 manager 后移除这些模块级函数定义。
- 这样高层流程就能在一个地方（`ensure_dependencies_preferred`）被完整看到，而不是在多个顶层辅助函数之间来回跳转。
---
### 2. 将底层导入操作拆分为更小的辅助函数
`_prefer_module_from_site_packages` 目前同时负责：
- 路径解析（包 vs 模块）、
- 创建 spec、
- `sys.modules` 对比/删除、
- 加载模块、
- 父子包关系修复、
- 失败时回滚。
可以通过职责分离让它更易读、更易于推理；在行为保持不变的前提下，可以这样拆分：
```python
def _resolve_module_location(module_name: str, site_packages_path: str) -> tuple[str | None, list[str] | None]:
base_path = os.path.join(site_packages_path, *module_name.split("."))
package_init = os.path.join(base_path, "__init__.py")
module_file = f"{base_path}.py"
if os.path.isfile(package_init):
return package_init, [os.path.dirname(package_init)]
if os.path.isfile(module_file):
return module_file, None
return None, None
def _swap_modules_for_preference(module_name: str) -> dict[str, types.ModuleType]:
matched_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
original_modules = {key: sys.modules[key] for key in matched_keys}
for key in matched_keys:
sys.modules.pop(key, None)
return original_modules
```
然后 `_prefer_module_from_site_packages` 在结构上会简单很多：
```python
def _prefer_module_from_site_packages(module_name: str, site_packages_path: str) -> bool:
with _SITE_PACKAGES_IMPORT_LOCK:
module_location, submodule_search_locations = _resolve_module_location(
module_name, site_packages_path
)
if not module_location:
return False
spec = importlib.util.spec_from_file_location(
module_name,
module_location,
submodule_search_locations=submodule_search_locations,
)
if spec is None or spec.loader is None:
return False
original_modules = _swap_modules_for_preference(module_name)
try:
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
_repair_parent_package_binding(module_name, module)
logger.info("Loaded %s from plugin site-packages: %s", module_name, module_location)
return True
except Exception:
_restore_modules_after_failure(module_name, original_modules)
raise
```
再配合这些小辅助函数：
```python
def _repair_parent_package_binding(module_name: str, module: types.ModuleType) -> None:
if "." not in module_name:
return
parent_name, child_name = module_name.rsplit(".", 1)
parent_module = sys.modules.get(parent_name)
if parent_module is not None:
setattr(parent_module, child_name, module)
def _restore_modules_after_failure(module_name: str, original_modules: dict[str, types.ModuleType]) -> None:
failed_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
for key in failed_keys:
sys.modules.pop(key, None)
sys.modules.update(original_modules)
```
这样既保留现有所有行为（包括回滚语义），又让每个操作都更容易单独理解和测试。
---
这些改动不会减少*行为*层面的功能或复杂度，但会将复杂性更加本地化，对外呈现一个更小、更清晰的 API。
</issue_to_address>
### 评论 2
<location> `astrbot/core/star/star_manager.py:648` </location>
<code_context>
self.failed_plugin_info = fail_rec
return False, fail_rec
+ async def _cleanup_failed_plugin_install(
+ self,
+ dir_name: str,
</code_context>
<issue_to_address>
**issue (complexity):** 请考虑将共享的插件查找/清理以及信息构建逻辑提取为一些小而聚焦的辅助函数，在保留新行为的同时简化安装流程中的代码路径。
你可以在保持新行为不变的前提下，通过提取重复逻辑并拆分当前“多职责”的辅助函数，大幅降低复杂度。
### 1. 将 `_cleanup_failed_plugin_install` 拆分为更聚焦的辅助函数
目前 `_cleanup_failed_plugin_install` 同时负责：查找、终止/解绑、目录清理、配置清理。可以保持行为不变，但让逻辑更易于跟踪和复用：
```python
async def _cleanup_failed_plugin_install(
self,
dir_name: str,
plugin_path: str,
) -> None:
plugin = self._find_plugin_by_dir(dir_name)
if plugin:
await self._terminate_and_unbind_plugin(plugin)
await self._remove_plugin_directory(plugin_path)
await self._remove_plugin_config(dir_name)
def _find_plugin_by_dir(self, dir_name: str):
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
return star
return None
async def _terminate_and_unbind_plugin(self, plugin) -> None:
if plugin and plugin.name and plugin.module_path:
try:
await self._terminate_plugin(plugin)
except Exception:
logger.warning(traceback.format_exc())
try:
await self._unbind_plugin(plugin.name, plugin.module_path)
except Exception:
logger.warning(traceback.format_exc())
async def _remove_plugin_directory(self, plugin_path: str) -> None:
if os.path.exists(plugin_path):
try:
remove_dir(plugin_path)
logger.warning(f"已清理安装失败的插件目录: {plugin_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件目录失败: {plugin_path}，原因: {e!s}",
)
async def _remove_plugin_config(self, dir_name: str) -> None:
plugin_config_path = os.path.join(
self.plugin_config_path,
f"{dir_name}_config.json",
)
if os.path.exists(plugin_config_path):
try:
os.remove(plugin_config_path)
logger.warning(f"已清理安装失败插件配置: {plugin_config_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}",
)
```
### 2. 抽取共享的 `plugin_info` 构建逻辑
`install_plugin` 和 `install_plugin_from_file` 都重复实现了 “查找插件 + 读取 README + 构建 dict” 的逻辑。可以将其集中到一个辅助函数中，在不改变行为的前提下复用：
```python
def _get_plugin_by_dir(self, dir_name: str):
plugin = self.context.get_registered_star(dir_name)
if plugin:
return plugin
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
return star
return None
def _read_plugin_readme(self, plugin_path: str, dir_name: str) -> str | None:
readme_path = os.path.join(plugin_path, "README.md")
if not os.path.exists(readme_path):
readme_path = os.path.join(plugin_path, "readme.md")
if not os.path.exists(readme_path):
return None
try:
with open(readme_path, encoding="utf-8") as f:
return f.read()
except Exception as e:
logger.warning(f"读取插件 {dir_name} 的 README.md 文件失败: {e!s}")
return None
def _build_plugin_info(self, dir_name: str, plugin_path: str):
plugin = self._get_plugin_by_dir(dir_name)
if not plugin:
return None
readme_content = self._read_plugin_readme(plugin_path, dir_name)
return {
"repo": plugin.repo,
"readme": readme_content,
"name": plugin.name,
}
```
这样 `install_plugin` 会简单很多：
```python
async def install_plugin(self, repo_url: str, proxy=""):
asyncio.create_task(
Metric.upload(et="install_star", repo=repo_url),
)
async with self._pm_lock:
plugin_path = ""
dir_name = ""
cleanup_required = False
try:
plugin_path = await self.updator.install(repo_url, proxy)
cleanup_required = True
dir_name = os.path.basename(plugin_path)
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return self._build_plugin_info(dir_name, plugin_path)
except Exception:
if cleanup_required and dir_name and plugin_path:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
)
raise
```
而 `install_plugin_from_file` 也可以复用同一个辅助函数：
```python
async def install_plugin_from_file(self, zip_file_path: str):
dir_name = os.path.basename(zip_file_path).replace(".zip", "")
dir_name = dir_name.removesuffix("-master").removesuffix("-main").lower()
desti_dir = os.path.join(self.plugin_store_path, dir_name)
cleanup_required = False
# existing_plugin termination logic unchanged...
# ...
try:
self.updator.unzip_file(zip_file_path, desti_dir)
cleanup_required = True
# metadata / same-name check logic unchanged...
# ...
try:
os.remove(zip_file_path)
except BaseException as e:
logger.warning(f"删除插件压缩包失败: {e!s}")
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
plugin_info = self._build_plugin_info(dir_name, desti_dir)
if plugin_info and plugin_info.get("repo"):
asyncio.create_task(
Metric.upload(
et="install_star_f",
repo=plugin_info["repo"],
),
)
return plugin_info
except Exception:
if cleanup_required:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
)
raise
```
在保留所有新的清理和错误处理行为的同时：
- `_cleanup_failed_plugin_install` 现在更像一个可读性良好的“总控”函数，而不是一个职责混杂的长函数。
- `install_plugin` 和 `install_plugin_from_file` 的主体更短，并且共享了一处关于插件查找 + README 读取 + 信息构建的“单一事实来源”。
</issue_to_address>帮我变得更有用！请在每条评论下点 👍 或 👎，我会根据你的反馈改进后续评审。
Original comment in English
Hey - I've found 2 issues, and left some high level feedback:
- The new dependency-preference helpers in
pip_installer.py(_collect_candidate_modules / _ensure_plugin_dependencies_preferred / _prefer_modules_from_site_packages` etc.) are quite large and interdependent; consider breaking them into smaller, focused utilities (e.g. requirement parsing vs. metadata expansion vs. import override) to make future reasoning and debugging easier. - Both plugin installation paths (
install_pluginandinstall_plugin_from_file) now contain similar success/cleanup and README-extraction logic around_cleanup_failed_plugin_install; it may be worth extracting a shared helper to reduce duplication and keep the plugin lifecycle behavior consistent in one place. _check_plugin_dept_updatenow unconditionally returnsTrueeven when individual dependency installs fail (only logging errors); if callers use this return value for control flow, consider whether you want to propagate a failure indicator instead of always signaling success.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The new dependency-preference helpers in `pip_installer.py` (_collect_candidate_modules / _ensure_plugin_dependencies_preferred / _prefer_modules_from_site_packages` etc.) are quite large and interdependent; consider breaking them into smaller, focused utilities (e.g. requirement parsing vs. metadata expansion vs. import override) to make future reasoning and debugging easier.
- Both plugin installation paths (`install_plugin` and `install_plugin_from_file`) now contain similar success/cleanup and README-extraction logic around `_cleanup_failed_plugin_install`; it may be worth extracting a shared helper to reduce duplication and keep the plugin lifecycle behavior consistent in one place.
- `_check_plugin_dept_update` now unconditionally returns `True` even when individual dependency installs fail (only logging errors); if callers use this return value for control flow, consider whether you want to propagate a failure indicator instead of always signaling success.
## Individual Comments
### Comment 1
<location> `astrbot/core/utils/pip_installer.py:412` </location>
<code_context>
+ return final_unresolved
+
+
+def _ensure_plugin_dependencies_preferred(
+ target_site_packages: str,
+ requested_requirements: set[str],
</code_context>
<issue_to_address>
**issue (complexity):** Consider encapsulating the dependency-preference orchestration into a manager class and breaking the low-level import manipulation into smaller helpers to present a simpler, clearer public flow.
You’ve added a lot of power here, but it’s now spread across many tightly-coupled helpers. You can reduce the cognitive load without changing behavior by:
1. **Encapsulating the orchestration into a small object**
2. **Splitting low-level import surgery into smaller, focused helpers**
Both keep all features (dependency expansion, recovery, retries) but present a smaller public surface and clearer flow.
---
### 1. Encapsulate orchestration into a manager
Right now, `_ensure_plugin_dependencies_preferred` + `_collect_candidate_modules` + `_ensure_preferred_modules` + `_prefer_modules_from_site_packages` + `_prefer_module_with_dependency_recovery` form a de-facto state machine.
Consider grouping these into a small manager class and exposing only a single “entry” method from the module. This keeps all existing logic but hides most of it behind an instance:
```python
class _PluginEnvManager:
def __init__(self, site_packages_path: str):
self.site_packages_path = site_packages_path
def ensure_dependencies_preferred(self, requested_requirements: set[str]) -> None:
if not requested_requirements:
return
candidate_modules = self._collect_candidate_modules(requested_requirements)
if not candidate_modules:
return
self._ensure_preferred_modules(candidate_modules)
# move existing helpers as methods, e.g.:
def _collect_candidate_modules(self, requirement_names: set[str]) -> set[str]:
return _collect_candidate_modules(requirement_names, self.site_packages_path)
def _ensure_preferred_modules(self, module_names: set[str]) -> None:
_ensure_preferred_modules(module_names, self.site_packages_path)
```
Then the external call site becomes simpler and the module-level surface shrinks:
```python
if target_site_packages:
_prepend_sys_path(target_site_packages)
manager = _PluginEnvManager(target_site_packages)
manager.ensure_dependencies_preferred(requested_requirements)
```
Next steps (optional, but helpful):
- Gradually inline/port `_collect_candidate_modules`, `_ensure_preferred_modules`, `_prefer_modules_from_site_packages`, `_prefer_module_with_dependency_recovery` into the class as private methods, and remove their module-level definitions once all call sites use the manager.
- This will make the high-level flow discoverable in one place (`ensure_dependencies_preferred`), instead of jumping through many top-level helpers.
---
### 2. Split low-level import manipulation into smaller helpers
`_prefer_module_from_site_packages` currently does:
- path resolution (package vs module),
- spec creation,
- `sys.modules` diffing/removal,
- module loading,
- parent-child repair,
- rollback on failure.
You can make this more readable and easier to reason about by separating concerns. Behavior can remain identical:
```python
def _resolve_module_location(module_name: str, site_packages_path: str) -> tuple[str | None, list[str] | None]:
base_path = os.path.join(site_packages_path, *module_name.split("."))
package_init = os.path.join(base_path, "__init__.py")
module_file = f"{base_path}.py"
if os.path.isfile(package_init):
return package_init, [os.path.dirname(package_init)]
if os.path.isfile(module_file):
return module_file, None
return None, None
def _swap_modules_for_preference(module_name: str) -> dict[str, types.ModuleType]:
matched_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
original_modules = {key: sys.modules[key] for key in matched_keys}
for key in matched_keys:
sys.modules.pop(key, None)
return original_modules
```
Then `_prefer_module_from_site_packages` becomes structurally simpler:
```python
def _prefer_module_from_site_packages(module_name: str, site_packages_path: str) -> bool:
with _SITE_PACKAGES_IMPORT_LOCK:
module_location, submodule_search_locations = _resolve_module_location(
module_name, site_packages_path
)
if not module_location:
return False
spec = importlib.util.spec_from_file_location(
module_name,
module_location,
submodule_search_locations=submodule_search_locations,
)
if spec is None or spec.loader is None:
return False
original_modules = _swap_modules_for_preference(module_name)
try:
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
_repair_parent_package_binding(module_name, module)
logger.info("Loaded %s from plugin site-packages: %s", module_name, module_location)
return True
except Exception:
_restore_modules_after_failure(module_name, original_modules)
raise
```
With small helpers like:
```python
def _repair_parent_package_binding(module_name: str, module: types.ModuleType) -> None:
if "." not in module_name:
return
parent_name, child_name = module_name.rsplit(".", 1)
parent_module = sys.modules.get(parent_name)
if parent_module is not None:
setattr(parent_module, child_name, module)
def _restore_modules_after_failure(module_name: str, original_modules: dict[str, types.ModuleType]) -> None:
failed_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
for key in failed_keys:
sys.modules.pop(key, None)
sys.modules.update(original_modules)
```
This keeps all the existing behavior (including rollback semantics) but makes each operation easier to reason about and test in isolation.
---
These changes don’t remove any functionality or complexity of the *behavior*, but they do localize the complexity and present a smaller, clearer API to the rest of the module.
</issue_to_address>
### Comment 2
<location> `astrbot/core/star/star_manager.py:648` </location>
<code_context>
self.failed_plugin_info = fail_rec
return False, fail_rec
+ async def _cleanup_failed_plugin_install(
+ self,
+ dir_name: str,
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the shared plugin lookup/cleanup and info-building logic into small focused helpers to keep the new behavior while simplifying the installer code paths.
You can keep the new behavior and significantly reduce complexity by extracting the duplicated pieces and splitting the mixed-responsibility helper.
### 1. Split `_cleanup_failed_plugin_install` into focused helpers
Right now `_cleanup_failed_plugin_install` mixes: lookup, terminate/unbind, directory cleanup, config cleanup. You can keep behavior but make it easier to follow and re‑use:
```python
async def _cleanup_failed_plugin_install(
self,
dir_name: str,
plugin_path: str,
) -> None:
plugin = self._find_plugin_by_dir(dir_name)
if plugin:
await self._terminate_and_unbind_plugin(plugin)
await self._remove_plugin_directory(plugin_path)
await self._remove_plugin_config(dir_name)
def _find_plugin_by_dir(self, dir_name: str):
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
return star
return None
async def _terminate_and_unbind_plugin(self, plugin) -> None:
if plugin and plugin.name and plugin.module_path:
try:
await self._terminate_plugin(plugin)
except Exception:
logger.warning(traceback.format_exc())
try:
await self._unbind_plugin(plugin.name, plugin.module_path)
except Exception:
logger.warning(traceback.format_exc())
async def _remove_plugin_directory(self, plugin_path: str) -> None:
if os.path.exists(plugin_path):
try:
remove_dir(plugin_path)
logger.warning(f"已清理安装失败的插件目录: {plugin_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件目录失败: {plugin_path}，原因: {e!s}",
)
async def _remove_plugin_config(self, dir_name: str) -> None:
plugin_config_path = os.path.join(
self.plugin_config_path,
f"{dir_name}_config.json",
)
if os.path.exists(plugin_config_path):
try:
os.remove(plugin_config_path)
logger.warning(f"已清理安装失败插件配置: {plugin_config_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}",
)
```
### 2. Extract shared `plugin_info` building logic
Both `install_plugin` and `install_plugin_from_file` duplicate “find plugin + read README + build dict”. You can centralize this into one helper that keeps the same behavior:
```python
def _get_plugin_by_dir(self, dir_name: str):
plugin = self.context.get_registered_star(dir_name)
if plugin:
return plugin
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
return star
return None
def _read_plugin_readme(self, plugin_path: str, dir_name: str) -> str | None:
readme_path = os.path.join(plugin_path, "README.md")
if not os.path.exists(readme_path):
readme_path = os.path.join(plugin_path, "readme.md")
if not os.path.exists(readme_path):
return None
try:
with open(readme_path, encoding="utf-8") as f:
return f.read()
except Exception as e:
logger.warning(f"读取插件 {dir_name} 的 README.md 文件失败: {e!s}")
return None
def _build_plugin_info(self, dir_name: str, plugin_path: str):
plugin = self._get_plugin_by_dir(dir_name)
if not plugin:
return None
readme_content = self._read_plugin_readme(plugin_path, dir_name)
return {
"repo": plugin.repo,
"readme": readme_content,
"name": plugin.name,
}
```
Then `install_plugin` becomes simpler:
```python
async def install_plugin(self, repo_url: str, proxy=""):
asyncio.create_task(
Metric.upload(et="install_star", repo=repo_url),
)
async with self._pm_lock:
plugin_path = ""
dir_name = ""
cleanup_required = False
try:
plugin_path = await self.updator.install(repo_url, proxy)
cleanup_required = True
dir_name = os.path.basename(plugin_path)
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return self._build_plugin_info(dir_name, plugin_path)
except Exception:
if cleanup_required and dir_name and plugin_path:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
)
raise
```
And `install_plugin_from_file` reuses the same helper:
```python
async def install_plugin_from_file(self, zip_file_path: str):
dir_name = os.path.basename(zip_file_path).replace(".zip", "")
dir_name = dir_name.removesuffix("-master").removesuffix("-main").lower()
desti_dir = os.path.join(self.plugin_store_path, dir_name)
cleanup_required = False
# existing_plugin termination logic unchanged...
# ...
try:
self.updator.unzip_file(zip_file_path, desti_dir)
cleanup_required = True
# metadata / same-name check logic unchanged...
# ...
try:
os.remove(zip_file_path)
except BaseException as e:
logger.warning(f"删除插件压缩包失败: {e!s}")
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
plugin_info = self._build_plugin_info(dir_name, desti_dir)
if plugin_info and plugin_info.get("repo"):
asyncio.create_task(
Metric.upload(
et="install_star_f",
repo=plugin_info["repo"],
),
)
return plugin_info
except Exception:
if cleanup_required:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
)
raise
```
This keeps all the new cleanup and error‑handling behavior, but:
- `_cleanup_failed_plugin_install` is now a readable orchestrator instead of a long mixed block.
- The `install_plugin` and `install_plugin_from_file` bodies are shorter and share a single source of truth for plugin lookup + README + info building.
</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 final_unresolved | ||
|
|
||
|
|
||
| def _ensure_plugin_dependencies_preferred( |
Contributor
There was a problem hiding this comment.
issue (complexity): 请考虑将依赖优先级的编排逻辑封装到一个管理类中，并把底层的导入操作拆分成更小的辅助函数，以提供更简单、更清晰的对外流程。
这里新增的能力很强大，但现在被分散在多个紧密耦合的辅助函数中。在不改变行为的前提下，可以通过以下方式降低理解成本：
- 将编排逻辑封装到一个小对象中
- 把底层的导入操作拆分为更小、更聚焦的辅助函数
这样既保留所有特性（依赖展开、恢复、重试），又能对外暴露更小的接口面，流程更加清晰。
1. 将编排逻辑封装进 manager
目前 _ensure_plugin_dependencies_preferred + _collect_candidate_modules + _ensure_preferred_modules + _prefer_modules_from_site_packages + _prefer_module_with_dependency_recovery 实际上组成了一个状态机。
可以考虑把这些组合为一个小的管理类，并从模块中只暴露一个“入口”方法。这样可以保留所有现有逻辑，但把大部分细节隐藏在实例内部：
class _PluginEnvManager:
def __init__(self, site_packages_path: str):
self.site_packages_path = site_packages_path
def ensure_dependencies_preferred(self, requested_requirements: set[str]) -> None:
if not requested_requirements:
return
candidate_modules = self._collect_candidate_modules(requested_requirements)
if not candidate_modules:
return
self._ensure_preferred_modules(candidate_modules)
# move existing helpers as methods, e.g.:
def _collect_candidate_modules(self, requirement_names: set[str]) -> set[str]:
return _collect_candidate_modules(requirement_names, self.site_packages_path)
def _ensure_preferred_modules(self, module_names: set[str]) -> None:
_ensure_preferred_modules(module_names, self.site_packages_path)这样外部调用点会更简单，模块级的对外接口也会缩小：
if target_site_packages:
_prepend_sys_path(target_site_packages)
manager = _PluginEnvManager(target_site_packages)
manager.ensure_dependencies_preferred(requested_requirements)后续步骤（可选但有帮助）：
- 逐步将
_collect_candidate_modules、_ensure_preferred_modules、_prefer_modules_from_site_packages、_prefer_module_with_dependency_recovery内联/迁移为该类的私有方法，并在所有调用点都使用 manager 后移除这些模块级函数定义。 - 这样高层流程就能在一个地方（
ensure_dependencies_preferred）被完整看到，而不是在多个顶层辅助函数之间来回跳转。
2. 将底层导入操作拆分为更小的辅助函数
_prefer_module_from_site_packages 目前同时负责：
- 路径解析（包 vs 模块）、
- 创建 spec、
sys.modules对比/删除、- 加载模块、
- 父子包关系修复、
- 失败时回滚。
可以通过职责分离让它更易读、更易于推理；在行为保持不变的前提下，可以这样拆分：
def _resolve_module_location(module_name: str, site_packages_path: str) -> tuple[str | None, list[str] | None]:
base_path = os.path.join(site_packages_path, *module_name.split("."))
package_init = os.path.join(base_path, "__init__.py")
module_file = f"{base_path}.py"
if os.path.isfile(package_init):
return package_init, [os.path.dirname(package_init)]
if os.path.isfile(module_file):
return module_file, None
return None, None
def _swap_modules_for_preference(module_name: str) -> dict[str, types.ModuleType]:
matched_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
original_modules = {key: sys.modules[key] for key in matched_keys}
for key in matched_keys:
sys.modules.pop(key, None)
return original_modules然后 _prefer_module_from_site_packages 在结构上会简单很多：
def _prefer_module_from_site_packages(module_name: str, site_packages_path: str) -> bool:
with _SITE_PACKAGES_IMPORT_LOCK:
module_location, submodule_search_locations = _resolve_module_location(
module_name, site_packages_path
)
if not module_location:
return False
spec = importlib.util.spec_from_file_location(
module_name,
module_location,
submodule_search_locations=submodule_search_locations,
)
if spec is None or spec.loader is None:
return False
original_modules = _swap_modules_for_preference(module_name)
try:
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
_repair_parent_package_binding(module_name, module)
logger.info("Loaded %s from plugin site-packages: %s", module_name, module_location)
return True
except Exception:
_restore_modules_after_failure(module_name, original_modules)
raise再配合这些小辅助函数：
def _repair_parent_package_binding(module_name: str, module: types.ModuleType) -> None:
if "." not in module_name:
return
parent_name, child_name = module_name.rsplit(".", 1)
parent_module = sys.modules.get(parent_name)
if parent_module is not None:
setattr(parent_module, child_name, module)
def _restore_modules_after_failure(module_name: str, original_modules: dict[str, types.ModuleType]) -> None:
failed_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
for key in failed_keys:
sys.modules.pop(key, None)
sys.modules.update(original_modules)这样既保留现有所有行为（包括回滚语义），又让每个操作都更容易单独理解和测试。
这些改动不会减少行为层面的功能或复杂度，但会将复杂性更加本地化，对外呈现一个更小、更清晰的 API。
Original comment in English
issue (complexity): Consider encapsulating the dependency-preference orchestration into a manager class and breaking the low-level import manipulation into smaller helpers to present a simpler, clearer public flow.
You’ve added a lot of power here, but it’s now spread across many tightly-coupled helpers. You can reduce the cognitive load without changing behavior by:
- Encapsulating the orchestration into a small object
- Splitting low-level import surgery into smaller, focused helpers
Both keep all features (dependency expansion, recovery, retries) but present a smaller public surface and clearer flow.
1. Encapsulate orchestration into a manager
Right now, _ensure_plugin_dependencies_preferred + _collect_candidate_modules + _ensure_preferred_modules + _prefer_modules_from_site_packages + _prefer_module_with_dependency_recovery form a de-facto state machine.
Consider grouping these into a small manager class and exposing only a single “entry” method from the module. This keeps all existing logic but hides most of it behind an instance:
class _PluginEnvManager:
def __init__(self, site_packages_path: str):
self.site_packages_path = site_packages_path
def ensure_dependencies_preferred(self, requested_requirements: set[str]) -> None:
if not requested_requirements:
return
candidate_modules = self._collect_candidate_modules(requested_requirements)
if not candidate_modules:
return
self._ensure_preferred_modules(candidate_modules)
# move existing helpers as methods, e.g.:
def _collect_candidate_modules(self, requirement_names: set[str]) -> set[str]:
return _collect_candidate_modules(requirement_names, self.site_packages_path)
def _ensure_preferred_modules(self, module_names: set[str]) -> None:
_ensure_preferred_modules(module_names, self.site_packages_path)Then the external call site becomes simpler and the module-level surface shrinks:
if target_site_packages:
_prepend_sys_path(target_site_packages)
manager = _PluginEnvManager(target_site_packages)
manager.ensure_dependencies_preferred(requested_requirements)Next steps (optional, but helpful):
- Gradually inline/port
_collect_candidate_modules,_ensure_preferred_modules,_prefer_modules_from_site_packages,_prefer_module_with_dependency_recoveryinto the class as private methods, and remove their module-level definitions once all call sites use the manager. - This will make the high-level flow discoverable in one place (
ensure_dependencies_preferred), instead of jumping through many top-level helpers.
2. Split low-level import manipulation into smaller helpers
_prefer_module_from_site_packages currently does:
- path resolution (package vs module),
- spec creation,
sys.modulesdiffing/removal,- module loading,
- parent-child repair,
- rollback on failure.
You can make this more readable and easier to reason about by separating concerns. Behavior can remain identical:
def _resolve_module_location(module_name: str, site_packages_path: str) -> tuple[str | None, list[str] | None]:
base_path = os.path.join(site_packages_path, *module_name.split("."))
package_init = os.path.join(base_path, "__init__.py")
module_file = f"{base_path}.py"
if os.path.isfile(package_init):
return package_init, [os.path.dirname(package_init)]
if os.path.isfile(module_file):
return module_file, None
return None, None
def _swap_modules_for_preference(module_name: str) -> dict[str, types.ModuleType]:
matched_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
original_modules = {key: sys.modules[key] for key in matched_keys}
for key in matched_keys:
sys.modules.pop(key, None)
return original_modulesThen _prefer_module_from_site_packages becomes structurally simpler:
def _prefer_module_from_site_packages(module_name: str, site_packages_path: str) -> bool:
with _SITE_PACKAGES_IMPORT_LOCK:
module_location, submodule_search_locations = _resolve_module_location(
module_name, site_packages_path
)
if not module_location:
return False
spec = importlib.util.spec_from_file_location(
module_name,
module_location,
submodule_search_locations=submodule_search_locations,
)
if spec is None or spec.loader is None:
return False
original_modules = _swap_modules_for_preference(module_name)
try:
module = importlib.util.module_from_spec(spec)
sys.modules[module_name] = module
spec.loader.exec_module(module)
_repair_parent_package_binding(module_name, module)
logger.info("Loaded %s from plugin site-packages: %s", module_name, module_location)
return True
except Exception:
_restore_modules_after_failure(module_name, original_modules)
raiseWith small helpers like:
def _repair_parent_package_binding(module_name: str, module: types.ModuleType) -> None:
if "." not in module_name:
return
parent_name, child_name = module_name.rsplit(".", 1)
parent_module = sys.modules.get(parent_name)
if parent_module is not None:
setattr(parent_module, child_name, module)
def _restore_modules_after_failure(module_name: str, original_modules: dict[str, types.ModuleType]) -> None:
failed_keys = [
key for key in list(sys.modules.keys())
if key == module_name or key.startswith(f"{module_name}.")
]
for key in failed_keys:
sys.modules.pop(key, None)
sys.modules.update(original_modules)This keeps all the existing behavior (including rollback semantics) but makes each operation easier to reason about and test in isolation.
These changes don’t remove any functionality or complexity of the behavior, but they do localize the complexity and present a smaller, clearer API to the rest of the module.
| self.failed_plugin_info = fail_rec | ||
| return False, fail_rec | ||
|
|
||
| async def _cleanup_failed_plugin_install( |
Contributor
There was a problem hiding this comment.
issue (complexity): 请考虑将共享的插件查找/清理以及信息构建逻辑提取为一些小而聚焦的辅助函数，在保留新行为的同时简化安装流程中的代码路径。
你可以在保持新行为不变的前提下，通过提取重复逻辑并拆分当前“多职责”的辅助函数，大幅降低复杂度。
1. 将 _cleanup_failed_plugin_install 拆分为更聚焦的辅助函数
目前 _cleanup_failed_plugin_install 同时负责：查找、终止/解绑、目录清理、配置清理。可以保持行为不变，但让逻辑更易于跟踪和复用：
async def _cleanup_failed_plugin_install(
self,
dir_name: str,
plugin_path: str,
) -> None:
plugin = self._find_plugin_by_dir(dir_name)
if plugin:
await self._terminate_and_unbind_plugin(plugin)
await self._remove_plugin_directory(plugin_path)
await self._remove_plugin_config(dir_name)
def _find_plugin_by_dir(self, dir_name: str):
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
return star
return None
async def _terminate_and_unbind_plugin(self, plugin) -> None:
if plugin and plugin.name and plugin.module_path:
try:
await self._terminate_plugin(plugin)
except Exception:
logger.warning(traceback.format_exc())
try:
await self._unbind_plugin(plugin.name, plugin.module_path)
except Exception:
logger.warning(traceback.format_exc())
async def _remove_plugin_directory(self, plugin_path: str) -> None:
if os.path.exists(plugin_path):
try:
remove_dir(plugin_path)
logger.warning(f"已清理安装失败的插件目录: {plugin_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件目录失败: {plugin_path}，原因: {e!s}",
)
async def _remove_plugin_config(self, dir_name: str) -> None:
plugin_config_path = os.path.join(
self.plugin_config_path,
f"{dir_name}_config.json",
)
if os.path.exists(plugin_config_path):
try:
os.remove(plugin_config_path)
logger.warning(f"已清理安装失败插件配置: {plugin_config_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}",
)2. 抽取共享的 plugin_info 构建逻辑
install_plugin 和 install_plugin_from_file 都重复实现了 “查找插件 + 读取 README + 构建 dict” 的逻辑。可以将其集中到一个辅助函数中，在不改变行为的前提下复用：
def _get_plugin_by_dir(self, dir_name: str):
plugin = self.context.get_registered_star(dir_name)
if plugin:
return plugin
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
return star
return None
def _read_plugin_readme(self, plugin_path: str, dir_name: str) -> str | None:
readme_path = os.path.join(plugin_path, "README.md")
if not os.path.exists(readme_path):
readme_path = os.path.join(plugin_path, "readme.md")
if not os.path.exists(readme_path):
return None
try:
with open(readme_path, encoding="utf-8") as f:
return f.read()
except Exception as e:
logger.warning(f"读取插件 {dir_name} 的 README.md 文件失败: {e!s}")
return None
def _build_plugin_info(self, dir_name: str, plugin_path: str):
plugin = self._get_plugin_by_dir(dir_name)
if not plugin:
return None
readme_content = self._read_plugin_readme(plugin_path, dir_name)
return {
"repo": plugin.repo,
"readme": readme_content,
"name": plugin.name,
}这样 install_plugin 会简单很多：
async def install_plugin(self, repo_url: str, proxy=""):
asyncio.create_task(
Metric.upload(et="install_star", repo=repo_url),
)
async with self._pm_lock:
plugin_path = ""
dir_name = ""
cleanup_required = False
try:
plugin_path = await self.updator.install(repo_url, proxy)
cleanup_required = True
dir_name = os.path.basename(plugin_path)
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return self._build_plugin_info(dir_name, plugin_path)
except Exception:
if cleanup_required and dir_name and plugin_path:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
)
raise而 install_plugin_from_file 也可以复用同一个辅助函数：
async def install_plugin_from_file(self, zip_file_path: str):
dir_name = os.path.basename(zip_file_path).replace(".zip", "")
dir_name = dir_name.removesuffix("-master").removesuffix("-main").lower()
desti_dir = os.path.join(self.plugin_store_path, dir_name)
cleanup_required = False
# existing_plugin termination logic unchanged...
# ...
try:
self.updator.unzip_file(zip_file_path, desti_dir)
cleanup_required = True
# metadata / same-name check logic unchanged...
# ...
try:
os.remove(zip_file_path)
except BaseException as e:
logger.warning(f"删除插件压缩包失败: {e!s}")
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
plugin_info = self._build_plugin_info(dir_name, desti_dir)
if plugin_info and plugin_info.get("repo"):
asyncio.create_task(
Metric.upload(
et="install_star_f",
repo=plugin_info["repo"],
),
)
return plugin_info
except Exception:
if cleanup_required:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
)
raise在保留所有新的清理和错误处理行为的同时：
_cleanup_failed_plugin_install现在更像一个可读性良好的“总控”函数，而不是一个职责混杂的长函数。install_plugin和install_plugin_from_file的主体更短，并且共享了一处关于插件查找 + README 读取 + 信息构建的“单一事实来源”。
Original comment in English
issue (complexity): Consider extracting the shared plugin lookup/cleanup and info-building logic into small focused helpers to keep the new behavior while simplifying the installer code paths.
You can keep the new behavior and significantly reduce complexity by extracting the duplicated pieces and splitting the mixed-responsibility helper.
1. Split _cleanup_failed_plugin_install into focused helpers
Right now _cleanup_failed_plugin_install mixes: lookup, terminate/unbind, directory cleanup, config cleanup. You can keep behavior but make it easier to follow and re‑use:
async def _cleanup_failed_plugin_install(
self,
dir_name: str,
plugin_path: str,
) -> None:
plugin = self._find_plugin_by_dir(dir_name)
if plugin:
await self._terminate_and_unbind_plugin(plugin)
await self._remove_plugin_directory(plugin_path)
await self._remove_plugin_config(dir_name)
def _find_plugin_by_dir(self, dir_name: str):
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
return star
return None
async def _terminate_and_unbind_plugin(self, plugin) -> None:
if plugin and plugin.name and plugin.module_path:
try:
await self._terminate_plugin(plugin)
except Exception:
logger.warning(traceback.format_exc())
try:
await self._unbind_plugin(plugin.name, plugin.module_path)
except Exception:
logger.warning(traceback.format_exc())
async def _remove_plugin_directory(self, plugin_path: str) -> None:
if os.path.exists(plugin_path):
try:
remove_dir(plugin_path)
logger.warning(f"已清理安装失败的插件目录: {plugin_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件目录失败: {plugin_path}，原因: {e!s}",
)
async def _remove_plugin_config(self, dir_name: str) -> None:
plugin_config_path = os.path.join(
self.plugin_config_path,
f"{dir_name}_config.json",
)
if os.path.exists(plugin_config_path):
try:
os.remove(plugin_config_path)
logger.warning(f"已清理安装失败插件配置: {plugin_config_path}")
except Exception as e:
logger.warning(
f"清理安装失败插件配置失败: {plugin_config_path}，原因: {e!s}",
)2. Extract shared plugin_info building logic
Both install_plugin and install_plugin_from_file duplicate “find plugin + read README + build dict”. You can centralize this into one helper that keeps the same behavior:
def _get_plugin_by_dir(self, dir_name: str):
plugin = self.context.get_registered_star(dir_name)
if plugin:
return plugin
for star in self.context.get_all_stars():
if star.root_dir_name == dir_name:
return star
return None
def _read_plugin_readme(self, plugin_path: str, dir_name: str) -> str | None:
readme_path = os.path.join(plugin_path, "README.md")
if not os.path.exists(readme_path):
readme_path = os.path.join(plugin_path, "readme.md")
if not os.path.exists(readme_path):
return None
try:
with open(readme_path, encoding="utf-8") as f:
return f.read()
except Exception as e:
logger.warning(f"读取插件 {dir_name} 的 README.md 文件失败: {e!s}")
return None
def _build_plugin_info(self, dir_name: str, plugin_path: str):
plugin = self._get_plugin_by_dir(dir_name)
if not plugin:
return None
readme_content = self._read_plugin_readme(plugin_path, dir_name)
return {
"repo": plugin.repo,
"readme": readme_content,
"name": plugin.name,
}Then install_plugin becomes simpler:
async def install_plugin(self, repo_url: str, proxy=""):
asyncio.create_task(
Metric.upload(et="install_star", repo=repo_url),
)
async with self._pm_lock:
plugin_path = ""
dir_name = ""
cleanup_required = False
try:
plugin_path = await self.updator.install(repo_url, proxy)
cleanup_required = True
dir_name = os.path.basename(plugin_path)
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
return self._build_plugin_info(dir_name, plugin_path)
except Exception:
if cleanup_required and dir_name and plugin_path:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=plugin_path,
)
raiseAnd install_plugin_from_file reuses the same helper:
async def install_plugin_from_file(self, zip_file_path: str):
dir_name = os.path.basename(zip_file_path).replace(".zip", "")
dir_name = dir_name.removesuffix("-master").removesuffix("-main").lower()
desti_dir = os.path.join(self.plugin_store_path, dir_name)
cleanup_required = False
# existing_plugin termination logic unchanged...
# ...
try:
self.updator.unzip_file(zip_file_path, desti_dir)
cleanup_required = True
# metadata / same-name check logic unchanged...
# ...
try:
os.remove(zip_file_path)
except BaseException as e:
logger.warning(f"删除插件压缩包失败: {e!s}")
success, error_message = await self.load(specified_dir_name=dir_name)
if not success:
raise Exception(
error_message
or f"安装插件 {dir_name} 失败，请检查插件依赖或兼容性。"
)
plugin_info = self._build_plugin_info(dir_name, desti_dir)
if plugin_info and plugin_info.get("repo"):
asyncio.create_task(
Metric.upload(
et="install_star_f",
repo=plugin_info["repo"],
),
)
return plugin_info
except Exception:
if cleanup_required:
await self._cleanup_failed_plugin_install(
dir_name=dir_name,
plugin_path=desti_dir,
)
raiseThis keeps all the new cleanup and error‑handling behavior, but:
_cleanup_failed_plugin_installis now a readable orchestrator instead of a long mixed block.- The
install_pluginandinstall_plugin_from_filebodies are shorter and share a single source of truth for plugin lookup + README + info building.
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.
背景
近期在 Electron 打包版（尤其是 Windows/macOS 安装包）中，出现了两类高频问题：
ImportError / ModuleNotFoundError（典型如 fastapi / starlette 链路）。
在误杀非目标进程风险；带空格路径的 frozen 重启参数在部分场景下不稳定。
本次改动点
1) 插件依赖加载与冲突处理（pip_installer.py）
2) 插件安装失败清理（star_manager.py）
3) 桌面端重启链路统一（desktop/* + dashboard/*）
4) Windows 重启与清理安全性（backend-manager.js / updator.py）
结果与收益