统一路径类，旧函数已弃用 #3279
Closed
统一路径类，旧函数已弃用 #3279
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
.env文件保存统一的路径 静态文件一起打包而不是从上上上级目录找... Co-Authored-By: 赵天乐(tyler zhao) <[email protected]>
![sourcery-ai[bot]](https://avatars.githubusercontent.com/in/48477?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Hey there - I've reviewed your changes and they look great!
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location> `astrbot/dashboard/routes/config.py:464-466` </location>
<code_context>
- get_astrbot_path(),
- "samples",
- "stt_health_check.wav",
+ sample_audio_path = str(
+ importlib.resources.files("astrbot")
+ / "samples"
+ / "stt_health_check.wav"
)
</code_context>
<issue_to_address>
**issue (bug_risk):** Switching to importlib.resources for file access may break if the resource is not packaged.
Confirm that 'samples/stt_health_check.wav' is packaged and accessible via importlib.resources, or add error handling for missing resources.
</issue_to_address>
### Comment 2
<location> `astrbot/base/paths.py:34-40` </location>
<code_context>
+ @classmethod
+ def getPaths(cls, name: str) -> AstrbotPaths:
+ """返回Paths实例,用于访问模块的各类目录."""
+ normalized_name: NormalizedName = canonicalize_name(name)
+ instance: AstrbotPaths = cls(normalized_name)
+ instance.name = normalized_name
</code_context>
<issue_to_address>
**suggestion (bug_risk):** Using canonicalize_name for directory names may introduce unexpected normalization.
Verify that canonicalize_name does not cause naming conflicts or break compatibility with existing directory naming conventions used by plugins or modules.
```suggestion
@classmethod
def getPaths(cls, name: str) -> AstrbotPaths:
"""返回Paths实例,用于访问模块的各类目录."""
instance: AstrbotPaths = cls(name)
instance.name = name
return instance
```
</issue_to_address>
### Comment 3
<location> `astrbot/base/paths.py:45-46` </location>
<code_context>
+ @property
+ def root(self) -> Path:
+ """返回根目录."""
+ return (
+ self.astrbot_root if self.astrbot_root.exists() else Path.cwd() / ".astrbot"
+ )
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Fallback to Path.cwd()/.astrbot may not be expected.
Switching to the working directory if astrbot_root is missing may cause unexpected results. Instead, raise an error or notify the user to ensure clarity.
</issue_to_address>
### Comment 4
<location> `astrbot/base/paths.py:21` </location>
<code_context>
+ from collections.abc import AsyncGenerator, Generator
+
+
+class AstrbotPaths(IAstrbotPaths):
+ """Class to manage and provide paths used by Astrbot Canary."""
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the class to use helper functions and unified logic for path properties, directory changes, and environment reloads to reduce boilerplate and nested abstractions.
```markdown
You can significantly cut down boilerplate and nested abstractions by
1. removing `getPaths` and normalizing the name in `__init__`
2. generating all “mkdir + return Path” properties from one helper
3. unifying sync/async `chdir` logic
4. moving the env-reload logic into a small helper
For example:
```python
from contextlib import contextmanager, asynccontextmanager
from pathlib import Path
from os import chdir, getenv
from dotenv import load_dotenv
from packaging.utils import canonicalize_name
def _path_property(subdir: str):
@property
def prop(self) -> Path:
p = self.astrbot_root / subdir / self.name
p.mkdir(parents=True, exist_ok=True)
return p
return prop
class AstrbotPaths:
home = _path_property("home")
config = _path_property("config")
data = _path_property("data")
log = _path_property("logs")
def __init__(self, name: str):
# normalize once
self.name = canonicalize_name(name)
load_dotenv() # ensure env is loaded
self.astrbot_root = Path(
getenv("ASTRBOT_ROOT", Path.home() / ".astrbot")
).absolute()
self.astrbot_root.mkdir(parents=True, exist_ok=True)
@contextmanager
def chdir(self, key: str = "home"):
original = Path.cwd()
target = getattr(self, key)
chdir(target)
try:
yield target
finally:
chdir(original)
@asynccontextmanager
async def achdir(self, key: str = "home"):
# reuse the sync chdir under the hood
with self.chdir(key) as target:
yield target
def reload_env(self) -> None:
load_dotenv()
self.astrbot_root = Path(
getenv("ASTRBOT_ROOT", Path.home() / ".astrbot")
).absolute()
```
Key benefits:
- **No more** duplicated `mkdir` blocks: one helper covers all subdirs.
- **Drop** `getPaths` and duplicate `.name` sets; `__init__` does normalization.
- **Single** sync `chdir`, async version just wraps it.
- **Env-reload** is now a focused `reload_env()` method.
- **All** original behavior remains intact.
```
</issue_to_address>
### Comment 5
<location> `astrbot/base/paths.py:89-91` </location>
<code_context>
@classmethod
def is_root(cls, path: Path) -> bool:
"""检查路径是否为 Astrbot 根目录."""
if not path.exists() or not path.is_dir():
return False
# 检查此目录内是是否包含.astrbot标记文件
if not (path / ".astrbot").exists():
return False
return True
</code_context>
<issue_to_address>
**suggestion (code-quality):** We've found these issues:
- Lift code into else after jump in control flow ([`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/))
- Replace if statement with if expression ([`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/))
- Simplify boolean if expression ([`boolean-if-exp-identity`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/boolean-if-exp-identity/))
```suggestion
return bool((path / ".astrbot").exists())
```
</issue_to_address>
### Comment 6
<location> `astrbot/core/pipeline/respond/stage.py:152` </location>
<code_context>
async def process(
self,
event: AstrMessageEvent,
) -> None:
result = event.get_result()
if result is None:
return
if result.result_content_type == ResultContentType.STREAMING_FINISH:
return
logger.info(
f"Prepare to send - {event.get_sender_name()}/{event.get_sender_id()}: {event._outline_chain(result.chain)}",
)
if result.result_content_type == ResultContentType.STREAMING_RESULT:
if result.async_stream is None:
logger.warning("async_stream 为空，跳过发送。")
return
# 流式结果直接交付平台适配器处理
use_fallback = self.config.get("provider_settings", {}).get(
"streaming_segmented",
False,
)
logger.info(f"应用流式输出({event.get_platform_id()})")
await event.send_streaming(result.async_stream, use_fallback)
return
if len(result.chain) > 0:
# 检查路径映射
if mappings := self.platform_settings.get("path_mapping", []):
for idx, component in enumerate(result.chain):
if isinstance(component, Comp.File) and component.file:
# 支持 File 消息段的路径映射。
component.file = path_Mapping(mappings, component.file)
event.get_result().chain[idx] = component
# 检查消息链是否为空
try:
if await self._is_empty_message_chain(result.chain):
logger.info("消息为空，跳过发送阶段")
return
except Exception as e:
logger.warning(f"空内容检查异常: {e}")
# 将 Plain 为空的消息段移除
result.chain = [
comp
for comp in result.chain
if not (
isinstance(comp, Comp.Plain)
and (not comp.text or not comp.text.strip())
)
]
# 发送消息链
# Record 需要强制单独发送
need_separately = {ComponentType.Record}
if self.is_seg_reply_required(event):
header_comps = self._extract_comp(
result.chain,
{ComponentType.Reply, ComponentType.At},
modify_raw_chain=True,
)
if not result.chain or len(result.chain) == 0:
# may fix #2670
logger.warning(
f"实际消息链为空, 跳过发送阶段。header_chain: {header_comps}, actual_chain: {result.chain}",
)
return
async with session_lock_manager.acquire_lock(event.unified_msg_origin):
for comp in result.chain:
i = await self._calc_comp_interval(comp)
await asyncio.sleep(i)
try:
if comp.type in need_separately:
await event.send(MessageChain([comp]))
else:
await event.send(MessageChain([*header_comps, comp]))
header_comps.clear()
except Exception as e:
logger.error(
f"发送消息链失败: chain = {MessageChain([comp])}, error = {e}",
exc_info=True,
)
else:
if all(
comp.type in {ComponentType.Reply, ComponentType.At}
for comp in result.chain
):
# may fix #2670
logger.warning(
f"消息链全为 Reply 和 At 消息段, 跳过发送阶段。chain: {result.chain}",
)
return
sep_comps = self._extract_comp(
result.chain,
need_separately,
modify_raw_chain=True,
)
for comp in sep_comps:
chain = MessageChain([comp])
try:
await event.send(chain)
except Exception as e:
logger.error(
f"发送消息链失败: chain = {chain}, error = {e}",
exc_info=True,
)
chain = MessageChain(result.chain)
if result.chain and len(result.chain) > 0:
try:
await event.send(chain)
except Exception as e:
logger.error(
f"发送消息链失败: chain = {chain}, error = {e}",
exc_info=True,
)
if await call_event_hook(event, EventType.OnAfterMessageSentEvent):
return
event.clear_result()
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Remove redundant conditional [×2] ([`remove-redundant-if`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-redundant-if/))
- Simplify sequence length comparison [×2] ([`simplify-len-comparison`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/simplify-len-comparison/))
- Low code quality found in RespondStage.process - 12% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))
<br/><details><summary>Explanation</summary>
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.</details>
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Member
Author
|
我蓄谋已久的路径统一 |
Co-Authored-By: 赵天乐(tyler zhao) <[email protected]>
Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>
Co-authored-by: 赵天乐(tyler zhao) <[email protected]>
Contributor
There was a problem hiding this comment.
Pull Request Overview
This PR refactors the AstrBot project's path management system by introducing a centralized AstrbotPaths class to replace scattered path utility functions. The refactoring moves core startup logic from main.py to astrbot/__main__.py and deprecates old path functions while maintaining backward compatibility.
Key changes:
- Introduces a new
astrbot.basepackage withAstrbotPathsclass for unified path management usingpathlib.Path - Replaces
get_astrbot_data_path()and related functions throughout the codebase withAstrbotPaths.astrbot_root - Migrates VERSION string from hardcoded value to dynamic retrieval via
importlib.metadata.version()
Reviewed Changes
Copilot reviewed 36 out of 37 changed files in this pull request and generated 13 comments.
Show a summary per file
| File | Description |
|---|---|
| pyproject.toml | Added dotenv>=0.9.9 dependency for environment variable loading |
| astrbot/base/paths.py | New centralized path management class with singleton pattern |
| astrbot/main.py | Moved application entry point logic from main.py |
| main.py | Simplified to delegate to astrbot.main.main() |
| astrbot/core/utils/astrbot_path.py | Deprecated functions with warnings, delegating to AstrbotPaths |
| Multiple plugin/core files | Updated to use AstrbotPaths instead of old path functions |
| astrbot/core/config/default.py | VERSION now uses importlib.metadata.version() |
| astrbot/core/message/components.py | Improved File component's .file property behavior |
astrbot/core/platform/sources/wechatpadpro/wechatpadpro_adapter.py
Outdated
Show resolved
Hide resolved
Contributor
There was a problem hiding this comment.
Pull Request Overview
Copilot reviewed 36 out of 37 changed files in this pull request and generated no new comments.
Comments suppressed due to low confidence (2)
astrbot/base/paths.py:1
- The documentation message is confusing and contains an awkward example
AstrbotPaths.getPaths('any!'). The proper usage should just beAstrbotPaths.astrbot_rootas shown in the implementation. Update the message to:\"不要再使用本函数!实际上就是获取data目录!等效于: AstrbotPaths.astrbot_root\"
from __future__ import annotations
astrbot/base/paths.py:1
- The deprecation warning message is unclear and contains incorrect syntax.
AstrbotPaths.getPaths('any!').root.envis not valid -.envis not a property of Path. The message should simply state:\"等效于: AstrbotPaths.astrbot_root (环境变量 ASTRBOT_ROOT 默认值: ./data)\"
from __future__ import annotations
Member
Author
|
@copilot 你可以使用ruff format --check帮我格式化吗？创建一个pr帮我格式化一下。 |
Contributor
|
@LIghtJUNction I've opened a new pull request, #3283, to work on those changes. Once the pull request is ready, I'll request review from you. |
* Initial plan * chore: 使用 ruff format 格式化代码 Co-authored-by: LIghtJUNction <[email protected]> --------- Co-authored-by: copilot-swe-agent[bot] <[email protected]> Co-authored-by: LIghtJUNction <[email protected]>
Member
Author
|
@copilot 创建一个pr，执行ruff check --fix，然后提交 |
Contributor
|
@LIghtJUNction I've opened a new pull request, #3285, to work on those changes. Once the pull request is ready, I'll request review from you. |
Member
Author
|
@copilot 运行ruff check --fix，并执行uv run main.py检查 |
Merged
Contributor
|
@LIghtJUNction I've opened a new pull request, #3286, to work on those changes. Once the pull request is ready, I'll request review from you. |
* Initial plan * fix: 移除未使用的临时目录变量 (ruff check --fix) Co-authored-by: LIghtJUNction <[email protected]> --------- Co-authored-by: copilot-swe-agent[bot] <[email protected]> Co-authored-by: LIghtJUNction <[email protected]>
Member
Author
|
@copilot 对路径类进行覆盖率测试，测试各种情况。覆盖率需达到100% |
12 tasks
Contributor
|
@LIghtJUNction I've opened a new pull request, #3287, to work on those changes. Once the pull request is ready, I'll request review from you. |
* Initial plan * feat: 为 AstrbotPaths 添加全面测试，覆盖率达到 100% Co-authored-by: LIghtJUNction <[email protected]> --------- Co-authored-by: copilot-swe-agent[bot] <[email protected]> Co-authored-by: LIghtJUNction <[email protected]>
Co-authored-by: Copilot <[email protected]>
Member
Author
|
@copilot 使用ruff修复格式等错误，使用mypy检查类型错误 |
Contributor
|
@LIghtJUNction I've opened a new pull request, #3289, to work on those changes. Once the pull request is ready, I'll request review from you. |
Member
Author
|
@Dt8333 check |
* Chore: Dockerfile (#3266) * fix: Dockerfile python main.py 改为uv run main.py * fix(dockerfile): 减少重复安装 * fix: 修复一些细节问题 * fix(.dockerignore): 需要git文件夹以获取astrbot版本（带git commit hash后缀) * fix(.dockerignore): uv run之前会uv sync * Replace insecure random with secrets module in cryptographic contexts (#3248) * Initial plan * Security fixes: Replace insecure random with secrets module and improve SSL context Co-authored-by: LIghtJUNction <[email protected]> * Address code review feedback: fix POST method and add named constants Co-authored-by: LIghtJUNction <[email protected]> * Improve documentation for random number generation constants Co-authored-by: LIghtJUNction <[email protected]> * Update astrbot/core/utils/io.py Co-authored-by: Copilot <[email protected]> * Update astrbot/core/platform/sources/wecom_ai_bot/WXBizJsonMsgCrypt.py Co-authored-by: Copilot <[email protected]> * Update tests/test_security_fixes.py Co-authored-by: Copilot <[email protected]> * Update astrbot/core/utils/io.py Co-authored-by: Copilot <[email protected]> * Update astrbot/core/utils/io.py Co-authored-by: Copilot <[email protected]> * Fix: Handle path parameter in SSL fallback for download_image_by_url Co-authored-by: LIghtJUNction <[email protected]> --------- Co-authored-by: copilot-swe-agent[bot] <[email protected]> Co-authored-by: LIghtJUNction <[email protected]> Co-authored-by: LIghtJUNction <[email protected]> Co-authored-by: Copilot <[email protected]> * chore: nodejs in Dockerfile * fix: typing error (#3267) * fix: 修复一些小错误。 修复aiocqhttp和slack中部分逻辑缺失的await。修复discord中错误的异常捕获类型。 * fix(core.platform): 修复discord适配器中错误的message_chain赋值 * fix(aiocqhttp): 更新convert_message方法的返回类型为AstrBotMessage | None --------- Co-authored-by: Soulter <[email protected]> * feat: support options to delete plugins config and data (#3280) * - 为插件管理页面中，删除插件提供一致的二次确认（原本只有卡片视图有二次确认） - 二次确认时可选删除插件配置和持久化数据 - 添加对应的i18n支持 * ruff * 移除未使用的 const $confirm = inject('$confirm'); --------- Co-authored-by: Copilot <[email protected]> Co-authored-by: Copilot <[email protected]> Co-authored-by: Soulter <[email protected]> Co-authored-by: Dt8333 <[email protected]> Co-authored-by: Soulter <[email protected]> Co-authored-by: Misaka Mikoto <[email protected]>
Member
Author
这个保留吧，我发现目前最大的用途就是手动部署时更新源代码时候用得到，其余时候都不应该使用。 |
Dt8333
approved these changes
Nov 5, 2025
Member
Dt8333
left a comment
Dt8333
left a comment
There was a problem hiding this comment.
目前行为基本与先前一致。
astrbot plug new计划弃用(By LIghtJUNction) ，ignore。
Contributor
There was a problem hiding this comment.
Hey there - I've reviewed your changes - here's some feedback:
- The dynamic VERSION calculation (subtracting one for dev builds) is brittle—consider centralizing version logic or simplifying it to avoid off-by-one mistakes.
- Deprecation warnings on every call to legacy path functions will spam logs; consider throttling them or issuing a one-time migration notice instead.
- The AstrbotPaths singleton pattern isn’t guarded against concurrent access—if you expect multi-threaded usage, add synchronization or document its thread-unsafe nature.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- The dynamic VERSION calculation (subtracting one for dev builds) is brittle—consider centralizing version logic or simplifying it to avoid off-by-one mistakes.
- Deprecation warnings on every call to legacy path functions will spam logs; consider throttling them or issuing a one-time migration notice instead.
- The AstrbotPaths singleton pattern isn’t guarded against concurrent access—if you expect multi-threaded usage, add synchronization or document its thread-unsafe nature.
## Individual Comments
### Comment 1
<location> `astrbot/core/utils/pip_installer.py:24-29` </location>
<code_context>
mirror: str | None = None,
- ):
+ ) -> None:
+ if requirements_path:
+ cwd = Path(requirements_path).parent.resolve()
+ elif project_path:
+ cwd = Path(project_path).resolve()
+ else:
+ cwd = Path().cwd() # 安装pip包时避免cwd变量未初始化
+
args = ["install"]
</code_context>
<issue_to_address>
**suggestion:** cwd is set based on requirements_path or project_path, but fallback is Path().cwd().
Path().cwd() may not reflect the intended working directory for pip installs. Please consider making cwd selection more explicit or allowing it to be configured.
Suggested implementation:
```python
async def install(
self,
package_name: str | None = None,
requirements_path: str | None = None,
project_path: str | None = None,
mirror: str | None = None,
cwd: str | Path | None = None,
) -> None:
if cwd is not None:
cwd_path = Path(cwd).resolve()
elif requirements_path:
cwd_path = Path(requirements_path).parent.resolve()
elif project_path:
cwd_path = Path(project_path).resolve()
else:
cwd_path = Path.cwd() # fallback to current working directory if nothing is provided
args = ["install"]
```
You will need to update any calls to `install()` elsewhere in your codebase to optionally pass the new `cwd` parameter if you want to explicitly control the working directory for pip installs.
Also, ensure that wherever `cwd` is used later in the function (e.g., in subprocess calls), you replace it with `cwd_path`.
</issue_to_address>
### Comment 2
<location> `tests/test_paths.py:190-196` </location>
<code_context>
+ assert paths.root == non_existent
+ assert non_existent.exists()
+
+ def test_root_property_fallback_to_cwd(
+ self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+ ) -> None:
+ """测试 root 属性在根目录被删除后回退到 cwd/.astrbot."""
+ import shutil
+
+ # 创建并设置一个根目录
+ temp_root = tmp_path / "test_root"
+ temp_root.mkdir(parents=True, exist_ok=True)
+
+ # 清除实例缓存
+ AstrbotPaths._instances.clear()
+ AstrbotPaths.astrbot_root = temp_root
+
+ # 创建实例
+ paths = AstrbotPaths.getPaths("test-fallback")
+
+ # 删除根目录（模拟被外部删除的情况）
+ shutil.rmtree(temp_root)
+
+ # 现在访问 root 应该回退到 cwd/.astrbot
+ expected = Path.cwd() / ".astrbot"
+ assert paths.root == expected
+
+ def test_home_property(self, paths_instance: AstrbotPaths, temp_root: Path) -> None:
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding tests for permission errors and invalid path types.
Adding tests for permission errors and invalid path types will improve coverage of error handling and ensure the code behaves correctly in these scenarios.
```suggestion
def test_home_property(self, paths_instance: AstrbotPaths, temp_root: Path) -> None:
"""测试 home 属性."""
home_path = paths_instance.home
expected = temp_root / "home" / paths_instance.name
assert home_path == expected
assert home_path.exists()
assert home_path.is_dir()
def test_root_permission_error(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""测试 root 属性在根目录权限不足时的行为."""
import os
import stat
temp_root = tmp_path / "perm_root"
temp_root.mkdir(parents=True, exist_ok=True)
AstrbotPaths._instances.clear()
AstrbotPaths.astrbot_root = temp_root
# 移除所有权限
temp_root.chmod(0)
paths = AstrbotPaths.getPaths("test-perm")
try:
# 访问 root 属性应引发 PermissionError
with pytest.raises(PermissionError):
_ = paths.root.exists()
finally:
# 恢复权限，避免影响后续测试
temp_root.chmod(stat.S_IRWXU)
def test_root_invalid_path_type(self, tmp_path: Path) -> None:
"""测试 root 属性为文件而非目录时的行为."""
temp_file = tmp_path / "not_a_dir"
temp_file.write_text("not a directory")
AstrbotPaths._instances.clear()
AstrbotPaths.astrbot_root = temp_file
paths = AstrbotPaths.getPaths("test-invalid-type")
# 访问 root 属性应检测到不是目录
assert paths.root == temp_file
assert temp_file.is_file()
# 可以根据实际 AstrbotPaths 行为补充断言，比如是否回退或报错
```
</issue_to_address>
### Comment 3
<location> `tests/test_paths.py:296-302` </location>
<code_context>
+
+ assert AstrbotPaths.is_root(temp_root) is False
+
+ def test_is_root_with_non_existent_path(self) -> None:
+ """测试不存在的路径."""
+ non_existent = Path("/definitely/not/exist/path")
+ assert AstrbotPaths.is_root(non_existent) is False
+
+ def test_is_root_with_file_not_directory(self, temp_root: Path) -> None:
</code_context>
<issue_to_address>
**suggestion (testing):** Missing test for is_root with symlinked directories.
Please add a test for AstrbotPaths.is_root with a symlink to a root directory to ensure correct handling of symlinks.
```suggestion
def test_is_root_with_file_not_directory(self, temp_root: Path) -> None:
"""测试路径是文件而非目录."""
test_file = temp_root / "test.txt"
test_file.touch()
assert AstrbotPaths.is_root(test_file) is False
def test_is_root_with_symlink_to_root(self, tmp_path: Path) -> None:
"""测试符号链接指向根目录."""
root_dir = tmp_path / "root"
root_dir.mkdir()
marker_file = root_dir / ".astrbot"
marker_file.touch()
symlink_dir = tmp_path / "symlink_to_root"
symlink_dir.symlink_to(root_dir, target_is_directory=True)
assert AstrbotPaths.is_root(symlink_dir) is True
def test_is_root_with_symlink_to_non_root(self, tmp_path: Path) -> None:
"""测试符号链接指向非根目录."""
non_root_dir = tmp_path / "non_root"
non_root_dir.mkdir()
symlink_dir = tmp_path / "symlink_to_non_root"
symlink_dir.symlink_to(non_root_dir, target_is_directory=True)
assert AstrbotPaths.is_root(symlink_dir) is False
```
</issue_to_address>
### Comment 4
<location> `astrbot/base/paths.py:21` </location>
<code_context>
+ from collections.abc import AsyncGenerator, Generator
+
+
+class AstrbotPaths(IAstrbotPaths):
+ """统一化路径获取."""
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring to use lru_cache, a helper for directory creation, and unified context manager logic to reduce boilerplate and indirection.
Here are a few ways to cut down on indirection and repeated boilerplate while keeping all of your functionality:
1. Replace manual instance‐cache with an `lru_cache` on your factory:
```python
from functools import lru_cache
class AstrbotPaths(IAstrbotPaths):
# … keep __init__, astrbot_root, load_dotenv, etc.
@classmethod
@lru_cache(maxsize=None)
def get_paths(cls, name: str) -> AstrbotPaths:
normalized = canonicalize_name(name)
return cls(normalized)
```
2. Factor out all of the identical “mk-dir + return” logic into one helper:
```python
class AstrbotPaths(IAstrbotPaths):
# …
def _ensure(self, section: str) -> Path:
p = self.astrbot_root / section / self.name
p.mkdir(parents=True, exist_ok=True)
return p
@property
def home(self) -> Path:
return self._ensure("home")
@property
def config(self) -> Path:
return self._ensure("config")
# same for data, logs, temp, plugins…
```
3. Unify your sync/async context managers to share the same body:
```python
from contextlib import contextmanager, asynccontextmanager
class AstrbotPaths(IAstrbotPaths):
# …
@contextmanager
def chdir(self, cwd: Path) -> Generator[Path, None, None]:
original = Path.cwd()
target = self.root / cwd
os.chdir(target)
try:
yield target
finally:
os.chdir(original)
@asynccontextmanager
async def achdir(self, cwd: Path) -> AsyncGenerator[Path, None]:
with self.chdir(cwd) as target:
yield target
```
These small refactors remove the manual cache dict, collapse six nearly‐identical properties into one helper, and let your async version simply wrap the sync one—all while preserving exactly the same behavior.
</issue_to_address>
### Comment 5
<location> `astrbot/core/utils/astrbot_path.py:16` </location>
<code_context>
+from astrbot.base import AstrbotPaths
def get_astrbot_path() -> str:
- """获取Astrbot项目路径"""
+ """获取Astrbot项目路径 --仅供手动部署时/更新源代码时使用.
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the repeated deprecation wrappers into a single factory function to centralize logic and avoid duplication.
Here’s how you can collapse all of these identical deprecation‐wrappers into a tiny factory (or `__getattr__`) so you don’t have to keep copy/pasting doc‐strings and `warnings.warn()` calls:
```python
import warnings
from astrbot.base import AstrbotPaths
_deprecated = {
"get_astrbot_path": ("astrbot_path",
"Only for manual source‐deploy/update. Use AstrbotPaths.astrbot_root for most cases."),
"get_astrbot_root": ("astrbot_root",
"Use AstrbotPaths.astrbot_root."),
"get_astrbot_data_path": ("astrbot_root",
"Data dir is the same as AstrbotPaths.astrbot_root."),
"get_astrbot_config_path": ("astrbot_root",
"Use AstrbotPaths.astrbot_root / 'config'."),
"get_astrbot_plugin_path": ("astrbot_root",
"Use AstrbotPaths.astrbot_root / 'plugins'."),
}
def _make_shim(old_name, attr_name, hint):
def shim() -> str:
warnings.warn(
f"{old_name} is deprecated, use AstrbotPaths.{attr_name}. {hint}",
DeprecationWarning,
stacklevel=2,
)
return str(getattr(AstrbotPaths, attr_name))
shim.__name__ = old_name
shim.__doc__ = f"DEPRECATED: {hint}"
return shim
# populate module‐globals
for old, (attr, hint) in _deprecated.items():
globals()[old] = _make_shim(old, attr, hint)
```
This keeps each stub 5 lines, centralizes the warning logic, and you never repeat yourself.
</issue_to_address>
### Comment 6
<location> `astrbot/core/message/components.py:658` </location>
<code_context>
@property
def file(self) -> str:
- """获取文件路径，如果文件不存在但有URL，则同步下载文件
+ """
</code_context>
<issue_to_address>
**issue (complexity):** Consider removing synchronous download logic from the deprecated .file property and centralizing temp-file handling in a helper function to simplify the code.
Here are two concrete ways to cut down on the complexity without removing any existing public API:
1) **Drop all of the synchronous-download branches from `.file`**
Since you’re already deprecating it, just emit one warning and forward everybody to `get_file()`. That will collapse nearly the entire `if self.url: …` block into a single line:
```python
@property
def file(self) -> str:
warnings.warn(
"File.file is deprecated; use `await get_file()` instead",
DeprecationWarning,
stacklevel=2,
)
return self.file_ if self.file_ and os.path.exists(self.file_) else ""
```
2) **Extract the “download / temp‐file” logic into a single helper**
Almost every `async def …` (“convert_to_base64”, “convert_to_path”, “register_to_file_service” in all components) does:
- make a temp dir
- pick a UUID filename
- download via URL, base64, or copy an existing file
- return an absolute path
Move that to one private function, then each method becomes a trivial dispatcher:
```python
# utils.py
async def _fetch_to_temp(source: str, ext: str="") -> str:
"""
If source is a file:// or existing local path, return abs path.
If http:// URL, download via download_file().
If base64://… decode and write to .jpg (or ext).
"""
if source.startswith("file:///"):
return source[len("file:///"):]
tmp_root = AstrbotPaths.astrbot_root / "temp"
os.makedirs(tmp_root, exist_ok=True)
target = tmp_root / f"{uuid.uuid4().hex}{ext}"
if source.startswith("http"):
await download_file(source, target)
elif source.startswith("base64://"):
data = base64.b64decode(source.removeprefix("base64://"))
target.write_bytes(data)
elif os.path.exists(source):
return os.path.abspath(source)
else:
raise ValueError(f"not a valid source: {source}")
return str(target.resolve())
```
```python
# in your component
async def convert_to_path(self) -> str:
return await _fetch_to_temp(self.file, ext="")
```
```python
async def convert_to_base64(self) -> str:
# if you need base64 out, first fetch to path then:
path = await _fetch_to_temp(self.file, ext=".jpg")
return file_to_base64(path)
```
This removes almost all of the repeated branching and temp-dir boilerplate from the component methods, while keeping every existing public method signature exactly the same.
</issue_to_address>
### Comment 7
<location> `astrbot/core/updator.py:120-122` </location>
<code_context>
</code_context>
<issue_to_address>
**issue (code-quality):** Raise a specific error instead of the general `Exception` or `BaseException` ([`raise-specific-error`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/raise-specific-error))
<details><summary>Explanation</summary>If a piece of code raises a specific exception type
rather than the generic
[`BaseException`](https://docs.python.org/3/library/exceptions.html#BaseException)
or [`Exception`](https://docs.python.org/3/library/exceptions.html#Exception),
the calling code can:
- get more information about what type of error it is
- define specific exception handling for it
This way, callers of the code can handle the error appropriately.
How can you solve this?
- Use one of the [built-in exceptions](https://docs.python.org/3/library/exceptions.html) of the standard library.
- [Define your own error class](https://docs.python.org/3/tutorial/errors.html#tut-userexceptions) that subclasses `Exception`.
So instead of having code raising `Exception` or `BaseException` like
```python
if incorrect_input(value):
raise Exception("The input is incorrect")
```
you can have code raising a specific error like
```python
if incorrect_input(value):
raise ValueError("The input is incorrect")
```
or
```python
class IncorrectInputError(Exception):
pass
if incorrect_input(value):
raise IncorrectInputError("The input is incorrect")
```
</details>
</issue_to_address>
### Comment 8
<location> `astrbot/__main__.py:15` </location>
<code_context>
def check_env():
if not (sys.version_info.major == 3 and sys.version_info.minor >= 10):
logger.error("请使用 Python3.10+ 运行本项目。")
exit()
# os.makedirs("data/config", exist_ok=True)
# os.makedirs("data/plugins", exist_ok=True)
# os.makedirs("data/temp", exist_ok=True)
# 针对问题 #181 的临时解决方案
mimetypes.add_type("text/javascript", ".js")
mimetypes.add_type("text/javascript", ".mjs")
mimetypes.add_type("application/json", ".json")
</code_context>
<issue_to_address>
**suggestion (code-quality):** Simplify logical expression using De Morgan identities ([`de-morgan`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/de-morgan/))
```suggestion
if sys.version_info.major != 3 or sys.version_info.minor < 10:
```
</issue_to_address>
### Comment 9
<location> `astrbot/cli/commands/cmd_init.py:33` </location>
<code_context>
async def initialize_astrbot(astrbot_root: Path) -> None:
"""执行 AstrBot 初始化逻辑"""
dot_astrbot = astrbot_root / ".astrbot"
if not dot_astrbot.exists():
click.echo(f"Current Directory: {astrbot_root}")
click.echo(
"如果你确认这是 Astrbot root directory, 你需要在当前目录下创建一个 .astrbot 文件标记该目录为 AstrBot 的数据目录。",
)
if click.confirm(
f"请检查当前目录是否正确，确认正确请回车: {astrbot_root}",
default=True,
abort=True,
):
dot_astrbot.touch()
click.echo(f"Created {dot_astrbot}")
paths = {
"config": astrbot_root / "config",
"plugins": astrbot_root / "plugins",
"temp": astrbot_root / "temp",
}
for _, path in paths.items():
path.mkdir(parents=True, exist_ok=True)
click.echo(f"{'Created' if not path.exists() else 'Directory exists'}: {path}")
await check_dashboard(astrbot_root)
</code_context>
<issue_to_address>
**suggestion (code-quality):** Replace calls to `dict.items` with `dict.values` when the keys are not used ([`replace-dict-items-with-values`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/replace-dict-items-with-values/))
```suggestion
for path in paths.values():
```
</issue_to_address>
### Comment 10
<location> `astrbot/core/pipeline/respond/stage.py:152` </location>
<code_context>
async def process(
self,
event: AstrMessageEvent,
) -> None:
result = event.get_result()
if result is None:
return
if result.result_content_type == ResultContentType.STREAMING_FINISH:
return
logger.info(
f"Prepare to send - {event.get_sender_name()}/{event.get_sender_id()}: {event._outline_chain(result.chain)}",
)
if result.result_content_type == ResultContentType.STREAMING_RESULT:
if result.async_stream is None:
logger.warning("async_stream 为空，跳过发送。")
return
# 流式结果直接交付平台适配器处理
use_fallback = self.config.get("provider_settings", {}).get(
"streaming_segmented",
False,
)
logger.info(f"应用流式输出({event.get_platform_id()})")
await event.send_streaming(result.async_stream, use_fallback)
return
if len(result.chain) > 0:
# 检查路径映射
if mappings := self.platform_settings.get("path_mapping", []):
for idx, component in enumerate(result.chain):
if isinstance(component, Comp.File) and component.file:
# 支持 File 消息段的路径映射。
component.file = path_Mapping(mappings, component.file)
event.get_result().chain[idx] = component
# 检查消息链是否为空
try:
if await self._is_empty_message_chain(result.chain):
logger.info("消息为空，跳过发送阶段")
return
except Exception as e:
logger.warning(f"空内容检查异常: {e}")
# 将 Plain 为空的消息段移除
result.chain = [
comp
for comp in result.chain
if not (
isinstance(comp, Comp.Plain)
and (not comp.text or not comp.text.strip())
)
]
# 发送消息链
# Record 需要强制单独发送
need_separately = {ComponentType.Record}
if self.is_seg_reply_required(event):
header_comps = self._extract_comp(
result.chain,
{ComponentType.Reply, ComponentType.At},
modify_raw_chain=True,
)
if not result.chain or len(result.chain) == 0:
# may fix #2670
logger.warning(
f"实际消息链为空, 跳过发送阶段。header_chain: {header_comps}, actual_chain: {result.chain}",
)
return
async with session_lock_manager.acquire_lock(event.unified_msg_origin):
for comp in result.chain:
i = await self._calc_comp_interval(comp)
await asyncio.sleep(i)
try:
if comp.type in need_separately:
await event.send(MessageChain([comp]))
else:
await event.send(MessageChain([*header_comps, comp]))
header_comps.clear()
except Exception as e:
logger.error(
f"发送消息链失败: chain = {MessageChain([comp])}, error = {e}",
exc_info=True,
)
else:
if all(
comp.type in {ComponentType.Reply, ComponentType.At}
for comp in result.chain
):
# may fix #2670
logger.warning(
f"消息链全为 Reply 和 At 消息段, 跳过发送阶段。chain: {result.chain}",
)
return
sep_comps = self._extract_comp(
result.chain,
need_separately,
modify_raw_chain=True,
)
for comp in sep_comps:
chain = MessageChain([comp])
try:
await event.send(chain)
except Exception as e:
logger.error(
f"发送消息链失败: chain = {chain}, error = {e}",
exc_info=True,
)
chain = MessageChain(result.chain)
if result.chain and len(result.chain) > 0:
try:
await event.send(chain)
except Exception as e:
logger.error(
f"发送消息链失败: chain = {chain}, error = {e}",
exc_info=True,
)
if await call_event_hook(event, EventType.OnAfterMessageSentEvent):
return
event.clear_result()
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Remove redundant conditional [×2] ([`remove-redundant-if`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/remove-redundant-if/))
- Simplify sequence length comparison [×2] ([`simplify-len-comparison`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/simplify-len-comparison/))
- Low code quality found in RespondStage.process - 12% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))
<br/><details><summary>Explanation</summary>
The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.
How can you solve this?
It might be worth refactoring this function to make it shorter and more readable.
- Reduce the function length by extracting pieces of functionality out into
their own functions. This is the most important thing you can do - ideally a
function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
sits together within the function rather than being scattered.</details>
</issue_to_address>
### Comment 11
<location> `astrbot/core/updator.py:100-108` </location>
<code_context>
def _generate_update_instruction(
self, latest: bool = True, version: str | None = None
) -> str:
"""私有辅助函数
Args:
latest: 是否更新到最新版本
version: 目标版本号，如果 latest=True 则忽略
Returns:
str: 更新指令字符串
"""
if latest:
pip_cmd = "pip install git+https://github.com/AstrBotDevs/AstrBot.git"
uv_cmd = "uv tool upgrade astrbot"
else:
if version:
pip_cmd = f"pip install git+https://github.com/AstrBotDevs/AstrBot.git@{version}"
uv_cmd = f"uv tool install --force git+https://github.com/AstrBotDevs/AstrBot.git@{version} astrbot"
else:
raise ValueError("当 latest=False 时，必须提供 version")
return (
"命令行启动时,请直接使用uv tool upgrade astrbot更新\n"
f"或者使用此命令更新: {pip_cmd}"
f"使用uv: {uv_cmd}"
)
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Merge else clause's nested if statement into elif ([`merge-else-if-into-elif`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/merge-else-if-into-elif/))
- Swap if/else branches [×2] ([`swap-if-else-branches`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/swap-if-else-branches/))
</issue_to_address>
### Comment 12
<location> `astrbot/core/utils/pip_installer.py:47` </location>
<code_context>
async def install(
self,
package_name: str | None = None,
requirements_path: str | None = None,
project_path: str | None = None,
mirror: str | None = None,
) -> None:
if requirements_path:
cwd = Path(requirements_path).parent.resolve()
elif project_path:
cwd = Path(project_path).resolve()
else:
cwd = Path().cwd() # 安装pip包时避免cwd变量未初始化
args = ["install"]
if package_name:
args.append(package_name)
elif requirements_path:
args.extend(["-r", requirements_path])
elif project_path:
args.extend(".")
index_url = mirror or self.pypi_index_url or "https://pypi.org/simple"
args.extend(["--trusted-host", "mirrors.aliyun.com", "-i", index_url])
if self.pip_install_arg:
args.extend(self.pip_install_arg.split())
logger.info(f"Pip 包管理器: pip {' '.join(args)}")
try:
async with self.paths.achdir(cwd):
process = await asyncio.create_subprocess_exec(
sys.executable,
"-m",
"pip",
*args,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.STDOUT,
)
assert process.stdout is not None
async for line in process.stdout:
logger.info(line.decode().strip())
await process.wait()
if process.returncode != 0:
raise Exception(f"安装失败，错误码：{process.returncode}")
except FileNotFoundError:
# 没有 pip
from pip import main as pip_main
result_code = await asyncio.to_thread(pip_main, args)
# 清除 pip.main 导致的多余的 logging handlers
for handler in logging.root.handlers[:]:
logging.root.removeHandler(handler)
if result_code != 0:
raise Exception(f"安装失败，错误码：{result_code}")
</code_context>
<issue_to_address>
**issue (code-quality):** Explicitly raise from a previous error ([`raise-from-previous-error`](https://docs.sourcery.ai/Reference/Default-Rules/suggestions/raise-from-previous-error/))
</issue_to_address>
### Comment 13
<location> `tests/test_paths.py:507-510` </location>
<code_context>
def test_singleton_pattern_thread_safe(self, temp_root: Path) -> None:
"""测试单例模式的基本行为（注意：不是真正的线程安全测试）."""
instances = []
for _ in range(10):
instances.append(AstrbotPaths.getPaths("singleton-test"))
# 所有实例应该是同一个对象
first = instances[0]
for instance in instances[1:]:
assert instance is first
</code_context>
<issue_to_address>
**suggestion (code-quality):** Convert for loop into list comprehension ([`list-comprehension`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/list-comprehension/))
```suggestion
instances = [AstrbotPaths.getPaths("singleton-test") for _ in range(10)]
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>
Co-authored-by: sourcery-ai[bot] <58596630+sourcery-ai[bot]@users.noreply.github.com>
Member
Author
|
@Soulter 最后再请求下你的意见 安装插件依赖时用上下文管理器，将cwd临时切换到插件目录 |
Dt8333
approved these changes
Nov 5, 2025
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
.env文件保存统一的路径
静态文件一起打包
fixes #XYZ
Motivation / 动机
Modifications / 改动点
Verification Steps / 验证步骤
Screenshots or Test Results / 运行截图或测试结果
Compatibility & Breaking Changes / 兼容性与破坏性变更
Checklist / 检查清单
requirements.txt和pyproject.toml文件相应位置。/ I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations inrequirements.txtandpyproject.toml.Sourcery 总结
引入统一的
AstrbotPaths类进行路径管理，弃用旧的路径工具函数，采用importlib.resources加载静态资源，并通过.env文件集中化配置。新功能:
AstrbotPaths类，带有IAstrbotPaths抽象基类，用于集中化和结构化的路径处理改进:
get_astrbot_*函数和 CLI 路径检查，转而使用AstrbotPaths并附带弃用警告importlib.resources，以实现更安全的包资源访问AstrBotUpdator，在解压更新时使用新的AstrbotPaths根路径构建:
pyproject.toml中添加dotenv依赖，用于基于环境的配置文档:
.env和.env.example文件，用于管理ASTRBOT_ROOTOriginal summary in English
Summary by Sourcery
Introduce a unified AstrbotPaths class for path management and deprecate legacy path utility functions, adopt importlib.resources for static asset loading, and centralize configuration via a .env file.
New Features:
Enhancements:
Build:
Documentation:
Sourcery 总结
通过引入
AstrbotPaths类来统一和集中路径管理，废弃旧的实用函数，更新代码库以使用新的路径 API，通过.env配置环境，并为新的路径管理器添加全面的测试。新功能：
AstrbotPaths类及IAstrbotPaths接口，用于统一和结构化的路径管理AstrbotPaths添加全面的测试，涵盖初始化、属性、上下文管理器和单例行为改进：
get_astrbot_*实用函数和 CLI 路径检查，转而使用AstrbotPaths并附带废弃警告AstrbotPathsimportlib.resources加载静态包资产，并更新仪表板 CLI 以使用集中式的、通过环境变量配置的根目录astrbot/__main__.py中，并简化 CLI 和 Python 模块的入口点构建：
pyproject.toml中添加dotenv依赖，并提供.env/.env.example文件用于集中化ASTRBOT_ROOT配置pyproject.toml中引入严格的 mypy 配置测试：
tests/test_paths.py用于测试AstrbotPaths的行为和模块隔离Original summary in English
Summary by Sourcery
Unify and centralize path management by introducing AstrbotPaths class, deprecate old utility functions, update codebase to use the new path API, configure environment via .env, and add comprehensive tests for the new path manager.
New Features:
Enhancements:
Build:
Tests: