Python实现Telegram批量封禁非管理员机器人报错求助

问题描述 #

开发Telegram群组机器人，目标是通过指令封禁群组内所有非管理员成员，当前遇到两个问题：

• 机器人在群组内对任何指令无响应；
• 私发机器人指令时触发错误，报错信息为can only concatenate list (not "str") to list。

实现代码 #

import con as keys
import asyncio
from telegram.ext import *
import responses as r
from telethon import TelegramClient
from telethon.errors import SessionPasswordNeededError
from telethon.tl.functions.channels import GetParticipantsRequest
from telethon.tl.types import ChannelParticipantsSearch
from telethon.tl.types import (
    PeerChannel
)
import pyrogram
from telethon.sync import TelegramClient
from telethon.tl.functions.messages import GetDialogsRequest
from telethon.tl.types import InputPeerEmpty
import os, sys
import time

print("Iniziamo...")

def pyro_command(update, context):
    chats = []
    last_date = None
    chunk_size = 900
    groups=[]
 
    result = client(GetDialogsRequest(
             offset_date=last_date,
             offset_id=0,
             offset_peer=InputPeerEmpty(),
             limit=chunk_size,
             hash = 0
         ))
    chats.extend(result.chats)
    for chat in chats:
        try:
            if chat.group== True:
                groups.append(chat)
        except:
            continue
    print('[+] scegli il gruppo dove vuoi cercare gli utenti:')
    i=0
    for g in groups:
        print([str(i)] + ' - '+ g.title)
        i+=1
 
    print('')
    g_index = input(gr+"[+] inserisci un numero : "+re)
    target_group=groups[int(g_index)]
 
    print(gr+'[+] creo il messaggio...')
    time.sleep(1)
    all_participants = []
    all_participants = client.get_participants(target_group, aggressive=True)
    print("creo il messaggio...")
    time.sleep(1) 
    for user in all_participants:
        if not update.message.from_user.id in get_admin_ids(bot, update.message.chat_id):#i check if the user is an admin for the message
            #ban the user     
    print('[+] Utenti bannati.')

def start_command(update, context):
    update.message.reply_text("INIZIALIZZAZIONE....")
    for x in range(1, 4):
        update.message.reply_text("Pyro Never Dies.")

def help_command(update, context):
    update.message.reply_text("Se hai bisogno di aiuto, chiedi a un Admin!")

def handle_message(update, context):
    text = str(update.message.text).lower()
    response = r.sample_responses(text)

    update.message.reply_text(response)

def error(update, context):
    print(f"update {update} ha causato un errore {context.error}")



try:
    api_id = "info"
    api_hash = "info"
    phone = 'myphonenumber'
    client = TelegramClient(phone, api_id, api_hash)
except KeyError:
    os.system('clear')
    print("Errore.")
client.connect()
print("Client creato e connesso....")
if not client.is_user_authorized():
    client.send_code_request(phone)
    os.system('clear')
    client.sign_in(phone, input('[+] Enter the code: '))
 
os.system('clear')

updater = Updater(keys.API_KEY, use_context = True)
dp = updater.dispatcher
dp.add_handler(CommandHandler("start", start_command))
dp.add_handler(CommandHandler("help", help_command))
dp.add_handler(CommandHandler("pyro", pyro_command))
dp.add_handler(MessageHandler(Filters.text, handle_message))
dp.add_error_handler(error)
updater.start_polling()
updater.idle()

报错信息 #

[+] scegli il gruppo dove vuoi bannare gli utenti:
update {'message': {'delete_chat_photo': False, 'chat': {'first_name': 'kernel', 'type': 'private', 'id': 5377477821, 'username': 'khagnaccio'}, 'photo': [], 'new_chat_photo': [], 'entities': [{'length': 5, 'type': 'bot_command', 'offset': 0}], 'text': '/pyro', 'group_chat_created': False, 'new_chat_members': [], 'supergroup_chat_created': False, 'date': 1661200420, 'channel_chat_created': False, 'message_id': 125, 'caption_entities': [], 'from': {'id': 5377477821, 'language_code': 'it', 'is_bot': False, 'first_name': 'kernel', 'username': 'khagnaccio'}}, 'update_id': 967684595} ha causato un errore can only concatenate list (not "str") to list

错误分析与修复方案 #

1. 解决can only concatenate list (not "str") to list错误 #

这个错误来自pyro_command中的这行代码：

print([str(i)] + ' - '+ g.title)

这里[str(i)]是一个列表，尝试和字符串' - '拼接会触发类型错误。修正为：

print(str(i) + ' - '+ g.title)

2. 解决群组内指令无响应及逻辑错误 #

原代码存在多个设计和逻辑问题，导致机器人无法正常工作：

• 混用两个Telegram库：同时使用python-telegram-bot（处理指令）和Telethon（获取成员），两者授权和上下文独立，造成逻辑混乱；
• 命令行输入交互：用input()获取群组选择，这在机器人场景中不适用，机器人需要通过Telegram消息交互；
• 未定义函数：get_admin_ids函数未实现，直接调用会报错；
• 无封禁逻辑：封禁用户的代码块为空，没有实际执行封禁操作；
• 权限检查错误：权限检查逻辑写在遍历成员的循环内，逻辑颠倒；
• 同步/异步不兼容：python-telegram-bot v13+默认异步，原代码用同步函数会导致阻塞。

修正后的核心代码

以下是完全基于python-telegram-bot异步版本的修复代码：

import con as keys
import asyncio
from telegram.ext import *
import responses as r
import time

print("Iniziamo...")

# 获取群组管理员ID列表
async def get_admin_ids(bot, chat_id):
    admin_ids = []
    async for member in bot.get_chat_administrators(chat_id):
        admin_ids.append(member.user.id)
    return admin_ids

async def pyro_command(update, context):
    chat_id = update.message.chat_id
    
    # 检查发送者是否为管理员
    admin_ids = await get_admin_ids(context.bot, chat_id)
    if update.message.from_user.id not in admin_ids:
        await update.message.reply_text("你没有权限执行这个指令！")
        return
    
    await update.message.reply_text("开始封禁非管理员成员...")
    
    try:
        banned_count = 0
        # 遍历群组所有成员
        async for member in context.bot.get_chat_members(chat_id):
            user_id = member.user.id
            # 跳过管理员和机器人
            if user_id not in admin_ids and not member.user.is_bot:
                await context.bot.ban_chat_member(chat_id, user_id)
                banned_count += 1
                await asyncio.sleep(0.5)  # 避免请求频率过高触发限制
        
        await update.message.reply_text(f"已封禁 {banned_count} 名非管理员成员。")
    except Exception as e:
        await update.message.reply_text(f"执行出错：{str(e)}")
        print(f"封禁操作错误：{e}")

async def start_command(update, context):
    await update.message.reply_text("INIZIALIZZAZIONE....")
    for x in range(1, 4):
        await update.message.reply_text("Pyro Never Dies.")

async def help_command(update, context):
    await update.message.reply_text("Se hai bisogno di aiuto, chiedi a un Admin!")

async def handle_message(update, context):
    text = str(update.message.text).lower()
    response = r.sample_responses(text)
    await update.message.reply_text(response)

async def error(update, context):
    print(f"update {update} 触发错误: {context.error}")

# 初始化机器人
updater = Updater(keys.API_KEY, use_context=True)
dp = updater.dispatcher

# 添加处理器
dp.add_handler(CommandHandler("start", start_command))
dp.add_handler(CommandHandler("help", help_command))
dp.add_handler(CommandHandler("pyro", pyro_command))
dp.add_handler(MessageHandler(Filters.text & ~Filters.command, handle_message))
dp.add_error_handler(error)

updater.start_polling()
updater.idle()

3. 额外注意事项 #

• 确保机器人已被添加为群组管理员，并授予封禁用户的权限；
• 移除原代码中无用的Telethon和pyrogram导入，避免依赖混乱；
• 测试时先在小群组中验证，避免误封大量用户；
• 遵守Telegram API的请求频率限制，避免触发封禁。

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