Как создать Telegram-бот для автопубликации сторис

Около месяца назад мы рассказывали о том, как воспроизвести весенний кейс с продвижением конфы путем одновременной публикации аватарок в разных Telegram-каналах, но не ручками, а в автоматическом режиме — с помощью бота. Сегодня логическое продолжение этой темы: бот для одновременной публикации сторис. Собственно, поехали!

Какие задачи выполняет бот для автоматической публикации сторис

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

• Массовая заливка сторис в одно и то же время — очень удобно, если это часть PR-кампании.

• Заливка сторис в неудобное время — зачем просыпаться ночью, если можно закинуть в бот.

• Диверсификация рисков — в XXI веке интернет если и пропадает, то обычно в самый неподходящий момент. Вот только заказчикам-то какое до этого дела? Бот не требует наличия интернета у вас лично. С учетом анонса постоянных отключений связи — мастхэв.

• Карусель — если зациклить код из статьи, можно будет запускать сторьки по кругу.

• Безопасность — свой бот всегда безопаснее стороннего.

Принцип работы бота для автоматической публикации сторис

С точки зрения алгоритма бот не самый сложный, пусть даже он и использует не только bot-api, но и user-api. Функционально он реализован следующим образом:

1. Прослушивание ЛС бота и обработка «неправильного формата» путем ответного приветствия с перечислением возможностей.
2. Старт создания отложки после отправки в ЛС бота картинки.
3. Сохранение картинки изображения, списка @usernames каналов и времени публикации сторис в task со своим ID.
4. Отмена создания task’а при получении неправильного формата инфы с одновременной отправкой просьбы писать в корректном формате
5. Отображение существующих task’ов при вводе /show.
6. Удаление task’а по ID при вводе /delete ID (цифра).
7. Создание резервной копии при любой манипуляции с task’ами.
8. Создание резервной копии для резервной копии — чтоб было из чего все восстанавливать, если бэкап скораптится при создании.
9. Применение резервной копии или бэкапа резервной копии при вводе /restore и /restore_old.
10. Автоматическая публикация сторис согласно task-листу в планировщике.

Ничего сложного. Единственный нюанс — из-за специфики самой Телеги есть еще два «важных алгоритмических момента», которые все же нужно отделить от основной логики бота:

1. У вас (странички, от которой api_id и api_hash) должна быть админка в каналах.
2. Каналы должны быть с бустом за звезды — иначе сторис в принципе невозможно будет опубликовать даже вручную.

Без этого бот, естественно, работать не сможет.

Пошаговая инструкция, как сделать бот для автоматической публикации сторис

Так как Телеграм не позволяет публиковать сторис непосредственно через bot-api, основной функционал нашего бота будет реализован через user-api. В свою очередь, bot-api будет использоваться для вспомогательных функций управления. Поэтому авторизации будет две — со странички юзера и в BotFather.

Начнем с получения токена в BotFather. Для этого:

1. Находим профиль сам BotFather.
2. Следуем отправленным им инструкциям.
3. Сохраняем полученный токен.

Затем получаем api_id и api_hash, для этого:

1. Открываем веб-версию.
2. Находим «API development tools», заполняем форму.
3. Сохраняем сгенерированные api_id и api_hash.

Далее настраиваем сервер для работы с ботом. Подразумевается, что базовые настройки уже выполнены, а сам сервер у вас имеется. Если нет — вы можете развернуть свой сервак бесплатно прямо на домашнем ПК, следуя инструкции из этой статьи. Или же арендовать готовый — дело ваше. Как бы там ни было, идем дальше:

1. Пишем в консоли сервера:

pip install asyncio

pip install telethon

pip install telebot

2. Создаем в корневой директории сервера файл bot.py

3. Открываем его текстовым редактором и вставляем следующий код:

import os

import random

import asyncio

import telebot

from telethon import TelegramClient

from telethon.tl.functions.stories import SendStoryRequest

from telethon.tl.types import InputPeerChannel, InputMediaUploadedPhoto

from apscheduler.schedulers.background import BackgroundScheduler

from datetime import datetime

import json

import shutil

api_id = "Ваш api_id"

api_hash = 'Ваш api_hash'

phone = 'Ваш телефон'

session_name = 'session'

BOT_TOKEN = 'Ваш токен бота'

bot = telebot.TeleBot(BOT_TOKEN)

scheduler = BackgroundScheduler()

scheduler.start()

pending_tasks = {}

user_state = {}

id_counter = 1

BACKUP_FILE = 'tasks_backup.json'

OLD_BACKUP_FILE = 'tasks_backup_old.json'

def save_backup():

try:

shutil.copy(BACKUP_FILE, OLD_BACKUP_FILE)

print("Старый бэкап сохранен.")

except FileNotFoundError:

pass

with open(BACKUP_FILE, 'w') as f:

json.dump(pending_tasks, f, indent=4)

print("Бэкап сохранен.")

def clear_all_jobs():

for job in scheduler.get_jobs():

scheduler.remove_job(job.id)

def restore_backup():

global pending_tasks

clear_all_jobs()

try:

with open(BACKUP_FILE, 'r') as f:

pending_tasks = json.load(f)

print("Бэкап восстановлен.")

except FileNotFoundError:

print("Бэкап не найден.")

return

for task_id, task in pending_tasks.items():

scheduler.add_job(

send_story_job,

'date',

run_date=datetime.strptime(task['time'], "%Y-%m-%d %H:%M"),

args=[task['chat_ids'], task['image_path'], task_id],

id=f"task_{task_id}"

)

def restore_old_backup():

global pending_tasks

clear_all_jobs()

try:

with open(OLD_BACKUP_FILE, 'r') as f:

pending_tasks = json.load(f)

print("Старый бэкап восстановлен.")

except FileNotFoundError:

print("Старый бэкап не найден.")

return

for task_id, task in pending_tasks.items():

scheduler.add_job(

send_story_job,

'date',

run_date=datetime.strptime(task['time'], "%Y-%m-%d %H:%M"),

args=[task['chat_ids'], task['image_path'], task_id],

id=f"task_{task_id}"

)

@bot.message_handler(commands=['start'])

def send_welcome(message):

bot.reply_to(message, "Привет! Отправь мне фото, потом каналы (username или ID через пробел), потом дату и время в формате YYYY-MM-DD HH:MM — и я выложу сторис во все эти каналы в нужное время.")

@bot.message_handler(content_types=['photo'])

def handle_photo(message):

file_id = message.photo[-1].file_id

file_info = bot.get_file(file_id)

downloaded_file = bot.download_file(file_info.file_path)

file_name = f"story_{random.randint(10000,99999)}.jpg"

with open(file_name, 'wb') as f:

f.write(downloaded_file)

user_state[message.from_user.id] = {'image_path': file_name, 'step': 'channels'}

bot.reply_to(message, "Отлично. Теперь отправь ID или username каналов через пробел.")

@bot.message_handler(func=lambda m: user_state.get(m.from_user.id, {}).get('step') == 'channels')

def get_channels(message):

chat_ids = message.text.split()

if not chat_ids:

bot.reply_to(message, "Пожалуйста, отправь хотя бы один канал.")

return

user_state[message.from_user.id]['chat_ids'] = chat_ids

user_state[message.from_user.id]['step'] = 'datetime'

bot.reply_to(message, "Теперь отправь дату и время в формате YYYY-MM-DD HH:MM")

@bot.message_handler(func=lambda m: user_state.get(m.from_user.id, {}).get('step') == 'datetime')

def get_datetime(message):

global id_counter

try:

target_time = datetime.strptime(message.text, "%Y-%m-%d %H:%M")

if target_time < datetime.now():

bot.reply_to(message, "Время не может быть в прошлом.")

return

task = user_state[message.from_user.id]

task['time'] = target_time.strftime("%Y-%m-%d %H:%M")

task_id = id_counter

id_counter += 1

task['id'] = task_id

scheduler.add_job(

send_story_job,

'date',

run_date=target_time,

args=[task['chat_ids'], task['image_path'], task_id],

id=f"task_{task_id}"

)

pending_tasks[task_id] = task

save_backup()

del user_state[message.from_user.id]

bot.reply_to(message, f"Задача запланирована на {target_time.strftime('%Y-%m-%d %H:%M')} с ID {task_id}.")

except ValueError:

bot.reply_to(message, "Неверный формат. Пожалуйста, используй YYYY-MM-DD HH:MM.")

async def send_story_to_channels(image_path, chat_ids):

client = TelegramClient(session_name, api_id, api_hash)

await client.start(phone)

file = await client.upload_file(image_path)

media = InputMediaUploadedPhoto(file)

for username in chat_ids:

try:

entity = await client.get_entity(username)

peer = InputPeerChannel(entity.id, entity.access_hash)

await client(SendStoryRequest(

peer=peer,

media=media,

random_id=random.randint(-(2**63), 2**63 - 1),

caption='',

privacy_rules=[],

pinned=False,

noforwards=False,

fwd_modified=False,

media_areas=None,

period=86400

))

print(f"Сторис отправлена в {username}")

except Exception as e:

print(f"Ошибка при отправке в {username}: {e}")

await client.disconnect()

def send_story_job(chat_ids, image_path, task_id):

asyncio.run(send_story_to_channels(image_path, chat_ids))

if os.path.exists(image_path):

os.remove(image_path)

if task_id in pending_tasks:

del pending_tasks[task_id]

scheduler.remove_job(f"task_{task_id}")

save_backup()

@bot.message_handler(commands=['show'])

def show_tasks(message):

if not pending_tasks:

bot.reply_to(message, "Нет запланированных задач.")

return

text = "Запланированные задачи:\n"

for task_id, task in pending_tasks.items():

text += f"ID: {task_id}\nКаналы: {', '.join(task['chat_ids'])}\nВремя: {task['time']}\n\n"

bot.reply_to(message, text)

@bot.message_handler(commands=['delete'])

def delete_task(message):

try:

task_id = int(message.text.split()[1])

if task_id in pending_tasks:

scheduler.remove_job(f"task_{task_id}")

os.remove(pending_tasks[task_id]['image_path'])

del pending_tasks[task_id]

save_backup()

bot.reply_to(message, f"Задача {task_id} удалена.")

else:

bot.reply_to(message, "Задача не найдена.")

except:

bot.reply_to(message, "Формат: /delete <id>")

@bot.message_handler(commands=['restore'])

def restore_now(message):

restore_backup()

bot.reply_to(message, "Задачи восстановлены из последнего бэкапа.")

@bot.message_handler(commands=['restore_old'])

def restore_old_now(message):

restore_old_backup()

bot.reply_to(message, "Задачи восстановлены из старого бэкапа.")

print("Бот запущен.")

bot.polling(none_stop=True)

4. Заменяем Ваш api_id, Ваш api_hash, Ваш телефон и Ваш токен бота на свои данные.

5. Сохраняем файл bot.py.

6. Запускаем бот командой python bot.py.

7. Проверяем работоспособность.

Демонстрация работы

Подводя итоги

Как видите, создать бот для автоматизации массовой публикации сторис в Телеге — не так уж и сложно. Следите за нашим Telegram-каналом, чтобы первыми получать доступ к ботам для арбитража трафика.