feat(platform): 新增企业微信消息推送机器人（原群机器人）适配器支持 #3858
Open
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
![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 - 我已经审查了你的修改，这里是一些反馈：
- 在
WecomGroupBotParser._should_parse_as_json中，在不去除空白字符的情况下使用payload.startswith('<')，可能会把前面带有空格/换行的 XML 错误识别为 JSON；建议改用payload.lstrip().startswith('<')（或类似方式），这样在prefer_format为xml时可以更可靠地识别 XML。 - 在
WecomGroupBotEvent.send_streaming中，目前的实现会先缓存整个流并在结束时一次性发送，这违背了流式（streaming）的语义；如果平台可以处理部分更新，建议在生成块时就增量发送，而不是在发送前把所有内容聚合起来。
面向 AI Agent 的提示词
请根据以下代码审查中的评论进行修改：
## 总体评论
- 在 `WecomGroupBotParser._should_parse_as_json` 中，在不去除空白字符的情况下使用 `payload.startswith('<')`，可能会把前面带有空格/换行的 XML 错误识别为 JSON；建议改用 `payload.lstrip().startswith('<')`（或类似方式），这样在 `prefer_format` 为 `xml` 时可以更可靠地识别 XML。
- 在 `WecomGroupBotEvent.send_streaming` 中，目前的实现会先缓存整个流并在结束时一次性发送，这违背了流式（streaming）的语义；如果平台可以处理部分更新，建议在生成块时就增量发送，而不是在发送前把所有内容聚合起来。
## 单独评论
### 评论 1
<location> `astrbot/core/platform/sources/wecom_group_bot/wecom_group_bot_event.py:64-73` </location>
<code_context>
+
+ await super().send(message)
+
+ async def send_streaming(self, generator, use_fallback: bool = False):
+ buffer = None
+ async for chain in generator:
+ if not buffer:
+ buffer = chain
+ else:
+ buffer.chain.extend(chain.chain)
+ if buffer:
+ await self.send(buffer)
+ return await super().send_streaming(generator, use_fallback)
+
+ def _extract_chat_id(self) -> str:
</code_context>
<issue_to_address>
**issue (bug_risk):** `send_streaming` 的实现会先耗尽 generator，然后仍然调用 `super().send_streaming(generator)`。
这意味着父类的 `send_streaming` 接收到的是一个已经被耗尽的 generator，实际上什么都不会做。如果这个适配器是为了非流式聚合发送，那么在 `await self.send(buffer)` 之后很可能应该直接 `return`，而不是调用 `super()`。如果需要真正的流式行为，这个方法应该在块到达时就向下游转发，而不是先聚合再发送。
</issue_to_address>
### 评论 2
<location> `astrbot/core/platform/sources/wecom_group_bot/wecom_group_bot_adapter.py:151-158` </location>
<code_context>
async def _build_message_components(
self,
msgtype: str,
payload: dict[str, Any],
) -> tuple[list, str]:
components: list = []
message_str = ""
if msgtype == "text":
content = str(payload.get("text", {}).get("content", "")).strip()
message_str = content
components.append(Plain(content))
elif msgtype == "image":
image_url = payload.get("image", {}).get("image_url") or payload.get("image", {}).get("url")
message_str = "[图片]"
if image_url:
components.append(Image(file=image_url, url=image_url))
elif msgtype == "mixed":
items = payload.get("mixed_message", {}).get("msg_item", [])
texts: list[str] = []
for item in items:
item_type = str(item.get("msg_type", "")).lower()
if item_type == "text":
text_content = item.get("text", {}).get("content", "")
texts.append(text_content)
components.append(Plain(text_content))
elif item_type == "image":
image_url = item.get("image", {}).get("image_url")
if image_url:
components.append(Image(file=image_url, url=image_url))
message_str = " ".join(texts)
elif msgtype == "event":
event_type = payload.get("event", {}).get("event_type")
message_str = f"[事件] {event_type}" if event_type else "[事件]"
components.append(Plain(message_str))
elif msgtype == "attachment":
callback_id = payload.get("attachment", {}).get("callback_id")
message_str = f"[按钮回调] {callback_id or ''}".strip()
components.append(Plain(message_str))
else:
message_str = f"[{msgtype}]"
components.append(Plain(message_str))
return components, message_str
</code_context>
<issue_to_address>
**issue (code-quality):** 我们发现了如下问题：
- 将条件逻辑简化为类似 switch 的形式（[`switch`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/switch/)）
- 使用命名表达式来简化赋值和条件判断（[`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/)）
</issue_to_address>
### 评论 3
<location> `astrbot/core/platform/sources/wecom_group_bot/wecom_group_bot_event.py:79-80` </location>
<code_context>
def _extract_chat_id(self) -> str:
raw = self.message_obj.raw_message or {}
if isinstance(raw, dict):
payload: dict[str, Any] = raw.get("payload") or raw
chat_id = payload.get("chatid") or payload.get("chat_id")
if chat_id:
return str(chat_id)
return self.message_obj.session_id or self.session_id or ""
</code_context>
<issue_to_address>
**suggestion (code-quality):** 使用命名表达式来简化赋值和条件判断（[`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/)）
```suggestion
if chat_id := payload.get("chatid") or payload.get("chat_id"):
```
</issue_to_address>
### 评论 4
<location> `astrbot/core/platform/sources/wecom_group_bot/wecom_group_bot_parser.py:17-19` </location>
<code_context>
def _camel_to_snake(value: str) -> str:
if not value:
return value
return _CAMEL_CASE_PATTERN.sub("_", value).lower()
</code_context>
<issue_to_address>
**suggestion (code-quality):** 我们发现了如下问题：
- 在控制流跳转后，将代码提升到对应的 else 分支中（[`reintroduce-else`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/reintroduce-else/)）
- 使用 if 表达式替换 if 语句（[`assign-if-exp`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/assign-if-exp/)）
```suggestion
return value if not value else _CAMEL_CASE_PATTERN.sub("_", value).lower()
```
</issue_to_address>
### 评论 5
<location> `astrbot/core/platform/sources/wecom_group_bot/wecom_group_bot_parser.py:74-76` </location>
<code_context>
def _parse_xml(self, payload: str) -> dict[str, Any]:
root = ET.fromstring(payload)
parsed: dict[str, Any] = {}
for child in root:
parsed[_camel_to_snake(child.tag)] = _element_to_data(child)
return parsed
</code_context>
<issue_to_address>
**suggestion (code-quality):** 将 for 循环转换为字典推导式（[`dict-comprehension`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/dict-comprehension/)）
```suggestion
parsed: dict[str, Any] = {
_camel_to_snake(child.tag): _element_to_data(child) for child in root
}
```
</issue_to_address>帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据反馈改进后续的审查。
Original comment in English
Hey there - I've reviewed your changes - here's some feedback:
- In
WecomGroupBotParser._should_parse_as_json, usingpayload.startswith('<')without stripping whitespace can misclassify XML with leading spaces/newlines as JSON; consider usingpayload.lstrip().startswith('<')(or similar) so XML is reliably detected whenprefer_formatisxml. - In
WecomGroupBotEvent.send_streaming, the implementation buffers the entire stream and sends once at the end, which defeats streaming semantics; if the platform can handle partial updates, consider emitting chunks incrementally instead of aggregating them all before sending.
Prompt for AI Agents
Please address the comments from this code review:
## Overall Comments
- In `WecomGroupBotParser._should_parse_as_json`, using `payload.startswith('<')` without stripping whitespace can misclassify XML with leading spaces/newlines as JSON; consider using `payload.lstrip().startswith('<')` (or similar) so XML is reliably detected when `prefer_format` is `xml`.
- In `WecomGroupBotEvent.send_streaming`, the implementation buffers the entire stream and sends once at the end, which defeats streaming semantics; if the platform can handle partial updates, consider emitting chunks incrementally instead of aggregating them all before sending.
## Individual Comments
### Comment 1
<location> `astrbot/core/platform/sources/wecom_group_bot/wecom_group_bot_event.py:64-73` </location>
<code_context>
+
+ await super().send(message)
+
+ async def send_streaming(self, generator, use_fallback: bool = False):
+ buffer = None
+ async for chain in generator:
+ if not buffer:
+ buffer = chain
+ else:
+ buffer.chain.extend(chain.chain)
+ if buffer:
+ await self.send(buffer)
+ return await super().send_streaming(generator, use_fallback)
+
+ def _extract_chat_id(self) -> str:
</code_context>
<issue_to_address>
**issue (bug_risk):** The `send_streaming` implementation exhausts the generator and then still calls `super().send_streaming(generator)`.
This means the superclass `send_streaming` receives an already-exhausted generator and effectively does nothing. If this adapter is meant to aggregate non-streamingly, it should likely just `return` after `await self.send(buffer)` and not call `super()`. If true streaming is required, this method should instead forward chunks as they arrive rather than aggregating them first.
</issue_to_address>
### Comment 2
<location> `astrbot/core/platform/sources/wecom_group_bot/wecom_group_bot_adapter.py:151-158` </location>
<code_context>
async def _build_message_components(
self,
msgtype: str,
payload: dict[str, Any],
) -> tuple[list, str]:
components: list = []
message_str = ""
if msgtype == "text":
content = str(payload.get("text", {}).get("content", "")).strip()
message_str = content
components.append(Plain(content))
elif msgtype == "image":
image_url = payload.get("image", {}).get("image_url") or payload.get("image", {}).get("url")
message_str = "[图片]"
if image_url:
components.append(Image(file=image_url, url=image_url))
elif msgtype == "mixed":
items = payload.get("mixed_message", {}).get("msg_item", [])
texts: list[str] = []
for item in items:
item_type = str(item.get("msg_type", "")).lower()
if item_type == "text":
text_content = item.get("text", {}).get("content", "")
texts.append(text_content)
components.append(Plain(text_content))
elif item_type == "image":
image_url = item.get("image", {}).get("image_url")
if image_url:
components.append(Image(file=image_url, url=image_url))
message_str = " ".join(texts)
elif msgtype == "event":
event_type = payload.get("event", {}).get("event_type")
message_str = f"[事件] {event_type}" if event_type else "[事件]"
components.append(Plain(message_str))
elif msgtype == "attachment":
callback_id = payload.get("attachment", {}).get("callback_id")
message_str = f"[按钮回调] {callback_id or ''}".strip()
components.append(Plain(message_str))
else:
message_str = f"[{msgtype}]"
components.append(Plain(message_str))
return components, message_str
</code_context>
<issue_to_address>
**issue (code-quality):** We've found these issues:
- Simplify conditional into switch-like form ([`switch`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/switch/))
- Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
</issue_to_address>
### Comment 3
<location> `astrbot/core/platform/sources/wecom_group_bot/wecom_group_bot_event.py:79-80` </location>
<code_context>
def _extract_chat_id(self) -> str:
raw = self.message_obj.raw_message or {}
if isinstance(raw, dict):
payload: dict[str, Any] = raw.get("payload") or raw
chat_id = payload.get("chatid") or payload.get("chat_id")
if chat_id:
return str(chat_id)
return self.message_obj.session_id or self.session_id or ""
</code_context>
<issue_to_address>
**suggestion (code-quality):** Use named expression to simplify assignment and conditional ([`use-named-expression`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/use-named-expression/))
```suggestion
if chat_id := payload.get("chatid") or payload.get("chat_id"):
```
</issue_to_address>
### Comment 4
<location> `astrbot/core/platform/sources/wecom_group_bot/wecom_group_bot_parser.py:17-19` </location>
<code_context>
def _camel_to_snake(value: str) -> str:
if not value:
return value
return _CAMEL_CASE_PATTERN.sub("_", value).lower()
</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/))
```suggestion
return value if not value else _CAMEL_CASE_PATTERN.sub("_", value).lower()
```
</issue_to_address>
### Comment 5
<location> `astrbot/core/platform/sources/wecom_group_bot/wecom_group_bot_parser.py:74-76` </location>
<code_context>
def _parse_xml(self, payload: str) -> dict[str, Any]:
root = ET.fromstring(payload)
parsed: dict[str, Any] = {}
for child in root:
parsed[_camel_to_snake(child.tag)] = _element_to_data(child)
return parsed
</code_context>
<issue_to_address>
**suggestion (code-quality):** Convert for loop into dictionary comprehension ([`dict-comprehension`](https://docs.sourcery.ai/Reference/Default-Rules/refactorings/dict-comprehension/))
```suggestion
parsed: dict[str, Any] = {
_camel_to_snake(child.tag): _element_to_data(child) for child in root
}
```
</issue_to_address>Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.
Author
|
附上成功截图和日志。目前该功能还没全部推送 |
|
看起来这个正在测试的消息推送(原群机器人)和新出智能机器人是一样的通讯方式 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
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.
为 AstrBot 增加企业微信消息推送机器人（原群机器人）适配器，抢先适配，使其能够接收企业微信推送内容并通过 AstrBot 的会话处理与其他平台统一管理。
适配 #3857
Modifications / 改动点
在
astrbot/core/config/default.py中新增wecom_group_bot平台配置模板和相关配置项说明，供仪表盘与配置文件启用。在
astrbot/core/platform/manager.py以及astrbot/core/platform/sources/wecom_ai_bot相关文件中扩展平台路由与消息加解密逻辑，以共享企业微信的安全校验处理。新增
astrbot/core/platform/sources/wecom_group_bot/目录，包含适配器的客户端、事件模型、解析器和服务端实现，实现企业微信回调解析、消息入队与 AstrBot 内部消息格式的转换。为前端配置面板
dashboard/src/utils/platformUtils.js加入新的平台条目和配置映射，让用户可以在仪表盘里开关和设置该适配器。This is NOT a breaking change. / 这不是一个破坏性变更。
Screenshots or Test Results / 运行截图或测试结果
（暂无运行截图，已在本地完成功能联调。）
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.Summary by Sourcery
为 AstrBot 新增一个平台适配器，将企业微信消息推送（群机器人）集成进来，包括服务端回调处理、消息解析和统一事件分发。
New Features:
WecomGroupBot平台适配器，通过 AstrBot 接收和发送企业微信消息推送（群机器人）消息。WecomGroupBot平台新增配置模板和默认值，以便启用和调优该适配器。WecomGroupBot适配器，使其可以在 UI 中被选择和配置。Enhancements:
Original summary in English
Summary by Sourcery
Add a new platform adapter integrating WeCom message push (group) bots into AstrBot, including server-side callback handling, message parsing, and unified event dispatch.
New Features:
Enhancements: