Бот для написания постов в Телеграм. Создание и запуск

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

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

Функциональность бота

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

Пример работы:

После завершения разработки кода бота мы запустим его в облаке Amvera. 

Amvera представляет собой «app engine» и предлагает

• Возможность деплоя через загрузку файлов в интерфейсе (или через push в привязанный репозиторий) с автоматической настройкой переменных, доменов с https и всего необходимого, что проще использования VPS.

• Встроенные бэкапы, облачное логирование с синтаксическим поиском, метрики и алерты.

• Бесплатное прокси до ведущих LLM и свой инференс LLaMA 70B.

• Стартовый баланс в 111 р. на тесты.

Исходный код бота доступен по ссылке репозитория GitHub

Поехали.

Создание бота:

Зарегистрируем бота в @BotFather и получим токен по данному примеру:

Успешно! Мы создали бота, нам необходимо сохранить его токен.

Написание бота с использованием aiogram v3

Установим все необходимые библиотеки на локальном уровне. Хотя можно воспользоваться библиотекой telebot, но aiogram будет более подходящим выбором для начинающих из-за своей простоты.

pip install datetime aiogram

Переходим в командную строку и прописываем данную команду выше. 

Теперь приступим к написанию самого бота

1. Создаём файл main.py

2. Импортируем все необходимые библиотеки.

import asyncio  # Для работы с асинхронным программированием import sqlite3  # Для работы с базами данных SQLite from datetime import datetime  # Для работы с датой и временем from aiogram import Bot, Dispatcher, F  # Импортируем классы для работы с Telegram API from aiogram.enums import ParseMode  # Для определения режима обработки текста from aiogram.types import (  # Импортируем типы данных для работы с сообщениями и клавиатурами     Message,     InlineKeyboardMarkup,     InlineKeyboardButton,     CallbackQuery,     ReplyKeyboardMarkup,     KeyboardButton, ) from aiogram.client.default import DefaultBotProperties  # Для задания свойств бота import os  # Для работы с переменными окружения

3. Создадим необходимые переменные

# Получаем идентификатор канала из переменной окружения CHANNEL_ID = os.environ['CHID'] # Получаем идентификаторы администраторов из переменной окружения и сохраняем в список ADMIN_IDS = [int(os.environ['ADID'])]   # Создаем экземпляр бота с токеном и устанавливаем режим обработки текста по умолчанию bot = Bot(token=os.environ['TOKEN'], default=DefaultBotProperties(parse_mode=ParseMode.HTML)) # Создаем экземпляр диспетчера для обработки сообщений и событий dp = Dispatcher()   # Устанавливаем соединение с базой данных SQLite. Важно производить сохранение именно в постоянное хранилище /data conn = sqlite3.connect('/data/posts.db') # Создаем объект курсора для выполнения SQL-запросов cursor = conn.cursor()   # Создаем таблицу для хранения постов, если она не существует cursor.execute('''     CREATE TABLE IF NOT EXISTS posts (         id INTEGER PRIMARY KEY AUTOINCREMENT,         text TEXT NOT NULL,         time TEXT NOT NULL     ) ''') # Сохраняем изменения в базе данных conn.commit() 

4. Напишем функцию, которая будет проверять пользователя на администратора

# Функция для проверки, является ли пользователь администратором def is_admin(user_id: int) -> bool:     return user_id in ADMIN_IDS  # Возвращает True, если пользователь в списке администраторов

5. Создадим экземпляр бота

async def main():  # Основная асинхронная функция для запуска бота     try:         await dp.start_polling(bot)  # Запускаем опрос для получения обновлений от Telegram     finally:         await bot.session.close()  # Закрываем сессию бота после завершения работы         conn.close()  # Закрываем соединение с базой данных   if __name__ == "__main__":  # Проверяем, является ли данный файл исполняемым модулем     asyncio.run(main())  # Запускаем основную функцию в асинхронном режиме 

6. Создадим обработчик команды /start

@dp.message(F.text == "/start") async def start_command(message: Message):     # Создаем клавиатуру с одной кнопкой "Предложить идею"     keyboard = ReplyKeyboardMarkup(         keyboard=[[KeyboardButton(text="Предложить идею")]],         resize_keyboard=True  # Автоматически подстраивает размер клавиатуры     )     # Отправляем приветственное сообщение с инструкциями     await message.reply(         "Привет! Я бот для публикации постов в канале.\n"         "Нажми на кнопку 'Предложить идею' или используй команду /post, чтобы предложить свой пост.",         reply_markup=keyboard  # Прикрепляем клавиатуру к сообщению     ) 

Это обработчик команды /start. В нём мы создаём дополнительно кнопку, которую в следующем этапе будем обрабатывать.

7. Следующим этапом напишем обработчик сообщения “Предложить идею”.

@dp.message(F.text == "Предложить идею") async def suggest_idea(message: Message):     # Отправляем сообщение с инструкциями по предложению поста     await message.reply(         "Пожалуйста, напиши свой пост, начиная с команды /post.\n"         "Например: /post Это мой пост для канала!"     ) 

8. Далее обработчик сообщений, начинающийся на /post.

@dp.message(F.text.startswith("/post")) async def handle_post(message: Message):     user_id = message.from_user.id  # Получаем идентификатор пользователя     post_text = message.text[6:].strip()  # Извлекаем текст поста, убирая команду /post       if not post_text:  # Проверяем, не пустой ли текст поста         await message.reply("Пожалуйста, добавьте текст после команды /post")  # Запрашиваем текст         return  # Завершаем выполнение функции       if is_admin(user_id):  # Проверяем, является ли пользователь администратором         # Вставляем новый пост в таблицу posts с текстом и текущим временем         cursor.execute('INSERT INTO posts (text, time) VALUES (?, ?)',                        (post_text, datetime.now().strftime("%H:%M")))         conn.commit()  # Сохраняем изменения в базе данных           # Отправляем пост в указанный канал         await bot.send_message(             chat_id=CHANNEL_ID,             text=post_text         )         await message.reply("Пост опубликован!")  # Уведомляем пользователя об успешной публикации     else:  # Если пользователь не является администратором         # Создаем инлайн-клавиатуру с кнопками "Одобрить" и "Отклонить"         keyboard = InlineKeyboardMarkup(inline_keyboard=[             [                 InlineKeyboardButton(text="Одобрить", callback_data=f"approve_{message.message_id}_{user_id}"),                 InlineKeyboardButton(text="Отклонить", callback_data=f"reject_{message.message_id}_{user_id}")             ]         ])                 for admin_id in ADMIN_IDS:  # Цикл по всем администраторам             # Отправляем сообщение каждому администратору о новом предложении поста             await bot.send_message(                 chat_id=admin_id,                 text=f"Новое предложение поста от {message.from_user.full_name}:\n\n{post_text}",                 reply_markup=keyboard  # Прикрепляем клавиатуру к сообщению             )                 await message.reply("Ваш пост отправлен администраторам на проверку!")  # Уведомляем пользователя о статусе его поста

С помощью функции, пользователи смогут отправить вам свою идею или же вы сами сможете опубликовать пост в телеграм-канал. Теперь есть проверка на аргументы после команды /post. Вот пример работы:

9. Далее напишем функцию, которая отвечает за одобрение/отклонение идей наших подписчиков.

@dp.callback_query(F.data.startswith("approve_"))  # Обработчик для колбек-запросов, начинающихся с "approve_" async def approve_post(callback: CallbackQuery):  # Асинхронная функция для обработки одобрения поста     if not is_admin(callback.from_user.id):  # Проверяем, является ли пользователь администратором         await callback.answer("У вас нет прав для одобрения постов")  # Если нет, отправляем ответ с сообщением         return  # Завершаем выполнение функции       data_parts = callback.data.split("_")  # Разделяем данные колбек-запроса на части     message_id = data_parts[1]  # Извлекаем идентификатор сообщения из данных     user_id = int(data_parts[2])  # Извлекаем идентификатор пользователя и преобразуем в целое число     suggested_text = callback.message.text.split("\n\n")[1]  # Извлекаем текст предложения поста       try:         await bot.send_message(  # Пытаемся отправить сообщение пользователю о том, что пост одобрен             chat_id=user_id,  # Указываем идентификатор пользователя             text="Ваш пост одобрен! Администратор скоро его опубликует."  # Текст сообщения         )     except Exception as e:  # Обрабатываем возможные исключения при отправке сообщения         print(f"Ошибка при отправке сообщения пользователю: {e}")  # Выводим ошибку в консоль       await callback.message.edit_text(  # Редактируем текст сообщения колбек-запроса         f"{callback.message.text}\n\nСтатус: Одобрен ✅\n\nДля публикации используйте команду:\n/post {suggested_text}"  # Добавляем статус и инструкцию для публикации     )   @dp.callback_query(F.data.startswith("reject_"))  # Обработчик для колбек-запросов, начинающихся с "reject_" async def reject_post(callback: CallbackQuery):  # Асинхронная функция для обработки отклонения поста     if not is_admin(callback.from_user.id):  # Проверяем, является ли пользователь администратором         await callback.answer("У вас нет прав для отклонения постов")  # Если нет, отправляем ответ с сообщением         return  # Завершаем выполнение функции       data_parts = callback.data.split("_")  # Разделяем данные колбек-запроса на части     message_id = data_parts[1]  # Извлекаем идентификатор сообщения из данных     user_id = int(data_parts[2])  # Извлекаем идентификатор пользователя и преобразуем в целое число         try:         await bot.send_message(  # Пытаемся отправить сообщение пользователю о том, что пост отклонен             chat_id=user_id,  # Указываем идентификатор пользователя             text="Ваш пост был отклонен."  # Текст сообщения         )     except Exception as e:  # Обрабатываем возможные исключения при отправке сообщения         print(f"Ошибка при отправке сообщения пользователю: {e}")  # Выводим ошибку в консоль         await callback.message.edit_text(  # Редактируем текст сообщения колбек-запроса         f"{callback.message.text}\n\nСтатус: Отклонен ❌"  # Добавляем статус отклонения     ) 

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

Готово! Наш код вышел на: 166 строк кода.

Чтобы развернуть нашего бота на удалённом сервере, нам необходимо написать requirements.txt, который отвечает за установку всех зависимостей

Выглядеть данный файл будет так:

datetime==5.5 aiogram==3.17.0

Подготовка к запуску на удалённом сервере

Мы осуществим деплой в облаке Amvera, которое предоставляет возможность развертывания через загрузку файлов в интерфейсе или с помощью git push. Кроме того, при регистрации вам будет начислен бонусный баланс в размере 111 рублей, который позволит вам бесплатно пользоваться сервисом в течение тестового периода.

Сам деплой займет у нас не более 5 минут.

Файл amvera.yaml

Мы можем воспользоваться генератором для создания этого файла или просто указать необходимые параметры в интерфейсе личного кабинета в разделе «Конфигурация».

Выбираем окружение Python

Далее вводим версию Python, в нашем случае 3.10
Указываем путь к файлу с зависимостями, requirements.txt

Вводим имя нашего основного файла, в нашем случае main.py

Нажимаем Generate YAML и закидываем получившейся файл в репозиторий с ботом.

Вот так выглядит файл конфигурации:

version: null meta:   environment: python   toolchain:     name: pip     version: 3.11 build:   requirementsPath: requirements.txt run:   scriptName: main.py   persistenceMount: /data   containerPort: 80   servicePort: 80

Деплой через интерфейс

1. Заходим в ЛК Amvera.ru

2. Нажимаем на кнопочку “Создать проект”.

3. Выбираем тип сервиса — приложение, вводим название проекта и выбираем тарифный план.

4. Нажимаем далее.

5. Выбираем файлы, которые нужно закинуть в проект, и перемещаем их в окно загрузки.

Нажмите кнопку «Далее». Появится окно для загрузки файлов, в которое нужно перетащить необходимые файлы. Обратите внимание, что папку venv загружать не следует, так как система создаст её автоматически на основе файла с зависимостями.

В данном случае ничего не нужно делать, так как мы уже загрузили файл amvera.yaml, и все настройки выполняются автоматически. Просто нажмите «Завершить». Начнется процесс сборки, но он завершится с ошибкой, так как мы не добавили токен нашего бота в секреты.

Теперь нам необходимо добавить секрет — токен нашего бота. В методе bot.run() мы указали только слово TOKEN, но теперь нужно создать переменную в качестве секрета. Для этого перейдите на вкладку «Переменные» в проекте и нажмите «Создать секрет». В поле «Название» введите переменную TOKEN, а в поле «Значение» — сам токен.

Помимо TOKEN мы добавляем ещё несколько переменных: ADID, CHID

ADID:

Сюда мы должны написать ваш ID Telegram Account

CHID:

А сюда мы прописываем username нашего телеграм-канала.

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

Рассмотрим альтернативный способ деплоя — деплой через Git

Не менее интересный способ деплоя и более удобный!

Процесс доставки кода с помощью команды git push amvera master позволит нам обновлять проект всего тремя командами в терминале, избавляя от необходимости заходить на сайт облачного сервиса, что делает работу гораздо более удобной.

1. Создаём папку для проекта и помещаем в неё все необходимые файлы.

2. Затем открываем командную строку и переходим в созданную папку с помощью команды cd «путь к папке». После этого инициализируем репозиторий, выполнив соответствующую команду.

git init

Далее заходим на Amvera и создаём новый проект

Теперь выбираем метод “Через git”.

Теперь нам нужно подключиться к существующему репозиторию, для этого копируем команду ниже и вставляем в командную строку.

Необходимо будет ввести логин и пароль от сервиса Amvera. Обратите внимание, что при вводе пароля в терминале он не будет отображаться — это сделано для повышения безопасности.

Затем вводим команды для добавления файлов и создания коммита:

git add .

Не забываем про точку в конце, она обязательна!

git commit -m "initial commit"

Запушим все файлы и сборка начнется автоматически. Для этого вводим команду:

git push amvera master

Если все сделано правильно, после сборки начнётся запуск бота для выкладки постов в Телеграм. В случае ошибки рекомендуется ознакомиться с логами сборки и приложения, а также изучить распространённые ошибки в документации сервиса.

Исходный код бота доступен в репозитории GitHub