Загрузка...

Python
Telegram bot for collecting usernames of group members

Thread in Your projects created by feyzis Dec 18, 2025. (bumped Jan 22, 2026) 232 views

  1. feyzis
    feyzis Topic starter Dec 18, 2025 16 Mar 15, 2024
    Решил создать простого бота - парсера юзернеймов . бот парсит юзеры только из открытых групп, где участники видны.
    Code
    import asyncio
    
import logging
    
from aiogram import Bot, Dispatcher, types
    
from aiogram.client.default import DefaultBotProperties
    
from aiogram.filters import Command
    
from aiogram.types import Message, InlineKeyboardMarkup, InlineKeyboardButton
    
from aiogram.enums import ParseMode
    
from telethon import TelegramClient
    
from telethon.tl.functions.channels import GetParticipantsRequest
    
from telethon.tl.types import ChannelParticipantsSearch
    
from telethon.errors import UsernameNotOccupiedError, ChannelInvalidError
    

    
logging.basicConfig(level=logging.INFO)
    
logger = logging.getLogger(__name__)
    

    
BOT_TOKEN = "854953632:AAEPD2k5YlLiLQpuHwF5CRsLc8OQZ8Pl5vA"
    
API_ID = 34708956
    
API_HASH = "146331b31f37fcba788171f6a654c3c8"
    

    
bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
    
dp = Dispatcher()
    
client = TelegramClient('session_name', API_ID, API_HASH)
    

    
def get_keyboard():
    
    keyboard = InlineKeyboardMarkup(
    
        inline_keyboard=[
    
            [InlineKeyboardButton(text=" Получить участников", callback_data="get_members")]
    
        ]
    
    )
    
    return keyboard
    

    
@dp.message(Command("start"))
    
async def cmd_start(message: Message):
    
    await message.answer(
    
        "Привет! Я бот для сбора юзернеймов из открытых групп\n"
    
        "Нажмите кнопку ниже",
    
        reply_markup=get_keyboard()
    
    )
    

    
@dp.callback_query(lambda c: c.data == "get_members")
    
async def process_callback_get_members(callback_query: types.CallbackQuery):
    
    await callback_query.answer()
    
    await callback_query.message.answer(
    
        "Отправьте мне ссылку на группу.\n\n"
    
        "<i>Примеры:</i>\n"
    
        "<code>@group</code>\n"
    
        "<code>https://t.me/group</code>"
    
    )
    

    
def extract_username(link):
    
    link = link.strip()
    
 
    
    if link.startswith('https://'):
    
        link = link[8:]
    
    elif link.startswith('http://'):
    
        link = link[7:]
    
 
    
    if link.startswith('t.me/'):
    
        link = link[5:]
    
 
    
    if link.startswith('@'):
    
        link = link[1:]
    
 
    
    link = link.split('?')[0]
    
 
    
    return link
    

    
async def get_channel_members(channel_username):
    
    try:
    
        await client.start()
    
   
    
        try:
    
            entity = await client.get_entity(channel_username)
    
        except (UsernameNotOccupiedError, ValueError):
    
            return None, "Группа не найдена. Проверьте правильность ссылки."
    
   
    
        participants = []
    
        offset = 0
    
        limit = 200
    
        total = 0
    
   
    
        while True:
    
            try:
    
                result = await client(GetParticipantsRequest(
    
                    channel=entity,
    
                    filter=ChannelParticipantsSearch(''),
    
                    offset=offset,
    
                    limit=limit,
    
                    hash=0
    
                ))
    
           
    
                if not result.users:
    
                    break
    
           
    
                for user in result.users:
    
                    if user.username:
    
                        participants.append(f"@{user.username}")
    
                        total += 1
    
           
    
                offset += len(result.users)
    
           
    
                if len(result.users) < limit or total >= 10000:
    
                    break
    
               
    
                await asyncio.sleep(1)
    
           
    
            except Exception as e:
    
                logger.error(f"Error getting participants: {e}")
    
                break
    
   
    
        return participants, f"Найдено участников: {total}"
    
   
    
    except Exception as e:
    
        logger.error(f"Error in get_channel_members: {e}")
    
        return None, f"Ошибка: {str(e)}"
    

    
def split_usernames_optimal(usernames):
    
    chunks = []
    
    current_chunk = []
    
    current_length = 0
    
 
    
    MAX_MESSAGE_LENGTH = 4096
    
 
    
    for username in usernames:
    
        username_with_space = username + " "
    
        username_len = len(username_with_space)
    
   
    
        if current_length + username_len > MAX_MESSAGE_LENGTH:
    
            if current_chunk:
    
                chunks.append(" ".join(current_chunk))
    
            current_chunk = [username]
    
            current_length = username_len - 1
    
        else:
    
            current_chunk.append(username)
    
            current_length += username_len
    
 
    
    if current_chunk:
    
        chunks.append(" ".join(current_chunk))
    
 
    
    return chunks
    

    
@dp.message()
    
async def handle_message(message: Message):
    
    user_input = message.text.strip()
    
 
    
    if not user_input:
    
        return
    
 
    
    if not ('t.me' in user_input or '@' in user_input):
    
        await message.answer("Пожалуйста, отправьте ссылку на группу.")
    
        return
    
 
    
    username = extract_username(user_input)
    
 
    
    if not username:
    
        await message.answer("Не удалось извлечь username из ссылки.")
    
        return
    
 
    
    processing_msg = await message.answer(f" Ищу участников в @{username}...\n<i>Это может занять некоторое время</i>")
    
 
    
    try:
    
        participants, status_message = await get_channel_members(username)
    
   
    
        if participants is None:
    
            await processing_msg.edit_text(status_message)
    
            return
    
   
    
        if not participants:
    
            await processing_msg.edit_text(f"В группе @{username} не найдено участников с юзернеймами или группа приватная.")
    
            return
    
   
    
        await processing_msg.edit_text(f" {status_message}\n\nГруппа: @{username}")
    
   
    
        username_chunks = split_usernames_optimal(participants)
    
   
    
        for chunk in username_chunks:
    
            try:
    
                await message.answer(f"<code>{chunk}</code>")
    
                await asyncio.sleep(0.3)
    
            except Exception as e:
    
                logger.error(f"Error sending chunk: {e}")
    
                continue
    
   
    
        await message.answer(f" Сбор завершен! Всего собрано: {len(participants)} юзернеймов.")
    
   
    
    except Exception as e:
    
        logger.error(f"Error processing group: {e}")
    
        await processing_msg.edit_text(f"Произошла ошибка при обработке группы: {str(e)}")
    

    
async def main():
    
    logger.info("Запуск бота...")
    
 
    
    await client.start()
    
    logger.info("Telethon клиент запущен")
    
 
    
    await dp.start_polling(bot)
    

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

    Данные нужно заменить на свои: BOT_TOKEN, API_ID, API_HASH (Бот работает за счет вашего телеграм аккаунта)

    pip install aiogram telethon
     
    Dec 18, 2025 Edited
Loading...