fix: 插件依赖自动安装逻辑与 Dashboard 安装体验优化#5954
fix: 插件依赖自动安装逻辑与 Dashboard 安装体验优化#5954zouyonghe wants to merge 23 commits intoAstrBotDevs:masterfrom
Conversation
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request significantly enhances the stability and usability of plugin management by ensuring that all plugin dependencies are installed at the appropriate lifecycle stages. It also improves the flexibility and robustness of the pip installer by allowing more natural and varied input formats for package specifications, making the system more resilient to user input and ensuring a smoother plugin experience. Highlights
Changelog
Activity
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here. You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension. Footnotes
|
There was a problem hiding this comment.
Hey - 我发现了 3 个问题
给 AI Agent 的提示
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="137-146" />
<code_context>
return names
+def _split_package_install_input(raw_input: str) -> list[str]:
+ normalized = raw_input.strip()
+ if not normalized:
+ return []
+
+ if "\n" in normalized or "\r" in normalized:
+ return _split_multiline_package_input(normalized)
+
+ if _is_valid_install_requirement(normalized):
+ return [normalized]
+
+ split_tokens = shlex.split(normalized)
+ if split_tokens and all(_is_valid_install_token(token) for token in split_tokens):
+ return split_tokens
</code_context>
<issue_to_address>
**issue (bug_risk):** 带有选项和参数的单行输入(例如 `-e .` 或 `--index-url URL`)可能被错误分类,并作为单个 pip 参数传入。
由于 `_is_valid_install_token` 只会自动接受以 `-` 开头的 token,其他情况则要求能被 `Requirement` 解析,因此任何选项+值的组合(例如 `-e .`、`--index-url https://…`)都会在 `all(_is_valid_install_token(...))` 检查中失败。然后代码会回退为 `[normalized]`,这样 pip 会看到一个带空格的单一参数(`"-e ."`),这是无效的。为避免这种情况,可以放宽对选项参数的校验(允许选项+值对),或者只要输入是以 `-` 开头,就跳过逐 token 的 requirement 校验,直接使用 `shlex.split` 的结果。
</issue_to_address>
### Comment 2
<location path="astrbot/core/utils/pip_installer.py" line_range="163-167" />
<code_context>
+ return True
+
+
+def _is_valid_install_token(token: str) -> bool:
+ if token.startswith("-"):
+ return True
+
+ return _is_valid_install_requirement(token)
+
+
</code_context>
<issue_to_address>
**suggestion (bug_risk):** 类路径或基于文件的规格(例如 `.` 或 `../pkg`)即使和合法选项一起使用,也会被拒绝为无效 token。
由于非选项 token 会传给 `_is_valid_install_requirement`,像 `.` 或 `../pkg` 这样的路径值会被判定为无效,从而导致整个拆分结果被丢弃。这会破坏可编辑安装(`-e .`)以及类似“选项 + 路径”的用法。可以考虑对一些常见模式做特殊处理(例如 `-e` 后面的 `.`),或者对明显是路径/URL 的 token 放宽验证。另一种选择是:仅在单包规格时做严格校验,而对任意包含 `-` 的片段,将其视为完整的 pip CLI 调用并原样透传。
</issue_to_address>
### Comment 3
<location path="tests/test_pip_installer.py" line_range="127-136" />
<code_context>
+
+
+@pytest.mark.asyncio
+async def test_install_keeps_single_requirement_with_version_range_intact(monkeypatch):
+ run_pip = AsyncMock(return_value=0)
+
+ monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
+
+ installer = PipInstaller("")
+ await installer.install(package_name="demo-package >= 1.0, < 2.0")
+
+ run_pip.assert_awaited_once()
+ recorded_args = run_pip.await_args_list[0].args[0]
+
+ assert recorded_args[0:2] == ["install", "demo-package >= 1.0, < 2.0"]
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** 可以考虑为“部分无效”的、以空格分隔的输入添加一个回归测试,以验证回退行为。
当前测试覆盖了合法的空格分隔输入、多行输入以及单条复杂 requirement 字符串,但从未触发 `_split_package_install_input` 遇到无效 token、并回退为将整个字符串视作一个规格的分支。请添加一个带混合/部分无效字符串的回归测试(例如 `"demo-package another package with spaces"`),以确认我们会回退为 `[raw_input]`,而不是丢弃或错误拆分 token。
</issue_to_address>帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进评审质量。
Original comment in English
Hey - I've found 3 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="137-146" />
<code_context>
return names
+def _split_package_install_input(raw_input: str) -> list[str]:
+ normalized = raw_input.strip()
+ if not normalized:
+ return []
+
+ if "\n" in normalized or "\r" in normalized:
+ return _split_multiline_package_input(normalized)
+
+ if _is_valid_install_requirement(normalized):
+ return [normalized]
+
+ split_tokens = shlex.split(normalized)
+ if split_tokens and all(_is_valid_install_token(token) for token in split_tokens):
+ return split_tokens
</code_context>
<issue_to_address>
**issue (bug_risk):** Single-line input with options and arguments (e.g. `-e .` or `--index-url URL`) can be misclassified and passed as a single pip arg.
Because `_is_valid_install_token` only auto-accepts tokens starting with `-` and otherwise requires them to be `Requirement`-parsable, any option+value pair (e.g. `-e .`, `--index-url https://…`) fails the `all(_is_valid_install_token(...))` check. The code then falls back to `[normalized]`, so pip sees a single argument with spaces (`"-e ."`), which is invalid. To avoid this, either relax validation for option arguments (so option+value pairs are allowed) or bypass per-token requirement validation whenever the input starts with `-` and just use `shlex.split` as-is.
</issue_to_address>
### Comment 2
<location path="astrbot/core/utils/pip_installer.py" line_range="163-167" />
<code_context>
+ return True
+
+
+def _is_valid_install_token(token: str) -> bool:
+ if token.startswith("-"):
+ return True
+
+ return _is_valid_install_requirement(token)
+
+
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Path-like or file-based specs (e.g. `.` or `../pkg`) are rejected as invalid tokens even when used with valid options.
Since non-option tokens are passed to `_is_valid_install_requirement`, path-like values such as `.` or `../pkg` will be rejected and the whole split discarded. This breaks editable installs (`-e .`) and similar option+path usages. Consider either special-casing common patterns (e.g. `.` after `-e`) or relaxing validation for tokens that clearly look like paths/URLs. Another option is to apply strict validation only for single-package specs and otherwise treat any fragment containing `-` as a full pip CLI invocation to pass through unchanged.
</issue_to_address>
### Comment 3
<location path="tests/test_pip_installer.py" line_range="127-136" />
<code_context>
+
+
+@pytest.mark.asyncio
+async def test_install_keeps_single_requirement_with_version_range_intact(monkeypatch):
+ run_pip = AsyncMock(return_value=0)
+
+ monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
+
+ installer = PipInstaller("")
+ await installer.install(package_name="demo-package >= 1.0, < 2.0")
+
+ run_pip.assert_awaited_once()
+ recorded_args = run_pip.await_args_list[0].args[0]
+
+ assert recorded_args[0:2] == ["install", "demo-package >= 1.0, < 2.0"]
+
+
</code_context>
<issue_to_address>
**suggestion (testing):** Consider a regression test for partially-invalid space-separated input to verify fallback behavior.
Current tests cover valid space-separated input, multiline input, and single complex requirement strings, but they never exercise the branch where `_split_package_install_input` encounters an invalid token and falls back to treating the whole string as one spec. Please add a regression test with a mixed/partially invalid string (e.g. `"demo-package another package with spaces"`) to confirm we fall back to `[raw_input]` instead of dropping or mis-splitting tokens.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
There was a problem hiding this comment.
Code Review
This Pull Request effectively addresses the issue of installing plugin dependencies before loading them, covering various scenarios such as new installations, installations from zip files, reloading after failures, and updates. While the code refactoring is clear, the pip command input parsing is more robust, and new regression tests improve quality, two critical security issues were identified in the pip_installer.py file. These include the logging of sensitive information (credentials in URLs) when executing pip commands and potential pip option injection if the dashboard input is not properly restricted. Addressing these vulnerabilities is crucial to prevent credential leakage and unauthorized code execution. A minor issue regarding exception handling was also noted in the specific comments.
| package_specs = _split_package_install_input(package_name) | ||
| args.extend(package_specs) |
There was a problem hiding this comment.
The PipInstaller.install method now allows for arbitrary pip options to be injected if the package_name input is user-controlled (e.g., from a dashboard). An attacker could use options like --index-url or --extra-index-url to redirect package installation to a malicious server, or use other options to execute arbitrary code during the installation process. This is a significant increase in attack surface compared to the previous implementation which only allowed a single package name. It is recommended to validate that each token in the input is a valid package name and not an option, unless the option is explicitly allowed and sanitized.
astrbot/core/utils/pip_installer.py
Outdated
| args.extend(self.pip_install_arg.split()) | ||
|
|
||
| logger.info(f"Pip 包管理器: pip {' '.join(args)}") | ||
| logger.info("Pip 包管理器 argv: %s", ["pip", *args]) |
There was a problem hiding this comment.
The code logs the full list of arguments passed to the pip command. These arguments can include URLs with embedded credentials (e.g., https://user:password@repo.com) or sensitive API tokens if they are provided as part of the package name or pip options. This can lead to the exposure of secrets in the application logs. It is recommended to sanitize the arguments before logging, for example by masking credentials in URLs using a regular expression.
astrbot/core/star/star_manager.py
Outdated
| plugin_path = os.path.join(self.plugin_store_path, dir_name) | ||
| await self._install_plugin_requirements(plugin_path, dir_name) |
There was a problem hiding this comment.
reload_failed_plugin 方法的返回类型协定是 (success, error_message) 元组,但对 _install_plugin_requirements 的调用缺少异常处理。如果依赖安装失败导致 _install_plugin_requirements 抛出异常,整个方法会直接崩溃,而不是按预期返回 (False, error_message)。这破坏了该方法的健壮性。建议添加 try...except 块来捕获此异常,并以约定的方式返回错误信息。
| plugin_path = os.path.join(self.plugin_store_path, dir_name) | |
| await self._install_plugin_requirements(plugin_path, dir_name) | |
| try: | |
| plugin_path = os.path.join(self.plugin_store_path, dir_name) | |
| await self._install_plugin_requirements(plugin_path, dir_name) | |
| except Exception as e: | |
| error_msg = f"安装插件 {dir_name} 的依赖失败: {e!s}" | |
| logger.error(error_msg) | |
| return False, error_msg |
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了 2 个问题,并给出了一些整体性的反馈:
- 在
reload_failed_plugin中,你现在调用_install_plugin_requirements时没有使用try/except,这与_check_plugin_dept_update不一致,并且意味着一旦依赖安装出错,会直接中断整个 reload 流程;建议将该调用也包在同样的错误日志模式中,这样有问题的 requirements 安装就不会意外破坏恢复逻辑。 _package_specs_override_index中用于检测 index 覆盖的逻辑并不能识别像-ihttps://example.com/simple这样的短格式选项,因此用户有可能以为自己已经指定了自定义 index,但实际上仍然使用的是你的默认 index;建议同时处理这种-i<url>形式的合并写法。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `reload_failed_plugin` you now call `_install_plugin_requirements` without a `try/except`, which is inconsistent with `_check_plugin_dept_update` and means a dependency install error will abort the reload path entirely; consider wrapping this call in the same error-logging pattern so a bad requirements install doesn’t unexpectedly break recovery logic.
- The index override detection in `_package_specs_override_index` doesn’t catch short-form options like `-ihttps://example.com/simple`, so users can still inadvertently get your default index even when they think they’ve specified their own; consider handling the concatenated `-i<url>` form as well.
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="137-146" />
<code_context>
return names
+def _split_package_install_input(raw_input: str) -> list[str]:
+ normalized = raw_input.strip()
+ if not normalized:
+ return []
+
+ if "\n" in normalized or "\r" in normalized:
+ return _split_multiline_package_input(normalized)
+
+ if _is_valid_install_requirement(normalized):
+ return [normalized]
+
+ split_tokens = shlex.split(normalized)
</code_context>
<issue_to_address>
**issue (bug_risk):** Single-line inputs with inline comments are treated as a single invalid requirement rather than being parsed token-wise.
For a single-line input like `"requests==2.31.0 # latest"`, we first fail `_is_valid_install_requirement`, then `shlex.split` returns tokens where the `#`/comment part is not a valid requirement. Because `all(_is_valid_install_token(...))` then fails, we fall back to returning `[normalized]`, sending the full string (including the comment) to pip and likely causing a parse error. Multiline inputs already strip `#` comments, but single-line inputs don’t. Consider either reusing the multiline comment-stripping for single-line inputs or relaxing the `all(...)` condition so we still return the split tokens even if one token is invalid, letting pip handle it.
</issue_to_address>
### Comment 2
<location path="tests/test_pip_installer.py" line_range="272-281" />
<code_context>
+
+
+@pytest.mark.asyncio
+async def test_install_falls_back_to_raw_input_for_invalid_token_string(monkeypatch):
+ run_pip = AsyncMock(return_value=0)
+
+ monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
+
+ installer = PipInstaller("")
+ raw_input = "demo-package !!! another-package"
+ await installer.install(package_name=raw_input)
+
+ run_pip.assert_awaited_once()
+ recorded_args = run_pip.await_args_list[0].args[0]
+
+ assert recorded_args[0:2] == ["install", raw_input]
</code_context>
<issue_to_address>
**suggestion (testing):** You may want a small test for empty/whitespace-only package input to lock in `_split_package_install_input` behavior.
One remaining gap is when `package_name` is empty or whitespace-only:
- `_split_package_install_input` returns `[]`.
- `install` then skips the `package_name` branch and, without a `requirements_path`, effectively runs a bare `pip install` with only index args.
A test like `test_install_ignores_empty_package_string` (e.g. `package_name=' '`) that asserts only index-related args are passed would lock in this behavior and catch regressions in input normalization.
</issue_to_address>Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Original comment in English
Hey - I've found 2 issues, and left some high level feedback:
- In
reload_failed_pluginyou now call_install_plugin_requirementswithout atry/except, which is inconsistent with_check_plugin_dept_updateand means a dependency install error will abort the reload path entirely; consider wrapping this call in the same error-logging pattern so a bad requirements install doesn’t unexpectedly break recovery logic. - The index override detection in
_package_specs_override_indexdoesn’t catch short-form options like-ihttps://example.com/simple, so users can still inadvertently get your default index even when they think they’ve specified their own; consider handling the concatenated-i<url>form as well.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `reload_failed_plugin` you now call `_install_plugin_requirements` without a `try/except`, which is inconsistent with `_check_plugin_dept_update` and means a dependency install error will abort the reload path entirely; consider wrapping this call in the same error-logging pattern so a bad requirements install doesn’t unexpectedly break recovery logic.
- The index override detection in `_package_specs_override_index` doesn’t catch short-form options like `-ihttps://example.com/simple`, so users can still inadvertently get your default index even when they think they’ve specified their own; consider handling the concatenated `-i<url>` form as well.
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="137-146" />
<code_context>
return names
+def _split_package_install_input(raw_input: str) -> list[str]:
+ normalized = raw_input.strip()
+ if not normalized:
+ return []
+
+ if "\n" in normalized or "\r" in normalized:
+ return _split_multiline_package_input(normalized)
+
+ if _is_valid_install_requirement(normalized):
+ return [normalized]
+
+ split_tokens = shlex.split(normalized)
</code_context>
<issue_to_address>
**issue (bug_risk):** Single-line inputs with inline comments are treated as a single invalid requirement rather than being parsed token-wise.
For a single-line input like `"requests==2.31.0 # latest"`, we first fail `_is_valid_install_requirement`, then `shlex.split` returns tokens where the `#`/comment part is not a valid requirement. Because `all(_is_valid_install_token(...))` then fails, we fall back to returning `[normalized]`, sending the full string (including the comment) to pip and likely causing a parse error. Multiline inputs already strip `#` comments, but single-line inputs don’t. Consider either reusing the multiline comment-stripping for single-line inputs or relaxing the `all(...)` condition so we still return the split tokens even if one token is invalid, letting pip handle it.
</issue_to_address>
### Comment 2
<location path="tests/test_pip_installer.py" line_range="272-281" />
<code_context>
+
+
+@pytest.mark.asyncio
+async def test_install_falls_back_to_raw_input_for_invalid_token_string(monkeypatch):
+ run_pip = AsyncMock(return_value=0)
+
+ monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
+
+ installer = PipInstaller("")
+ raw_input = "demo-package !!! another-package"
+ await installer.install(package_name=raw_input)
+
+ run_pip.assert_awaited_once()
+ recorded_args = run_pip.await_args_list[0].args[0]
+
+ assert recorded_args[0:2] == ["install", raw_input]
</code_context>
<issue_to_address>
**suggestion (testing):** You may want a small test for empty/whitespace-only package input to lock in `_split_package_install_input` behavior.
One remaining gap is when `package_name` is empty or whitespace-only:
- `_split_package_install_input` returns `[]`.
- `install` then skips the `package_name` branch and, without a `requirements_path`, effectively runs a bare `pip install` with only index args.
A test like `test_install_ignores_empty_package_string` (e.g. `package_name=' '`) that asserts only index-related args are passed would lock in this behavior and catch regressions in input normalization.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| async def test_install_falls_back_to_raw_input_for_invalid_token_string(monkeypatch): | ||
| run_pip = AsyncMock(return_value=0) | ||
|
|
||
| monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip) | ||
|
|
||
| installer = PipInstaller("") | ||
| raw_input = "demo-package !!! another-package" | ||
| await installer.install(package_name=raw_input) | ||
|
|
||
| run_pip.assert_awaited_once() |
There was a problem hiding this comment.
suggestion (testing): 建议增加一个关于空字符串或仅包含空白字符的 package 输入的小测试,以锁定 _split_package_install_input 的行为。
目前还有一个缺口是当 package_name 为空或只包含空白字符时:
_split_package_install_input返回[];install随后会跳过package_name分支,在没有requirements_path的情况下,实际上执行的是只带 index 参数的裸pip install。
可以添加一个类似 test_install_ignores_empty_package_string 的测试(例如 package_name=' '),断言只传入与 index 相关的参数,这样可以锁定当前行为,并在输入规范化发生回归时把问题暴露出来。
Original comment in English
suggestion (testing): You may want a small test for empty/whitespace-only package input to lock in _split_package_install_input behavior.
One remaining gap is when package_name is empty or whitespace-only:
_split_package_install_inputreturns[].installthen skips thepackage_namebranch and, without arequirements_path, effectively runs a barepip installwith only index args.
A test like test_install_ignores_empty_package_string (e.g. package_name=' ') that asserts only index-related args are passed would lock in this behavior and catch regressions in input normalization.
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了 1 个问题
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="137" />
<code_context>
return names
+def _split_package_install_input(raw_input: str) -> list[str]:
+ normalized = raw_input.strip()
+ if not normalized:
</code_context>
<issue_to_address>
**issue (complexity):** 请考虑通过使索引覆盖检查更加显式、内联简单的辅助函数,以及集中处理行清理逻辑来简化新的 pip 参数解析辅助函数,从而降低理解复杂度。
你可以在不改变行为的前提下降低这里的认知负担,方式包括:
1. **让 `_package_specs_override_index` 逻辑线性且显式**
语义保持不变,但更容易浏览和理解:
```python
def _package_specs_override_index(package_specs: list[str]) -> bool:
for spec in package_specs:
if spec in {"-i", "--index-url"}:
return True
if spec.startswith("--index-url="):
return True
if spec.startswith("-i"):
return True
return False
```
这样可以保持原有逻辑(`spec.startswith("-i")` 仍然会匹配 `-ignore-something`),但去掉了复合的 `any(...)` 表达式。
---
2. **移除对 `_is_valid_install_token` 的细微间接调用**
`_is_valid_install_token` 是一个只在单一位置使用的一行函数。将其逻辑内联可以减少在辅助函数之间跳转,同时不改变行为:
```python
def _split_package_install_input(raw_input: str) -> list[str]:
normalized = raw_input.strip()
if not normalized:
return []
if "\n" in normalized or "\r" in normalized:
return _split_multiline_package_input(normalized)
normalized = _strip_inline_requirement_comment(normalized)
if not normalized:
return []
if _is_valid_install_requirement(normalized):
return [normalized]
split_tokens = shlex.split(normalized)
if split_tokens and split_tokens[0].startswith("-"):
return split_tokens
if split_tokens and all(
token.startswith("-") or _is_valid_install_requirement(token)
for token in split_tokens
):
return split_tokens
return [normalized]
```
然后你就可以完全删除 `_is_valid_install_token`。
---
3. **为单行和多行输入共享“去注释/拆分”的处理路径**
你已经有 `_strip_inline_requirement_comment`;可以把通用的“逐行 -> 清洗后的候选值”逻辑集中到一起,并在 `_split_package_install_input` 和 `_split_multiline_package_input` 中复用,使行为在单一位置定义:
```python
def _iter_clean_requirement_lines(raw_input: str) -> list[str]:
for line in raw_input.splitlines():
candidate = _strip_inline_requirement_comment(line)
if not candidate or candidate.startswith("#"):
continue
yield candidate
def _split_multiline_package_input(raw_input: str) -> list[str]:
requirements: list[str] = []
for candidate in _iter_clean_requirement_lines(raw_input):
if candidate.startswith("-"):
requirements.extend(shlex.split(candidate))
else:
requirements.append(candidate)
return requirements
```
这样 `_split_package_install_input` 的多行处理路径就可以变成:
```python
if "\n" in normalized or "\r" in normalized:
return _split_multiline_package_input(normalized)
```
行为保持不变,但逐行清理和注释处理的规则集中在一个辅助函数中,使整体流程更易理解。
</issue_to_address>帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据这些反馈改进后续的评审。
Original comment in English
Hey - I've found 1 issue
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="137" />
<code_context>
return names
+def _split_package_install_input(raw_input: str) -> list[str]:
+ normalized = raw_input.strip()
+ if not normalized:
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the new pip argument‑parsing helpers by making index override checks explicit, inlining trivial helpers, and centralizing line-cleaning logic to reduce cognitive load.
You can reduce the cognitive load here without changing behavior by:
1. **Making `_package_specs_override_index` linear and explicit**
Same semantics, but easier to scan and reason about:
```python
def _package_specs_override_index(package_specs: list[str]) -> bool:
for spec in package_specs:
if spec in {"-i", "--index-url"}:
return True
if spec.startswith("--index-url="):
return True
if spec.startswith("-i"):
return True
return False
```
This keeps the exact logic (`spec.startswith("-i")` still matches `-ignore-something`) but removes the compound `any(...)` expression.
---
2. **Removing tiny indirection for `_is_valid_install_token`**
`_is_valid_install_token` is a one-liner used only in one place. Inlining the logic reduces helper hopping without changing behavior:
```python
def _split_package_install_input(raw_input: str) -> list[str]:
normalized = raw_input.strip()
if not normalized:
return []
if "\n" in normalized or "\r" in normalized:
return _split_multiline_package_input(normalized)
normalized = _strip_inline_requirement_comment(normalized)
if not normalized:
return []
if _is_valid_install_requirement(normalized):
return [normalized]
split_tokens = shlex.split(normalized)
if split_tokens and split_tokens[0].startswith("-"):
return split_tokens
if split_tokens and all(
token.startswith("-") or _is_valid_install_requirement(token)
for token in split_tokens
):
return split_tokens
return [normalized]
```
Then you can drop `_is_valid_install_token` completely.
---
3. **Sharing the comment‑stripping/splitting path for single‑line and multi‑line input**
You already have `_strip_inline_requirement_comment`; you can centralize the common “per-line -> cleaned candidate” logic and reuse it in both `_split_package_install_input` and `_split_multiline_package_input` so the behavior is defined in one place:
```python
def _iter_clean_requirement_lines(raw_input: str) -> list[str]:
for line in raw_input.splitlines():
candidate = _strip_inline_requirement_comment(line)
if not candidate or candidate.startswith("#"):
continue
yield candidate
def _split_multiline_package_input(raw_input: str) -> list[str]:
requirements: list[str] = []
for candidate in _iter_clean_requirement_lines(raw_input):
if candidate.startswith("-"):
requirements.extend(shlex.split(candidate))
else:
requirements.append(candidate)
return requirements
```
Then `_split_package_install_input`’s multiline path becomes:
```python
if "\n" in normalized or "\r" in normalized:
return _split_multiline_package_input(normalized)
```
Behavior stays the same, but the rules for per-line cleaning and comment handling live in one helper, which makes the flow easier to follow.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了 1 个问题,并给出了一些高层反馈:
- 在
_strip_inline_requirement_comment中,按[ \t]+#进行 split 会把 URL 中合法的#片段也去掉(例如:--extra-index-url https://example.com/simple#frag);建议将注释剥离限制在“不是选项”的行上,或者使用更精确的启发式规则。 - 新的
_install_plugin_requirementshelper 在部分调用路径(_check_plugin_dept_update、reload_failed_plugin)中被 try/except 包裹,而在其他路径(install_plugin、install_plugin_from_file)中则没有,从而改变了失败时的行为;建议对这些调用点的错误处理逻辑进行统一,以便在依赖安装失败时插件安装行为保持一致。
AI 代理使用的提示词
Please address the comments from this code review:
## Overall Comments
- In `_strip_inline_requirement_comment`, splitting on `[ \t]+#` will also strip legitimate `#` fragments in URLs (e.g. `--extra-index-url https://example.com/simple#frag`); consider limiting comment stripping to lines that are not options or using a more precise heuristic.
- The new `_install_plugin_requirements` helper is wrapped in try/except in some flows (`_check_plugin_dept_update`, `reload_failed_plugin`) but not in others (`install_plugin`, `install_plugin_from_file`), which changes failure behavior; consider aligning error handling across these call sites so plugin installs behave consistently when dependency installation fails.
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="183" />
<code_context>
+ return requirements
+
+
+def _package_specs_override_index(package_specs: list[str]) -> bool:
+ for spec in package_specs:
+ if spec in {"-i", "--index-url"}:
</code_context>
<issue_to_address>
**issue (complexity):** 请考虑收紧 `_package_specs_override_index` 的实现,并将多行解析 helper 内联,这样可以让“index 覆盖逻辑”和“多行 requirement 处理”更加清晰、易于理解。
你可以在两个关键位置降低复杂度,同时无需改变 `install` 的对外行为,也不需要移除任何特性。
### 1. 收紧并简化 `_package_specs_override_index`
当前基于前缀的 `-i` 检测既不直观也过于宽泛:
```python
def _package_specs_override_index(package_specs: list[str]) -> bool:
for spec in package_specs:
if spec in {"-i", "--index-url"}:
return True
if spec.startswith("--index-url="):
return True
if spec.startswith("-i"):
return True
return False
```
你可以通过基于索引的扫描逻辑,仅处理合法的 `pip` 形态,使表达更明确、更安全:
```python
def _package_specs_override_index(package_specs: list[str]) -> bool:
for i, spec in enumerate(package_specs):
if spec in {"-i", "--index-url"}:
# 像 pip 一样要求后面必须跟一个值
if i + 1 < len(package_specs):
return True
continue
# 内联值:--index-url=https://... 或 -ihttps://...
if spec.startswith("--index-url="):
return True
if spec.startswith("-i") and spec != "-i":
return True
return False
```
这样可以保留所有能力(短/长选项、内联形式),同时消除“任何以 `-i` 开头的字符串都匹配”的含糊之处,让意图更易于推断。
### 2. 合并 `_split_multiline_package_input` 和 `_iter_clean_requirement_lines`
目前多行路径被拆成两层:
```python
def _split_multiline_package_input(raw_input: str) -> list[str]:
requirements: list[str] = []
for candidate in _iter_clean_requirement_lines(raw_input):
if candidate.startswith("-"):
requirements.extend(shlex.split(candidate))
continue
requirements.append(candidate)
return requirements
def _iter_clean_requirement_lines(raw_input: str):
for line in raw_input.splitlines():
candidate = _strip_inline_requirement_comment(line)
if not candidate or candidate.startswith("#"):
continue
yield candidate
```
你可以将 `_iter_clean_requirement_lines` 内联到 `_split_multiline_package_input` 中,在不改变行为的前提下减少间接层和分支:
```python
def _split_multiline_package_input(raw_input: str) -> list[str]:
requirements: list[str] = []
for line in raw_input.splitlines():
candidate = _strip_inline_requirement_comment(line)
if not candidate or candidate.startswith("#"):
continue
if candidate.startswith("-"):
requirements.extend(shlex.split(candidate))
else:
requirements.append(candidate)
return requirements
```
这会保留相同语义(空行、注释、行内注释、选项支持),同时让流程更易理解,并把所有多行处理逻辑集中在一个函数里。
</issue_to_address>帮我变得更有用!请在每条评论上点 👍 或 👎,我会用这些反馈来改进后续的 review。
Original comment in English
Hey - I've found 1 issue, and left some high level feedback:
- In
_strip_inline_requirement_comment, splitting on[ \t]+#will also strip legitimate#fragments in URLs (e.g.--extra-index-url https://example.com/simple#frag); consider limiting comment stripping to lines that are not options or using a more precise heuristic. - The new
_install_plugin_requirementshelper is wrapped in try/except in some flows (_check_plugin_dept_update,reload_failed_plugin) but not in others (install_plugin,install_plugin_from_file), which changes failure behavior; consider aligning error handling across these call sites so plugin installs behave consistently when dependency installation fails.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `_strip_inline_requirement_comment`, splitting on `[ \t]+#` will also strip legitimate `#` fragments in URLs (e.g. `--extra-index-url https://example.com/simple#frag`); consider limiting comment stripping to lines that are not options or using a more precise heuristic.
- The new `_install_plugin_requirements` helper is wrapped in try/except in some flows (`_check_plugin_dept_update`, `reload_failed_plugin`) but not in others (`install_plugin`, `install_plugin_from_file`), which changes failure behavior; consider aligning error handling across these call sites so plugin installs behave consistently when dependency installation fails.
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="183" />
<code_context>
+ return requirements
+
+
+def _package_specs_override_index(package_specs: list[str]) -> bool:
+ for spec in package_specs:
+ if spec in {"-i", "--index-url"}:
</code_context>
<issue_to_address>
**issue (complexity):** Consider tightening `_package_specs_override_index` and inlining the multiline parsing helper to make the index-override logic and multiline requirement handling more explicit and easier to follow.
You can reduce complexity in two focused spots without changing the public behavior of `install` or removing any features.
### 1. Tighten and simplify `_package_specs_override_index`
The current prefix-based `-i` detection is both non-obvious and over-general:
```python
def _package_specs_override_index(package_specs: list[str]) -> bool:
for spec in package_specs:
if spec in {"-i", "--index-url"}:
return True
if spec.startswith("--index-url="):
return True
if spec.startswith("-i"):
return True
return False
```
You can make this more explicit and safer by scanning with index-aware logic and handling only valid `pip` shapes:
```python
def _package_specs_override_index(package_specs: list[str]) -> bool:
for i, spec in enumerate(package_specs):
if spec in {"-i", "--index-url"}:
# require a following value, like pip does
if i + 1 < len(package_specs):
return True
continue
# inline values: --index-url=https://... or -ihttps://...
if spec.startswith("--index-url="):
return True
if spec.startswith("-i") and spec != "-i":
return True
return False
```
This keeps all capabilities (short/long options, inline forms) but removes the “any string starting with `-i`” ambiguity and makes the intent easier to reason about.
### 2. Merge `_split_multiline_package_input` and `_iter_clean_requirement_lines`
Right now the multiline path has two layers:
```python
def _split_multiline_package_input(raw_input: str) -> list[str]:
requirements: list[str] = []
for candidate in _iter_clean_requirement_lines(raw_input):
if candidate.startswith("-"):
requirements.extend(shlex.split(candidate))
continue
requirements.append(candidate)
return requirements
def _iter_clean_requirement_lines(raw_input: str):
for line in raw_input.splitlines():
candidate = _strip_inline_requirement_comment(line)
if not candidate or candidate.startswith("#"):
continue
yield candidate
```
You can inline `_iter_clean_requirement_lines` into `_split_multiline_package_input` to reduce indirection and branching without changing behavior:
```python
def _split_multiline_package_input(raw_input: str) -> list[str]:
requirements: list[str] = []
for line in raw_input.splitlines():
candidate = _strip_inline_requirement_comment(line)
if not candidate or candidate.startswith("#"):
continue
if candidate.startswith("-"):
requirements.extend(shlex.split(candidate))
else:
requirements.append(candidate)
return requirements
```
This keeps the same semantics (blank lines, comments, inline comments, options support) while making the flow easier to follow and localizing all multiline handling to one function.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了两个问题,并给出了一些高层次的反馈:
_install_plugin_requirements_with_logging中的错误日志总是显示更新插件 {plugin_label} 的依赖失败,当从安装或重新加载路径调用时,这会产生误导;建议将该消息改为更通用的,或者传入操作类型,使日志能准确反映当前工作流程。- 在
PipInstaller.install中,_package_specs_override_index只在追加pip_install_arg之前检查args[1:],因此用户在pip_install_arg中指定的--index-url不会覆盖默认的-i/--trusted-host;如果这不是有意为之,建议在覆盖检查中同时考虑pip_install_arg。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- `_install_plugin_requirements_with_logging` 中的错误日志总是显示 `更新插件 {plugin_label} 的依赖失败`,当从安装或重新加载路径调用时,这会产生误导;建议将该消息改为更通用的,或者传入操作类型,使日志能准确反映当前工作流程。
- 在 `PipInstaller.install` 中,`_package_specs_override_index` 只在追加 `pip_install_arg` 之前检查 `args[1:]`,因此用户在 `pip_install_arg` 中指定的 `--index-url` 不会覆盖默认的 `-i`/`--trusted-host`;如果这不是有意为之,建议在覆盖检查中同时考虑 `pip_install_arg`。
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="173-182" />
<code_context>
+ return True
+
+
+def _split_multiline_package_input(raw_input: str) -> list[str]:
+ requirements: list[str] = []
+ for line in raw_input.splitlines():
+ candidate = _strip_inline_requirement_comment(line)
+ if not candidate or candidate.startswith("#"):
+ continue
+ if candidate.startswith("-"):
+ requirements.extend(_split_option_input(candidate))
+ continue
+ requirements.append(candidate)
+ return requirements
+
</code_context>
<issue_to_address>
**issue (bug_risk):** 多行输入中,当同一行包含多个包时,其处理方式与单行输入不一致。
在这里,像 `"foo bar"` 这样的行会被当作单个 requirement,而单行路径会使用 `shlex.split` 将 `foo` 和 `bar` 视为两个独立的 requirement,并对每个 token 进行校验。对于类似 `"foo bar\nspam"` 的输入,第一行会以 `"foo bar"` 的形式传给 pip,这很可能是无效的。为了保持行为一致,要么在这里也对非选项行使用 `shlex.split`,要么拒绝包含多个 token 的行。
</issue_to_address>
### Comment 2
<location path="astrbot/core/star/star_manager.py" line_range="216-224" />
<code_context>
+ logger.info(f"正在安装插件 {plugin_label} 所需的依赖库: {requirements_path}")
+ await pip_installer.install(requirements_path=requirements_path)
+
+ async def _install_plugin_requirements_with_logging(
+ self,
+ plugin_dir_path: str,
+ plugin_label: str,
+ ) -> None:
+ try:
+ await self._install_plugin_requirements(plugin_dir_path, plugin_label)
+ except Exception as e:
+ logger.error(f"更新插件 {plugin_label} 的依赖失败。Code: {e!s}")
+
async def _import_plugin_with_dependency_recovery(
</code_context>
<issue_to_address>
**issue (bug_risk):** 在依赖安装过程中吞掉所有异常会隐藏可操作的错误,并且让调试更加困难。
`_install_plugin_requirements_with_logging` 捕获了 `Exception` 却只记录了 `e!s`,之后调用方会假定安装成功。在 `reload_failed_plugin` 或 `install_plugin` 等流程中,这可能导致插件处于部分更新的状态,却没有清晰的诊断信息。至少应使用 `logger.exception` 来保留 traceback,并考虑重新抛出异常或返回结构化错误,以便调用方在依赖安装失败时中止插件加载/更新。
</issue_to_address>帮我变得更有用!请对每条评论点 👍 或 👎,我会根据你的反馈改进后续评审。
Original comment in English
Hey - I've found 2 issues, and left some high level feedback:
- The error log in
_install_plugin_requirements_with_loggingalways says更新插件 {plugin_label} 的依赖失败, which is misleading when called from install or reload paths; consider making the message generic or passing in the operation type so logs accurately reflect the workflow. - In
PipInstaller.install,_package_specs_override_indexonly inspectsargs[1:]beforepip_install_argis appended, so a user-specified--index-urlinpip_install_argwill not suppress the default-i/--trusted-host; if that’s not intentional, consider includingpip_install_argin the override check.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The error log in `_install_plugin_requirements_with_logging` always says `更新插件 {plugin_label} 的依赖失败`, which is misleading when called from install or reload paths; consider making the message generic or passing in the operation type so logs accurately reflect the workflow.
- In `PipInstaller.install`, `_package_specs_override_index` only inspects `args[1:]` before `pip_install_arg` is appended, so a user-specified `--index-url` in `pip_install_arg` will not suppress the default `-i`/`--trusted-host`; if that’s not intentional, consider including `pip_install_arg` in the override check.
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="173-182" />
<code_context>
+ return True
+
+
+def _split_multiline_package_input(raw_input: str) -> list[str]:
+ requirements: list[str] = []
+ for line in raw_input.splitlines():
+ candidate = _strip_inline_requirement_comment(line)
+ if not candidate or candidate.startswith("#"):
+ continue
+ if candidate.startswith("-"):
+ requirements.extend(_split_option_input(candidate))
+ continue
+ requirements.append(candidate)
+ return requirements
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Multiline input with multiple packages on the same line is handled inconsistently vs single-line input.
Here, a line like `"foo bar"` becomes a single requirement, while the single-line path uses `shlex.split` to treat `foo` and `bar` as separate requirements and validate each token. For input like `"foo bar\nspam"`, the first line would be passed to pip as `"foo bar"`, which is likely invalid. To keep behavior consistent, either shlex-split non-option lines here as well or reject lines that contain multiple tokens.
</issue_to_address>
### Comment 2
<location path="astrbot/core/star/star_manager.py" line_range="216-224" />
<code_context>
+ logger.info(f"正在安装插件 {plugin_label} 所需的依赖库: {requirements_path}")
+ await pip_installer.install(requirements_path=requirements_path)
+
+ async def _install_plugin_requirements_with_logging(
+ self,
+ plugin_dir_path: str,
+ plugin_label: str,
+ ) -> None:
+ try:
+ await self._install_plugin_requirements(plugin_dir_path, plugin_label)
+ except Exception as e:
+ logger.error(f"更新插件 {plugin_label} 的依赖失败。Code: {e!s}")
+
async def _import_plugin_with_dependency_recovery(
</code_context>
<issue_to_address>
**issue (bug_risk):** Swallowing all exceptions during requirement installation may hide actionable errors and makes debugging harder.
`_install_plugin_requirements_with_logging` catches `Exception` and only logs `e!s`, after which callers assume installation succeeded. In flows like `reload_failed_plugin` or `install_plugin`, this can leave a plugin partially updated with no clear diagnostics. At least use `logger.exception` to retain the traceback, and consider re‑raising or returning a structured error so callers can abort plugin load/update when dependencies fail.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了 1 个问题,并给出了一些整体性的反馈:
- 若干新的插件管理器测试都遵循相同的模式:针对成功 vs 依赖安装失败记录
events(例如 install、install_from_file、reload_failed_plugin、update_plugin),并断言相同的事件序列;可以考虑把这些用例参数化,这样既能减少重复,又便于在一个地方新增/修改测试场景。
给 AI Agent 的提示
请根据这次代码评审中的评论进行修改:
## 整体性评论
- 若干新的插件管理器测试都遵循相同的模式:针对成功 vs 依赖安装失败记录 `events`(例如 install、install_from_file、reload_failed_plugin、update_plugin),并断言相同的事件序列;可以考虑把这些用例参数化,这样既能减少重复,又便于在一个地方新增/修改测试场景。
## 单独评论
### 评论 1
<location path="astrbot/core/star/star_manager.py" line_range="216-224" />
<code_context>
+ logger.info(f"正在安装插件 {plugin_label} 所需的依赖库: {requirements_path}")
+ await pip_installer.install(requirements_path=requirements_path)
+
+ async def _install_plugin_requirements_with_logging(
+ self,
+ plugin_dir_path: str,
+ plugin_label: str,
+ ) -> None:
+ try:
+ await self._install_plugin_requirements(plugin_dir_path, plugin_label)
+ except Exception as e:
+ logger.exception(f"插件 {plugin_label} 依赖安装失败: {e!s}")
+
async def _import_plugin_with_dependency_recovery(
</code_context>
<issue_to_address>
**issue (bug_risk):** 在这里捕获宽泛的 `Exception` 可能会无意中吞掉 `asyncio.CancelledError` 以及其他关键异常。
由于这个包装器会记录并屏蔽 `_install_plugin_requirements` 抛出的所有 `Exception`,它可能会把任务取消当作普通的插件安装失败来处理并忽略,从而破坏关机/超时行为。建议在通用的 `Exception` 处理之前,显式地重新抛出 `asyncio.CancelledError`(以及你关心的其他关键异常),例如:
```python
try:
await self._install_plugin_requirements(...)
except asyncio.CancelledError:
raise
except Exception as e:
logger.exception("插件 %s 依赖安装失败: %s", plugin_label, e)
```
这样既能保留取消语义,又能继续记录真实的安装错误。
</issue_to_address>帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进后续评审。
Original comment in English
Hey - I've found 1 issue, and left some high level feedback:
- Several of the new plugin manager tests follow the pattern of recording
eventsfor success vs. dependency-install failure (e.g., install, install_from_file, reload_failed_plugin, update_plugin) and assert the same sequence; consider parametrizing these cases to reduce duplication and make it easier to add/modify scenarios in one place.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- Several of the new plugin manager tests follow the pattern of recording `events` for success vs. dependency-install failure (e.g., install, install_from_file, reload_failed_plugin, update_plugin) and assert the same sequence; consider parametrizing these cases to reduce duplication and make it easier to add/modify scenarios in one place.
## Individual Comments
### Comment 1
<location path="astrbot/core/star/star_manager.py" line_range="216-224" />
<code_context>
+ logger.info(f"正在安装插件 {plugin_label} 所需的依赖库: {requirements_path}")
+ await pip_installer.install(requirements_path=requirements_path)
+
+ async def _install_plugin_requirements_with_logging(
+ self,
+ plugin_dir_path: str,
+ plugin_label: str,
+ ) -> None:
+ try:
+ await self._install_plugin_requirements(plugin_dir_path, plugin_label)
+ except Exception as e:
+ logger.exception(f"插件 {plugin_label} 依赖安装失败: {e!s}")
+
async def _import_plugin_with_dependency_recovery(
</code_context>
<issue_to_address>
**issue (bug_risk):** Catching broad `Exception` here may inadvertently swallow `asyncio.CancelledError` and other critical exceptions.
Because this wrapper logs and suppresses every `Exception` from `_install_plugin_requirements`, it may treat task cancellation as a normal plugin failure and then ignore it, which can break shutdown/timeout behavior. Consider explicitly re-raising `asyncio.CancelledError` (and any other critical exceptions you care about) before the generic `Exception` handler, for example:
```python
try:
await self._install_plugin_requirements(...)
except asyncio.CancelledError:
raise
except Exception as e:
logger.exception("插件 %s 依赖安装失败: %s", plugin_label, e)
```
This preserves cancellation semantics while still logging real install errors.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我已经留下了一些整体性的反馈:
- 在插件流程(
install_plugin、install_plugin_from_file、reload_failed_plugin、update_plugin)中,_install_plugin_requirements_with_logging会有意吞掉依赖安装失败(CancelledError之外)的错误,并仍然继续执行load/reload;如果依赖安装失败应该阻止插件激活,建议从这个辅助函数返回一个状态值,并在失败时提前中止后续的 load/reload,以便让这种行为更加明确。 - 新增的包解析辅助函数
_split_package_install_input和_split_multiline_package_input会相互调用(例如多行 → 按行 → 再次调用_split_package_install_input),虽然能工作,但逻辑有点难以跟踪;建议重构为一个小型的、非递归的单行解析器,再配合一个顶层的多行处理器,从而让控制流以及优先级规则(marker vs options vs comments)更容易理解。
供 AI 代理使用的提示词
Please address the comments from this code review:
## Overall Comments
- In the plugin flows (`install_plugin`, `install_plugin_from_file`, `reload_failed_plugin`, `update_plugin`), `_install_plugin_requirements_with_logging` intentionally swallows dependency installation failures (other than `CancelledError`) and still proceeds to `load`/`reload`; if dependency install failure should block plugin activation, consider returning a status from the helper and short‑circuiting the load/reload on failure so the behavior is explicit.
- The new package parsing helpers `_split_package_install_input` and `_split_multiline_package_input` call each other (e.g. multiline → per‑line → `_split_package_install_input` again), which works but is a bit hard to follow; consider refactoring into a small, non‑recursive line parser plus a top‑level multiline handler to make the control flow and precedence rules (marker vs options vs comments) easier to reason about.帮我变得更有用!请在每条评论上点击 👍 或 👎,我会根据你的反馈改进后续的评审。
Original comment in English
Hey - I've left some high level feedback:
- In the plugin flows (
install_plugin,install_plugin_from_file,reload_failed_plugin,update_plugin),_install_plugin_requirements_with_loggingintentionally swallows dependency installation failures (other thanCancelledError) and still proceeds toload/reload; if dependency install failure should block plugin activation, consider returning a status from the helper and short‑circuiting the load/reload on failure so the behavior is explicit. - The new package parsing helpers
_split_package_install_inputand_split_multiline_package_inputcall each other (e.g. multiline → per‑line →_split_package_install_inputagain), which works but is a bit hard to follow; consider refactoring into a small, non‑recursive line parser plus a top‑level multiline handler to make the control flow and precedence rules (marker vs options vs comments) easier to reason about.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In the plugin flows (`install_plugin`, `install_plugin_from_file`, `reload_failed_plugin`, `update_plugin`), `_install_plugin_requirements_with_logging` intentionally swallows dependency installation failures (other than `CancelledError`) and still proceeds to `load`/`reload`; if dependency install failure should block plugin activation, consider returning a status from the helper and short‑circuiting the load/reload on failure so the behavior is explicit.
- The new package parsing helpers `_split_package_install_input` and `_split_multiline_package_input` call each other (e.g. multiline → per‑line → `_split_package_install_input` again), which works but is a bit hard to follow; consider refactoring into a small, non‑recursive line parser plus a top‑level multiline handler to make the control flow and precedence rules (marker vs options vs comments) easier to reason about.Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
@sourcery-ai review |
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了 1 个问题
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="55-71" />
<code_context>
+ self._buffer += text.replace("\r", "\n")
+ while "\n" in self._buffer:
+ line, self._buffer = self._buffer.split("\n", 1)
+ line = line.strip()
+ if line:
+ self._log_func(line)
+ return len(text)
+
+ def flush(self) -> None:
+ line = self._buffer.strip()
+ if line:
+ self._log_func(line)
</code_context>
<issue_to_address>
**suggestion (bug_risk):** 在 `_StreamingLogWriter` 中对行进行裁剪可能会丢失有意义的前后空白,并改变 pip 输出的格式。
`write()` 和 `flush()` 都会对每一行调用 `strip()`,这会移除行首/行尾空格、压扁本来有意缩进的行(例如 `' WARNING:'`),并且完全丢弃空行。如果你希望在保留 pip 输出的同时只规范换行符,请使用 `rstrip("\n")`/`rstrip("\r\n")` 来替代 `strip()`,并只在检查是否为空时使用 `if line.strip():`,记录日志时仍使用原始行内容。
```suggestion
def write(self, text: str) -> int:
if not text:
return 0
# Normalize Windows newlines to Unix style but preserve other whitespace
self._buffer += text.replace("\r", "\n")
while "\n" in self._buffer:
raw_line, self._buffer = self._buffer.split("\n", 1)
# Ensure any trailing newline characters are removed without
# touching other leading/trailing whitespace
line = raw_line.rstrip("\r\n")
if line.strip():
self._log_func(line)
return len(text)
def flush(self) -> None:
# Flush any remaining buffered text, preserving leading/trailing spaces
line = self._buffer.rstrip("\r\n")
if line.strip():
self._log_func(line)
self._buffer = ""
```
</issue_to_address>帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进后续评审。
Original comment in English
Hey - I've found 1 issue
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="55-71" />
<code_context>
+ self._buffer += text.replace("\r", "\n")
+ while "\n" in self._buffer:
+ line, self._buffer = self._buffer.split("\n", 1)
+ line = line.strip()
+ if line:
+ self._log_func(line)
+ return len(text)
+
+ def flush(self) -> None:
+ line = self._buffer.strip()
+ if line:
+ self._log_func(line)
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Trimming lines in `_StreamingLogWriter` may drop meaningful leading/trailing whitespace and alter pip output formatting.
Both `write()` and `flush()` call `strip()` on each line, which removes leading/trailing spaces, collapses intentionally indented lines (e.g. `' WARNING:'`), and drops blank lines entirely. If you want to preserve pip’s output while just normalizing newlines, use `rstrip("\n")`/`rstrip("\r\n")` instead of `strip()`, and use `if line.strip():` only for the emptiness check while logging the original line.
```suggestion
def write(self, text: str) -> int:
if not text:
return 0
# Normalize Windows newlines to Unix style but preserve other whitespace
self._buffer += text.replace("\r", "\n")
while "\n" in self._buffer:
raw_line, self._buffer = self._buffer.split("\n", 1)
# Ensure any trailing newline characters are removed without
# touching other leading/trailing whitespace
line = raw_line.rstrip("\r\n")
if line.strip():
self._log_func(line)
return len(text)
def flush(self) -> None:
# Flush any remaining buffered text, preserving leading/trailing spaces
line = self._buffer.rstrip("\r\n")
if line.strip():
self._log_func(line)
self._buffer = ""
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了 1 个问题
给 AI 代理的提示
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="tests/test_pip_installer.py" line_range="648-657" />
<code_context>
+
+
+@pytest.mark.asyncio
+async def test_install_respects_index_override_in_pip_install_arg(monkeypatch):
+ run_pip = AsyncMock(return_value=(0, []))
+
+ monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
+
+ installer = PipInstaller("--index-url https://example.com/simple")
+ await installer.install(package_name="demo-package")
+
+ run_pip.assert_awaited_once()
+ recorded_args = run_pip.await_args_list[0].args[0]
+
+ assert "install" in recorded_args
+ assert "demo-package" in recorded_args
+ assert "--index-url" in recorded_args
+ assert "https://example.com/simple" in recorded_args
</code_context>
<issue_to_address>
**suggestion (testing):** 加强对索引覆盖行为的测试,以确保在通过 `pip_install_arg` 覆盖索引时不会再添加默认索引。
目前测试只验证了自定义的 `--index-url` 是否存在。为了完整覆盖覆盖行为,还应该断言当 `pip_install_arg` 指定了索引时,默认索引配置在 `recorded_args` 中是*不存在*的,例如:
```python
assert "mirrors.aliyun.com" not in recorded_args
assert "https://pypi.org/simple" not in recorded_args
```
这可以确保在该场景下,`_package_specs_override_index` 的确阻止默认索引配置被添加。
</issue_to_address>帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进后续的评审。
Original comment in English
Hey - I've found 1 issue
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="tests/test_pip_installer.py" line_range="648-657" />
<code_context>
+
+
+@pytest.mark.asyncio
+async def test_install_respects_index_override_in_pip_install_arg(monkeypatch):
+ run_pip = AsyncMock(return_value=(0, []))
+
+ monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip)
+
+ installer = PipInstaller("--index-url https://example.com/simple")
+ await installer.install(package_name="demo-package")
+
+ run_pip.assert_awaited_once()
+ recorded_args = run_pip.await_args_list[0].args[0]
+
+ assert "install" in recorded_args
+ assert "demo-package" in recorded_args
+ assert "--index-url" in recorded_args
+ assert "https://example.com/simple" in recorded_args
</code_context>
<issue_to_address>
**suggestion (testing):** Strengthen the index override test to ensure the default index is not added when overridden via pip_install_arg.
Right now the test only verifies that the custom `--index-url` is present. To fully cover the override behavior, it should also assert that the default index settings are *absent* from `recorded_args` when `pip_install_arg` specifies an index, e.g.:
```python
assert "mirrors.aliyun.com" not in recorded_args
assert "https://pypi.org/simple" not in recorded_args
```
This will ensure `_package_specs_override_index` truly prevents adding the default index configuration in this scenario.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| async def test_install_respects_index_override_in_pip_install_arg(monkeypatch): | ||
| run_pip = AsyncMock(return_value=(0, [])) | ||
|
|
||
| monkeypatch.setattr(PipInstaller, "_run_pip_in_process", run_pip) | ||
|
|
||
| installer = PipInstaller("--index-url https://example.com/simple") | ||
| await installer.install(package_name="demo-package") | ||
|
|
||
| run_pip.assert_awaited_once() | ||
| recorded_args = run_pip.await_args_list[0].args[0] |
There was a problem hiding this comment.
suggestion (testing): 加强对索引覆盖行为的测试,以确保在通过 pip_install_arg 覆盖索引时不会再添加默认索引。
目前测试只验证了自定义的 --index-url 是否存在。为了完整覆盖覆盖行为,还应该断言当 pip_install_arg 指定了索引时,默认索引配置在 recorded_args 中是不存在的,例如:
assert "mirrors.aliyun.com" not in recorded_args
assert "https://pypi.org/simple" not in recorded_args这可以确保在该场景下,_package_specs_override_index 的确阻止默认索引配置被添加。
Original comment in English
suggestion (testing): Strengthen the index override test to ensure the default index is not added when overridden via pip_install_arg.
Right now the test only verifies that the custom --index-url is present. To fully cover the override behavior, it should also assert that the default index settings are absent from recorded_args when pip_install_arg specifies an index, e.g.:
assert "mirrors.aliyun.com" not in recorded_args
assert "https://pypi.org/simple" not in recorded_argsThis will ensure _package_specs_override_index truly prevents adding the default index configuration in this scenario.
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了 1 个问题
给 AI Agents 的提示
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="50" />
<code_context>
-def _run_pip_main_with_output(pip_main, args: list[str]) -> tuple[int, str]:
- stream = io.StringIO()
- with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
+class _StreamingLogWriter(io.TextIOBase):
+ def __init__(self, log_func) -> None:
+ self._log_func = log_func
</code_context>
<issue_to_address>
**issue (complexity):** 考虑通过为日志流式输出、参数构建、约束处理以及冲突错误构造提取更聚焦的辅助函数,来重构新的 pip 安装逻辑,从而简化 `install` 以及相关代码路径。
你可以在不损失任何功能的前提下,通过分离职责并抽取少量聚焦的辅助函数来降低复杂度。
### 1. 将流式输出与错误解析解耦
目前 `_StreamingLogWriter` 同时负责日志流式输出和错误分类。你可以保留流式输出行为,但让它返回*所有*行,然后把分类逻辑移动到一个纯函数中。
```python
class _StreamingLogWriter(io.TextIOBase):
def __init__(self, log_func) -> None:
self._log_func = log_func
self._buffer = ""
self.lines: list[str] = [] # store all lines in order
def write(self, text: str) -> int:
if not text:
return 0
self._buffer += text.replace("\r", "\n")
while "\n" in self._buffer:
raw_line, self._buffer = self._buffer.split("\n", 1)
line = raw_line.rstrip("\r\n")
if line.strip():
self._log_func(line)
self.lines.append(line)
return len(text)
def flush(self) -> None:
line = self._buffer.rstrip("\r\n")
if line.strip():
self._log_func(line)
self.lines.append(line)
self._buffer = ""
```
```python
def _run_pip_main_streaming(pip_main, args: list[str]) -> tuple[int, list[str]]:
stream = _StreamingLogWriter(logger.info)
with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
result_code = pip_main(args)
stream.flush()
return result_code, stream.lines
```
然后把错误/冲突检测集中到一个地方:
```python
def _classify_pip_errors(lines: list[str]) -> tuple[list[str], bool, bool]:
"""Return (error_lines, is_conflict, is_core_conflict)."""
error_lines = [
line for line in lines
if (
line.startswith("ERROR:")
or "The user requested" in line
or "ResolutionImpossible" in line
)
]
is_conflict = any(
"conflict" in line.lower() or "resolutionimpossible" in line.lower()
for line in error_lines
)
is_core_conflict = any("(constraint)" in line for line in error_lines)
return error_lines, is_conflict, is_core_conflict
```
这样 `install` 会更简单:
```python
result_code, lines = await self._run_pip_in_process(args)
error_lines, is_conflict, is_core_conflict = _classify_pip_errors(lines)
if result_code != 0:
if is_conflict:
raise _build_dependency_conflict_error(error_lines, is_core_conflict)
raise Exception(f"安装失败,错误码:{result_code}")
```
### 2. 从 `install` 中抽取参数构建逻辑
目前 `install` 主体的大部分工作是构建 `args`。你可以把这部分移到一个辅助函数中,让 `install` 只负责整体编排:
```python
def _build_pip_args(
self,
package_name: str | None,
requirements_path: str | None,
mirror: str | None,
) -> tuple[list[str], set[str], str | None]:
args = ["install"]
requested_requirements: set[str] = set()
pip_install_args = shlex.split(self.pip_install_arg) if self.pip_install_arg else []
if package_name:
package_specs = _split_package_install_input(package_name)
args.extend(package_specs)
for spec in package_specs:
name = _extract_requirement_name(spec)
if name:
requested_requirements.add(name)
elif requirements_path:
args.extend(["-r", requirements_path])
requested_requirements = _extract_requirement_names(requirements_path)
if not _package_specs_override_index([*args[1:], *pip_install_args]):
index_url = mirror or self.pypi_index_url or "https://pypi.org/simple"
args.extend(["--trusted-host", "mirrors.aliyun.com", "-i", index_url])
return args, requested_requirements, " ".join(pip_install_args)
```
然后 `install` 可以这样写:
```python
args, requested_requirements, _ = self._build_pip_args(
package_name, requirements_path, mirror
)
target_site_packages = None
if is_packaged_desktop_runtime():
target_site_packages = get_astrbot_site_packages_path()
os.makedirs(target_site_packages, exist_ok=True)
_prepend_sys_path(target_site_packages)
args.extend(["--target", target_site_packages])
args.extend(["--upgrade", "--upgrade-strategy", "only-if-needed"])
pip_install_args = shlex.split(self.pip_install_arg) if self.pip_install_arg else []
if pip_install_args:
args.extend(pip_install_args)
```
### 3. 抽离约束文件处理逻辑
约束相关逻辑可以提取到一个小辅助函数中,以缩短 `install` 并让清理行为更一致:
```python
def _attach_core_constraints(args: list[str]) -> str | None:
core_constraints = _get_core_constraints()
if not core_constraints:
return None
try:
import tempfile
with tempfile.NamedTemporaryFile(
mode="w", suffix="_constraints.txt", delete=False, encoding="utf-8"
) as f:
f.write("\n".join(core_constraints))
constraints_file_path = f.name
args.extend(["-c", constraints_file_path])
logger.info("已启用核心依赖版本保护 (%d 个约束)", len(core_constraints))
return constraints_file_path
except Exception as exc:
logger.warning("创建临时约束文件失败: %s", exc)
return None
```
在 `install` 中使用它:
```python
constraints_file_path = _attach_core_constraints(args)
try:
# run pip, interpret errors, etc.
...
finally:
if constraints_file_path and os.path.exists(constraints_file_path):
with contextlib.suppress(Exception):
os.remove(constraints_file_path)
```
### 4. 将冲突错误的构造逻辑集中起来
目前消息构造逻辑是内联在 `install` 中的。把它提取到一个小辅助函数中,可以让正常路径的代码更简洁:
```python
def _build_dependency_conflict_error(
error_lines: list[str], is_core_conflict: bool
) -> DependencyConflictError:
constraints = [l.strip() for l in error_lines if "(constraint)" in l]
requested = [
l.strip()
for l in error_lines
if "The user requested" in l and "(constraint)" not in l
]
detail = ""
if constraints and requested:
detail = (
" 冲突详情: "
f"{requested[0].replace('The user requested ', '')} vs "
f"{constraints[0].replace('The user requested ', '')}。"
)
if is_core_conflict:
msg = (
f"检测到核心依赖版本保护冲突。{detail}"
"插件要求的依赖版本与 AstrBot 核心不兼容,"
"为了系统稳定,已阻止该降级行为。请联系插件作者或调整 requirements.txt。"
)
else:
msg = f"检测到依赖冲突。{detail}"
return DependencyConflictError(msg, error_lines)
```
然后 `install` 只需要调用这个辅助函数即可。
### 5. 将依赖处理相关辅助函数作为一个子系统归类
你已经有一组内聚的辅助函数:`_iter_requirement_specs`、`_strip_inline_requirement_comment`、`_split_package_install_input`、`_package_specs_override_index`、`_find_missing_requirements`、`_specifier_contains_version`。即使你继续把它们放在当前文件中,通过一个小节标题把它们归组在一起也能让意图更清晰:
```python
# === Requirements parsing and analysis utilities ===
# These helpers are responsible for:
# - parsing requirements files (including nested -r),
# - extracting normalized requirement names/specifiers,
# - checking installed versions vs SpecifierSet,
# - detecting when user-provided specs override the index URL.
def _strip_inline_requirement_comment(...): ...
def _split_package_install_input(...): ...
def _iter_requirement_specs(...): ...
def _specifier_contains_version(...): ...
def _find_missing_requirements(...): ...
def _package_specs_override_index(...): ...
```
或者,如果你更倾向的话,可以把它们移动到一个 `requirements_utils.py` 中,同时保持对外行为不变。
</issue_to_address>帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据这些反馈改进之后的 Review。
Original comment in English
Hey - I've found 1 issue
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="50" />
<code_context>
-def _run_pip_main_with_output(pip_main, args: list[str]) -> tuple[int, str]:
- stream = io.StringIO()
- with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
+class _StreamingLogWriter(io.TextIOBase):
+ def __init__(self, log_func) -> None:
+ self._log_func = log_func
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the new pip install logic by extracting focused helpers for streaming, argument building, constraints handling, and conflict error construction to simplify `install` and related code paths.
You can reduce the complexity without losing any functionality by separating responsibilities and extracting a few focused helpers.
### 1. Decouple streaming from error parsing
Right now `_StreamingLogWriter` both streams logs and classifies errors. You can keep streaming behavior but return *all* lines and move classification into a pure function.
```python
class _StreamingLogWriter(io.TextIOBase):
def __init__(self, log_func) -> None:
self._log_func = log_func
self._buffer = ""
self.lines: list[str] = [] # store all lines in order
def write(self, text: str) -> int:
if not text:
return 0
self._buffer += text.replace("\r", "\n")
while "\n" in self._buffer:
raw_line, self._buffer = self._buffer.split("\n", 1)
line = raw_line.rstrip("\r\n")
if line.strip():
self._log_func(line)
self.lines.append(line)
return len(text)
def flush(self) -> None:
line = self._buffer.rstrip("\r\n")
if line.strip():
self._log_func(line)
self.lines.append(line)
self._buffer = ""
```
```python
def _run_pip_main_streaming(pip_main, args: list[str]) -> tuple[int, list[str]]:
stream = _StreamingLogWriter(logger.info)
with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
result_code = pip_main(args)
stream.flush()
return result_code, stream.lines
```
Then error/conflict detection lives in one place:
```python
def _classify_pip_errors(lines: list[str]) -> tuple[list[str], bool, bool]:
"""Return (error_lines, is_conflict, is_core_conflict)."""
error_lines = [
line for line in lines
if (
line.startswith("ERROR:")
or "The user requested" in line
or "ResolutionImpossible" in line
)
]
is_conflict = any(
"conflict" in line.lower() or "resolutionimpossible" in line.lower()
for line in error_lines
)
is_core_conflict = any("(constraint)" in line for line in error_lines)
return error_lines, is_conflict, is_core_conflict
```
And `install` becomes simpler:
```python
result_code, lines = await self._run_pip_in_process(args)
error_lines, is_conflict, is_core_conflict = _classify_pip_errors(lines)
if result_code != 0:
if is_conflict:
raise _build_dependency_conflict_error(error_lines, is_core_conflict)
raise Exception(f"安装失败,错误码:{result_code}")
```
### 2. Extract argument building from `install`
Most of the `install` body is building `args`. You can move that into a helper so `install` reads as orchestration only:
```python
def _build_pip_args(
self,
package_name: str | None,
requirements_path: str | None,
mirror: str | None,
) -> tuple[list[str], set[str], str | None]:
args = ["install"]
requested_requirements: set[str] = set()
pip_install_args = shlex.split(self.pip_install_arg) if self.pip_install_arg else []
if package_name:
package_specs = _split_package_install_input(package_name)
args.extend(package_specs)
for spec in package_specs:
name = _extract_requirement_name(spec)
if name:
requested_requirements.add(name)
elif requirements_path:
args.extend(["-r", requirements_path])
requested_requirements = _extract_requirement_names(requirements_path)
if not _package_specs_override_index([*args[1:], *pip_install_args]):
index_url = mirror or self.pypi_index_url or "https://pypi.org/simple"
args.extend(["--trusted-host", "mirrors.aliyun.com", "-i", index_url])
return args, requested_requirements, " ".join(pip_install_args)
```
Then `install` can do:
```python
args, requested_requirements, _ = self._build_pip_args(
package_name, requirements_path, mirror
)
target_site_packages = None
if is_packaged_desktop_runtime():
target_site_packages = get_astrbot_site_packages_path()
os.makedirs(target_site_packages, exist_ok=True)
_prepend_sys_path(target_site_packages)
args.extend(["--target", target_site_packages])
args.extend(["--upgrade", "--upgrade-strategy", "only-if-needed"])
pip_install_args = shlex.split(self.pip_install_arg) if self.pip_install_arg else []
if pip_install_args:
args.extend(pip_install_args)
```
### 3. Isolate constraints file handling
The constraints logic can be pulled into a tiny helper to shorten `install` and make cleanup consistent:
```python
def _attach_core_constraints(args: list[str]) -> str | None:
core_constraints = _get_core_constraints()
if not core_constraints:
return None
try:
import tempfile
with tempfile.NamedTemporaryFile(
mode="w", suffix="_constraints.txt", delete=False, encoding="utf-8"
) as f:
f.write("\n".join(core_constraints))
constraints_file_path = f.name
args.extend(["-c", constraints_file_path])
logger.info("已启用核心依赖版本保护 (%d 个约束)", len(core_constraints))
return constraints_file_path
except Exception as exc:
logger.warning("创建临时约束文件失败: %s", exc)
return None
```
Use it in `install`:
```python
constraints_file_path = _attach_core_constraints(args)
try:
# run pip, interpret errors, etc.
...
finally:
if constraints_file_path and os.path.exists(constraints_file_path):
with contextlib.suppress(Exception):
os.remove(constraints_file_path)
```
### 4. Centralize conflict error construction
The message-building logic is currently inline in `install`. Moving it into a small helper makes the happy-path shorter:
```python
def _build_dependency_conflict_error(
error_lines: list[str], is_core_conflict: bool
) -> DependencyConflictError:
constraints = [l.strip() for l in error_lines if "(constraint)" in l]
requested = [
l.strip()
for l in error_lines
if "The user requested" in l and "(constraint)" not in l
]
detail = ""
if constraints and requested:
detail = (
" 冲突详情: "
f"{requested[0].replace('The user requested ', '')} vs "
f"{constraints[0].replace('The user requested ', '')}。"
)
if is_core_conflict:
msg = (
f"检测到核心依赖版本保护冲突。{detail}"
"插件要求的依赖版本与 AstrBot 核心不兼容,"
"为了系统稳定,已阻止该降级行为。请联系插件作者或调整 requirements.txt。"
)
else:
msg = f"检测到依赖冲突。{detail}"
return DependencyConflictError(msg, error_lines)
```
Then `install` just calls this helper.
### 5. Group requirement helpers as a subsystem
You already have a coherent set of helpers: `_iter_requirement_specs`, `_strip_inline_requirement_comment`, `_split_package_install_input`, `_package_specs_override_index`, `_find_missing_requirements`, `_specifier_contains_version`. Even if you keep them in this file, a small section header and grouping makes intent clearer:
```python
# === Requirements parsing and analysis utilities ===
# These helpers are responsible for:
# - parsing requirements files (including nested -r),
# - extracting normalized requirement names/specifiers,
# - checking installed versions vs SpecifierSet,
# - detecting when user-provided specs override the index URL.
def _strip_inline_requirement_comment(...): ...
def _split_package_install_input(...): ...
def _iter_requirement_specs(...): ...
def _specifier_contains_version(...): ...
def _find_missing_requirements(...): ...
def _package_specs_override_index(...): ...
```
Or, if you prefer, move them into a `requirements_utils.py` while keeping the public behavior identical.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
SourceryAI
left a comment
There was a problem hiding this comment.
Hey - 我发现了 1 个问题
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="50" />
<code_context>
-def _run_pip_main_with_output(pip_main, args: list[str]) -> tuple[int, str]:
- stream = io.StringIO()
- with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
+class _StreamingLogWriter(io.TextIOBase):
+ def __init__(self, log_func) -> None:
+ self._log_func = log_func
</code_context>
<issue_to_address>
**issue (complexity):** 建议将 pip 日志流/错误处理以及核心约束临时文件的相关逻辑,拆分到更聚焦的辅助函数或上下文管理器中,以简化 install() 中的控制流并减少内联复杂度。
你可以在保留现有全部行为的前提下,通过把部分内联逻辑提取到小而专注的辅助函数里来降低复杂度。
### 1. 将流式日志记录与错误分类解耦
目前 `_StreamingLogWriter` 既负责记录日志,又负责判断哪些行是“错误”行,这会让 `write`/`flush` 更加复杂,并且把它和 pip 的错误启发式规则耦合在一起。
你可以通过让 writer 只负责标准化 + 记录 + 收集日志行,并把错误分类移动到单独的辅助函数中,来简化 IO 路径:
```python
class _StreamingLogWriter(io.TextIOBase):
def __init__(self, log_func) -> None:
self._log_func = log_func
self._buffer = ""
self.lines: list[str] = [] # collect all normalized lines
def write(self, text: str) -> int:
if not text:
return 0
self._buffer += text.replace("\r", "\n")
while "\n" in self._buffer:
raw_line, self._buffer = self._buffer.split("\n", 1)
line = raw_line.rstrip("\r\n")
if line.strip():
self._log_func(line)
self.lines.append(line)
return len(text)
def flush(self) -> None:
line = self._buffer.rstrip("\r\n")
if line.strip():
self._log_func(line)
self.lines.append(line)
self._buffer = ""
```
然后 `_run_pip_main_streaming` 返回所有日志行:
```python
def _run_pip_main_streaming(pip_main, args: list[str]) -> tuple[int, list[str]]:
stream = _StreamingLogWriter(logger.info)
with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
result_code = pip_main(args)
stream.flush()
return result_code, stream.lines
```
接着,把错误/冲突的启发式判断移动到专门的函数里,从而将与 pip 输出相关的逻辑隔离开来:
```python
class PipFailureInfo:
def __init__(self, is_conflict: bool, is_core_conflict: bool, message: str):
self.is_conflict = is_conflict
self.is_core_conflict = is_core_conflict
self.message = message
def _classify_pip_failure(lines: list[str]) -> PipFailureInfo | None:
error_lines = [
line for line in lines
if line.startswith("ERROR:")
or "The user requested" in line
or "ResolutionImpossible" in line
]
if not error_lines:
return None
is_conflict = any(
"conflict" in line.lower() or "resolutionimpossible" in line.lower()
for line in error_lines
)
if not is_conflict:
return None
is_core_conflict = any("(constraint)" in line for line in error_lines)
constraints = [l.strip() for l in error_lines if "(constraint)" in l]
requested = [
l.strip()
for l in error_lines
if "The user requested" in l and "(constraint)" not in l
]
detail = ""
if constraints and requested:
detail = (
" 冲突详情: "
f"{requested[0].removeprefix('The user requested ')} vs "
f"{constraints[0].removeprefix('The user requested ')}。"
)
if is_core_conflict:
message = (
f"检测到核心依赖版本保护冲突。{detail}"
"插件要求的依赖版本与 AstrBot 核心不兼容,"
"为了系统稳定,已阻止该降级行为。请联系插件作者或调整 requirements.txt。"
)
else:
message = f"检测到依赖冲突。{detail}"
return PipFailureInfo(is_conflict=True, is_core_conflict=is_core_conflict, message=message)
```
这样一来,`install` 在错误处理方面会更加清晰:
```python
result_code, lines = await self._run_pip_in_process(args)
if result_code != 0:
failure = _classify_pip_failure(lines)
if failure and failure.is_conflict:
raise DependencyConflictError(failure.message, lines)
raise Exception(f"安装失败,错误码:{result_code}")
```
这在保持现有行为的同时(甚至通过集中启发式规则让逻辑更健壮),简化了 `_StreamingLogWriter` 和 `install`。
### 2. 将核心约束文件的生命周期提取为一个上下文管理器
当前 `constraints_file_path` 的 `try/finally` 代码块让 `install` 更难阅读。可以通过一个小的上下文管理器来封装核心约束文件的逻辑和清理工作:
```python
@contextlib.contextmanager
def _core_constraints_file() -> Iterator[str | None]:
constraints = _get_core_constraints()
if not constraints:
yield None
return
path: str | None = None
try:
import tempfile
with tempfile.NamedTemporaryFile(
mode="w", suffix="_constraints.txt", delete=False, encoding="utf-8"
) as f:
f.write("\n".join(constraints))
path = f.name
logger.info("已启用核心依赖版本保护 (%d 个约束)", len(constraints))
yield path
except Exception as exc:
logger.warning("创建临时约束文件失败: %s", exc)
yield None
finally:
if path and os.path.exists(path):
with contextlib.suppress(Exception):
os.remove(path)
```
然后 `install` 可以简化为:
```python
with _core_constraints_file() as constraints_file_path:
if constraints_file_path:
args.extend(["-c", constraints_file_path])
logger.info("Pip 包管理器 argv: %s", ["pip", *args])
result_code, lines = await self._run_pip_in_process(args)
if result_code != 0:
failure = _classify_pip_failure(lines)
if failure and failure.is_conflict:
raise DependencyConflictError(failure.message, lines)
raise Exception(f"安装失败,错误码:{result_code}")
if target_site_packages:
_prepend_sys_path(target_site_packages)
_ensure_plugin_dependencies_preferred(target_site_packages, requested_requirements)
importlib.invalidate_caches()
```
这样既保留了功能(核心保护 + 清理),又减少了样板代码,使 `install` 中的控制流更易于理解。
</issue_to_address>Hi @zouyonghe! 👋
感谢你通过评论 @sourcery-ai review 来试用 Sourcery!🚀
安装 sourcery-ai bot 以在每个 pull request 上自动获取代码审查 ✨
帮我变得更有用!请在每条评论上点击 👍 或 👎,我会根据这些反馈改进之后的审查。Original comment in English
Hey - I've found 1 issue
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="50" />
<code_context>
-def _run_pip_main_with_output(pip_main, args: list[str]) -> tuple[int, str]:
- stream = io.StringIO()
- with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
+class _StreamingLogWriter(io.TextIOBase):
+ def __init__(self, log_func) -> None:
+ self._log_func = log_func
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the pip streaming/error-handling and core-constraints temp-file logic into focused helpers/context managers to simplify control flow and reduce inline complexity in install().
You can keep all current behavior but reduce complexity by pushing some of the inline logic into small, focused helpers.
### 1. Decouple streaming logger from error classification
Right now `_StreamingLogWriter` both logs and decides which lines are “error” lines, which makes `write`/`flush` more complex and couples it to pip’s error heuristics.
You can simplify the IO path by letting the writer just normalize + log + collect lines, and move the classification into a separate helper:
```python
class _StreamingLogWriter(io.TextIOBase):
def __init__(self, log_func) -> None:
self._log_func = log_func
self._buffer = ""
self.lines: list[str] = [] # collect all normalized lines
def write(self, text: str) -> int:
if not text:
return 0
self._buffer += text.replace("\r", "\n")
while "\n" in self._buffer:
raw_line, self._buffer = self._buffer.split("\n", 1)
line = raw_line.rstrip("\r\n")
if line.strip():
self._log_func(line)
self.lines.append(line)
return len(text)
def flush(self) -> None:
line = self._buffer.rstrip("\r\n")
if line.strip():
self._log_func(line)
self.lines.append(line)
self._buffer = ""
```
Then `_run_pip_main_streaming` returns all lines:
```python
def _run_pip_main_streaming(pip_main, args: list[str]) -> tuple[int, list[str]]:
stream = _StreamingLogWriter(logger.info)
with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
result_code = pip_main(args)
stream.flush()
return result_code, stream.lines
```
And the error/conflict heuristics move into a dedicated function, isolating pip-output-specific logic:
```python
class PipFailureInfo:
def __init__(self, is_conflict: bool, is_core_conflict: bool, message: str):
self.is_conflict = is_conflict
self.is_core_conflict = is_core_conflict
self.message = message
def _classify_pip_failure(lines: list[str]) -> PipFailureInfo | None:
error_lines = [
line for line in lines
if line.startswith("ERROR:")
or "The user requested" in line
or "ResolutionImpossible" in line
]
if not error_lines:
return None
is_conflict = any(
"conflict" in line.lower() or "resolutionimpossible" in line.lower()
for line in error_lines
)
if not is_conflict:
return None
is_core_conflict = any("(constraint)" in line for line in error_lines)
constraints = [l.strip() for l in error_lines if "(constraint)" in l]
requested = [
l.strip()
for l in error_lines
if "The user requested" in l and "(constraint)" not in l
]
detail = ""
if constraints and requested:
detail = (
" 冲突详情: "
f"{requested[0].removeprefix('The user requested ')} vs "
f"{constraints[0].removeprefix('The user requested ')}。"
)
if is_core_conflict:
message = (
f"检测到核心依赖版本保护冲突。{detail}"
"插件要求的依赖版本与 AstrBot 核心不兼容,"
"为了系统稳定,已阻止该降级行为。请联系插件作者或调整 requirements.txt。"
)
else:
message = f"检测到依赖冲突。{detail}"
return PipFailureInfo(is_conflict=True, is_core_conflict=is_core_conflict, message=message)
```
`install` then becomes much clearer on error handling:
```python
result_code, lines = await self._run_pip_in_process(args)
if result_code != 0:
failure = _classify_pip_failure(lines)
if failure and failure.is_conflict:
raise DependencyConflictError(failure.message, lines)
raise Exception(f"安装失败,错误码:{result_code}")
```
This keeps all existing behavior (and even makes it a bit more robust by centralizing the heuristics) while simplifying `_StreamingLogWriter` and `install`.
### 2. Extract core-constraints file lifecycle into a context manager
The `constraints_file_path` `try/finally` block makes `install` harder to scan. A small context manager can encapsulate the core-constraints logic and cleanup:
```python
@contextlib.contextmanager
def _core_constraints_file() -> Iterator[str | None]:
constraints = _get_core_constraints()
if not constraints:
yield None
return
path: str | None = None
try:
import tempfile
with tempfile.NamedTemporaryFile(
mode="w", suffix="_constraints.txt", delete=False, encoding="utf-8"
) as f:
f.write("\n".join(constraints))
path = f.name
logger.info("已启用核心依赖版本保护 (%d 个约束)", len(constraints))
yield path
except Exception as exc:
logger.warning("创建临时约束文件失败: %s", exc)
yield None
finally:
if path and os.path.exists(path):
with contextlib.suppress(Exception):
os.remove(path)
```
Then `install` simplifies to:
```python
with _core_constraints_file() as constraints_file_path:
if constraints_file_path:
args.extend(["-c", constraints_file_path])
logger.info("Pip 包管理器 argv: %s", ["pip", *args])
result_code, lines = await self._run_pip_in_process(args)
if result_code != 0:
failure = _classify_pip_failure(lines)
if failure and failure.is_conflict:
raise DependencyConflictError(failure.message, lines)
raise Exception(f"安装失败,错误码:{result_code}")
if target_site_packages:
_prepend_sys_path(target_site_packages)
_ensure_plugin_dependencies_preferred(target_site_packages, requested_requirements)
importlib.invalidate_caches()
```
This keeps the feature set (core protection + cleanup) while reducing bookkeeping and making the control flow in `install` easier to follow.
</issue_to_address>Hi @zouyonghe! 👋
Thanks for trying out Sourcery by commenting with @sourcery-ai review! 🚀
Install the sourcery-ai bot to get automatic code reviews on every pull request ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.…ification and constraint file management
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了 1 个问题,并给出了一些整体性的反馈:
- 在
star_manager._install_plugin_requirements_with_logging中,你捕获了DependencyConflictError,但该符号既没有在当前模块中导入,也没有加上命名空间限定;运行时会触发NameError——请从pip_installer中导入它,或者改为引用pip_installer.DependencyConflictError。
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `star_manager._install_plugin_requirements_with_logging` you catch `DependencyConflictError`, but that symbol is not imported or qualified in this module; this will raise a `NameError` at runtime—either import it from `pip_installer` or reference it as `pip_installer.DependencyConflictError`.
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="50" />
<code_context>
-def _run_pip_main_with_output(pip_main, args: list[str]) -> tuple[int, str]:
- stream = io.StringIO()
- with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
+class _StreamingLogWriter(io.TextIOBase):
+ def __init__(self, log_func) -> None:
+ self._log_func = log_func
</code_context>
<issue_to_address>
**issue (complexity):** 请考虑通过减少缓冲/格式化逻辑、合并错误分类结构、将 requirements 文件递归和行解析拆开,以及把参数构造从 `install` 中抽离出来,来简化新的 pip 安装辅助工具,使整体执行流程更加线性、易读。
1. **_StreamingLogWriter 和 `_run_pip_in_process` 可进一步简化**
目前 `_StreamingLogWriter` 同时在做缓冲、CR/LF 归一化、空白字符策略、日志记录,以及为后续错误分类进行累积。
你可以在保持现有行为的前提下,将其简化为「按行切分、忽略空行、写日志并保存」,并交给 `str.splitlines()` 处理换行细节:
```python
class _StreamingLogWriter(io.TextIOBase):
def __init__(self, log_func) -> None:
self._log_func = log_func
self.lines: list[str] = []
def write(self, text: str) -> int:
if not text:
return 0
for raw_line in text.splitlines():
line = raw_line.rstrip("\r\n")
if not line.strip():
continue
self._log_func(line)
self.lines.append(line)
return len(text)
def flush(self) -> None:
# Nothing to do; we don't buffer across writes
pass
```
然后,`_run_pip_main_streaming` 可以变成一个很薄的包装:
```python
def _run_pip_main_streaming(pip_main, args: list[str]) -> tuple[int, list[str]]:
stream = _StreamingLogWriter(logger.info)
with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
result_code = pip_main(args)
return result_code, stream.lines
```
这样可以保留:
- 仅记录非空行日志
- 使用 `splitlines()` 统一 CR/LF 处理
- 为错误分类累积日志行
同时移除手动缓冲和换行处理逻辑。
---
2. **合并 `PipFailureInfo` 并简化错误处理流程**
`PipFailureInfo` + `_classify_pip_failure` + `DependencyConflictError` 再加上 `(code, lines)` 这种返回类型,会引入较多间接层。
你可以通过让 `_classify_pip_failure` 仅返回一个异常消息(或 `None`),并将 conflict / core-conflict 这类布尔信息移到异常里,甚至直接在 `_classify_pip_failure` 中抛出异常,来简化结构。
例如:保留现有的 `DependencyConflictError`,但移除 `PipFailureInfo`:
```python
class DependencyConflictError(Exception):
def __init__(self, message: str, errors: list[str], *, is_core_conflict: bool) -> None:
super().__init__(message)
self.errors = errors
self.is_core_conflict = is_core_conflict
def _classify_pip_failure(lines: list[str]) -> DependencyConflictError | None:
error_lines = [
line for line in lines
if line.startswith("ERROR:")
or "The user requested" in line
or "ResolutionImpossible" in line
]
if not error_lines:
return None
is_conflict = any(
"conflict" in line.lower() or "resolutionimpossible" in line.lower()
for line in error_lines
)
if not is_conflict:
return None
is_core_conflict = any("(constraint)" in line for line in error_lines)
constraints = [l.strip() for l in error_lines if "(constraint)" in l]
requested = [
l.strip()
for l in error_lines
if "The user requested" in l and "(constraint)" not in l
]
detail = ""
if constraints and requested:
detail = (
" 冲突详情: "
f"{requested[0].removeprefix('The user requested ')} vs "
f"{constraints[0].removeprefix('The user requested ')}。"
)
if is_core_conflict:
message = (
f"检测到核心依赖版本保护冲突。{detail}"
"插件要求的依赖版本与 AstrBot 核心不兼容,"
"为了系统稳定,已阻止该降级行为。请联系插件作者或调整 requirements.txt。"
)
else:
message = f"检测到依赖冲突。{detail}"
return DependencyConflictError(message, error_lines, is_core_conflict=is_core_conflict)
```
之后,`PipInstaller.install` 可以精简为:
```python
result_code, lines = await self._run_pip_in_process(args)
if result_code != 0:
conflict = _classify_pip_failure(lines)
if conflict:
raise conflict
raise Exception(f"安装失败,错误码:{result_code}")
```
并且可以删除那个只做「捕获然后立刻重新抛出」的冗余 try/except:
```python
# before
try:
...
except DependencyConflictError:
raise
except Exception:
raise
# after: just let exceptions propagate
...
result_code, lines = await self._run_pip_in_process(args)
...
```
这样可以去掉 `PipFailureInfo` 和一层分支,但保留结构化的冲突异常信息。
---
3. **拆分 `_iter_requirement_specs` 的职责**
`_iter_requirement_specs` 当前同时在处理:文件递归、选项解析、marker 评估,以及「尽力而为」的名称提取。你可以先将文件展平为已做注释去除和规范化的行,再对这些行做解析,从而降低单个函数的复杂度。
示例拆分:
```python
def _iter_requirement_lines(
requirements_path: str,
*,
_visited_paths: set[str] | None = None,
) -> Iterator[str]:
visited_paths = _visited_paths or set()
resolved_path = os.path.realpath(requirements_path)
if resolved_path in visited_paths:
return
visited_paths.add(resolved_path)
with open(resolved_path, encoding="utf-8") as requirements_file:
for raw_line in requirements_file:
line = _strip_inline_requirement_comment(raw_line)
if not line or line.startswith("#"):
continue
tokens = shlex.split(line)
if not tokens:
continue
nested_path = None
if tokens[0] in {"-r", "--requirement"} and len(tokens) > 1:
nested_path = tokens[1]
elif tokens[0].startswith("--requirement="):
nested_path = tokens[0].split("=", 1)[1]
if nested_path:
if not os.path.isabs(nested_path):
nested_path = os.path.join(os.path.dirname(resolved_path), nested_path)
yield from _iter_requirement_lines(nested_path, _visited_paths=visited_paths)
continue
yield line
```
这样 `_iter_requirement_specs` 只需要关心「一条逻辑上的 requirement 行 → (name, specifier)」:
```python
def _iter_requirement_specs(requirements_path: str) -> Iterator[tuple[str, SpecifierSet | None]]:
for line in _iter_requirement_lines(requirements_path):
tokens = shlex.split(line)
if not tokens:
continue
# constraints and pure options
if tokens[0] in {"-c", "--constraint"}:
continue
if tokens[0].startswith("-"):
requirement_name = _extract_requirement_name(line)
if requirement_name:
yield requirement_name, None
continue
try:
requirement = Requirement(line)
except InvalidRequirement:
requirement_name = _extract_requirement_name(line)
if requirement_name:
yield requirement_name, None
continue
if requirement.marker and not requirement.marker.evaluate():
continue
yield (
_canonicalize_distribution_name(requirement.name),
requirement.specifier,
)
```
这样既能保持你当前的行为(嵌套 `-r`、注释去除、选项处理、markers、兜底名称提取),又让每个函数更易于理解和测试。
---
4. **将参数构造逻辑从 `PipInstaller.install` 中抽取出来**
`install` 目前在做很多不相干的事情:解析用户输入、执行索引 URL 规则、桌面环境专用的 target 配置、核心约束文件处理,以及错误处理等。
可以通过把参数构造逻辑抽成一个小函数,来减少其中的分支,使「输入归一化」和「执行流程」的边界更加清晰:
```python
def _build_pip_args(
self,
package_name: str | None,
requirements_path: str | None,
mirror: str | None,
) -> tuple[list[str], set[str]]:
args = ["install"]
requested_requirements: set[str] = set()
pip_install_args = shlex.split(self.pip_install_arg) if self.pip_install_arg else []
if package_name:
package_specs = _split_package_install_input(package_name)
args.extend(package_specs)
for package_spec in package_specs:
requirement_name = _extract_requirement_name(package_spec)
if requirement_name:
requested_requirements.add(requirement_name)
elif requirements_path:
args.extend(["-r", requirements_path])
requested_requirements = _extract_requirement_names(requirements_path)
if not _package_specs_override_index([*args[1:], *pip_install_args]):
index_url = mirror or self.pypi_index_url or "https://pypi.org/simple"
args.extend(["--trusted-host", "mirrors.aliyun.com", "-i", index_url])
if pip_install_args:
args.extend(pip_install_args)
return args, requested_requirements
```
然后 `install` 可以简化为:
```python
async def install(...):
args, requested_requirements = self._build_pip_args(
package_name, requirements_path, mirror
)
target_site_packages = None
if is_packaged_desktop_runtime():
target_site_packages = get_astrbot_site_packages_path()
os.makedirs(target_site_packages, exist_ok=True)
_prepend_sys_path(target_site_packages)
args.extend(["--target", target_site_packages, "--upgrade", "--upgrade-strategy", "only-if-needed"])
with _core_constraints_file() as constraints_file_path:
if constraints_file_path:
args.extend(["-c", constraints_file_path])
logger.info("Pip 包管理器 argv: %s", ["pip", *args])
result_code, lines = await self._run_pip_in_process(args)
if result_code != 0:
conflict = _classify_pip_failure(lines)
if conflict:
raise conflict
raise Exception(f"安装失败,错误码:{result_code}")
if target_site_packages:
_prepend_sys_path(target_site_packages)
_ensure_plugin_dependencies_preferred(target_site_packages, requested_requirements)
importlib.invalidate_caches()
```
这样既保留了你新增的功能(索引覆盖检测、核心约束、桌面端 target 支持),又让 `install` 流程更线性,并在参数构造和 pip 执行之间划清了职责边界。
</issue_to_address>帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进后续评审。
Original comment in English
Hey - I've found 1 issue, and left some high level feedback:
- In
star_manager._install_plugin_requirements_with_loggingyou catchDependencyConflictError, but that symbol is not imported or qualified in this module; this will raise aNameErrorat runtime—either import it frompip_installeror reference it aspip_installer.DependencyConflictError.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `star_manager._install_plugin_requirements_with_logging` you catch `DependencyConflictError`, but that symbol is not imported or qualified in this module; this will raise a `NameError` at runtime—either import it from `pip_installer` or reference it as `pip_installer.DependencyConflictError`.
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="50" />
<code_context>
-def _run_pip_main_with_output(pip_main, args: list[str]) -> tuple[int, str]:
- stream = io.StringIO()
- with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
+class _StreamingLogWriter(io.TextIOBase):
+ def __init__(self, log_func) -> None:
+ self._log_func = log_func
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the new pip installation helpers by reducing buffering/formatting logic, collapsing error-classification structures, separating requirement-file recursion from line parsing, and extracting argument-building out of `install` to make the flow more linear and readable.
1. **_StreamingLogWriter + `_run_pip_in_process` can be simplified**
Right now `_StreamingLogWriter` is doing buffering, CR/LF normalization, whitespace policies, logging, and accumulation for later classification.
You can keep the same behavior while simplifying it to “split on newlines, ignore blank lines, log, and store”, and let `str.splitlines()` handle newline details:
```python
class _StreamingLogWriter(io.TextIOBase):
def __init__(self, log_func) -> None:
self._log_func = log_func
self.lines: list[str] = []
def write(self, text: str) -> int:
if not text:
return 0
for raw_line in text.splitlines():
line = raw_line.rstrip("\r\n")
if not line.strip():
continue
self._log_func(line)
self.lines.append(line)
return len(text)
def flush(self) -> None:
# Nothing to do; we don't buffer across writes
pass
```
Then `_run_pip_main_streaming` becomes a thin wrapper:
```python
def _run_pip_main_streaming(pip_main, args: list[str]) -> tuple[int, list[str]]:
stream = _StreamingLogWriter(logger.info)
with contextlib.redirect_stdout(stream), contextlib.redirect_stderr(stream):
result_code = pip_main(args)
return result_code, stream.lines
```
This preserves:
- logging only non-empty lines
- normalized CR/LF via `splitlines()`
- accumulation for error classification
but removes the manual buffer and newline handling.
---
2. **Collapse `PipFailureInfo` and simplify error flow**
`PipFailureInfo` + `_classify_pip_failure` + `DependencyConflictError` + the `(code, lines)` return type adds a lot of indirection.
You can simplify by having `_classify_pip_failure` just return a message (or `None`) and move the conflict/core-conflict booleans into the exception, or even have `_classify_pip_failure` raise directly.
Example: keep your `DependencyConflictError` but drop `PipFailureInfo`:
```python
class DependencyConflictError(Exception):
def __init__(self, message: str, errors: list[str], *, is_core_conflict: bool) -> None:
super().__init__(message)
self.errors = errors
self.is_core_conflict = is_core_conflict
def _classify_pip_failure(lines: list[str]) -> DependencyConflictError | None:
error_lines = [
line for line in lines
if line.startswith("ERROR:")
or "The user requested" in line
or "ResolutionImpossible" in line
]
if not error_lines:
return None
is_conflict = any(
"conflict" in line.lower() or "resolutionimpossible" in line.lower()
for line in error_lines
)
if not is_conflict:
return None
is_core_conflict = any("(constraint)" in line for line in error_lines)
constraints = [l.strip() for l in error_lines if "(constraint)" in l]
requested = [
l.strip()
for l in error_lines
if "The user requested" in l and "(constraint)" not in l
]
detail = ""
if constraints and requested:
detail = (
" 冲突详情: "
f"{requested[0].removeprefix('The user requested ')} vs "
f"{constraints[0].removeprefix('The user requested ')}。"
)
if is_core_conflict:
message = (
f"检测到核心依赖版本保护冲突。{detail}"
"插件要求的依赖版本与 AstrBot 核心不兼容,"
"为了系统稳定,已阻止该降级行为。请联系插件作者或调整 requirements.txt。"
)
else:
message = f"检测到依赖冲突。{detail}"
return DependencyConflictError(message, error_lines, is_core_conflict=is_core_conflict)
```
Then `PipInstaller.install` can be simplified to:
```python
result_code, lines = await self._run_pip_in_process(args)
if result_code != 0:
conflict = _classify_pip_failure(lines)
if conflict:
raise conflict
raise Exception(f"安装失败,错误码:{result_code}")
```
And you can drop the redundant try/except that immediately re-raises:
```python
# before
try:
...
except DependencyConflictError:
raise
except Exception:
raise
# after: just let exceptions propagate
...
result_code, lines = await self._run_pip_in_process(args)
...
```
This removes `PipFailureInfo` and one layer of branching while keeping your structured conflict error.
---
3. **Split `_iter_requirement_specs` responsibilities**
`_iter_requirement_specs` is handling file recursion, option parsing, marker evaluation, and “best-effort” name extraction all at once. You can reduce cognitive load by first flattening files into normalized, comment-stripped lines, then parsing those lines.
Example split:
```python
def _iter_requirement_lines(
requirements_path: str,
*,
_visited_paths: set[str] | None = None,
) -> Iterator[str]:
visited_paths = _visited_paths or set()
resolved_path = os.path.realpath(requirements_path)
if resolved_path in visited_paths:
return
visited_paths.add(resolved_path)
with open(resolved_path, encoding="utf-8") as requirements_file:
for raw_line in requirements_file:
line = _strip_inline_requirement_comment(raw_line)
if not line or line.startswith("#"):
continue
tokens = shlex.split(line)
if not tokens:
continue
nested_path = None
if tokens[0] in {"-r", "--requirement"} and len(tokens) > 1:
nested_path = tokens[1]
elif tokens[0].startswith("--requirement="):
nested_path = tokens[0].split("=", 1)[1]
if nested_path:
if not os.path.isabs(nested_path):
nested_path = os.path.join(os.path.dirname(resolved_path), nested_path)
yield from _iter_requirement_lines(nested_path, _visited_paths=visited_paths)
continue
yield line
```
Then `_iter_requirement_specs` only has to worry about “a single logical requirement line → (name, specifier)”:
```python
def _iter_requirement_specs(requirements_path: str) -> Iterator[tuple[str, SpecifierSet | None]]:
for line in _iter_requirement_lines(requirements_path):
tokens = shlex.split(line)
if not tokens:
continue
# constraints and pure options
if tokens[0] in {"-c", "--constraint"}:
continue
if tokens[0].startswith("-"):
requirement_name = _extract_requirement_name(line)
if requirement_name:
yield requirement_name, None
continue
try:
requirement = Requirement(line)
except InvalidRequirement:
requirement_name = _extract_requirement_name(line)
if requirement_name:
yield requirement_name, None
continue
if requirement.marker and not requirement.marker.evaluate():
continue
yield (
_canonicalize_distribution_name(requirement.name),
requirement.specifier,
)
```
This preserves your current behavior (nested `-r`, comment stripping, options handling, markers, fallback name extraction) but makes each function easier to reason about and test.
---
4. **Factor argument building out of `PipInstaller.install`**
`install` is doing a lot of unrelated work: parsing user input, enforcing index URL rules, desktop-specific target setup, core constraints, and error handling.
A small extraction for arg building will reduce its branching and make it easier to see what’s “input normalization” vs “execution”:
```python
def _build_pip_args(
self,
package_name: str | None,
requirements_path: str | None,
mirror: str | None,
) -> tuple[list[str], set[str]]:
args = ["install"]
requested_requirements: set[str] = set()
pip_install_args = shlex.split(self.pip_install_arg) if self.pip_install_arg else []
if package_name:
package_specs = _split_package_install_input(package_name)
args.extend(package_specs)
for package_spec in package_specs:
requirement_name = _extract_requirement_name(package_spec)
if requirement_name:
requested_requirements.add(requirement_name)
elif requirements_path:
args.extend(["-r", requirements_path])
requested_requirements = _extract_requirement_names(requirements_path)
if not _package_specs_override_index([*args[1:], *pip_install_args]):
index_url = mirror or self.pypi_index_url or "https://pypi.org/simple"
args.extend(["--trusted-host", "mirrors.aliyun.com", "-i", index_url])
if pip_install_args:
args.extend(pip_install_args)
return args, requested_requirements
```
Then `install` becomes:
```python
async def install(...):
args, requested_requirements = self._build_pip_args(
package_name, requirements_path, mirror
)
target_site_packages = None
if is_packaged_desktop_runtime():
target_site_packages = get_astrbot_site_packages_path()
os.makedirs(target_site_packages, exist_ok=True)
_prepend_sys_path(target_site_packages)
args.extend(["--target", target_site_packages, "--upgrade", "--upgrade-strategy", "only-if-needed"])
with _core_constraints_file() as constraints_file_path:
if constraints_file_path:
args.extend(["-c", constraints_file_path])
logger.info("Pip 包管理器 argv: %s", ["pip", *args])
result_code, lines = await self._run_pip_in_process(args)
if result_code != 0:
conflict = _classify_pip_failure(lines)
if conflict:
raise conflict
raise Exception(f"安装失败,错误码:{result_code}")
if target_site_packages:
_prepend_sys_path(target_site_packages)
_ensure_plugin_dependencies_preferred(target_site_packages, requested_requirements)
importlib.invalidate_caches()
```
This keeps all of your new features (index override detection, core constraints, desktop target) but makes `install` more linear and separates concerns between building args and executing pip.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了两个问题
给 AI 代理的提示
请根据本次代码审查中的评论进行修改:
## 单独评论
### 评论 1
<location path="astrbot/core/utils/pip_installer.py" line_range="248-257" />
<code_context>
return names
+def _iter_requirement_lines(
+ requirements_path: str,
+ *,
+ _visited_paths: set[str] | None = None,
+) -> Iterator[str]:
+ visited_paths = _visited_paths or set()
+ resolved_path = os.path.realpath(requirements_path)
+ if resolved_path in visited_paths:
+ return
+ visited_paths.add(resolved_path)
</code_context>
<issue_to_address>
**suggestion:** 对循环的 -r include 进行静默早返回会让调试变得困难。
目前循环 include 会被静默忽略:当某个文件被再次访问时,函数会直接返回,而不会 yield 或记录日志。请在检测到循环时至少记录一个警告(包含有问题的路径),以便用户可以理解为什么某些 requirements 被跳过。
```suggestion
def _iter_requirement_lines(
requirements_path: str,
*,
_visited_paths: set[str] | None = None,
) -> Iterator[str]:
visited_paths = _visited_paths or set()
resolved_path = os.path.realpath(requirements_path)
if resolved_path in visited_paths:
logger.warning(
"检测到循环依赖的 requirements 包含: %s,将跳过该文件",
resolved_path,
)
return
visited_paths.add(resolved_path)
```
</issue_to_address>
### 评论 2
<location path="astrbot/core/star/star_manager.py" line_range="204" />
<code_context>
+ await self._install_plugin_requirements_with_logging(plugin_path, p)
return True
+ async def _install_plugin_requirements(
+ self,
+ plugin_dir_path: str,
</code_context>
<issue_to_address>
**issue (complexity):** 建议将新的依赖安装辅助函数合并为一个语义明确的方法,并把哨兵值处理下放到安装器中,以简化分支逻辑和调用点。
1. 将两个辅助方法合并为一个 “ensure” 方法
```python
async def _ensure_plugin_requirements(
self,
plugin_dir_path: str,
plugin_label: str,
) -> None:
requirements_path = os.path.join(plugin_dir_path, "requirements.txt")
if not os.path.exists(requirements_path):
return
try:
missing = pip_installer.find_missing_requirements(requirements_path)
if missing == set():
logger.info(f"插件 {plugin_label} 的依赖已满足,跳过安装。")
return
if missing is None:
logger.info(
f"正在安装插件 {plugin_label} 的依赖库(预检查失败,回退到完整安装): "
f"{requirements_path}"
)
else:
logger.info(
f"正在安装插件 {plugin_label} 缺失的依赖库: "
f"{requirements_path} -> {sorted(missing)}"
)
await pip_installer.install(requirements_path=requirements_path)
except asyncio.CancelledError:
raise
except DependencyConflictError as e:
logger.error(f"插件 {plugin_label} 依赖冲突: {e!s}")
except Exception as e:
logger.exception(f"插件 {plugin_label} 依赖安装失败: {e!s}")
```
然后所有调用点都可以变得更简单、统一:
```python
plugin_path = os.path.join(self.plugin_store_path, dir_name)
await self._ensure_plugin_requirements(plugin_path, dir_name)
```
这样可以在保持所有行为(预检查 + 日志记录 + 错误处理)的同时,去掉一层间接调用,并将其封装在一个语义清晰的操作中。
2. 在安装器内部隐藏 `None`/`set()` 协议(可选,但可以减少分支)
如果可以修改 `pip_installer`,可以把对 `None` 哨兵值的处理下放到安装器中,让 `StarManager` 只关心“是否有缺失依赖”:
```python
# in pip_installer.py
class RequirementsPrecheckFailed(Exception):
pass
def find_missing_requirements_or_raise(requirements_path: str) -> set[str]:
missing = find_missing_requirements(requirements_path)
if missing is None:
raise RequirementsPrecheckFailed(f"预检查失败: {requirements_path}")
return missing
```
然后 StarManager 这边就可以简化为:
```python
try:
missing = pip_installer.find_missing_requirements_or_raise(requirements_path)
except RequirementsPrecheckFailed:
logger.info(
f"正在安装插件 {plugin_label} 的依赖库(预检查失败,回退到完整安装): "
f"{requirements_path}"
)
await pip_installer.install(requirements_path=requirements_path)
return
if not missing:
logger.info(f"插件 {plugin_label} 的依赖已满足,跳过安装。")
return
logger.info(
f"正在安装插件 {plugin_label} 缺失的依赖库: "
f"{requirements_path} -> {sorted(missing)}"
)
await pip_installer.install(requirements_path=requirements_path)
```
这样既保留了现有行为(包括当预检查失败时不同的日志信息),又避免了让 `StarManager` 理解 `None` 作为特殊哨兵值的需要,从而减少了本文件中的分支和理解负担。
</issue_to_address>帮我变得更有用!请在每条评论上点击 👍 或 👎,我会根据反馈改进你的代码审查体验。
Original comment in English
Hey - I've found 2 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="248-257" />
<code_context>
return names
+def _iter_requirement_lines(
+ requirements_path: str,
+ *,
+ _visited_paths: set[str] | None = None,
+) -> Iterator[str]:
+ visited_paths = _visited_paths or set()
+ resolved_path = os.path.realpath(requirements_path)
+ if resolved_path in visited_paths:
+ return
+ visited_paths.add(resolved_path)
</code_context>
<issue_to_address>
**suggestion:** Silent early-return on cyclic -r includes can make debugging hard.
Right now cyclic includes are silently ignored: the function returns without yielding or logging when a file is revisited. Please log at least a warning (including the problematic path) when a cycle is detected so users can understand why some requirements are being skipped.
```suggestion
def _iter_requirement_lines(
requirements_path: str,
*,
_visited_paths: set[str] | None = None,
) -> Iterator[str]:
visited_paths = _visited_paths or set()
resolved_path = os.path.realpath(requirements_path)
if resolved_path in visited_paths:
logger.warning(
"检测到循环依赖的 requirements 包含: %s,将跳过该文件",
resolved_path,
)
return
visited_paths.add(resolved_path)
```
</issue_to_address>
### Comment 2
<location path="astrbot/core/star/star_manager.py" line_range="204" />
<code_context>
+ await self._install_plugin_requirements_with_logging(plugin_path, p)
return True
+ async def _install_plugin_requirements(
+ self,
+ plugin_dir_path: str,
</code_context>
<issue_to_address>
**issue (complexity):** Consider collapsing the new requirement-installation helpers into a single, clearly named method and delegating sentinel handling to the installer to simplify branching and call sites.
1. Collapse the two helpers into a single “ensure” method
```python
async def _ensure_plugin_requirements(
self,
plugin_dir_path: str,
plugin_label: str,
) -> None:
requirements_path = os.path.join(plugin_dir_path, "requirements.txt")
if not os.path.exists(requirements_path):
return
try:
missing = pip_installer.find_missing_requirements(requirements_path)
if missing == set():
logger.info(f"插件 {plugin_label} 的依赖已满足,跳过安装。")
return
if missing is None:
logger.info(
f"正在安装插件 {plugin_label} 的依赖库(预检查失败,回退到完整安装): "
f"{requirements_path}"
)
else:
logger.info(
f"正在安装插件 {plugin_label} 缺失的依赖库: "
f"{requirements_path} -> {sorted(missing)}"
)
await pip_installer.install(requirements_path=requirements_path)
except asyncio.CancelledError:
raise
except DependencyConflictError as e:
logger.error(f"插件 {plugin_label} 依赖冲突: {e!s}")
except Exception as e:
logger.exception(f"插件 {plugin_label} 依赖安装失败: {e!s}")
```
Then all call sites become simpler and uniform:
```python
plugin_path = os.path.join(self.plugin_store_path, dir_name)
await self._ensure_plugin_requirements(plugin_path, dir_name)
```
This removes one layer of indirection while keeping all behavior (pre-check + logging + error handling) in a single clearly named operation.
2. Hide the `None`/`set()` contract inside the installer (optional but reduces branching)
If you can touch `pip_installer`, you can push the `None` sentinel handling down so `StarManager` only deals with “missing requirements or not”:
```python
# in pip_installer.py
class RequirementsPrecheckFailed(Exception):
pass
def find_missing_requirements_or_raise(requirements_path: str) -> set[str]:
missing = find_missing_requirements(requirements_path)
if missing is None:
raise RequirementsPrecheckFailed(f"预检查失败: {requirements_path}")
return missing
```
Then the StarManager side simplifies to:
```python
try:
missing = pip_installer.find_missing_requirements_or_raise(requirements_path)
except RequirementsPrecheckFailed:
logger.info(
f"正在安装插件 {plugin_label} 的依赖库(预检查失败,回退到完整安装): "
f"{requirements_path}"
)
await pip_installer.install(requirements_path=requirements_path)
return
if not missing:
logger.info(f"插件 {plugin_label} 的依赖已满足,跳过安装。")
return
logger.info(
f"正在安装插件 {plugin_label} 缺失的依赖库: "
f"{requirements_path} -> {sorted(missing)}"
)
await pip_installer.install(requirements_path=requirements_path)
```
This keeps the existing behavior (including the distinct log message when the pre-check fails) but removes the need for `StarManager` to understand `None` as a special sentinel value and reduces the branching/mental overhead in this file.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了 1 个问题,并留下了一些总体反馈:
PipInstaller同时暴露了模块级的find_missing_requirements*辅助函数和同名的实例方法,而这些实例方法只是简单地委托调用;可以考虑去掉这些实例包装器,在StarManager中直接使用模块级 API,以避免多余的间接调用以及对逻辑实际所在位置的困惑。- 在
_get_core_constraints中,发行版名称被硬编码为"AstrBot";如果已安装包的名字发生变化(例如变成小写、加 extras 或者重命名),这一保护机制会在无声无息中失效——可以考虑从一个共享常量或者__package__/元数据中推导核心包名,以提高健壮性。
给 AI Agent 的提示
Please address the comments from this code review:
## Overall Comments
- The `PipInstaller` exposes both module-level `find_missing_requirements*` helpers and identically named instance methods that just delegate; consider dropping the instance wrappers and using the module API directly in `StarManager` to avoid redundant indirection and confusion about where the logic lives.
- In `_get_core_constraints` the distribution name is hard-coded as `"AstrBot"`; if the installed package name ever diverges (e.g., lowercase, extras, or rebranding), this protection silently disables itself—consider deriving the core package name from a single shared constant or from `__package__`/metadata to keep this robust.
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="254" />
<code_context>
return names
+def _iter_requirement_lines(
+ requirements_path: str,
+ *,
</code_context>
<issue_to_address>
**issue (complexity):** 请考虑重构 requirements 解析辅助函数,使分词只在单个迭代器中进行一次,然后下游函数基于这些 token 操作,而不是反复对行进行拆分。
你可以通过避免双重 `shlex.split` 调用并将分词逻辑集中在一起,来降低当前 requirements 解析流程新增的复杂度。
目前的情况是:
- `_iter_requirement_lines` 使用 `shlex.split(line)` 来检测嵌套的 `-r/--requirement`,但最终返回的是**原始字符串行**。
- `_iter_requirement_specs` 会**再次**调用 `shlex.split(line)`,然后可能回退到 `_extract_requirement_name(line)`,而后者本身又会重新去除注释。
你可以让 `_iter_requirement_lines` 成为分词和处理嵌套的单一真源,而 `_iter_requirement_specs` 只在 token 上工作。这样可以消除重复解析,并让整体流程更易理解。
### 建议的重构(小范围、局部修改)
将 `_iter_requirement_lines` 修改为返回 token 而不是文本行:
```python
def _iter_requirement_tokens(
requirements_path: str,
*,
_visited_paths: set[str] | None = None,
) -> Iterator[list[str]]:
visited_paths = _visited_paths or set()
resolved_path = os.path.realpath(requirements_path)
if resolved_path in visited_paths:
logger.warning(
"检测到循环依赖的 requirements 包含: %s,将跳过该文件", resolved_path
)
return
visited_paths.add(resolved_path)
with open(resolved_path, encoding="utf-8") as requirements_file:
for raw_line in requirements_file:
line = _strip_inline_requirement_comment(raw_line)
if not line or line.startswith("#"):
continue
tokens = shlex.split(line)
if not tokens:
continue
nested_path: str | None = None
if tokens[0] in {"-r", "--requirement"} and len(tokens) > 1:
nested_path = tokens[1]
elif tokens[0].startswith("--requirement="):
nested_path = tokens[0].split("=", 1)[1]
if nested_path:
if not os.path.isabs(nested_path):
nested_path = os.path.join(os.path.dirname(resolved_path), nested_path)
yield from _iter_requirement_tokens(nested_path, _visited_paths=visited_paths)
continue
yield tokens
```
然后将 `_iter_requirement_specs` 简化为在这些 token 上工作,并只在需要时重建整行字符串:
```python
def _iter_requirement_specs(
requirements_path: str,
) -> Iterator[tuple[str, SpecifierSet | None]]:
for tokens in _iter_requirement_tokens(requirements_path):
# constraints and pure options
if tokens[0] in {"-c", "--constraint"}:
continue
if tokens[0].startswith("-"):
requirement_name = _extract_requirement_name(" ".join(tokens))
if requirement_name:
yield requirement_name, None
continue
line = " ".join(tokens)
try:
requirement = Requirement(line)
except InvalidRequirement:
requirement_name = _extract_requirement_name(line)
if requirement_name:
yield requirement_name, None
continue
if requirement.marker and not requirement.marker.evaluate():
continue
yield (
_canonicalize_distribution_name(requirement.name),
requirement.specifier,
)
```
通过这一改动:
- 分词(`shlex.split`)和嵌套 `-r` 的处理被集中在同一个地方。
- `_iter_requirement_specs` 在 token 上做一次线性遍历,并且拥有清晰的回退路径。
- `_extract_requirement_name` 仍然可以用于处理异常的行,但你避免了对同一字符串重复去注释和重复解析。
这样可以保持当前所有功能(嵌套 requirements、选项、markers、specifiers),同时减少另一位审查者指出的模块间耦合和重复解析问题。
</issue_to_address>帮我变得更有用!请对每条评论点 👍 或 👎,我会根据反馈改进后续的审查。
Original comment in English
Hey - I've found 1 issue, and left some high level feedback:
- The
PipInstallerexposes both module-levelfind_missing_requirements*helpers and identically named instance methods that just delegate; consider dropping the instance wrappers and using the module API directly inStarManagerto avoid redundant indirection and confusion about where the logic lives. - In
_get_core_constraintsthe distribution name is hard-coded as"AstrBot"; if the installed package name ever diverges (e.g., lowercase, extras, or rebranding), this protection silently disables itself—consider deriving the core package name from a single shared constant or from__package__/metadata to keep this robust.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The `PipInstaller` exposes both module-level `find_missing_requirements*` helpers and identically named instance methods that just delegate; consider dropping the instance wrappers and using the module API directly in `StarManager` to avoid redundant indirection and confusion about where the logic lives.
- In `_get_core_constraints` the distribution name is hard-coded as `"AstrBot"`; if the installed package name ever diverges (e.g., lowercase, extras, or rebranding), this protection silently disables itself—consider deriving the core package name from a single shared constant or from `__package__`/metadata to keep this robust.
## Individual Comments
### Comment 1
<location path="astrbot/core/utils/pip_installer.py" line_range="254" />
<code_context>
return names
+def _iter_requirement_lines(
+ requirements_path: str,
+ *,
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the requirements parsing helpers so that tokenization happens once in a single iterator and downstream functions operate on those tokens instead of repeatedly re-splitting lines.
You can reduce some of the new complexity in the requirements parsing pipeline by avoiding the double `shlex.split` pass and centralizing tokenization.
Right now:
- `_iter_requirement_lines` does `shlex.split(line)` to detect nested `-r/--requirement`, but ultimately yields the **string line**.
- `_iter_requirement_specs` calls `shlex.split(line)` **again** and may then fall back to `_extract_requirement_name(line)`, which itself re-strips comments.
You can make `_iter_requirement_lines` the single source of truth for tokenization and nesting, and let `_iter_requirement_specs` work purely on tokens. That removes the repeated parsing and makes the flow easier to follow.
### Suggested refactor (small and localized)
Change `_iter_requirement_lines` to yield tokens instead of lines:
```python
def _iter_requirement_tokens(
requirements_path: str,
*,
_visited_paths: set[str] | None = None,
) -> Iterator[list[str]]:
visited_paths = _visited_paths or set()
resolved_path = os.path.realpath(requirements_path)
if resolved_path in visited_paths:
logger.warning(
"检测到循环依赖的 requirements 包含: %s,将跳过该文件", resolved_path
)
return
visited_paths.add(resolved_path)
with open(resolved_path, encoding="utf-8") as requirements_file:
for raw_line in requirements_file:
line = _strip_inline_requirement_comment(raw_line)
if not line or line.startswith("#"):
continue
tokens = shlex.split(line)
if not tokens:
continue
nested_path: str | None = None
if tokens[0] in {"-r", "--requirement"} and len(tokens) > 1:
nested_path = tokens[1]
elif tokens[0].startswith("--requirement="):
nested_path = tokens[0].split("=", 1)[1]
if nested_path:
if not os.path.isabs(nested_path):
nested_path = os.path.join(os.path.dirname(resolved_path), nested_path)
yield from _iter_requirement_tokens(nested_path, _visited_paths=visited_paths)
continue
yield tokens
```
Then simplify `_iter_requirement_specs` to operate on those tokens and only reconstruct the line once when needed:
```python
def _iter_requirement_specs(
requirements_path: str,
) -> Iterator[tuple[str, SpecifierSet | None]]:
for tokens in _iter_requirement_tokens(requirements_path):
# constraints and pure options
if tokens[0] in {"-c", "--constraint"}:
continue
if tokens[0].startswith("-"):
requirement_name = _extract_requirement_name(" ".join(tokens))
if requirement_name:
yield requirement_name, None
continue
line = " ".join(tokens)
try:
requirement = Requirement(line)
except InvalidRequirement:
requirement_name = _extract_requirement_name(line)
if requirement_name:
yield requirement_name, None
continue
if requirement.marker and not requirement.marker.evaluate():
continue
yield (
_canonicalize_distribution_name(requirement.name),
requirement.specifier,
)
```
With this change:
- Tokenization (`shlex.split`) and nested `-r` handling live in one place.
- `_iter_requirement_specs` has a single linear pass over tokens, with a clear fallback path.
- `_extract_requirement_name` is still available for strange lines, but you avoid re-stripping / re-parsing the same string multiple times.
This keeps all current functionality (nested requirements, options, markers, specifiers) while reducing the cross‑coupling and repeated parsing that the other reviewer flagged.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
|
@sourcery-ai review |
There was a problem hiding this comment.
Hey - 我发现了两个问题,并给出了一些总体反馈:
_StreamingLogWriter.write()目前会丢弃仅包含空白字符的行(通过line.strip()),这可能会隐藏有意义的 pip 输出(例如缩进或特殊格式的行);建议保留这些行,或者至少只跳过真正的空字符串。_get_core_constraints()会扫描所有已安装的 distribution,并对每一个调用read_text('top_level.txt'),在大型环境中这会非常耗时;你可以先直接尝试默认的'AstrBot'distribution,如果该查找失败,再回退到扫描top_level.txt的逻辑。
给 AI Agents 的提示
请根据以下代码审查意见进行修改:
## 总体意见
- `_StreamingLogWriter.write()` 目前会丢弃仅包含空白字符的行(通过 `line.strip()`),这可能会隐藏有意义的 pip 输出(例如缩进或特殊格式的行);建议保留这些行,或者至少只跳过真正的空字符串。
- `_get_core_constraints()` 会扫描所有已安装的 distribution,并对每一个调用 `read_text('top_level.txt')`,在大型环境中这会非常耗时;你可以先直接尝试默认的 `'AstrBot'` distribution,如果该查找失败,再回退到扫描 `top_level.txt` 的逻辑。
## 单独评论
### 评论 1
<location path="astrbot/core/star/star_manager.py" line_range="211-206" />
<code_context>
+ async def _ensure_plugin_requirements(
</code_context>
<issue_to_address>
**issue (bug_risk):** 吞掉 `DependencyConflictError` 和通用异常可能会让调用方看不到严重错误。
由于 `_ensure_plugin_requirements` 只是在记录日志后就吞掉 `DependencyConflictError` 和其他异常并直接返回,像 `install_plugin`、`update_plugin` 和 `reload_failed_plugin` 这样的调用方会把依赖检查视为成功。这可能导致插件在依赖冲突未解决的情况下被加载,并在之后以晦涩的导入错误形式失败。建议要么重新抛出一个具体的错误(或包一层后再抛出),让调用方可以中止并展示清晰的错误信息;要么返回一个明确的成功/失败结果,要求调用方在继续之前必须检查该结果。
</issue_to_address>
### 评论 2
<location path="astrbot/core/utils/pip_installer.py" line_range="254" />
<code_context>
return names
+def _iter_requirement_tokens(
+ requirements_path: str,
+ *,
</code_context>
<issue_to_address>
**issue (complexity):** 建议通过统一为单个迭代器来解析 requirements,并让核心 distribution 名称显式且可注入,从而简化 requirements 处理和 core-constraints 的发现逻辑。
你可以在不损失现有行为的前提下,降低新增的复杂度:
---
### 1. 合并 requirements 解析流程
目前,在预检查和名称提取时,处理流程是:
- `_extract_requirement_names` → `_iter_requirement_specs` → `_iter_requirement_tokens` → `_strip_inline_requirement_comment`
- `_find_missing_requirements` → `_iter_requirement_specs` → `_iter_requirement_tokens` → `_strip_inline_requirement_comment`
你可以把它们折叠为一个迭代器,该迭代器:
- 读取文件(支持 `-r` 递归和循环检测),
- 去除行内注释,
- 使用 `Requirement` 解析,
- 直接产出 `(name, specifier | None)`。
这样 `_extract_requirement_names` 和 `_find_missing_requirements` 基本都可以简化成一两行代码,并且可以删除 `_iter_requirement_tokens`、`_iter_requirement_specs`、`_strip_inline_requirement_comment`(针对文件的版本),以及多个位置上的正则回退逻辑。
例如:
```python
def _iter_requirements(
requirements_path: str,
_visited: set[str] | None = None,
) -> Iterator[tuple[str, SpecifierSet | None]]:
visited = _visited or set()
resolved_path = os.path.realpath(requirements_path)
if resolved_path in visited:
logger.warning("检测到循环依赖的 requirements 包含: %s,将跳过该文件", resolved_path)
return
visited.add(resolved_path)
with open(resolved_path, encoding="utf-8") as f:
for raw_line in f:
# 统一:文件里的行统一在这里做注释和空行过滤
line = re.split(r"[ \t]+#", raw_line, maxsplit=1)[0].strip()
if not line or line.startswith("#"):
continue
tokens = shlex.split(line)
if not tokens:
continue
# 处理嵌套 -r
nested: str | None = None
if tokens[0] in {"-r", "--requirement"} and len(tokens) > 1:
nested = tokens[1]
elif tokens[0].startswith("--requirement="):
nested = tokens[0].split("=", 1)[1]
if nested:
if not os.path.isabs(nested):
nested = os.path.join(os.path.dirname(resolved_path), nested)
yield from _iter_requirements(nested, _visited=visited)
continue
# 纯选项:跳过
if tokens[0] in {"-c", "--constraint"}:
continue
if tokens[0].startswith("-"):
# 选项形式但依然可能带有包名,比如 -e git+...#egg=xxx
name = _extract_requirement_name(line)
if name:
yield name, None
continue
try:
req = Requirement(line)
if req.marker and not req.marker.evaluate():
continue
yield _canonicalize_distribution_name(req.name), req.specifier or None
except InvalidRequirement:
name = _extract_requirement_name(line)
if name:
yield name, None
```
然后:
```python
def _extract_requirement_names(requirements_path: str) -> set[str]:
try:
return {name for name, _ in _iter_requirements(requirements_path)}
except Exception as exc:
logger.warning("读取依赖文件失败,跳过冲突检测: %s", exc)
return set()
```
`_find_missing_requirements` 也会变得更线性:
```python
def _find_missing_requirements(requirements_path: str) -> set[str] | None:
try:
required = list(_iter_requirements(requirements_path))
except Exception as exc:
logger.warning("预检查缺失依赖失败,将回退到完整安装: %s", exc)
return None
if not required:
return set()
installed = _collect_installed_distribution_versions(_get_requirement_check_paths())
if installed is None:
return None
missing: set[str] = set()
for name, specifier in required:
installed_version = installed.get(name)
if not installed_version:
missing.add(name)
continue
if specifier and not _specifier_contains_version(specifier, installed_version):
missing.add(name)
return missing
```
这样可以保留当前的所有行为(递归 `-r`、markers、egg 片段等),但去掉多步骤的 token/spec 处理管线,让核心的“requirements → (name, spec)” 操作更加直观。
---
### 2. 让核心约束解析显式且可注入
当前 `_get_core_constraints`:
- 通过扫描 `top_level.txt` 来“猜测” AstrBot 的 distribution,
- 然后再解析 requirements 和已安装版本。
这种“猜测”逻辑增加了环境复杂度,也不利于测试。你可以保留现有行为,但让“核心 distribution 名称”变得显式且可覆盖,同时保留当前的启发式作为兜底方案。
示例:
```python
class PipInstaller:
def __init__(
self,
pip_install_arg: str,
pypi_index_url: str | None = None,
core_dist_name: str | None = "AstrBot",
) -> None:
self.pip_install_arg = pip_install_arg
self.pypi_index_url = pypi_index_url
self.core_dist_name = core_dist_name
```
然后将其传入 `_core_constraints_file` / `_get_core_constraints`:
```python
@contextlib.contextmanager
def _core_constraints_file(core_dist_name: str | None) -> Iterator[str | None]:
constraints = _get_core_constraints(core_dist_name)
...
```
```python
def _get_core_constraints(core_dist_name: str | None) -> list[str]:
constraints: list[str] = []
try:
if core_dist_name is None:
core_dist_name = "AstrBot"
try:
if __package__:
top_pkg = __package__.split(".")[0]
for dist in importlib_metadata.distributions():
if top_pkg in (dist.read_text("top_level.txt") or "").splitlines():
core_dist_name = dist.metadata["Name"]
break
except Exception:
pass
try:
dist = importlib_metadata.distribution(core_dist_name)
except importlib_metadata.PackageNotFoundError:
return []
if not dist.requires:
return []
installed = _collect_installed_distribution_versions(
_get_requirement_check_paths()
)
if not installed:
return []
for req_str in dist.requires:
try:
req = Requirement(req_str)
if req.marker and not req.marker.evaluate():
continue
name = _canonicalize_distribution_name(req.name)
if name in installed:
constraints.append(f"{name}=={installed[name]}")
except Exception:
continue
except Exception as exc:
logger.warning("获取核心依赖约束失败: %s", exc)
return constraints
```
在 `install` 里:
```python
with _core_constraints_file(self.core_dist_name) as constraints_file_path:
...
```
这样可以:
- 保留当前的动态发现逻辑作为回退,
- 让核心 distribution 显式可测,
- 允许部署环境配置核心 distribution 名称,从而避免扫描所有 distributions。
</issue_to_address>帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进后续的 review。
Original comment in English
Hey - I've found 2 issues, and left some high level feedback:
- _StreamingLogWriter.write() currently discards lines that are only whitespace (via line.strip()), which can hide meaningful pip output (e.g., indented or formatted lines); consider preserving such lines or at least only skipping truly empty strings.
- _get_core_constraints() scans all installed distributions and calls read_text('top_level.txt') on each, which can be expensive in large environments; you could first try the default 'AstrBot' distribution directly and only fall back to the top_level.txt scan if that lookup fails.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- _StreamingLogWriter.write() currently discards lines that are only whitespace (via line.strip()), which can hide meaningful pip output (e.g., indented or formatted lines); consider preserving such lines or at least only skipping truly empty strings.
- _get_core_constraints() scans all installed distributions and calls read_text('top_level.txt') on each, which can be expensive in large environments; you could first try the default 'AstrBot' distribution directly and only fall back to the top_level.txt scan if that lookup fails.
## Individual Comments
### Comment 1
<location path="astrbot/core/star/star_manager.py" line_range="211-206" />
<code_context>
+ async def _ensure_plugin_requirements(
</code_context>
<issue_to_address>
**issue (bug_risk):** Swallowing `DependencyConflictError` and generic exceptions may hide hard failures from callers.
Because `_ensure_plugin_requirements` only logs `DependencyConflictError` and other exceptions and then returns, callers like `install_plugin`, `update_plugin`, and `reload_failed_plugin` will treat dependency checks as successful. This can lead to plugins loading with unresolved dependency conflicts and failing later with opaque import errors. Consider either re-raising a specific error (or wrapping it) so callers can abort and show a clear message, or returning an explicit success/failure result that callers must check before proceeding.
</issue_to_address>
### Comment 2
<location path="astrbot/core/utils/pip_installer.py" line_range="254" />
<code_context>
return names
+def _iter_requirement_tokens(
+ requirements_path: str,
+ *,
</code_context>
<issue_to_address>
**issue (complexity):** Consider simplifying the requirements handling and core-constraints discovery by unifying parsing into a single iterator and making the core distribution name explicit and injectable.
You can reduce the new complexity without losing any behavior by:
---
### 1. Collapse the requirements parsing pipeline
Right now, for pre-checks and name extraction, you go through:
- `_extract_requirement_names` → `_iter_requirement_specs` → `_iter_requirement_tokens` → `_strip_inline_requirement_comment`
- `_find_missing_requirements` → `_iter_requirement_specs` → `_iter_requirement_tokens` → `_strip_inline_requirement_comment`
You can collapse this into a single iterator that:
- Reads a file (with `-r` recursion and cycle detection),
- Strips inline comments,
- Parses with `Requirement`,
- Yields `(name, specifier | None)` directly.
That makes both `_extract_requirement_names` and `_find_missing_requirements` essentially one-liners, and allows you to delete `_iter_requirement_tokens`, `_iter_requirement_specs`, `_strip_inline_requirement_comment` (for files), and the regex fallback in multiple places.
For example:
```python
def _iter_requirements(
requirements_path: str,
_visited: set[str] | None = None,
) -> Iterator[tuple[str, SpecifierSet | None]]:
visited = _visited or set()
resolved_path = os.path.realpath(requirements_path)
if resolved_path in visited:
logger.warning("检测到循环依赖的 requirements 包含: %s,将跳过该文件", resolved_path)
return
visited.add(resolved_path)
with open(resolved_path, encoding="utf-8") as f:
for raw_line in f:
# 统一:文件里的行统一在这里做注释和空行过滤
line = re.split(r"[ \t]+#", raw_line, maxsplit=1)[0].strip()
if not line or line.startswith("#"):
continue
tokens = shlex.split(line)
if not tokens:
continue
# 处理嵌套 -r
nested: str | None = None
if tokens[0] in {"-r", "--requirement"} and len(tokens) > 1:
nested = tokens[1]
elif tokens[0].startswith("--requirement="):
nested = tokens[0].split("=", 1)[1]
if nested:
if not os.path.isabs(nested):
nested = os.path.join(os.path.dirname(resolved_path), nested)
yield from _iter_requirements(nested, _visited=visited)
continue
# 纯选项:跳过
if tokens[0] in {"-c", "--constraint"}:
continue
if tokens[0].startswith("-"):
# 选项形式但依然可能带有包名,比如 -e git+...#egg=xxx
name = _extract_requirement_name(line)
if name:
yield name, None
continue
try:
req = Requirement(line)
if req.marker and not req.marker.evaluate():
continue
yield _canonicalize_distribution_name(req.name), req.specifier or None
except InvalidRequirement:
name = _extract_requirement_name(line)
if name:
yield name, None
```
Then:
```python
def _extract_requirement_names(requirements_path: str) -> set[str]:
try:
return {name for name, _ in _iter_requirements(requirements_path)}
except Exception as exc:
logger.warning("读取依赖文件失败,跳过冲突检测: %s", exc)
return set()
```
and `_find_missing_requirements` becomes much more linear:
```python
def _find_missing_requirements(requirements_path: str) -> set[str] | None:
try:
required = list(_iter_requirements(requirements_path))
except Exception as exc:
logger.warning("预检查缺失依赖失败,将回退到完整安装: %s", exc)
return None
if not required:
return set()
installed = _collect_installed_distribution_versions(_get_requirement_check_paths())
if installed is None:
return None
missing: set[str] = set()
for name, specifier in required:
installed_version = installed.get(name)
if not installed_version:
missing.add(name)
continue
if specifier and not _specifier_contains_version(specifier, installed_version):
missing.add(name)
return missing
```
This keeps all current behavior (recursive `-r`, markers, egg fragments, etc.) but removes the multi-step token/spec pipeline and makes the core “requirements → (name, spec)” operation straightforward.
---
### 2. Make core constraints resolution explicit and injectable
`_get_core_constraints` currently:
- Guess-detects the AstrBot distribution via `top_level.txt` scanning,
- Then resolves requirements and installed versions.
That guessing logic adds ambient complexity and is hard to test. You can keep the behavior but make the “core distribution name” explicit and overrideable, while keeping the current heuristic as a fallback.
Example:
```python
class PipInstaller:
def __init__(
self,
pip_install_arg: str,
pypi_index_url: str | None = None,
core_dist_name: str | None = "AstrBot",
) -> None:
self.pip_install_arg = pip_install_arg
self.pypi_index_url = pypi_index_url
self.core_dist_name = core_dist_name
```
Then pass it into `_core_constraints_file` / `_get_core_constraints`:
```python
@contextlib.contextmanager
def _core_constraints_file(core_dist_name: str | None) -> Iterator[str | None]:
constraints = _get_core_constraints(core_dist_name)
...
```
```python
def _get_core_constraints(core_dist_name: str | None) -> list[str]:
constraints: list[str] = []
try:
if core_dist_name is None:
core_dist_name = "AstrBot"
try:
if __package__:
top_pkg = __package__.split(".")[0]
for dist in importlib_metadata.distributions():
if top_pkg in (dist.read_text("top_level.txt") or "").splitlines():
core_dist_name = dist.metadata["Name"]
break
except Exception:
pass
try:
dist = importlib_metadata.distribution(core_dist_name)
except importlib_metadata.PackageNotFoundError:
return []
if not dist.requires:
return []
installed = _collect_installed_distribution_versions(
_get_requirement_check_paths()
)
if not installed:
return []
for req_str in dist.requires:
try:
req = Requirement(req_str)
if req.marker and not req.marker.evaluate():
continue
name = _canonicalize_distribution_name(req.name)
if name in installed:
constraints.append(f"{name}=={installed[name]}")
except Exception:
continue
except Exception as exc:
logger.warning("获取核心依赖约束失败: %s", exc)
return constraints
```
And in `install`:
```python
with _core_constraints_file(self.core_dist_name) as constraints_file_path:
...
```
This:
- Keeps the current dynamic discovery as a fallback,
- Makes the core dist explicit and testable,
- Allows deployments to configure the core dist name and avoid scanning all distributions.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
…and improve exception propagation
核心目标
解决插件安装/更新/上传流程中,依赖安装滞后于首次加载尝试的问题。同时优化 Dashboard 手动 pip 安装的输入解析能力,并新增核心依赖防降级保护机制与精准冲突告警。
主要变更
1. 核心依赖防降级保护与精准告警
pip install前,自动提取核心依赖。支持快速路径(直接查找 "AstrBot")与动态包名发现回退机制。-c constraints.txt) 传递给pip,确保插件安装过程中不会意外降级 AstrBot 的核心库。pip输出中自动提取具体的冲突版本信息,并直接展示在错误消息中。2. 代码架构重构与逻辑精简 (Final Refined)
_iter_requirements迭代器,将文件读取、注释剥离、Token 拆分和Requirement解析整合为单一高效管线,消除了重复分词并修复了 marker 引号丢失问题。StarManager现在会捕获并正确重新抛出依赖安装异常。这意味着依赖冲突将硬性阻止受损插件的加载,避免了运行时的不可预知错误。PipInstaller的冗余包装方法,对外直接暴露模块级 API,降低了调用方 (StarManager) 的认知负担。strip()丢弃空白行,从而完整保留了pip输出的缩进和原始格式。3. 安全扫描与测试增强
PluginManager构造函数签名、异常透传行为以及日志缓冲区合并逻辑。验证与测试
ruff格式化与 Lint 检查。关联工作
基于
docs/plans/2026-03-09-plugin-dependency-install-design.md完成实现。