Чистим накрутку

Приветствую! На связи админ канала Falcon Bytes! После долгого отсутствия наконец выкладываю этот пост об очистке накрутки. Приступим!

Накрутка - это когда на ваш ресурс тем или иным образом за короткое время переходит большое количество пользователей. Цель накрутки может быть разная. В этом посте мы рассмотрим случай, когда на ваш телеграм канал какой-то бомж накрутил подписчиков без вашего ведома с целью сноса вашего канала или просто чтобы насолить. И то и другое для нас нежелательно, поэтому перейдем к решению проблемы.

Как мы будем это делать?

Для решения нашей задачи мы будем использовать мозги, яп python, и несколько библиотек под него.

Если с момента накрутки вашего канала прошло меньше 48 часов, то лог всех входов можно увидеть открыв вкладку "недавние действия" в настройках канала.

При помощи модуля "telethon" мы будем открывать логи канала через скрипт, что позволит нам программно найти и отфильтровать ID накруточных ботов по времени.

После того, как мы спарсили всех ботов, мы сохраняем это все в список, в нашем случае это будет обычный текстовый файл, и затем при помощи телеграм бота забаним все айди из этого списка.

Почему не забанить при помощи того же Telethon?

Да потому, что телеграм накладывает ограничение на количество забаненных пользователей от аккаунта. Вы сможете забанить максимум 100 ботов, после чего телега вас пошлет нахуй минимум на 2 минуты. И 2 минуты - это только начало, при последующих попытках время ограничения будет только увеличиваться. У ботов же это ограничение наступает куда позднее, что и позволит нам за короткое время очистить много ботов.

С теорией разобрались, приступим к практике.

Если вы совсем раньше не работали с API телеграма, вам понадобится получить API_ID и API_HASH для вашего аккаунта.

Переходим по ссылке: https://my.telegram.org/auth?to=apps

Вводим номер, вводим код, который пришел на наш аккаунт

Если вы делаете это в первый раз, у вас появится такая страница:

Вводим случайное уникальное название в App title, заполняем short name, затем нажимаем "Create application". В некоторых случаях телеграм может послать вас нахуй и вместо желаемых ключиков появится какая-то поебота с ошибкой. Универсального решения нет. Просто ебите кнопку,меняйте названия, повторяйте попытку позже, колдуйте над сайтом, пока не получится.

Еще вы можете на гитхабе посмотреть официальное приложение Telegram и спиздить оттуда уже готовые официальные ключи (нахуя вам этот пост, если вы такое умеете?)

Если у вас все получится, вас должна приветствовать такая страница:

копируем эти данные и записываем в файл config.py

Теперь напишем мини-скрипт, который генерирует ключ, по которому можно легко получить доступ к аккаунту. Почему не обычную сессию использовать? Потому, что обычные файлы сессии долгие, с ними часто проблемы и вообще это говнище.

from telethon import TelegramClient
from telethon.sessions import StringSession

import config
import asyncio

api_id = config.api_id
api_hash = config.api_hash


async def main():
    client = TelegramClient(
        "zalupa", 
        api_id, 
        api_hash
        )
    
    await client.start()
    print("[+] Connected successfuly")

    string_session = str(StringSession.save(client.session))
    
    print(f"String session: {string_session}")


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

При запуске этого скрипта у вас спросит номер телефона от аккаунта, с которого вы будете парсить, затем попросит код подтверждения и пароль, если он активен.

Если все прошло успешно, Скрпт вам выдаст вашу стринг сессию. Ее мы запишем прямо в конфиг:

Теперь сохраняем полученную стрингу в config.py:

Добавьте аккаунт, сессию которого получили в администраторы вашего канала, мы могли в дальнейшем просматривать историю вступлений со скрипта.

Продолжаем подготовку.

Теперь нужно получить айди канала, который будет чиститься

Чтобы получить айди, можно включить параметр в расширенных настройках. Это будет выглядеть примерно так:

Теперь вы с легкостью сможете получить айди канала.

Айди не должен начинаться с -100. Если у вас так, то просто удалите "-100" из числа

channel_id = 1488228 #замени на свой 

Теперь последний шаг. Получим токен бота.

1. Переходим в @BotFather
2. Пишем /newbot
3. Назначаем боту имя
4. Устанавливаем боту юзернейм (должен быть уникальным) с припиской _bot
5. Полученный токен копируем

Создадим в конфиге переменную, в которой будет храниться время в часах, за которое зашли боты. В моем примере будет использоваться 2 часа, но вы можете ввести 48 часов, тогда вы получите все логи.

В результате окончательный вариант конфиг файла будет выглядеть так:

api_hash = 'govno123boobs'
api_id = 1488
session = "стрингсессиия="

bot_token = "614880:Aтокенбота0"

channel_id = 1488228

hours_filter = 2

Теперь приступим к первой части скрипта, а именно к парсеру ботов.

Создадим файл ids_parser.py

Импортируем библиотеки и конфиг:

from telethon import TelegramClient   #для работы с клиентом
from telethon.sessions import StringSession # для конвертации стринги

import datetime # для работы со временем
import asyncio # для работы с аскинхронными функциями

from rich import print # для пиздатого вывода текста в терминал

import config # важные значения

Теперь создадим асинхронную функцию и сделаем подключение к аккаунту:

async def main():
    client = TelegramClient(
        StringSession(config.session), 
        config.api_id, 
        config.api_hash
        )
    await client.connect()
    me = await client.get_me()

Разберем подробнее:

client = TelegramClient(
StringSession(config.session),
config.api_id,
config.api_hash
)

Инициализирует класс из библиотеки telethon для работы с аккаунтом. Для него нужно указать сессию, api_id и api_hash, что мы собственно и делаем. Там также есть необязательные параметры, о которых вы можете подробнее прочитать здесь.

После чего выполняется команда connect, для подсоединения к уже авторизованному аккаунту и get_me, для получения информации о аккаунте (айди, имя и т д)

Теперь получим объект нашего канала, чтобы можно было с ним проводить дальнейшие манипуляции. Это можно представить себе как если бы вы просто руками открыли список диалогов и выбрали бы ваш канал:

    await client.get_dialogs()
    channel_peer = await client.get_entity(config.channel_id2)

    result = client.iter_admin_log(
        entity=channel_peer,
        join=True,
    )

Также мы указали параметр join=True, этот параметр отвечает за фильтр (только входы) это было бы аналогично:

Теперь откроем текстовый файл и зарезервируем переменную, в которую будем сохранять список айди ботов:

ids_file = open("ids.txt", "a+", encoding= "utf-8")
ids = []

Параметр "a+" отвечает за то, чтобы текстовый файл открылся и дописывался при действиях.

Теперь добавим обработку ботов:

async for i in result:
    join_date = i.date
    user_id = i.user_id

    start_time = datetime.datetime.now(tz=datetime.timezone.utc) - datetime.timedelta(hours=config.hours_filter)
    time_now = datetime.datetime.now(tz=datetime.timezone.utc)
        
    if start_time <= join_date <= time_now:
        if not user_id in ids:
            if not user_id == me.id:
                ids.append(user_id)
                print(f"[{len(ids)}] Added userid [bold]{user_id}[/bold] who joined at [bold]{join_date}[/bold]")

Здесь мы проходимся по каждому элементу, который мы ранее сохранили при получении админ лога results

Для удобства сохраняем время входа пользователя в join_date и его айди в user_id

Теперь считаем предполагаемое время старта накрутки:

start_time = datetime.datetime.now(tz=datetime.timezone.utc) - datetime.timedelta(hours=config.hours_filter)

Здесь мы берем текущее время, и от него отнимаем 2 часа, которые мы указали ранее. Просто отнять число нельзя, так как это времена. Для этого используется timedelta.

Например, если предположить, что текущее время: 04.04.2025 12:03, то время в start_time будет: 04.04.2025 10:03

Для удобства сохраним текущее время в time_now.

Теперь все, что нам остается сделать - это проверить, находится ли время входа в промежутке между стартом накрутки (за 2 часа до текущего времени)

Также нам нужно проверить, не является ли найденный айди нашим, иначе в противном случае мы сами же себя захуярим при бане ботом.

Если время совпадает, сохраняем айди этого пользователя в список.

Теперь все, что остается сделать, это сохранить полученные айди в текстовый файл:

for user_id in ids:
    ids_file.write(f"{user_id}\n")
    
print(f"Total: {len(ids)}")

ids_file.close()

По окончанию процесса закрываем файл при помощи ids_file.close()

Добавляем код, запускающий асинхронную функцию и получаем такой результат:

from telethon import TelegramClient
from telethon.sessions import StringSession

import datetime
import asyncio

from rich import print

import config



async def main():
    client = TelegramClient(
        StringSession(config.session), 
        config.api_id, 
        config.api_hash
        )
    await client.connect()

    me = await client.get_me()

    await client.get_dialogs()
    channel_peer = await client.get_entity(config.channel_id2)

    result = client.iter_admin_log(
        entity=channel_peer,
        join=True,
    )


    ids_file = open("ids.txt", "a+", encoding= "utf-8")

    ids = []

    async for i in result:
        join_date = i.date
        user_id = i.user_id

        start_time = datetime.datetime.now(tz=datetime.timezone.utc) - datetime.timedelta(hours=config.hours_filter)
        time_now = datetime.datetime.now(tz=datetime.timezone.utc)
        
        if start_time <= join_date <= time_now:
            if not user_id in ids:
                if not user_id == me.id:
                    ids.append(user_id)
                    print(f"[{len(ids)}] Added userid [bold]{user_id}[/bold] who joined at [bold]{join_date}[/bold]")

    for user_id in ids:
        ids_file.write(f"{user_id}\n")

    print(f"Total: {len(ids)}")

    ids_file.close()



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

Теперь все, что нам остается сделать - запустить этот код. Если вы не проебланили и настроили питон правиильно, то скрипт успешно сохранит айди ботов в файл.

При запуске получится примерно так:

Как видите, в моем примере скрипт насчитал 22002 бота.

Пол дела уже сделано. Теперь осталось только очистить этот высер с канала.

Напишем мини-модуль, который будет отвечать за удаление юзеров с канала

import asyncio
from aiogram import Bot
import datetime
from rich import print
from aiogram.exceptions import TelegramRetryAfter


def chunk_list(input_list, chunk_size):
    """
    Разбивает список на чанки заданного размера.

    input_list: list - список, который нужно разбить
    chunk_size: int - размер чанка

    Возвращает: list[list] - список чанков
    """
    return [input_list[i:i + chunk_size] for i in range(0, len(input_list), chunk_size)]



class Kicker:
    def __init__(self, token):
        self.bot = Bot(token=token)
    
    async def kick_users(self, chat_id: int, users_to_kick: list[int], timeout: int = 1, chunk_size: int = 200):
        """
        chat_id: int id where to kick users
        users_to_kick: list[int] list userids to kick
        """
        u = datetime.datetime.now() + datetime.timedelta(minutes=1)

        async def banuser(userid):
            while True:
                try:
                    await self.bot.ban_chat_member(
                        chat_id=chat_id,
                        user_id=userid,
                        until_date=u
                    )
                    print(f'[+] Kicked successfully: [bold]{userid}[/bold] from [bold]{chat_id}[/bold]')
                    break  
                except TelegramRetryAfter as e:
                    print(f'[!] FloodWait! Sleeping for {e.retry_after} seconds before retrying...')
                    await asyncio.sleep(e.retry_after)  
                except Exception as e:
                    print(f'[!] Error kicking user {userid}: {e}')
                    break  

        chunks = chunk_list(users_to_kick, chunk_size)

        for chunk in chunks:
            await asyncio.gather(*[banuser(userid=userid) for userid in chunk])
            print(f"Chunk done, sleeping for [bold]{timeout}[/bold] seconds...")
            await asyncio.sleep(timeout)

Пройдемся по основным моментам:

Для быстрого удаления юзеров мы будем использовать такую конструкцию:

await asyncio.gather(*[banuser(userid=userid) for userid in chunk])

Почему не использовать обычный for-цикл? Потому,что это долго, а у нас 22к говна.

Представленная конструкция, в отличии от for цикла, запускает кик пользователя паралельно сразу большую пачку. Размер пачки: 200 пользователей.

for цикл запускал бы по очереди бан каждого и ждал бы выполнения, asyncio.gather сразу запускает все функции одновременно, что и обеспечивает высокую скорость выполнения.

Как вы могли заметить, мы не закидываем в asyncio.gather сразу все 22к пользователей, а делим их на пачки (чанки). Это сделано по той простой причине, что телеграм увидев запрос от бота на бан 22к пользователей на свой сервер знатно прихуеет и пошлет нас нахуй. Кроме того, для того, мы используем тайм аут в 1 секунду, когда забаним каждых 200 пользователей.

Кик в понимании телеграма - это бан пользователя на 1 минуту. Таким образом даже если мы случайно забаним реального пользователя, он сможет вернуться на канал через какое-то время.

Также в коде предусмотрено повторение попытки, если телега послала нас нахуй за флуд:

В этой функци мы прямо из самой ошибки получаем количество секунд, которые нужно подождать перед отправкой новой пачки запросов.

Все это помещено в цикл while True. Break прерывает выполнение этого цикла, и происходит при неизвестной ошибке или при удачном бане пользователя. В случае ошибки тайм аута - скрипт просто подождет нужное количество секунд и повторит попытку.

Модуль дописали, пора бы и очистить канал.

Напишем мини-скрипт запуска кика, в котором укажем, откуда должны быть прочитаны айди и в каком чате должны быть забанены пользователи.

from kicker import Kicker

import asyncio
import config

kicker = Kicker(token=config.bot_token)

async def main():
    raw_ids = open("ids.txt", "r", encoding= "utf-8").readlines()

    users_list = [int(i.replace("\n", "")) for i in raw_ids]

    await kicker.kick_users(
        chat_id=int(f"-100{config.channel_id}"),
        users_to_kick=users_list,
    )

asyncio.run(main())

У ботов немного по другому представлены айди чатов. Все они начинаются с -100, поэтому один из способов конвертировать айди канала - использовать эту конструкцию:

chat_id=int(f"-100{config.channel_id}")

В этой строке мы берем сырой текст, взятый из текстового файла и удаляем в нем лишние переносы знаков. Решение не самое лучшее, но для масштабов нашего скрипта этого более чем достаточно

users_list = [int(i.replace("\n", "")) for i in raw_ids]

Теперь все, что нам остается сделать - добавить бота в администраторы канала и выдать ему права на бан пользователей.

На примере у меня был такой канал:

Видим большое количество накрутки. Запускаем скрипт:

Смотрим в логи канала:

Наблюдаем удачную работу бота. Итого, он забанил больше 20к пользователей менее чем за 5 минут

Получаем результат:

Вы можете улучшить этот скрипт и добавить в него много нового.

Заключение

Теперь у вас в арсенале есть уверенный и быстрый скрипт для очистки высеров малолетних обиженных говноедов за короткое время.

Если эта статья была вам полезна — не забудьте поделиться этим постом с друзьями. Оставляйте свой фидбек в комментариях, это помогает каналу.

На связи был админ канала Falcon Bytes. Еще увидимся!