release: v2.1.3 - #100
Conversation
OneBotMessageSender 构建合并转发 Node 时未传 uin,导致 Node.uin 默认
为 "0",序列化后 user_id="0"。新版 NapCat / Lagrange 会拒绝 user_id
为 0 或空的转发节点,返回 retcode=1200 ("forward node user_id/uin is
required"),使 RSS 推送到 QQ 群失败。
修复:
- onebot_sender: 新增 _resolve_bot_self_id,优先从 event.message_obj.self_id
读取(命令响应场景),其次通过 get_bot_self_id provider 解析(主动推送
场景),最后兜底非零占位 "10000";构建与回退 Node 时均传入 uin。
- bootstrap: 注册 set_bot_self_id_provider,从 CQHttp 的 _wsr_api_clients
解析已连接 bot 的 self_id(types.py 已预留 hook 但未注册)。
同时将 bot_client 解析上移到节点构建之前,使 uin 在构建节点时即可用。
(cherry picked from commit 6d5f7af)
(cherry picked from commit 62ce5ee)
Reviewer's Guide实现了一种安全机制,用于在命令响应和主动推送两种场景中解析并注入 OneBot 机器人 self_id(QQ 号)到合并转发节点中;通过 bootstrap 注册全局 self_id provider;相应更新测试和文档;并将插件版本号提升至 v2.1.3。 Sequence diagram for resolving bot self_id during OneBot send_to_usersequenceDiagram
participant RSSJob
participant OneBotMessageSender
participant MessageContext
participant get_bot_self_id
participant _bot_self_id_provider
RSSJob->>OneBotMessageSender: send_to_user(request, context)
OneBotMessageSender->>OneBotMessageSender: _resolve_bot_self_id(context)
alt event message has self_id and != 0
OneBotMessageSender->>MessageContext: getattr(event.message_obj, self_id)
OneBotMessageSender-->>OneBotMessageSender: return self_id
else platform_name is set
OneBotMessageSender->>get_bot_self_id: get_bot_self_id(platform_name)
get_bot_self_id->>_bot_self_id_provider: _bot_self_id_provider(platform_name)
_bot_self_id_provider-->>get_bot_self_id: self_id or ""
get_bot_self_id-->>OneBotMessageSender: self_id or ""
else cannot resolve
OneBotMessageSender-->>OneBotMessageSender: return ""
end
OneBotMessageSender->>OneBotMessageSender: _build_forward_node(content, nickname, bot_self_id)
Note over OneBotMessageSender: Only sets Node.uin when bot_self_id is non-empty
OneBotMessageSender-->>RSSJob: SendResult
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your Experience访问你的 dashboard 以:
Getting HelpOriginal review guide in EnglishReviewer's GuideImplements a safe mechanism for resolving and injecting the OneBot bot self_id (QQ number) into merged forward nodes for both command-response and active-push scenarios, wires a global self_id provider via bootstrap, adjusts tests and docs accordingly, and bumps the plugin version to v2.1.3. Sequence diagram for resolving bot self_id during OneBot send_to_usersequenceDiagram
participant RSSJob
participant OneBotMessageSender
participant MessageContext
participant get_bot_self_id
participant _bot_self_id_provider
RSSJob->>OneBotMessageSender: send_to_user(request, context)
OneBotMessageSender->>OneBotMessageSender: _resolve_bot_self_id(context)
alt event message has self_id and != 0
OneBotMessageSender->>MessageContext: getattr(event.message_obj, self_id)
OneBotMessageSender-->>OneBotMessageSender: return self_id
else platform_name is set
OneBotMessageSender->>get_bot_self_id: get_bot_self_id(platform_name)
get_bot_self_id->>_bot_self_id_provider: _bot_self_id_provider(platform_name)
_bot_self_id_provider-->>get_bot_self_id: self_id or ""
get_bot_self_id-->>OneBotMessageSender: self_id or ""
else cannot resolve
OneBotMessageSender-->>OneBotMessageSender: return ""
end
OneBotMessageSender->>OneBotMessageSender: _build_forward_node(content, nickname, bot_self_id)
Note over OneBotMessageSender: Only sets Node.uin when bot_self_id is non-empty
OneBotMessageSender-->>RSSJob: SendResult
File-Level Changes
Tips and commandsInteracting with Sourcery
Customizing Your ExperienceAccess your dashboard to:
Getting Help
|
📝 WalkthroughWalkthrough本次更新发布 2.1.3,新增 OneBot bot self_id 解析与合并转发节点 ChangesOneBot self_id 解析与转发节点构建
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant MessageContext
participant OneBotMessageSender
participant SelfIdProvider
participant ForwardNode
MessageContext->>OneBotMessageSender: 提供事件 self_id
OneBotMessageSender->>SelfIdProvider: 事件账号无效时解析 self_id
SelfIdProvider-->>OneBotMessageSender: 返回唯一账号或空字符串
OneBotMessageSender->>ForwardNode: 条件写入 uin
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
There was a problem hiding this comment.
Hey - I've found 4 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="bootstrap.py" line_range="282-297" />
<code_context>
)
return None
+ def _resolve_bot_self_id(platform_name: str) -> str:
+ """从已连接的反向 WebSocket 客户端中解析 bot 的 self_id(QQ 号)。
+
+ aiocqhttp 的 CQHttp 实例以 self_id 为键保存已连接的 API 客户端
+ (``_wsr_api_clients``)。只有连接中恰有一个非零 self_id 时才返回,
+ 避免多 bot 平台把其他账号误写入合并转发节点;解析失败时返回空串,
+ 由调用方保留 SDK 的默认值。
+ """
+ client = _resolve_bot_client(platform_name)
+ if client is None:
+ return ""
+ api_clients = getattr(client, "_wsr_api_clients", None)
+ if isinstance(api_clients, dict):
+ self_ids = [
+ str(self_id).strip()
+ for self_id in list(api_clients)
+ if str(self_id or "").strip() not in {"", "0"}
+ ]
+ if len(self_ids) == 1:
+ return self_ids[0]
+ if len(self_ids) > 1:
+ logger.warning(
+ "无法唯一解析 bot self_id,保留合并转发节点默认值: "
</code_context>
<issue_to_address>
**suggestion:** self_id 提取逻辑做了一些可以简化的额外工作,影响清晰度和效率。
在 `_resolve_bot_self_id` 中,下列列表推导式
```python
self_ids = [
str(self_id).strip()
for self_id in list(api_clients)
if str(self_id or "").strip() not in {"", "0"}
]
```
包含多余的 `str(...)` 调用以及不必要的 `list(api_clients)` 转换。你可以直接遍历 `api_clients`(或 `api_clients.keys()`),对每个键只归一化一次,再应用非空 / 非零过滤。这样可以在保持现有唯一性检查逻辑不变的前提下,提高可读性和效率。
```suggestion
api_clients = getattr(client, "_wsr_api_clients", None)
if isinstance(api_clients, dict):
self_ids: list[str] = []
for self_id in api_clients:
normalized = str(self_id or "").strip()
if normalized and normalized != "0":
self_ids.append(normalized)
if len(self_ids) == 1:
return self_ids[0]
if len(self_ids) > 1:
logger.warning(
"无法唯一解析 bot self_id,保留合并转发节点默认值: "
"platform=%s, candidates=%s",
platform_name,
self_ids,
)
```
</issue_to_address>
### Comment 2
<location path="tests/unit/infrastructure/test_onebot_sender.py" line_range="417-426" />
<code_context>
+@pytest.mark.asyncio
</code_context>
<issue_to_address>
**suggestion (testing):** 请增加一个同时存在 event.self_id 和 provider self_id 的测试用例,以验证优先级。
当前测试已经覆盖了仅事件提供 self_id、仅 provider 提供 self_id,以及未知 `self_id` 的情况,但还缺少验证「事件值应覆盖 provider 值」这一优先级规则的用例。建议新增一个用例,在其中 `context.event.message_obj.self_id` 与 `get_bot_self_id` 返回值不同,并断言转发节点的 `uin` 使用事件中的值,以防止将来修改不小心颠倒这个优先级。
建议实现如下:
```python
monkeypatch.setattr(sender, "_send_chain", fake_send_chain)
@pytest.mark.asyncio
async def test_onebot_sender_prefers_event_self_id_over_provider(monkeypatch):
"""当事件和 provider 都提供 self_id 时,应优先使用事件中的 bot QQ 号写入转发节点。"""
sender = OneBotMessageSender()
calls: list[tuple[str, list]] = []
async def fake_send_chain(session_id: str, chain: list, **_kwargs):
calls.append((session_id, chain))
return SendResult(ok=True)
# provider 返回的 bot self_id 与事件中的不同,用于验证优先级
monkeypatch.setattr(sender, "_send_chain", fake_send_chain)
monkeypatch.setattr(sender, "get_bot_self_id", lambda context: "222222")
# 构造同时具有 event.self_id 和 provider self_id 的上下文
# 这里假设已有的测试工具/fixture可以设置 event.message_obj.self_id
context = make_dummy_context() # 占位:请替换为项目中实际的上下文构造方式
context.event.message_obj.self_id = "111111"
# 触发构建转发消息的逻辑(占位:请替换为项目中实际的发送/构建调用)
await sender.send(context, "测试带转发节点的消息")
# 断言转发节点的 uin 使用事件中的 self_id,而不是 provider 返回的值
assert calls, "预期 fake_send_chain 被调用以生成转发节点"
forward_nodes = [
segment for segment in calls[0][1]
if getattr(segment, "type", None) == "node"
]
assert forward_nodes, "预期消息链中包含至少一个转发节点"
for node in forward_nodes:
# uin 字段应该等于事件中的 self_id,而不是 get_bot_self_id 返回的值
assert getattr(node, "data", {}).get("uin") == "111111"
```
1. 将 `make_dummy_context()` 替换为项目中实际用于构造上下文 / 事件的 helper 或 fixture,确保能够设置 `context.event.message_obj.self_id`。
2. 将 `await sender.send(context, "测试带转发节点的消息")` 替换为实际会走到「构建转发节点」逻辑的调用(例如发送包含转发 / 引用 / 多段消息的接口)。
3. `forward_nodes` 的筛选条件和 `node.data["uin"]` 的访问方式需要根据现有消息段结构调整:如果转发节点类型、字段名不同,请改为匹配当前实现(例如 `type == "forward"` 或使用 `segment.data["sender"]["uin"]` 等)。
4. 如现有代码中 `get_bot_self_id` 的签名不同(例如需要 `provider`、`context` 或其他参数),请相应调整 `monkeypatch.setattr(sender, "get_bot_self_id", ...)` 的 lambda 以匹配实际签名。
</issue_to_address>
### Comment 3
<location path="tests/unit/test_bootstrap_runtime.py" line_range="36-45" />
<code_context>
+def test_bot_self_id_provider_returns_only_a_unique_connected_account(monkeypatch):
</code_context>
<issue_to_address>
**suggestion (testing):** 请增加一个在没有 API 客户端或只有零值 self_id 时的用例,以验证 provider 会返回空字符串。
该测试已经覆盖了单个非零 self_id 和多个 self_id 的场景。为了完整覆盖 `_resolve_bot_self_id`,建议再添加 `_wsr_api_clients` 为空,以及仅包含零值(例如 `{}` 和 `{"0": object()}`)两种情况,并在这两种情况下都断言 provider 返回 `""`,以符合文档中描述的回退行为。
建议实现如下:
```python
from unittest.mock import AsyncMock, MagicMock
import pytest
from nonebot_plugin_saa.bootstrap.runtime import _resolve_bot_self_id
```
```python
def meta(self):
def test_bot_self_id_provider_returns_empty_string_for_missing_or_zero_self_ids(monkeypatch):
"""当不存在已连接 bot 或仅存在零值 self_id 时,应返回空字符串作为回退行为。"""
class FakePlatform:
def __init__(self, self_ids: list[str]):
# 仿照运行时结构,构造 _wsr_api_clients 字典
self._client = SimpleNamespace(
_wsr_api_clients={self_id: object() for self_id in self_ids}
)
def meta(self):
# 假定 _resolve_bot_self_id 只关心 _client._wsr_api_clients,
# 这里返回 self 以满足可能的接口需求
return self
# 情况一:没有任何已连接的 API client
empty_platform = FakePlatform([])
assert _resolve_bot_self_id(empty_platform) == ""
# 情况二:仅存在零值 / 占位 self_id(例如 "0")
zero_only_platform = FakePlatform(["0"])
assert _resolve_bot_self_id(zero_only_platform) == ""
```
上述实现假定:
1. `_resolve_bot_self_id` 定义在 `nonebot_plugin_saa.bootstrap.runtime` 中,并接收平台 / 适配器实例(或一个其 `meta()` / `_client._wsr_api_clients` 结构与真实运行时一致的对象)。
2. `_resolve_bot_self_id` 通过 `platform._client._wsr_api_clients`(或经由 `platform.meta()`)读取映射,并在映射为空或仅包含类似零值的键(例如 `"0"`)时返回 `""`。
为了让这个测试与你的代码库精准对齐:
1. 如果 `_resolve_bot_self_id` 位于其他模块,请调整导入路径。
2. 如果 `_resolve_bot_self_id` 期望的对象结构不同(如 `platform.meta().client._wsr_api_clients`),请相应修改 `FakePlatform.meta()` 以镜像真实的运行时对象结构。
3. 如果你文档中约定的「零值」 self_id 有不同表示方式(例如 `None`、`""` 或数值 `0`),建议扩展测试以覆盖这些表示形式,并同样断言返回值为 `""`。
</issue_to_address>
### Comment 4
<location path="tests/unit/infrastructure/test_paths.py" line_range="8-16" />
<code_context>
-def test_plugin_data_dir_avoids_plugin_local_astrbot_data(monkeypatch):
- plugin_root = paths.PLUGIN_ROOT
- astrbot_root = plugin_root.parents[2]
+def test_plugin_data_dir_avoids_plugin_local_astrbot_data(monkeypatch, tmp_path):
+ """本地插件目录运行时,数据应回落到实际 AstrBot 根目录。"""
+ astrbot_root = tmp_path / "astrbot"
+ plugin_root = astrbot_root / "data" / "plugins" / paths.PLUGIN_NAME
+ plugin_root.mkdir(parents=True)
plugin_local_data = plugin_root / "data" / "plugin_data"
+ monkeypatch.setattr(paths, "PLUGIN_ROOT", plugin_root)
monkeypatch.setattr(
paths, "_resolve_explicit_astrbot_data_dir", lambda: plugin_local_data
)
</code_context>
<issue_to_address>
**suggestion (testing):** 建议显式断言解析出的数据目录位于构造的 AstrBot 根目录下,以使测试意图更清晰。
使用 `tmp_path` 是一个不错的改进。为了更完整地体现预期行为,建议在断言 `paths.PLUGIN_DATA_DIR` 不等于 `plugin_local_data` 之外,再额外断言 `paths.PLUGIN_DATA_DIR` 位于构造出的 `astrbot_root` 之下。这样可以在测试中同时记录并确保「避免使用插件本地目录」和「回落到 AstrBot 根目录」这两层语义。
</issue_to_address>Sourcery is free for open source - if you like our reviews please consider sharing them ✨
Original comment in English
Hey - I've found 4 issues
Prompt for AI Agents
Please address the comments from this code review:
## Individual Comments
### Comment 1
<location path="bootstrap.py" line_range="282-297" />
<code_context>
)
return None
+ def _resolve_bot_self_id(platform_name: str) -> str:
+ """从已连接的反向 WebSocket 客户端中解析 bot 的 self_id(QQ 号)。
+
+ aiocqhttp 的 CQHttp 实例以 self_id 为键保存已连接的 API 客户端
+ (``_wsr_api_clients``)。只有连接中恰有一个非零 self_id 时才返回,
+ 避免多 bot 平台把其他账号误写入合并转发节点;解析失败时返回空串,
+ 由调用方保留 SDK 的默认值。
+ """
+ client = _resolve_bot_client(platform_name)
+ if client is None:
+ return ""
+ api_clients = getattr(client, "_wsr_api_clients", None)
+ if isinstance(api_clients, dict):
+ self_ids = [
+ str(self_id).strip()
+ for self_id in list(api_clients)
+ if str(self_id or "").strip() not in {"", "0"}
+ ]
+ if len(self_ids) == 1:
+ return self_ids[0]
+ if len(self_ids) > 1:
+ logger.warning(
+ "无法唯一解析 bot self_id,保留合并转发节点默认值: "
</code_context>
<issue_to_address>
**suggestion:** The self_id extraction logic does extra work that could be simplified for clarity and efficiency.
In `_resolve_bot_self_id`, the list comprehension
```python
self_ids = [
str(self_id).strip()
for self_id in list(api_clients)
if str(self_id or "").strip() not in {"", "0"}
]
```
performs redundant `str(...)` calls and an unnecessary `list(api_clients)` conversion. You can iterate directly over `api_clients` (or `api_clients.keys()`), normalize once per key, and then apply the non-empty/non-zero filter. This keeps the uniqueness check logic the same while improving readability and efficiency.
```suggestion
api_clients = getattr(client, "_wsr_api_clients", None)
if isinstance(api_clients, dict):
self_ids: list[str] = []
for self_id in api_clients:
normalized = str(self_id or "").strip()
if normalized and normalized != "0":
self_ids.append(normalized)
if len(self_ids) == 1:
return self_ids[0]
if len(self_ids) > 1:
logger.warning(
"无法唯一解析 bot self_id,保留合并转发节点默认值: "
"platform=%s, candidates=%s",
platform_name,
self_ids,
)
```
</issue_to_address>
### Comment 2
<location path="tests/unit/infrastructure/test_onebot_sender.py" line_range="417-426" />
<code_context>
+@pytest.mark.asyncio
</code_context>
<issue_to_address>
**suggestion (testing):** Add a test where both event.self_id and provider self_id exist to verify precedence
Current tests cover event-only, provider-only, and unknown `self_id` cases, but not the precedence rule where the event value should override the provider value. Please add a test where `context.event.message_obj.self_id` and `get_bot_self_id` return different values, and assert that the forward node’s `uin` uses the event value to protect against future changes that might invert this precedence.
Suggested implementation:
```python
monkeypatch.setattr(sender, "_send_chain", fake_send_chain)
@pytest.mark.asyncio
async def test_onebot_sender_prefers_event_self_id_over_provider(monkeypatch):
"""当事件和 provider 都提供 self_id 时,应优先使用事件中的 bot QQ 号写入转发节点。"""
sender = OneBotMessageSender()
calls: list[tuple[str, list]] = []
async def fake_send_chain(session_id: str, chain: list, **_kwargs):
calls.append((session_id, chain))
return SendResult(ok=True)
# provider 返回的 bot self_id 与事件中的不同,用于验证优先级
monkeypatch.setattr(sender, "_send_chain", fake_send_chain)
monkeypatch.setattr(sender, "get_bot_self_id", lambda context: "222222")
# 构造同时具有 event.self_id 和 provider self_id 的上下文
# 这里假设已有的测试工具/fixture可以设置 event.message_obj.self_id
context = make_dummy_context() # 占位:请替换为项目中实际的上下文构造方式
context.event.message_obj.self_id = "111111"
# 触发构建转发消息的逻辑(占位:请替换为项目中实际的发送/构建调用)
await sender.send(context, "测试带转发节点的消息")
# 断言转发节点的 uin 使用事件中的 self_id,而不是 provider 返回的值
assert calls, "预期 fake_send_chain 被调用以生成转发节点"
forward_nodes = [
segment for segment in calls[0][1]
if getattr(segment, "type", None) == "node"
]
assert forward_nodes, "预期消息链中包含至少一个转发节点"
for node in forward_nodes:
# uin 字段应该等于事件中的 self_id,而不是 get_bot_self_id 返回的值
assert getattr(node, "data", {}).get("uin") == "111111"
```
1. 将 `make_dummy_context()` 替换为项目中实际用于构造上下文/事件的 helper 或 fixture,确保能够设置 `context.event.message_obj.self_id`。
2. 将 `await sender.send(context, "测试带转发节点的消息")` 替换为实际会走到「构建转发节点」逻辑的调用(例如发送包含转发/引用/多段消息的接口)。
3. `forward_nodes` 的筛选条件和 `node.data["uin"]` 的访问方式需要根据现有消息段结构调整:如果转发节点类型、字段名不同,请改为匹配当前实现(例如 `type == "forward"` 或使用 `segment.data["sender"]["uin"]` 等)。
4. 如现有代码中 `get_bot_self_id` 的签名不同(例如需要 `provider`、`context` 或其他参数),请相应调整 `monkeypatch.setattr(sender, "get_bot_self_id", ...)` 的 lambda 以匹配实际签名。
</issue_to_address>
### Comment 3
<location path="tests/unit/test_bootstrap_runtime.py" line_range="36-45" />
<code_context>
+def test_bot_self_id_provider_returns_only_a_unique_connected_account(monkeypatch):
</code_context>
<issue_to_address>
**suggestion (testing):** Add a case where no API clients or only zero-valued self_ids exist to verify provider returns an empty string
This test already covers the single non-zero and multiple self_id scenarios. To fully exercise `_resolve_bot_self_id`, please also add coverage for when `_wsr_api_clients` is empty and when it only contains a zero-like entry (e.g. `{}` and `{"0": object()}`), asserting that the provider returns `""` in both cases, consistent with the documented fallback behavior.
Suggested implementation:
```python
from unittest.mock import AsyncMock, MagicMock
import pytest
from nonebot_plugin_saa.bootstrap.runtime import _resolve_bot_self_id
```
```python
def meta(self):
def test_bot_self_id_provider_returns_empty_string_for_missing_or_zero_self_ids(monkeypatch):
"""当不存在已连接 bot 或仅存在零值 self_id 时,应返回空字符串作为回退行为。"""
class FakePlatform:
def __init__(self, self_ids: list[str]):
# 仿照运行时结构,构造 _wsr_api_clients 字典
self._client = SimpleNamespace(
_wsr_api_clients={self_id: object() for self_id in self_ids}
)
def meta(self):
# 假定 _resolve_bot_self_id 只关心 _client._wsr_api_clients,
# 这里返回 self 以满足可能的接口需求
return self
# 情况一:没有任何已连接的 API client
empty_platform = FakePlatform([])
assert _resolve_bot_self_id(empty_platform) == ""
# 情况二:仅存在零值 / 占位 self_id(例如 "0")
zero_only_platform = FakePlatform(["0"])
assert _resolve_bot_self_id(zero_only_platform) == ""
```
The above implementation assumes:
1. `_resolve_bot_self_id` is defined in `nonebot_plugin_saa.bootstrap.runtime` and accepts the platform/adapter instance (or an object whose `meta()`/`_client._wsr_api_clients` matches the real runtime structure).
2. `_resolve_bot_self_id` reads `platform._client._wsr_api_clients` (or through `platform.meta()`), and returns `""` when the mapping is empty or when only a zero-like key (e.g. `"0"`) is present.
To align this test precisely with your codebase:
1. Adjust the import path for `_resolve_bot_self_id` if it lives in a different module.
2. If `_resolve_bot_self_id` expects a different object shape (e.g. `platform.meta().client._wsr_api_clients`), adapt `FakePlatform.meta()` accordingly to mirror the real runtime object graph.
3. If your documented "zero-like" self_id is represented differently (e.g. `None`, `""`, or numeric `0`), extend the test to include those specific representations and still assert `""` as the fallback return value.
</issue_to_address>
### Comment 4
<location path="tests/unit/infrastructure/test_paths.py" line_range="8-16" />
<code_context>
-def test_plugin_data_dir_avoids_plugin_local_astrbot_data(monkeypatch):
- plugin_root = paths.PLUGIN_ROOT
- astrbot_root = plugin_root.parents[2]
+def test_plugin_data_dir_avoids_plugin_local_astrbot_data(monkeypatch, tmp_path):
+ """本地插件目录运行时,数据应回落到实际 AstrBot 根目录。"""
+ astrbot_root = tmp_path / "astrbot"
+ plugin_root = astrbot_root / "data" / "plugins" / paths.PLUGIN_NAME
+ plugin_root.mkdir(parents=True)
plugin_local_data = plugin_root / "data" / "plugin_data"
+ monkeypatch.setattr(paths, "PLUGIN_ROOT", plugin_root)
monkeypatch.setattr(
paths, "_resolve_explicit_astrbot_data_dir", lambda: plugin_local_data
)
</code_context>
<issue_to_address>
**suggestion (testing):** Assert explicitly that the resolved data dir is under the constructed AstrBot root to make the intent clearer
Using `tmp_path` is a good improvement. To fully capture the intended behavior, please also assert that `paths.PLUGIN_DATA_DIR` is located under the constructed `astrbot_root`, in addition to not being equal to `plugin_local_data`. This will document and enforce both "avoid plugin-local" and "fall back to AstrBot root" semantics in the test.
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
| api_clients = getattr(client, "_wsr_api_clients", None) | ||
| if isinstance(api_clients, dict): | ||
| self_ids = [ | ||
| str(self_id).strip() | ||
| for self_id in list(api_clients) | ||
| if str(self_id or "").strip() not in {"", "0"} | ||
| ] | ||
| if len(self_ids) == 1: | ||
| return self_ids[0] | ||
| if len(self_ids) > 1: | ||
| logger.warning( | ||
| "无法唯一解析 bot self_id,保留合并转发节点默认值: " | ||
| "platform=%s, candidates=%s", | ||
| platform_name, | ||
| self_ids, | ||
| ) |
There was a problem hiding this comment.
suggestion: self_id 提取逻辑做了一些可以简化的额外工作,影响清晰度和效率。
在 _resolve_bot_self_id 中,下列列表推导式
self_ids = [
str(self_id).strip()
for self_id in list(api_clients)
if str(self_id or "").strip() not in {"", "0"}
]包含多余的 str(...) 调用以及不必要的 list(api_clients) 转换。你可以直接遍历 api_clients(或 api_clients.keys()),对每个键只归一化一次,再应用非空 / 非零过滤。这样可以在保持现有唯一性检查逻辑不变的前提下,提高可读性和效率。
| api_clients = getattr(client, "_wsr_api_clients", None) | |
| if isinstance(api_clients, dict): | |
| self_ids = [ | |
| str(self_id).strip() | |
| for self_id in list(api_clients) | |
| if str(self_id or "").strip() not in {"", "0"} | |
| ] | |
| if len(self_ids) == 1: | |
| return self_ids[0] | |
| if len(self_ids) > 1: | |
| logger.warning( | |
| "无法唯一解析 bot self_id,保留合并转发节点默认值: " | |
| "platform=%s, candidates=%s", | |
| platform_name, | |
| self_ids, | |
| ) | |
| api_clients = getattr(client, "_wsr_api_clients", None) | |
| if isinstance(api_clients, dict): | |
| self_ids: list[str] = [] | |
| for self_id in api_clients: | |
| normalized = str(self_id or "").strip() | |
| if normalized and normalized != "0": | |
| self_ids.append(normalized) | |
| if len(self_ids) == 1: | |
| return self_ids[0] | |
| if len(self_ids) > 1: | |
| logger.warning( | |
| "无法唯一解析 bot self_id,保留合并转发节点默认值: " | |
| "platform=%s, candidates=%s", | |
| platform_name, | |
| self_ids, | |
| ) |
Original comment in English
suggestion: The self_id extraction logic does extra work that could be simplified for clarity and efficiency.
In _resolve_bot_self_id, the list comprehension
self_ids = [
str(self_id).strip()
for self_id in list(api_clients)
if str(self_id or "").strip() not in {"", "0"}
]performs redundant str(...) calls and an unnecessary list(api_clients) conversion. You can iterate directly over api_clients (or api_clients.keys()), normalize once per key, and then apply the non-empty/non-zero filter. This keeps the uniqueness check logic the same while improving readability and efficiency.
| api_clients = getattr(client, "_wsr_api_clients", None) | |
| if isinstance(api_clients, dict): | |
| self_ids = [ | |
| str(self_id).strip() | |
| for self_id in list(api_clients) | |
| if str(self_id or "").strip() not in {"", "0"} | |
| ] | |
| if len(self_ids) == 1: | |
| return self_ids[0] | |
| if len(self_ids) > 1: | |
| logger.warning( | |
| "无法唯一解析 bot self_id,保留合并转发节点默认值: " | |
| "platform=%s, candidates=%s", | |
| platform_name, | |
| self_ids, | |
| ) | |
| api_clients = getattr(client, "_wsr_api_clients", None) | |
| if isinstance(api_clients, dict): | |
| self_ids: list[str] = [] | |
| for self_id in api_clients: | |
| normalized = str(self_id or "").strip() | |
| if normalized and normalized != "0": | |
| self_ids.append(normalized) | |
| if len(self_ids) == 1: | |
| return self_ids[0] | |
| if len(self_ids) > 1: | |
| logger.warning( | |
| "无法唯一解析 bot self_id,保留合并转发节点默认值: " | |
| "platform=%s, candidates=%s", | |
| platform_name, | |
| self_ids, | |
| ) |
| @pytest.mark.asyncio | ||
| async def test_onebot_sender_uses_event_self_id_for_forward_nodes(monkeypatch): | ||
| """命令响应场景应把事件中的 bot QQ 号写入转发节点。""" | ||
| sender = OneBotMessageSender() | ||
| calls: list[tuple[str, list]] = [] | ||
|
|
||
| async def fake_send_chain(session_id: str, chain: list, **_kwargs): | ||
| calls.append((session_id, chain)) | ||
| return SendResult(ok=True) | ||
|
|
There was a problem hiding this comment.
suggestion (testing): 请增加一个同时存在 event.self_id 和 provider self_id 的测试用例,以验证优先级。
当前测试已经覆盖了仅事件提供 self_id、仅 provider 提供 self_id,以及未知 self_id 的情况,但还缺少验证「事件值应覆盖 provider 值」这一优先级规则的用例。建议新增一个用例,在其中 context.event.message_obj.self_id 与 get_bot_self_id 返回值不同,并断言转发节点的 uin 使用事件中的值,以防止将来修改不小心颠倒这个优先级。
建议实现如下:
monkeypatch.setattr(sender, "_send_chain", fake_send_chain)
@pytest.mark.asyncio
async def test_onebot_sender_prefers_event_self_id_over_provider(monkeypatch):
"""当事件和 provider 都提供 self_id 时,应优先使用事件中的 bot QQ 号写入转发节点。"""
sender = OneBotMessageSender()
calls: list[tuple[str, list]] = []
async def fake_send_chain(session_id: str, chain: list, **_kwargs):
calls.append((session_id, chain))
return SendResult(ok=True)
# provider 返回的 bot self_id 与事件中的不同,用于验证优先级
monkeypatch.setattr(sender, "_send_chain", fake_send_chain)
monkeypatch.setattr(sender, "get_bot_self_id", lambda context: "222222")
# 构造同时具有 event.self_id 和 provider self_id 的上下文
# 这里假设已有的测试工具/fixture可以设置 event.message_obj.self_id
context = make_dummy_context() # 占位:请替换为项目中实际的上下文构造方式
context.event.message_obj.self_id = "111111"
# 触发构建转发消息的逻辑(占位:请替换为项目中实际的发送/构建调用)
await sender.send(context, "测试带转发节点的消息")
# 断言转发节点的 uin 使用事件中的 self_id,而不是 provider 返回的值
assert calls, "预期 fake_send_chain 被调用以生成转发节点"
forward_nodes = [
segment for segment in calls[0][1]
if getattr(segment, "type", None) == "node"
]
assert forward_nodes, "预期消息链中包含至少一个转发节点"
for node in forward_nodes:
# uin 字段应该等于事件中的 self_id,而不是 get_bot_self_id 返回的值
assert getattr(node, "data", {}).get("uin") == "111111"- 将
make_dummy_context()替换为项目中实际用于构造上下文 / 事件的 helper 或 fixture,确保能够设置context.event.message_obj.self_id。 - 将
await sender.send(context, "测试带转发节点的消息")替换为实际会走到「构建转发节点」逻辑的调用(例如发送包含转发 / 引用 / 多段消息的接口)。 forward_nodes的筛选条件和node.data["uin"]的访问方式需要根据现有消息段结构调整:如果转发节点类型、字段名不同,请改为匹配当前实现(例如type == "forward"或使用segment.data["sender"]["uin"]等)。- 如现有代码中
get_bot_self_id的签名不同(例如需要provider、context或其他参数),请相应调整monkeypatch.setattr(sender, "get_bot_self_id", ...)的 lambda 以匹配实际签名。
Original comment in English
suggestion (testing): Add a test where both event.self_id and provider self_id exist to verify precedence
Current tests cover event-only, provider-only, and unknown self_id cases, but not the precedence rule where the event value should override the provider value. Please add a test where context.event.message_obj.self_id and get_bot_self_id return different values, and assert that the forward node’s uin uses the event value to protect against future changes that might invert this precedence.
Suggested implementation:
monkeypatch.setattr(sender, "_send_chain", fake_send_chain)
@pytest.mark.asyncio
async def test_onebot_sender_prefers_event_self_id_over_provider(monkeypatch):
"""当事件和 provider 都提供 self_id 时,应优先使用事件中的 bot QQ 号写入转发节点。"""
sender = OneBotMessageSender()
calls: list[tuple[str, list]] = []
async def fake_send_chain(session_id: str, chain: list, **_kwargs):
calls.append((session_id, chain))
return SendResult(ok=True)
# provider 返回的 bot self_id 与事件中的不同,用于验证优先级
monkeypatch.setattr(sender, "_send_chain", fake_send_chain)
monkeypatch.setattr(sender, "get_bot_self_id", lambda context: "222222")
# 构造同时具有 event.self_id 和 provider self_id 的上下文
# 这里假设已有的测试工具/fixture可以设置 event.message_obj.self_id
context = make_dummy_context() # 占位:请替换为项目中实际的上下文构造方式
context.event.message_obj.self_id = "111111"
# 触发构建转发消息的逻辑(占位:请替换为项目中实际的发送/构建调用)
await sender.send(context, "测试带转发节点的消息")
# 断言转发节点的 uin 使用事件中的 self_id,而不是 provider 返回的值
assert calls, "预期 fake_send_chain 被调用以生成转发节点"
forward_nodes = [
segment for segment in calls[0][1]
if getattr(segment, "type", None) == "node"
]
assert forward_nodes, "预期消息链中包含至少一个转发节点"
for node in forward_nodes:
# uin 字段应该等于事件中的 self_id,而不是 get_bot_self_id 返回的值
assert getattr(node, "data", {}).get("uin") == "111111"- 将
make_dummy_context()替换为项目中实际用于构造上下文/事件的 helper 或 fixture,确保能够设置context.event.message_obj.self_id。 - 将
await sender.send(context, "测试带转发节点的消息")替换为实际会走到「构建转发节点」逻辑的调用(例如发送包含转发/引用/多段消息的接口)。 forward_nodes的筛选条件和node.data["uin"]的访问方式需要根据现有消息段结构调整:如果转发节点类型、字段名不同,请改为匹配当前实现(例如type == "forward"或使用segment.data["sender"]["uin"]等)。- 如现有代码中
get_bot_self_id的签名不同(例如需要provider、context或其他参数),请相应调整monkeypatch.setattr(sender, "get_bot_self_id", ...)的 lambda 以匹配实际签名。
| def test_bot_self_id_provider_returns_only_a_unique_connected_account(monkeypatch): | ||
| """多 bot 连接时不应把任意账号误用为合并转发节点 UIN。""" | ||
| providers: dict[str, object] = {} | ||
|
|
||
| class FakePlatform: | ||
| def __init__(self, self_ids: list[str]): | ||
| self._client = SimpleNamespace( | ||
| _wsr_api_clients={self_id: object() for self_id in self_ids} | ||
| ) | ||
|
|
There was a problem hiding this comment.
suggestion (testing): 请增加一个在没有 API 客户端或只有零值 self_id 时的用例,以验证 provider 会返回空字符串。
该测试已经覆盖了单个非零 self_id 和多个 self_id 的场景。为了完整覆盖 _resolve_bot_self_id,建议再添加 _wsr_api_clients 为空,以及仅包含零值(例如 {} 和 {"0": object()})两种情况,并在这两种情况下都断言 provider 返回 "",以符合文档中描述的回退行为。
建议实现如下:
from unittest.mock import AsyncMock, MagicMock
import pytest
from nonebot_plugin_saa.bootstrap.runtime import _resolve_bot_self_id def meta(self):
def test_bot_self_id_provider_returns_empty_string_for_missing_or_zero_self_ids(monkeypatch):
"""当不存在已连接 bot 或仅存在零值 self_id 时,应返回空字符串作为回退行为。"""
class FakePlatform:
def __init__(self, self_ids: list[str]):
# 仿照运行时结构,构造 _wsr_api_clients 字典
self._client = SimpleNamespace(
_wsr_api_clients={self_id: object() for self_id in self_ids}
)
def meta(self):
# 假定 _resolve_bot_self_id 只关心 _client._wsr_api_clients,
# 这里返回 self 以满足可能的接口需求
return self
# 情况一:没有任何已连接的 API client
empty_platform = FakePlatform([])
assert _resolve_bot_self_id(empty_platform) == ""
# 情况二:仅存在零值 / 占位 self_id(例如 "0")
zero_only_platform = FakePlatform(["0"])
assert _resolve_bot_self_id(zero_only_platform) == ""上述实现假定:
_resolve_bot_self_id定义在nonebot_plugin_saa.bootstrap.runtime中,并接收平台 / 适配器实例(或一个其meta()/_client._wsr_api_clients结构与真实运行时一致的对象)。_resolve_bot_self_id通过platform._client._wsr_api_clients(或经由platform.meta())读取映射,并在映射为空或仅包含类似零值的键(例如"0")时返回""。
为了让这个测试与你的代码库精准对齐:
- 如果
_resolve_bot_self_id位于其他模块,请调整导入路径。 - 如果
_resolve_bot_self_id期望的对象结构不同(如platform.meta().client._wsr_api_clients),请相应修改FakePlatform.meta()以镜像真实的运行时对象结构。 - 如果你文档中约定的「零值」 self_id 有不同表示方式(例如
None、""或数值0),建议扩展测试以覆盖这些表示形式,并同样断言返回值为""。
Original comment in English
suggestion (testing): Add a case where no API clients or only zero-valued self_ids exist to verify provider returns an empty string
This test already covers the single non-zero and multiple self_id scenarios. To fully exercise _resolve_bot_self_id, please also add coverage for when _wsr_api_clients is empty and when it only contains a zero-like entry (e.g. {} and {"0": object()}), asserting that the provider returns "" in both cases, consistent with the documented fallback behavior.
Suggested implementation:
from unittest.mock import AsyncMock, MagicMock
import pytest
from nonebot_plugin_saa.bootstrap.runtime import _resolve_bot_self_id def meta(self):
def test_bot_self_id_provider_returns_empty_string_for_missing_or_zero_self_ids(monkeypatch):
"""当不存在已连接 bot 或仅存在零值 self_id 时,应返回空字符串作为回退行为。"""
class FakePlatform:
def __init__(self, self_ids: list[str]):
# 仿照运行时结构,构造 _wsr_api_clients 字典
self._client = SimpleNamespace(
_wsr_api_clients={self_id: object() for self_id in self_ids}
)
def meta(self):
# 假定 _resolve_bot_self_id 只关心 _client._wsr_api_clients,
# 这里返回 self 以满足可能的接口需求
return self
# 情况一:没有任何已连接的 API client
empty_platform = FakePlatform([])
assert _resolve_bot_self_id(empty_platform) == ""
# 情况二:仅存在零值 / 占位 self_id(例如 "0")
zero_only_platform = FakePlatform(["0"])
assert _resolve_bot_self_id(zero_only_platform) == ""The above implementation assumes:
_resolve_bot_self_idis defined innonebot_plugin_saa.bootstrap.runtimeand accepts the platform/adapter instance (or an object whosemeta()/_client._wsr_api_clientsmatches the real runtime structure)._resolve_bot_self_idreadsplatform._client._wsr_api_clients(or throughplatform.meta()), and returns""when the mapping is empty or when only a zero-like key (e.g."0") is present.
To align this test precisely with your codebase:
- Adjust the import path for
_resolve_bot_self_idif it lives in a different module. - If
_resolve_bot_self_idexpects a different object shape (e.g.platform.meta().client._wsr_api_clients), adaptFakePlatform.meta()accordingly to mirror the real runtime object graph. - If your documented "zero-like" self_id is represented differently (e.g.
None,"", or numeric0), extend the test to include those specific representations and still assert""as the fallback return value.
| def test_plugin_data_dir_avoids_plugin_local_astrbot_data(monkeypatch, tmp_path): | ||
| """本地插件目录运行时,数据应回落到实际 AstrBot 根目录。""" | ||
| astrbot_root = tmp_path / "astrbot" | ||
| plugin_root = astrbot_root / "data" / "plugins" / paths.PLUGIN_NAME | ||
| plugin_root.mkdir(parents=True) | ||
| plugin_local_data = plugin_root / "data" / "plugin_data" | ||
|
|
||
| monkeypatch.setattr(paths, "PLUGIN_ROOT", plugin_root) | ||
| monkeypatch.setattr( |
There was a problem hiding this comment.
suggestion (testing): 建议显式断言解析出的数据目录位于构造的 AstrBot 根目录下,以使测试意图更清晰。
使用 tmp_path 是一个不错的改进。为了更完整地体现预期行为,建议在断言 paths.PLUGIN_DATA_DIR 不等于 plugin_local_data 之外,再额外断言 paths.PLUGIN_DATA_DIR 位于构造出的 astrbot_root 之下。这样可以在测试中同时记录并确保「避免使用插件本地目录」和「回落到 AstrBot 根目录」这两层语义。
Original comment in English
suggestion (testing): Assert explicitly that the resolved data dir is under the constructed AstrBot root to make the intent clearer
Using tmp_path is a good improvement. To fully capture the intended behavior, please also assert that paths.PLUGIN_DATA_DIR is located under the constructed astrbot_root, in addition to not being equal to plugin_local_data. This will document and enforce both "avoid plugin-local" and "fall back to AstrBot root" semantics in the test.
There was a problem hiding this comment.
🧹 Nitpick comments (1)
bootstrap.py (1)
271-301: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win为
_resolve_bot_self_id补充与_resolve_bot_client一致的异常保护。
_resolve_bot_self_id直接访问 aiocqhttp 内部私有属性_wsr_api_clients并做列表推导,但没有像同级的_resolve_bot_client(Lines 251-269)那样用try/except包裹。一旦该属性的内部结构在未来版本中发生非预期变化(例如键类型变化导致str()转换异常),异常会一路传播到send_to_user的 try 块中,被外层except Exception捕获,导致本可成功的发送被误判为失败,而不是安全回退到空字符串。
_wsr_api_clients属于 aiocqhttp 的私有实现细节(未出现在官方文档 API 中),建议同时为该依赖增加与_resolve_bot_client一致的防御性处理。🛡️ 建议修复
def _resolve_bot_self_id(platform_name: str) -> str: """从已连接的反向 WebSocket 客户端中解析 bot 的 self_id(QQ 号)。 ... """ - client = _resolve_bot_client(platform_name) - if client is None: - return "" - api_clients = getattr(client, "_wsr_api_clients", None) - if isinstance(api_clients, dict): - self_ids = [ - str(self_id).strip() - for self_id in list(api_clients) - if str(self_id or "").strip() not in {"", "0"} - ] - if len(self_ids) == 1: - return self_ids[0] - if len(self_ids) > 1: - logger.warning( - "无法唯一解析 bot self_id,保留合并转发节点默认值: " - "platform=%s, candidates=%s", - platform_name, - self_ids, - ) - return "" + try: + client = _resolve_bot_client(platform_name) + if client is None: + return "" + api_clients = getattr(client, "_wsr_api_clients", None) + if isinstance(api_clients, dict): + self_ids = [ + str(self_id).strip() + for self_id in list(api_clients) + if str(self_id or "").strip() not in {"", "0"} + ] + if len(self_ids) == 1: + return self_ids[0] + if len(self_ids) > 1: + logger.warning( + "无法唯一解析 bot self_id,保留合并转发节点默认值: " + "platform=%s, candidates=%s", + platform_name, + self_ids, + ) + except Exception as exc: + logger.debug( + "解析 bot self_id 失败: platform=%s, err=%s", platform_name, exc + ) + return ""🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@bootstrap.py` around lines 271 - 301, 为 _resolve_bot_self_id 增加与 _resolve_bot_client 一致的 try/except 防御,将访问 _wsr_api_clients、类型检查及 self_ids 列表推导包裹其中;发生任何非预期异常时记录适当日志并返回空字符串,确保解析失败不会向 send_to_user 传播异常。
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@bootstrap.py`:
- Around line 271-301: 为 _resolve_bot_self_id 增加与 _resolve_bot_client 一致的
try/except 防御,将访问 _wsr_api_clients、类型检查及 self_ids
列表推导包裹其中;发生任何非预期异常时记录适当日志并返回空字符串,确保解析失败不会向 send_to_user 传播异常。
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 333ded1e-8711-4781-8701-515267a210f7
📒 Files selected for processing (9)
CHANGELOG.mdbootstrap.pydocs/project/platforms.mdmetadata.yamlsrc/infrastructure/messaging/senders/onebot_sender.pysrc/infrastructure/messaging/senders/types.pytests/unit/infrastructure/test_onebot_sender.pytests/unit/infrastructure/test_paths.pytests/unit/test_bootstrap_runtime.py
发布 v2.1.3:将 PR #93 的 OneBot 合并转发节点 UIN 修复安全移植到当前 master(v2.1.2)上。\n\n### 内容\n- 合并转发节点在可可靠识别时写入 bot self_id,修复新版 NapCat / Lagrange 的 retcode=1200。\n- 主动推送仅在 bot 账号唯一时使用解析结果;多 bot 或未知场景保留 SDK 默认值,避免写错账号。\n- 更新 CHANGELOG.md 与 metadata.yaml 至 v2.1.3。\n- 修正数据目录测试对普通 checkout 路径的硬编码,使其可在 Git worktree 中稳定运行。\n\n### 验证\n- pytest tests/unit -q:724 passed, 2 skipped\n- pre-commit run --all-files:passed\n- 仓库体积:约 9 MB,低于 16 MB 上限\n\n- [x] 这不是破坏性变更。
Summary by Sourcery
发布 v2.1.3,改进 OneBot 转发消息处理,并在 QQ 平台上实现更安全的机器人账号解析。
Bug 修复:
self_id,防止在新版 NapCat/Lagrange 实现上出现retcode=1200错误。self_id时,回退为保留 SDK 默认 UIN,避免在多机器人或未知场景中出现错误归属。功能改进:
self_id,用于主动推送场景,并接入到 bootstrap 和 OneBot sender 中。文档:
测试:
self_id情况下的self_id解析。self_idprovider 才会返回值。杂项:
Original summary in English
Summary by Sourcery
Release v2.1.3 with improved OneBot forward-message handling and safer bot account resolution for QQ platforms.
Bug Fixes:
Enhancements:
Documentation:
Tests:
Chores:
Summary by CodeRabbit
新功能
问题修复
文档
版本