Ранее мы уже рассказывали, как создать бот для простановки реакций в Telegram. Однако после публикации статьи уже в наш бот обратной связи поступил запрос рассказать о создании бота для простановки реакций на посты сразу же после их выхода. Что ж, рассказываем.
Какие задачи выполняет бот для моментальных реакций
Несмотря на функциональную схожесть, по своему целевому назначению этот бот отличается от того бота для простановки реакций, что мы публиковали ранее. В частности, он не подходит для:
Акцента внимания на конкретных постах — когда реакции ставятся автоматически на все новые публикации, пропадает сам смысл акцента с их помощью.
Простановки «каках» — технически это возможно, но такое сразу будет обнаружено админом. Да и обмазывать ими весь канал — это уже кринж.
Впрочем, он все же позволяет:
Стимулировать живые реакции — самостоятельно юзеры их ставят редко. А вот когда уже какие-то реакции есть, шанс живого действия выше.
Имитировать активность — одно дело, когда реакции ставятся выборочно, и совсем другое, когда они появляются под всеми новыми публикациями. Это создает видимость активности.
В остальном же целевое назначение обоих ботов идентично. Перейдем к алгоритму.
Принцип работы бота для моментальных реакций
Алгоритмически бот реализован следующим образом:
- Считывание файла с реквизитами для входа в аккаунты.
- Прохождение авторизации.
- Считывание списка постов для простановки реакций (используются в качестве входной точки для следующей функции).
- Получение перечня реакций, которые включены в канале.
- Запоминание выбранных реакций и их рандомная простановка на посты из списка.
- Автоматическая задержка простановки, если новый пост появляется чаще, чем раз в 3 секунды (необходимо для корректной обработки постов с вложениями, где каждое вложение — это новый пост).
- Цикличное повторение для каждого аккаунта.
- Прослушивание каналов из списка с постами.
- Автоматическая простановка реакций на пост после его появления.
- Цикличное повторение для каждого аккаунта.
Как видите, бот схож с прошлой версией, но имеет некоторые отличия. Это, конечно, немного усложняет понимание его алгоритмов новичками, но все же в пределах разумного.
Пошаговая инструкция, как сделать бот для моментальных реакций
Так же, как и бот для автоматизации селективной простановки реакций, бот для простановки моментальных реакций — это не классический Telegram-бот, а скорее автоматизация взаимодействия с user-api. Почему именно так? Потому что «просто боты» априори не могут ставить реакции. А вот профили пользователей могут. Но все связанные с этим сложности заканчиваются на этапе ручного получения api_id и api_hash — после этого шага все происходит автоматически.
Но сначала все же нужно для каждого файла получить api_id и api_hash:
- Открываем версию для разработчиков.
- Заходим в «API development tools».
- Заполняем все поля.
- Сохраняем api_id и api_hash.
- Проделываем шаги 1–4 для каждого профиля.
- После этого нужно развернуть и настроить сервер, где наш бот будет «жить». Можно арендовать, а можно развернуть свой. Для тестов советуем свой. Для полноценной работы — аренду.
- После развертывания — настройка. Вводим в консоль:
pip install asyncio
pip install telethon - Создаем в корневой папке .py файл, пишем в нем:
from telethon import TelegramClient, events
from telethon.errors import SessionPasswordNeededError
from telethon.tl.functions.messages import SendReactionRequest
from telethon.tl.functions.channels import GetFullChannelRequest
from telethon.tl.types import InputPeerChannel, ReactionEmoji
import asyncio
import random
import time
def create_client(api_id, api_hash, phone_number):
session_name = f'session_{phone_number}'
return TelegramClient(session_name, api_id, api_hash)
def load_accounts(filename='accounts.txt'):
accounts = []
with open(filename, 'r') as f:
for line in f:
api_id, api_hash, phone_number = line.strip().split()
accounts.append((int(api_id), api_hash, phone_number))
return accounts
def load_links(filename='links.txt'):
with open(filename, 'r') as f:
return [line.strip() for line in f if line.strip()]
async def get_available_reactions(client, channel_name):
try:
await client.connect()
channel = await client.get_entity(channel_name)
full_chat = await client(GetFullChannelRequest(channel))
available_reactions = full_chat.full_chat.available_reactions
if available_reactions:
if hasattr(available_reactions, 'reactions'):
emoji_list = [reaction.emoticon for reaction in available_reactions.reactions]
return emoji_list
else:
emoji_list = ["👍", "❤️", "😂", "😮", "😢", "🔥"]
return emoji_list
else:
print(f"Нет доступных реакций для канала {channel_name}.")
return []
except Exception as e:
print(f"Error while getting available reactions: {e}")
return []
finally:
await client.disconnect()
async def send_reaction(client, phone_number, post_url, emoji):
channel_name = post_url.split('/')[3]
message_id = int(post_url.split('/')[-1])
try:
await client.connect()
if not await client.is_user_authorized():
await client.sign_in(phone_number)
print(f"Введите код для {phone_number}: ")
code = input('> ')
try:
await client.sign_in(phone_number, code)
except SessionPasswordNeededError:
password = input("Введите пароль двухфакторной аутентификации: ")
await client.sign_in(password=password)
channel = await client.get_entity(channel_name)
result = await client(SendReactionRequest(
peer=InputPeerChannel(channel.id, channel.access_hash),
msg_id=message_id,
big=True,
add_to_recent=True,
reaction=[ReactionEmoji(emoticon=emoji)]
))
except Exception as e:
print(f"Error with {client.session.filename}: {e}")
finally:
await client.disconnect()
def get_reaction_choices(emoji_list):
while True:
print(f"Доступные реакции для канала: {emoji_list}")
choices = input("Введите номера смайлов через запятую (например, 1,4,5): ")
choices = choices.split(',')
try:
chosen_indices = [int(choice.strip()) - 1 for choice in choices]
if all(0 <= index < len(emoji_list) for index in chosen_indices):
return chosen_indices
else:
print("Некоторые номера не существуют. Пожалуйста, попробуйте снова.")
except ValueError:
print("Некорректный ввод. Убедитесь, что вы ввели числа.")
last_post_time = None
async def listen_for_new_posts(client, channel_name, channel_reactions):
global last_post_time
@client.on(events.NewMessage(chats=channel_name))
async def handler(event):
global last_post_time
current_time = event.message.date.timestamp()
if last_post_time is not None and current_time - last_post_time < 3:
print(f"Пропущен пост на канале {channel_name} из-за короткого времени между постами.")
return
last_post_time = current_time
post_url = f"https://t.me/{channel_name}/{event.message.id}"
print(f"Новый пост на канале {channel_name}: {post_url}") # Выводим ссылку на новый пост
tasks = []
chosen_indices, emoji_list = channel_reactions.get(channel_name, ([], []))
if emoji_list:
for api_id, api_hash, phone_number in load_accounts()[1:]: # Пропускаем первый аккаунт
client_to_use = create_client(api_id, api_hash, phone_number)
emoji = random.choice([emoji_list[index] for index in chosen_indices])
tasks.append(send_reaction(client_to_use, phone_number, post_url, emoji))
await asyncio.gather(*tasks)
async def main():
accounts = load_accounts()
links = load_links()
channel_reactions = {}
for post_url in links:
channel_name = post_url.split('/')[3]
if channel_name not in channel_reactions:
first_api_id, first_api_hash, first_phone_number = accounts[0]
first_client = create_client(first_api_id, first_api_hash, first_phone_number)
emoji_list = await get_available_reactions(first_client, channel_name)
if emoji_list:
chosen_indices = get_reaction_choices(emoji_list)
if len(chosen_indices) == 0:
print("Вы должны выбрать хотя бы одну реакцию.")
return
channel_reactions[channel_name] = chosen_indices, emoji_list
else:
print(f"Не удалось получить реакции для канала {channel_name}. Пропускаем.")
continue
tasks = []
for api_id, api_hash, phone_number in accounts:
client = create_client(api_id, api_hash, phone_number)
chosen_indices, emoji_list = channel_reactions[channel_name]
emoji = random.choice([emoji_list[index] for index in chosen_indices])
tasks.append(send_reaction(client, phone_number, post_url, emoji))
await asyncio.gather(*tasks)
first_api_id, first_api_hash, first_phone_number = accounts[0]
first_client = create_client(first_api_id, first_api_hash, first_phone_number)
for post_url in links:
channel_name = post_url.split('/')[3]
await listen_for_new_posts(first_client, channel_name, channel_reactions)
await first_client.start()
await first_client.run_until_disconnected()
asyncio.run(main()) - Сохраняем этот файл как bot.py.
- Создаем в корневой папке файл, accounts.txt, пишем в него: 'ВАШ API_ID', 'ВАШ API_HASH' и 'ВАШ НОМЕР' без кавычек. Каждый акк — новая строка. После последнего проверяем, не поставили ли вы случайно перенос строки (его быть не должно).
- Создаем в корневой папке файл links.txt, пишем в него «входные посты» целевых каналов.
- Пишем в консоль python bot.py — происходит запуск бота.
- Задаем реакции каждому каналу.
- Проверяем работу.
Демонстрация работы
Подводя итоги
Несмотря на простоту реализации и запрос от наших читателей на создание такого бота, мы советуем использовать прошлую версию с селективным выбором постов. Да, это геморно, но так ваши аккаунты проживут дольше, а канал вряд ли привлечет внимание модерации. Если же накрутка будет на каждый новый пост, скорее всего, это приведет к бану. Но тут, как говорится, надо тестить — причем на дистанции в несколько недель.
