fix(settings): detect Telegram group chat IDs via my_chat_member

Extend getUpdates parsing for group joins and update settings UI hints so Chat ID detection works for supergroups, not only private DMs.
This commit is contained in:
Nexus Agent
2026-06-13 02:04:59 +08:00
parent 821cf12ade
commit 40cba6a567
4 changed files with 152 additions and 26 deletions
@@ -0,0 +1,29 @@
# Telegram Chat ID 群聊检测修复
**日期**2026-06-13
## 变更摘要
修复设置页「检测 Chat ID」无法识别群聊的问题:扩展 `getUpdates` 解析,覆盖 Bot 被拉入群时的 `my_chat_member` 等更新类型;更新前端操作说明。
## 动机
用户反馈群聊 Chat ID 检测不到。原实现仅从 `message` / `channel_post``chat`,而群聊常见首条可见更新为 `my_chat_member`Bot 入群),导致列表为空。
## 涉及文件
- `server/api/settings.py``_telegram_chat_entry``_extract_chats_from_telegram_updates``_telegram_fetch_chats`limit 100,最多返回 10 条,群组优先)
- `frontend/src/pages/SettingsPage.vue` — 群聊检测说明、空状态文案、类型中文标签
- `tests/test_settings_telegram.py` — 群聊/编辑消息/callback 解析单测
## 迁移 / 重启
无需数据库迁移;部署后端 + 前端后生效。
## 验证方式
```bash
pytest tests/test_settings_telegram.py -q
```
生产:将 Bot 拉入测试群 → 群内发送 `/start@Bot用户名` → 设置页点击「检测 Chat ID」应出现群组项(负 ID)。
+18 -5
View File
@@ -134,7 +134,7 @@
<v-chip v-if="telegramTokenSet && !telegramTokenDraft" size="small" color="success" variant="tonal" class="mb-3">Token 已配置</v-chip>
<v-text-field v-model="settings.telegram_chat_id" label="Chat ID" variant="outlined" density="compact" class="mb-3" />
<p class="text-caption text-medium-emphasis mb-3">
Telegram Bot 发送任意消息点击下方按钮自动检测 Chat ID
私聊 Bot 发送任意消息群聊 Bot 拉入群后发送 <code>/start@你的Bot用户名</code> @提及 Bot点击下方检测
</p>
<div class="d-flex flex-wrap ga-2 mb-4">
<v-btn size="small" variant="tonal" prepend-icon="mdi-radar" @click="detectTelegramChats" :loading="detectingTgChats">检测 Chat ID</v-btn>
@@ -197,6 +197,9 @@
density="compact"
class="mb-3"
/>
<p class="text-caption text-medium-emphasis mb-3">
检测规则与默认 Bot 相同:群聊需拉入专用 Bot 后在群内发送 <code>/start@Bot用户名</code>。
</p>
<div class="d-flex flex-wrap ga-2 mb-2">
<v-btn
size="small"
@@ -450,14 +453,14 @@
<v-card-title>选择 Chat ID</v-card-title>
<v-card-text>
<div v-if="!tgChats.length" class="text-body-2 text-medium-emphasis">
未检测到对话请先在 Telegram Bot 发送一条消息然后重新检测
未检测到对话私聊请向 Bot 发消息群聊请将 Bot 拉入群后在群内发送 <code>/start@Bot用户名</code>然后重新检测
</div>
<v-list v-else density="compact" lines="two">
<v-list-item
v-for="chat in tgChats"
:key="chat.id"
:title="chat.title || chat.username || `Chat ${chat.id}`"
:subtitle="`${chat.type}${chat.username ? ' · @' + chat.username : ''} · ID ${chat.id}`"
:subtitle="`${formatTelegramChatType(chat.type)}${chat.username ? ' · @' + chat.username : ''} · ID ${chat.id}`"
@click="applyTelegramChat(chat)"
/>
</v-list>
@@ -572,6 +575,16 @@ const notifyToggleGroups = computed(() => {
return groups
})
function formatTelegramChatType(type: string): string {
const labels: Record<string, string> = {
private: '私聊',
group: '群组',
supergroup: '群组',
channel: '频道',
}
return labels[type] || type
}
function parseNotifyEnabled(val: unknown): boolean {
const s = String(val ?? 'true').trim().toLowerCase()
return !['false', '0', 'no', 'off'].includes(s)
@@ -836,7 +849,7 @@ async function detectOfflineTelegramChats() {
const res = await http.get<TelegramChatsResponse>('/settings/telegram/offline/chats')
tgChats.value = res.chats || []
if (!tgChats.value.length) {
snackbar('未检测到对话,请先向离线专用 Bot 发送一条消息后再试', 'warning')
snackbar('未检测到对话:私聊发消息,或群聊拉入 Bot 发送 /start@Bot用户名', 'warning')
return
}
tgChatTarget.value = 'offline'
@@ -948,7 +961,7 @@ async function detectTelegramChats() {
const res = await http.get<TelegramChatsResponse>('/settings/telegram/chats')
tgChats.value = res.chats || []
if (!tgChats.value.length) {
snackbar('未检测到对话,请先向 Bot 发送一条消息后再试', 'warning')
snackbar('未检测到对话:私聊发消息,或群聊拉入 Bot 发送 /start@Bot用户名', 'warning')
return
}
tgChatTarget.value = 'default'
+57 -21
View File
@@ -1185,6 +1185,56 @@ async def reveal_telegram_token(
return {"key": "telegram_bot_token", "value": value}
def _telegram_chat_entry(chat: dict) -> dict | None:
cid = chat.get("id")
if cid is None:
return None
return {
"id": cid,
"type": chat.get("type", ""),
"title": chat.get("title") or chat.get("first_name") or chat.get("username") or "",
"username": chat.get("username") or "",
}
def _extract_chats_from_telegram_updates(updates: list) -> list[dict]:
"""Extract unique chats from getUpdates (DM, group join, channel, callbacks)."""
seen: set[int] = set()
chats: list[dict] = []
def add(chat: object) -> None:
if not isinstance(chat, dict):
return
entry = _telegram_chat_entry(chat)
if not entry or entry["id"] in seen:
return
seen.add(entry["id"])
chats.append(entry)
for update in updates:
if not isinstance(update, dict):
continue
for key in ("message", "edited_message", "channel_post", "edited_channel_post"):
msg = update.get(key)
if isinstance(msg, dict):
add(msg.get("chat"))
mcm = update.get("my_chat_member")
if isinstance(mcm, dict):
add(mcm.get("chat"))
cm = update.get("chat_member")
if isinstance(cm, dict):
add(cm.get("chat"))
cq = update.get("callback_query")
if isinstance(cq, dict):
msg = cq.get("message")
if isinstance(msg, dict):
add(msg.get("chat"))
type_order = {"supergroup": 0, "group": 1, "channel": 2, "private": 3}
chats.sort(key=lambda c: (type_order.get(c.get("type", ""), 9), c.get("id", 0)))
return chats
async def _telegram_fetch_chats(token: str) -> dict:
import httpx
from server.infrastructure.telegram import TELEGRAM_API_BASE
@@ -1195,7 +1245,7 @@ async def _telegram_fetch_chats(token: str) -> dict:
await _telegram_clear_webhook_if_set(client, base_url)
resp = await client.get(
f"{base_url}/getUpdates",
params={"limit": 50},
params={"limit": 100},
)
if resp.status_code != 200:
raise HTTPException(status_code=502, detail=f"Telegram API 返回 {resp.status_code}")
@@ -1208,23 +1258,7 @@ async def _telegram_fetch_chats(token: str) -> dict:
except Exception as e:
raise HTTPException(status_code=502, detail=f"请求失败: {e}") from e
seen: set[int] = set()
chats: list[dict] = []
for update in data.get("result", []):
msg = update.get("message") or update.get("channel_post") or {}
chat = msg.get("chat", {})
cid = chat.get("id")
if cid and cid not in seen:
seen.add(cid)
chats.append({
"id": cid,
"type": chat.get("type", ""),
"title": chat.get("title") or chat.get("first_name") or "",
"username": chat.get("username") or "",
})
if len(chats) >= 5:
break
chats = _extract_chats_from_telegram_updates(data.get("result", []))[:10]
return {"chats": chats, "count": len(chats)}
@@ -1234,8 +1268,10 @@ async def telegram_get_chats(
):
"""Call getUpdates to detect recent chats and extract unique chat_ids.
User must send at least one message to the Bot before calling this.
Returns up to 5 distinct chats from recent messages.
For private chats: send any message to the Bot.
For groups: add the Bot to the group, then send /start@BotName (privacy mode)
or any message mentioning the Bot. Bot join events (my_chat_member) are included.
Returns up to 10 distinct chats, groups listed first.
"""
from server.config import settings as _settings
@@ -1336,7 +1372,7 @@ async def reveal_offline_telegram_token(
async def telegram_offline_get_chats(
admin: Admin = Depends(get_current_admin),
):
"""Detect chats for offline-dedicated Bot via getUpdates."""
"""Detect chats for offline-dedicated Bot via getUpdates (same rules as /telegram/chats)."""
token = _offline_telegram_token_or_400()
return await _telegram_fetch_chats(token)
+48
View File
@@ -134,6 +134,54 @@ async def test_telegram_chats_deletes_webhook_before_get_updates():
assert call_args[1]["params"]["drop_pending_updates"] == "false"
@pytest.mark.asyncio
async def test_telegram_chats_includes_my_chat_member_supergroup():
from server.api.settings import _extract_chats_from_telegram_updates
updates = [
{
"my_chat_member": {
"chat": {"id": -1001234567890, "title": "Ops Group", "type": "supergroup"},
},
},
{
"message": {
"chat": {"id": 42, "type": "private", "first_name": "Alice"},
},
},
]
chats = _extract_chats_from_telegram_updates(updates)
assert len(chats) == 2
assert chats[0]["id"] == -1001234567890
assert chats[0]["type"] == "supergroup"
assert chats[0]["title"] == "Ops Group"
assert chats[1]["id"] == 42
@pytest.mark.asyncio
async def test_telegram_chats_includes_edited_message_and_callback():
from server.api.settings import _extract_chats_from_telegram_updates
updates = [
{
"edited_message": {
"chat": {"id": -10099, "title": "Alert Room", "type": "group"},
},
},
{
"callback_query": {
"message": {
"chat": {"id": -10088, "title": "Channel", "type": "channel"},
},
},
},
]
chats = _extract_chats_from_telegram_updates(updates)
ids = {c["id"] for c in chats}
assert -10099 in ids
assert -10088 in ids
@pytest.mark.asyncio
async def test_notify_toggle_put_hot_reload():
from server.api.schemas import SettingUpdatePayload