feat: add send_chat_action for Telegram platform adapter

[evpeople]

Add typing/upload indicator when sending messages via Telegram.

• Added _send_chat_action helper method for sending chat actions
• Send appropriate action (typing, upload_photo, upload_document, upload_voice) before sending different message types
• Support streaming mode with typing indicator
• Support supergroup with message_thread_id

[Feature]为Telegram渠道更新 Bot is typing 的状态 #5020

Modifications / 改动点

仅修改 astrbot/core/platform/sources/telegram/tg_event.py 文件，实现在Bot思考时，更新telegram的状态为Typing

• This is NOT a breaking change. / 这不是一个破坏性变更。

Screenshots or Test Results / 运行截图或测试结果

---

Checklist / 检查清单

• 😊 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。/ If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
• 👀 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。/ My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
• 🤓 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到了 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 in requirements.txt and pyproject.toml.
• 😮 我的更改没有引入恶意代码。/ My changes do not introduce malicious code.

Summary by Sourcery

为 Telegram 添加聊天状态（chat action）支持，在发送消息（包括流式响应和超级群组线程中的消息）时显示机器人活动。

新功能：

• 在发送不同类型的消息之前，发送相应的 Telegram 聊天状态（typing、upload_photo、upload_document、upload_voice）。
• 在通过 Telegram 适配器发送流式响应时显示“正在输入”指示。

增强：

• 在通过 Telegram 适配器为超级群组线程消息发送聊天状态时，支持 message_thread_id。

Original summary in English

Summary by Sourcery

Add Telegram chat action support to show bot activity while messages are being sent, including streaming responses and supergroup threads.

New Features:

• Send appropriate Telegram chat actions (typing, upload_photo, upload_document, upload_voice) before sending different message types.
• Display typing indicators during streaming responses from the Telegram adapter.

Enhancements:

• Support message_thread_id when sending chat actions for supergroup thread messages through the Telegram adapter.

[dosubot]

Related Documentation

Checked 1 published document(s) in 1 knowledge base(s). No updates required.

^{How did I do? Any feedback?}  

[sourcery-ai]

Hey - 我发现了 1 个问题，并给出了一些整体性反馈：

• _send_chat_action 辅助函数的类型标注为 action: str，但实际总是传入 ChatAction 的值；建议更新类型标注（例如 ChatAction | str），以更好地反映真实用法并帮助静态检查。
• 在 send_streaming 中，几乎每次编辑和发送媒体之前都会发送 chat action，这可能过于频繁；建议复用现有的 throttle_interval，或者为 chat action 单独引入一个冷却间隔，以避免触发 Telegram 的频率限制。
• 在 send_streaming 里，对图片、文件和语音消息重复出现“发送 upload_* chat action → 发送媒体 → 发送 typing chat action”的模式，可以抽取成一个小的辅助函数，以减少重复并让控制流更易理解。

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The `_send_chat_action` helper is typed with `action: str` but is always passed `ChatAction` values; consider updating the type annotation (e.g., `ChatAction | str`) to better reflect actual usage and aid static checking.
- In `send_streaming`, chat actions are sent before almost every edit and media send, which may be unnecessarily frequent; consider reusing the existing `throttle_interval` or introducing a separate cooldown for chat actions to avoid hitting Telegram rate limits.
- The repeated pattern of 'send upload_* chat action → send media → send typing chat action' for images, files, and voice messages in `send_streaming` could be extracted into a small helper to reduce duplication and make the control flow easier to follow.

## Individual Comments

### Comment 1
<location> `astrbot/core/platform/sources/telegram/tg_event.py:113` </location>
<code_context>
             # it's a supergroup chat with message_thread_id
             user_name, message_thread_id = user_name.split("#")
+
+        # 先确定要发送的 action 类型
+        has_text = any(isinstance(i, Plain) for i in message.chain)
+        has_image = any(isinstance(i, Image) for i in message.chain)
</code_context>

<issue_to_address>
**issue (complexity):** 请考虑为选择 chat action、发送媒体以及保证 typing 状态提取共享的辅助函数，以去除重复并理清控制流。

你可以保持当前新增的行为，只需要通过几个小的辅助函数来降低复杂度和重复度。

### 1. 替换 `send_with_client` 中硬编码的 action 链

与其写 4 次独立的 `any(...)` 检查并手动处理优先级，不如在一个辅助函数中集中管理类型与优先级映射。这样也更方便在未来扩展新的段类型：

```python
# near top of the class/module
ACTION_BY_TYPE: dict[type, str] = {
    Record: ChatAction.UPLOAD_VOICE,
    File: ChatAction.UPLOAD_DOCUMENT,
    Image: ChatAction.UPLOAD_PHOTO,
    Plain: ChatAction.TYPING,
}


@classmethod
def _get_chat_action_for_chain(cls, chain: list[Any]) -> str:
    # priority is defined by the order in ACTION_BY_TYPE
    for seg_type, action in ACTION_BY_TYPE.items():
        if any(isinstance(seg, seg_type) for seg in chain):
            return action
    return ChatAction.TYPING
```

然后在 `send_with_client` 中：

```python
# replace the has_text/has_image/has_file/has_voice block
action = cls._get_chat_action_for_chain(message.chain)
await cls._send_chat_action(client, user_name, action, message_thread_id)
```

这会保留当前的优先级顺序（Record > File > Image > Plain），同时也更容易调整或扩展。

### 2. 抽取 `send_streaming` 中“upload + send + restore typing”的模式

各个媒体分支几乎完全相同。你可以把这个模式集中到一个辅助函数中，同时保持现有行为：

```python
async def _send_media_with_action(
    self,
    upload_action: str,
    send_coro: Callable[..., Awaitable[Any]],
    *,
    user_name: str,
    message_thread_id: str | None,
    **payload: Any,
) -> None:
    await self._send_chat_action(self.client, user_name, upload_action, message_thread_id)
    await send_coro(**payload)
    await self._send_chat_action(self.client, user_name, ChatAction.TYPING, message_thread_id)
```

然后媒体相关的分支可以写成：

```python
elif isinstance(i, Image):
    image_path = await i.convert_to_file_path()
    await self._send_media_with_action(
        ChatAction.UPLOAD_PHOTO,
        self.client.send_photo,
        user_name=user_name,
        message_thread_id=message_thread_id,
        photo=image_path,
        **cast(Any, payload),
    )
    continue

elif isinstance(i, File):
    path = await i.get_file()
    name = i.name or os.path.basename(path)
    await self._send_media_with_action(
        ChatAction.UPLOAD_DOCUMENT,
        self.client.send_document,
        user_name=user_name,
        message_thread_id=message_thread_id,
        document=path,
        filename=name,
        **cast(Any, payload),
    )
    continue

elif isinstance(i, Record):
    path = await i.convert_to_file_path()
    await self._send_media_with_action(
        ChatAction.UPLOAD_VOICE,
        self.client.send_voice,
        user_name=user_name,
        message_thread_id=message_thread_id,
        voice=path,
        **cast(Any, payload),
    )
    continue
```

这样可以保持完全一致的顺序：先发送 upload action → 发送媒体 → 恢复为 typing。

### 3. 把“在文本编辑/发送前确保 typing 状态”集中处理

你目前在多处编辑和发送之前调用 `_send_chat_action(..., TYPING, ...)`。可以抽取一个小的辅助函数，让意图更清晰：

```python
async def _ensure_typing(
    self,
    user_name: str,
    message_thread_id: str | None,
) -> None:
    await self._send_chat_action(
        self.client, user_name, ChatAction.TYPING, message_thread_id
    )
```

然后在 `send_streaming` 中：

```python
# initial typing
await self._ensure_typing(user_name, message_thread_id)

# before throttled edit
if time_since_last_edit >= throttle_interval:
    await self._ensure_typing(user_name, message_thread_id)
    try:
        await self.client.edit_message_text(...)
    ...

# before first send
else:
    await self._ensure_typing(user_name, message_thread_id)
    try:
        msg = await self.client.send_message(...)
    ...
```

这会保持所有现有行为，但让控制流不那么分散，更容易理解。
</issue_to_address>

---

Sourcery 对开源项目是免费的——如果你觉得这些 Review 有帮助，欢迎分享 ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{帮我变得更有用！请在每条评论上点 👍 或 👎，我会根据反馈改进 Review 质量。}

Original comment in English

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

• The _send_chat_action helper is typed with action: str but is always passed ChatAction values; consider updating the type annotation (e.g., ChatAction | str) to better reflect actual usage and aid static checking.
• In send_streaming, chat actions are sent before almost every edit and media send, which may be unnecessarily frequent; consider reusing the existing throttle_interval or introducing a separate cooldown for chat actions to avoid hitting Telegram rate limits.
• The repeated pattern of 'send upload_* chat action → send media → send typing chat action' for images, files, and voice messages in send_streaming could be extracted into a small helper to reduce duplication and make the control flow easier to follow.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- The `_send_chat_action` helper is typed with `action: str` but is always passed `ChatAction` values; consider updating the type annotation (e.g., `ChatAction | str`) to better reflect actual usage and aid static checking.
- In `send_streaming`, chat actions are sent before almost every edit and media send, which may be unnecessarily frequent; consider reusing the existing `throttle_interval` or introducing a separate cooldown for chat actions to avoid hitting Telegram rate limits.
- The repeated pattern of 'send upload_* chat action → send media → send typing chat action' for images, files, and voice messages in `send_streaming` could be extracted into a small helper to reduce duplication and make the control flow easier to follow.

## Individual Comments

### Comment 1
<location> `astrbot/core/platform/sources/telegram/tg_event.py:113` </location>
<code_context>
             # it's a supergroup chat with message_thread_id
             user_name, message_thread_id = user_name.split("#")
+
+        # 先确定要发送的 action 类型
+        has_text = any(isinstance(i, Plain) for i in message.chain)
+        has_image = any(isinstance(i, Image) for i in message.chain)
</code_context>

<issue_to_address>
**issue (complexity):** Consider extracting shared helper functions for choosing chat actions, sending media, and ensuring typing status to remove duplication and clarify the control flow.

You can keep the new behavior but reduce complexity and duplication with a couple of small helpers.

### 1. Replace hard‑coded action chain in `send_with_client`

Instead of 4 separate `any(...)` checks and manual priority, centralize the mapping and priority in a helper. This makes it easier to extend with new segment types:

```python
# near top of the class/module
ACTION_BY_TYPE: dict[type, str] = {
    Record: ChatAction.UPLOAD_VOICE,
    File: ChatAction.UPLOAD_DOCUMENT,
    Image: ChatAction.UPLOAD_PHOTO,
    Plain: ChatAction.TYPING,
}


@classmethod
def _get_chat_action_for_chain(cls, chain: list[Any]) -> str:
    # priority is defined by the order in ACTION_BY_TYPE
    for seg_type, action in ACTION_BY_TYPE.items():
        if any(isinstance(seg, seg_type) for seg in chain):
            return action
    return ChatAction.TYPING
```

Then in `send_with_client`:

```python
# replace the has_text/has_image/has_file/has_voice block
action = cls._get_chat_action_for_chain(message.chain)
await cls._send_chat_action(client, user_name, action, message_thread_id)
```

This preserves the existing priority (Record > File > Image > Plain) but makes it easy to adjust/extend.

### 2. Factor out “upload + send + restore typing” in `send_streaming`

The media branches are nearly identical. You can centralize the pattern and keep the current behavior:

```python
async def _send_media_with_action(
    self,
    upload_action: str,
    send_coro: Callable[..., Awaitable[Any]],
    *,
    user_name: str,
    message_thread_id: str | None,
    **payload: Any,
) -> None:
    await self._send_chat_action(self.client, user_name, upload_action, message_thread_id)
    await send_coro(**payload)
    await self._send_chat_action(self.client, user_name, ChatAction.TYPING, message_thread_id)
```

Then the media branches become:

```python
elif isinstance(i, Image):
    image_path = await i.convert_to_file_path()
    await self._send_media_with_action(
        ChatAction.UPLOAD_PHOTO,
        self.client.send_photo,
        user_name=user_name,
        message_thread_id=message_thread_id,
        photo=image_path,
        **cast(Any, payload),
    )
    continue

elif isinstance(i, File):
    path = await i.get_file()
    name = i.name or os.path.basename(path)
    await self._send_media_with_action(
        ChatAction.UPLOAD_DOCUMENT,
        self.client.send_document,
        user_name=user_name,
        message_thread_id=message_thread_id,
        document=path,
        filename=name,
        **cast(Any, payload),
    )
    continue

elif isinstance(i, Record):
    path = await i.convert_to_file_path()
    await self._send_media_with_action(
        ChatAction.UPLOAD_VOICE,
        self.client.send_voice,
        user_name=user_name,
        message_thread_id=message_thread_id,
        voice=path,
        **cast(Any, payload),
    )
    continue
```

This keeps the exact sequence: upload action → send media → restore typing.

### 3. Centralize “ensure typing before text edit/send”

You currently call `_send_chat_action(..., TYPING, ...)` multiple times before edits and sends. Extract a small helper to make intent explicit:

```python
async def _ensure_typing(
    self,
    user_name: str,
    message_thread_id: str | None,
) -> None:
    await self._send_chat_action(
        self.client, user_name, ChatAction.TYPING, message_thread_id
    )
```

Then in `send_streaming`:

```python
# initial typing
await self._ensure_typing(user_name, message_thread_id)

# before throttled edit
if time_since_last_edit >= throttle_interval:
    await self._ensure_typing(user_name, message_thread_id)
    try:
        await self.client.edit_message_text(...)
    ...

# before first send
else:
    await self._ensure_typing(user_name, message_thread_id)
    try:
        msg = await self.client.send_message(...)
    ...
```

This keeps all behavior but makes the control flow less scattered and easier to understand.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}