X Tutup
Skip to content

fix: 插件依赖自动安装逻辑与 Dashboard 安装体验优化#5954

Open
zouyonghe wants to merge 23 commits intoAstrBotDevs:masterfrom
zouyonghe:fix/plugin-deps-pr
Open

fix: 插件依赖自动安装逻辑与 Dashboard 安装体验优化#5954
zouyonghe wants to merge 23 commits intoAstrBotDevs:masterfrom
zouyonghe:fix/plugin-deps-pr

Conversation

@zouyonghe
Copy link
Member

@zouyonghe zouyonghe commented Mar 9, 2026

核心目标

解决插件安装/更新/上传流程中,依赖安装滞后于首次加载尝试的问题。同时优化 Dashboard 手动 pip 安装的输入解析能力,并新增核心依赖防降级保护机制与精准冲突告警

主要变更

1. 核心依赖防降级保护与精准告警

  • 动态约束生成:在执行 pip install 前,自动提取核心依赖。支持快速路径(直接查找 "AstrBot")与动态包名发现回退机制。
  • 强制版本锁定:通过临时生成的约束文件 (-c constraints.txt) 传递给 pip,确保插件安装过程中不会意外降级 AstrBot 的核心库。
  • 冲突细节展示:当检测到依赖冲突时,系统会从 pip 输出中自动提取具体的冲突版本信息,并直接展示在错误消息中。

2. 代码架构重构与逻辑精简 (Final Refined)

  • 统一解析迭代器 (New):引入 _iter_requirements 迭代器,将文件读取、注释剥离、Token 拆分和 Requirement 解析整合为单一高效管线,消除了重复分词并修复了 marker 引号丢失问题。
  • 异常驱动生命周期 (New)StarManager 现在会捕获并正确重新抛出依赖安装异常。这意味着依赖冲突将硬性阻止受损插件的加载,避免了运行时的不可预知错误。
  • 极简 API 接口:移除了 PipInstaller 的冗余包装方法,对外直接暴露模块级 API,降低了调用方 (StarManager) 的认知负担。
  • 完善的日志格式:优化了流式日志记录器,不再通过 strip() 丢弃空白行,从而完整保留了 pip 输出的缩进和原始格式。

3. 安全扫描与测试增强

  • 消除 CodeQL 警报:将测试代码中潜在的 URL 子字符串匹配逻辑改为精确的序列断言。
  • 测试套件同步 (Updated):更新了所有测试用例以适配最新的 PluginManager 构造函数签名、异常透传行为以及日志缓冲区合并逻辑。

验证与测试

  • 单元测试 (Pytest):41 个测试用例全部通过,验证了性能优化后的 Token 解析流、防降级约束及所有新特性。
  • 代码规范:所有代码均通过 ruff 格式化与 Lint 检查。

关联工作

基于 docs/plans/2026-03-09-plugin-dependency-install-design.md 完成实现。

@auto-assign auto-assign bot requested review from Soulter and anka-afk March 9, 2026 13:46
@dosubot dosubot bot added the size:L This PR changes 100-499 lines, ignoring generated files. label Mar 9, 2026
@gemini-code-assist
Copy link
Contributor

Summary of Changes

Hello, 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

  • Plugin Dependency Installation: Plugin requirements.txt files are now installed before critical first-load flows, including repository installs, zip installs, failed-plugin reloads, and plugin updates, ensuring dependencies are met prior to plugin activation.
  • Improved Pip Input Parsing: The system now robustly accepts multiline or whitespace-separated pip input from the dashboard, correctly handling valid single requirement specifiers, comments, and option lines without errors.
  • Enhanced Test Coverage: New regression tests have been added to cover plugin dependency installation order and various edge cases in pip input parsing, improving the reliability of the plugin management system.
Changelog
  • astrbot/core/star/star_manager.py
    • Extracted plugin requirement installation logic into a new private method _install_plugin_requirements for better modularity.
    • Integrated _install_plugin_requirements into _check_plugin_dept_update to ensure dependencies are updated when plugins are updated.
    • Added calls to _install_plugin_requirements in reload_failed_plugin, install_plugin, update_plugin, and install_plugin_from_file to ensure dependencies are installed before plugin loading or reloading in these critical paths.
  • astrbot/core/utils/pip_installer.py
    • Imported shlex, InvalidRequirement, and Requirement for advanced package parsing.
    • Introduced new utility functions (_split_package_install_input, _is_valid_install_requirement, _is_valid_install_token, _split_multiline_package_input) to handle diverse pip input formats, including multiline and space-separated requirements.
    • Modified the install method to utilize the new input parsing logic, allowing for more flexible package name inputs.
    • Updated the logging format for pip commands to improve clarity.
  • tests/test_pip_installer.py
    • Added new test cases to verify the install method correctly splits space-separated package names.
    • Included tests to ensure multiline package inputs are tracked and parsed correctly for desktop clients.
    • Added tests to confirm that single requirements with markers or version ranges are kept intact during parsing.
    • Implemented a test to validate that multiline input correctly strips comments and splits options.
  • tests/test_plugin_manager.py
    • Added a helper function _write_requirements to facilitate testing of plugin dependency installation.
    • Added assertions for plugin_name being non-None in test_update_plugin and test_uninstall_plugin.
    • Introduced new test cases to confirm that plugin requirements are installed before loading after initial installation, installation from file, and when reloading failed plugins.
    • Added a test to verify that plugin requirements are installed after an update but before the plugin is reloaded.
Activity
  • No human activity has been recorded on this pull request yet.
Using Gemini Code Assist

The 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 /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature Command Description
Code Review /gemini review Performs a code review for the current pull request in its current state.
Pull Request Summary /gemini summary Provides a summary of the current pull request in its current state.
Comment @gemini-code-assist Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help /gemini help Displays a list of available commands.

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 .gemini/ folder in the base of the repository. Detailed instructions can be found here.

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

  1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution.

@dosubot dosubot bot added the feature:plugin The bug / feature is about AstrBot plugin system. label Mar 9, 2026
Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目免费——如果你觉得这些评审有帮助,欢迎分享 ✨
帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进评审质量。
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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

Copy link
Contributor

@gemini-code-assist gemini-code-assist bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Comment on lines +596 to +597
package_specs = _split_package_install_input(package_name)
args.extend(package_specs)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

security-high high

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.

args.extend(self.pip_install_arg.split())

logger.info(f"Pip 包管理器: pip {' '.join(args)}")
logger.info("Pip 包管理器 argv: %s", ["pip", *args])
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

security-medium medium

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.

Comment on lines +507 to +508
plugin_path = os.path.join(self.plugin_store_path, dir_name)
await self._install_plugin_requirements(plugin_path, dir_name)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

reload_failed_plugin 方法的返回类型协定是 (success, error_message) 元组,但对 _install_plugin_requirements 的调用缺少异常处理。如果依赖安装失败导致 _install_plugin_requirements 抛出异常,整个方法会直接崩溃,而不是按预期返回 (False, error_message)。这破坏了该方法的健壮性。建议添加 try...except 块来捕获此异常,并以约定的方式返回错误信息。

Suggested change
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

@dosubot dosubot bot added size:XL This PR changes 500-999 lines, ignoring generated files. and removed size:L This PR changes 100-499 lines, ignoring generated files. labels Mar 9, 2026
@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Original comment in English

Hey - I've found 2 issues, and left some high level feedback:

  • 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.
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 ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

Comment on lines +272 to +281
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()
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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_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.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目是免费的——如果你觉得我们的评审有帮助,请考虑分享 ✨
帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据这些反馈改进后续的评审。
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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hey - 我发现了 1 个问题,并给出了一些高层反馈:

  • _strip_inline_requirement_comment 中,按 [ \t]+# 进行 split 会把 URL 中合法的 # 片段也去掉(例如:--extra-index-url https://example.com/simple#frag);建议将注释剥离限制在“不是选项”的行上,或者使用更精确的启发式规则。
  • 新的 _install_plugin_requirements helper 在部分调用路径(_check_plugin_dept_updatereload_failed_plugin)中被 try/except 包裹,而在其他路径(install_plugininstall_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>

Sourcery 对开源项目免费——如果你觉得我们的 review 有帮助,欢迎分享 ✨
帮我变得更有用!请在每条评论上点 👍 或 👎,我会用这些反馈来改进后续的 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_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.
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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目免费使用——如果你觉得我们的评审有帮助,欢迎分享给更多人 ✨
帮我变得更有用!请对每条评论点 👍 或 👎,我会根据你的反馈改进后续评审。
Original comment in English

Hey - I've found 2 issues, and left some high level feedback:

  • 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.
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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目免费 —— 如果你觉得这次评审有帮助,欢迎分享 ✨
帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进后续评审。
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 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.
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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hey - 我已经留下了一些整体性的反馈:

  • 在插件流程(install_plugininstall_plugin_from_filereload_failed_pluginupdate_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.

Sourcery 对开源项目免费——如果你喜欢我们的代码评审,请考虑分享给他人 ✨
帮我变得更有用!请在每条评论上点击 👍 或 👎,我会根据你的反馈改进后续的评审。
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_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.
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.

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目是免费的——如果你觉得我们的评审有帮助,请考虑分享 ✨
帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进后续评审。
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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目是免费的——如果你觉得我们的评审有帮助,欢迎帮忙分享 ✨
帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进后续的评审。
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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

Comment on lines +648 to +657
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]
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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_args

This will ensure _package_specs_override_index truly prevents adding the default index configuration in this scenario.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目是免费的 —— 如果你觉得我们的 Review 有帮助,欢迎分享 ✨
帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据这些反馈改进之后的 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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

Copy link

@SourceryAI SourceryAI left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目是免费的——如果你觉得这次评审有帮助,欢迎分享 ✨
帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进后续评审。
Original comment in English

Hey - I've found 1 issue, and left some high level feedback:

  • 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.
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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目是免费的——如果你喜欢我们的评审,请考虑分享 ✨
帮我变得更有用!请在每条评论上点击 👍 或 👎,我会根据反馈改进你的代码审查体验。
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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目免费——如果你觉得我们的审查有帮助,欢迎分享 ✨
帮我变得更有用!请对每条评论点 👍 或 👎,我会根据反馈改进后续的审查。
Original comment in English

Hey - I've found 1 issue, and left some high level feedback:

  • 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.
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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

@zouyonghe
Copy link
Member Author

@sourcery-ai review

Copy link
Contributor

@sourcery-ai sourcery-ai bot left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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>

Sourcery 对开源项目免费——如果你觉得这些 review 有帮助,欢迎分享 ✨
帮我变得更有用!请在每条评论上点 👍 或 👎,我会根据你的反馈改进后续的 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>

Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

feature:plugin The bug / feature is about AstrBot plugin system. size:XXL This PR changes 1000+ lines, ignoring generated files.

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants

X Tutup