如何在aiogram中实现Telegram机器人被加群时自动获取group_id

现有Telegram机器人代码如下，当用户私聊发送图片时，机器人会将图片转发至chat_ids列表中的群组（转发前检查机器人是否为群管理员）。但chat_ids需手动填充，无法实现机器人被加入任意群组时自动追加对应group_id至列表。目标是：用户私聊发送任意图片后，机器人将图片转发至所有其为管理员的群组。

bot = Bot(token=TOKEN)
dp = Dispatcher()
chat_ids = []

async def check_if_bot_is_admin(chat_id):
    member = await bot.get_chat_member(chat_id, bot.id)
    resp = member.status
    return resp

@dp.message(F.photo)
async def handle_photo(message: types.Message):
    for group_id in chat_ids:
        is_admin = await check_if_bot_is_admin(group_id)
        if message.chat.type == 'private' and is_admin:
            await message.copy_to(chat_id=group_id)
async def main():
    await dp.start_polling(bot)

1. 核心思路 #

通过监听机器人的群组状态变化事件，自动维护群组ID列表；同时将列表持久化存储，避免重启机器人后丢失数据。处理图片转发时，遍历列表并验证权限后执行转发操作。

2. 修改后的完整代码 #

from aiogram import Bot, Dispatcher, F, types
from aiogram.types import ChatMemberUpdated
import json

TOKEN = "你的机器人令牌"
bot = Bot(token=TOKEN)
dp = Dispatcher()
CHAT_IDS_FILE = "chat_ids.json"

# 加载已存储的群组ID
def load_chat_ids():
    try:
        with open(CHAT_IDS_FILE, "r") as f:
            return json.load(f)
    except FileNotFoundError:
        return []

# 保存群组ID到文件
def save_chat_ids(chat_ids):
    with open(CHAT_IDS_FILE, "w") as f:
        json.dump(chat_ids, f)

chat_ids = load_chat_ids()

async def check_if_bot_is_admin(chat_id):
    member = await bot.get_chat_member(chat_id, bot.id)
    # 覆盖群主和管理员两种权限状态
    return member.status in ["administrator", "creator"]

@dp.my_chat_member()
async def handle_chat_member_update(event: ChatMemberUpdated):
    global chat_ids
    chat_id = event.chat.id
    # 仅处理群组/超级群的状态变化
    if event.chat.type in ["group", "supergroup"]:
        # 机器人被加入群组时，添加ID到列表
        if event.new_chat_member.status in ["member", "administrator", "creator"]:
            if chat_id not in chat_ids:
                chat_ids.append(chat_id)
                save_chat_ids(chat_ids)
        # 机器人被踢出/离开群组时，移除对应ID
        elif event.new_chat_member.status in ["kicked", "left"]:
            if chat_id in chat_ids:
                chat_ids.remove(chat_id)
                save_chat_ids(chat_ids)

@dp.message(F.photo)
async def handle_photo(message: types.Message):
    if message.chat.type != 'private':
        return
    # 遍历列表副本，避免修改原列表引发遍历异常
    for group_id in chat_ids.copy():
        try:
            is_admin = await check_if_bot_is_admin(group_id)
            if is_admin:
                await message.copy_to(chat_id=group_id)
        except Exception as e:
            # 处理转发失败的情况，自动移除无效群组ID
            print(f"转发至群组{group_id}失败: {str(e)}")
            if group_id in chat_ids:
                chat_ids.remove(group_id)
                save_chat_ids(chat_ids)

async def main():
    await dp.start_polling(bot)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

关键优化点 #

• 自动维护群组列表：通过my_chat_member事件监听机器人的加入/移除操作，实时更新群组ID列表。
• 数据持久化：用JSON文件存储群组ID，重启机器人后不会丢失已记录的群组信息。
• 权限检查完善：同时判断群主(creator)和管理员(administrator)状态，覆盖所有有权限转发的场景。
• 异常容错：转发失败时自动移除无效群组ID，避免后续重复尝试无效操作。

内容的提问来源于stack exchange，提问作者Yunusbek Yusupov