Python Telegram Bot自动销毁命令异常问题求助

问题1：群组内发送命令无响应，仅私聊有效 #

原因 #

• 机器人在群组中未被授予读取消息权限，无法接收群组内的命令
• CommandHandler未明确指定允许处理群组内的消息，默认仅监听私聊场景

修复方案 #

1. 检查机器人权限：确保机器人在目标群组中拥有「读取消息」权限（可通过群组设置→管理员权限调整）
2. 修改CommandHandler配置：为命令处理器添加群组聊天类型过滤，让机器人响应群组内的命令：

   # 替换原有的CommandHandler代码，以/pyro为例
   dp.add_handler(CommandHandler("pyro", pyro_command, filters=Filters.chat_type.groups))
   # 若start/help命令也需在群组使用，同理修改
   dp.add_handler(CommandHandler("start", start_command, filters=Filters.chat_type.all))
   dp.add_handler(CommandHandler("help", help_command, filters=Filters.chat_type.all))
   

问题2：执行/pyro命令触发事件循环错误 #

错误原因 #

• Telethon的get_participants是异步协程方法，在python-telegram-bot的调度线程中没有运行事件循环，导致协程未被正确执行
• 代码中使用input()进行交互式输入，属于阻塞式操作，会导致机器人进程卡住
• 当前代码仅获取了成员列表，未实现封禁非管理员成员的核心逻辑

修复方案 #

1. 重构命令逻辑，移除阻塞式输入

改用机器人消息交互让用户选择目标群组，而非控制台输入

2. 正确处理Telethon的异步方法

在回调函数中手动创建事件循环，或改用异步回调（python-telegram-bot v13+支持异步处理函数）

3. 补充封禁非管理员成员的逻辑

修改后的完整代码 #

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, EditBannedRequest
from telethon.tl.types import ChannelParticipantsSearch, ChatBannedRights
from telethon.tl.types import PeerChannel
from telethon.tl.functions.messages import GetDialogsRequest
from telethon.tl.types import InputPeerEmpty
import os, sys
import time

print("Iniziamo...")

# 存储群组列表，供用户选择
global groups
groups = []

async def pyro_command(update, context):
    chat_id = update.effective_chat.id
    if not groups:
        update.message.reply_text("Nessun gruppo trovato.")
        return
    
    # 发送群组选择菜单
    group_list = "\n".join([f"[{i}] - {g.title}" for i, g in enumerate(groups)])
    update.message.reply_text(f"Scegli il gruppo dove vuoi bannare gli utenti:\n{group_list}")
    
    # 标记等待用户回复群组编号
    context.user_data['waiting_for_group'] = True

async def handle_group_selection(update, context):
    if not context.user_data.get('waiting_for_group'):
        return
    
    try:
        g_index = int(update.message.text)
        if g_index < 0 or g_index >= len(groups):
            update.message.reply_text("Numero gruppo non valido.")
            return
        
        target_group = groups[g_index]
        update.message.reply_text(f"Selezionato gruppo: {target_group.title}\nInizio selezione utenti...")
        
        # 创建单独事件循环处理Telethon异步操作
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        
        # 获取所有成员
        all_participants = loop.run_until_complete(client.get_participants(target_group, aggressive=False))
        update.message.reply_text(f"Trovati {len(all_participants)} utenti. Inizio bannaggio non amministratori...")
        
        # 获取群组管理员ID集合
        admins = loop.run_until_complete(client.get_participants(target_group, filter=ChannelParticipantsSearch(""), admins_only=True))
        admin_ids = {admin.id for admin in admins}
        
        # 封禁非管理员成员
        banned_count = 0
        for participant in all_participants:
            if participant.id not in admin_ids:
                try:
                    # 设置永久封禁权限
                    rights = ChatBannedRights(until_date=None, view_messages=True)
                    loop.run_until_complete(client(EditBannedRequest(target_group, participant, rights)))
                    banned_count += 1
                    time.sleep(0.5)  # 避免触发API频率限制
                except Exception as e:
                    print(f"Errore nel bannare {participant.id}: {e}")
        
        update.message.reply_text(f"Operazione completata! Bannati {banned_count} non amministratori.")
        context.user_data['waiting_for_group'] = False
        
    except ValueError:
        update.message.reply_text("Inserisci un numero valido.")
    except Exception as e:
        update.message.reply_text(f"Errore durante l'operazione: {str(e)}")
        context.user_data['waiting_for_group'] = False

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 = "api id"
    api_hash = "api hash"
    phone = 'phone number'
    client = TelegramClient(phone, api_id, api_hash)
except KeyError:
    os.system('clear')
    print("Errore.")

# 初始化Telethon客户端并获取群组列表
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')

# 获取群组列表
chats = []
last_date = None
chunk_size = 900

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.megagroup == True:
            groups.append(chat)
    except:
        continue

# 初始化python-telegram-bot
updater = Updater(keys.API_KEY, use_context=True)
dp = updater.dispatcher

# 添加命令处理器，允许群组内使用
dp.add_handler(CommandHandler("start", start_command, filters=Filters.chat_type.all))
dp.add_handler(CommandHandler("help", help_command, filters=Filters.chat_type.all))
dp.add_handler(CommandHandler("pyro", pyro_command, filters=Filters.chat_type.groups))

# 添加群组选择的消息处理器
dp.add_handler(MessageHandler(Filters.text & ~Filters.command, handle_group_selection))

dp.add_handler(MessageHandler(Filters.text & ~Filters.command, handle_message))
dp.add_error_handler(error)

updater.start_polling()
updater.idle()

关键修改说明 #

• 将pyro_command改为异步函数，通过机器人消息交互获取用户选择的群组，替代控制台input()
• 添加handle_group_selection函数处理用户的群组编号回复
• 使用asyncio.new_event_loop()在单独的事件循环中运行Telethon的异步方法，解决事件循环缺失的问题
• 补充了获取管理员列表、封禁非管理员成员的核心逻辑
• 为CommandHandler添加聊天类型过滤，确保机器人响应群组内的命令

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