Помимо написания постов в канал, должна быть обратная связь для предложений улучшения канала или идей для новый постов. Для обратной связи часто используются специальные сервисы, которые помогают поддерживать связь с подписчиками канала. Но не всегда хочется давать сторонним сервисам права администратора в канале. И сегодня мы попробуем не просто использовать такой сервис, а написать свой, который сможем улучшать под собственные требования и полностью контролировать его работу. В этой статье мы напишем и развернем на удаленном сервере Telegram бота, который обладает обратной связью и помогает выкладывать посты в канал через него.
Бот будет выкладывать посты через команду, либо же присылать нам предложения от подписчиков.
Функциональность бота
Бот будет присылать вам уведомления, когда будут приходить предложения от подписчиков, а также легко вы сможете отправить свой пост в канал с помощью него.
Пример работы:
После завершения разработки кода бота мы запустим его в облаке Amvera.
Amvera представляет собой "app engine" и предлагает
Возможность деплоя через загрузку файлов в интерфейсе (или через push в привязанный репозиторий) с автоматической настройкой переменных, доменов с https и всего необходимого, что проще использования VPS.
Встроенные бэкапы, облачное логирование с синтаксическим поиском, метрики и алерты.
Бесплатное прокси до ведущих LLM и свой инференс LLaMA 70B.
Стартовый баланс в 111 р. на тесты.
Исходный код бота доступен по ссылке репозитория GitHub
Поехали.
Создание бота:
Зарегистрируем бота в @BotFather и получим токен по данному примеру:
Успешно! Мы создали бота, нам необходимо сохранить его токен.
Написание бота с использованием aiogram v3
Установим все необходимые библиотеки на локальном уровне. Хотя можно воспользоваться библиотекой telebot, но aiogram будет более подходящим выбором для начинающих из-за своей простоты.
pip install datetime aiogramПереходим в командную строку и прописываем данную команду выше.
Теперь приступим к написанию самого бота
Создаём файл main.py
Импортируем все необходимые библиотеки.
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Деплой через интерфейс
Заходим в ЛК Amvera.ru
Нажимаем на кнопочку “Создать проект”.
Выбираем тип сервиса — приложение, вводим название проекта и выбираем тарифный план.
Нажимаем далее.
Выбираем файлы, которые нужно закинуть в проект, и перемещаем их в окно загрузки.
Нажмите кнопку "Далее". Появится окно для загрузки файлов, в которое нужно перетащить необходимые файлы. Обратите внимание, что папку venv загружать не следует, так как система создаст её автоматически на основе файла с зависимостями.
В данном случае ничего не нужно делать, так как мы уже загрузили файл amvera.yaml, и все настройки выполняются автоматически. Просто нажмите «Завершить». Начнется процесс сборки, но он завершится с ошибкой, так как мы не добавили токен нашего бота в секреты.
Теперь нам необходимо добавить секрет — токен нашего бота. В методе bot.run() мы указали только слово TOKEN, но теперь нужно создать переменную в качестве секрета. Для этого перейдите на вкладку «Переменные» в проекте и нажмите «Создать секрет». В поле «Название» введите переменную TOKEN, а в поле «Значение» — сам токен.
Помимо TOKEN мы добавляем ещё несколько переменных: ADID, CHID
ADID:
Сюда мы должны написать ваш ID Telegram Account
CHID:
А сюда мы прописываем username нашего телеграм-канала.
После добавления всех необходимых переменных перезапустим проект и получим вот такой статус.
Рассмотрим альтернативный способ деплоя - деплой через Git
Не менее интересный способ деплоя и более удобный!
Процесс доставки кода с помощью команды git push amvera master позволит нам обновлять проект всего тремя командами в терминале, избавляя от необходимости заходить на сайт облачного сервиса, что делает работу гораздо более удобной.
Создаём папку для проекта и помещаем в неё все необходимые файлы.
Затем открываем командную строку и переходим в созданную папку с помощью команды cd "путь к папке". После этого инициализируем репозиторий, выполнив соответствующую команду.
git initДалее заходим на Amvera и создаём новый проект
Теперь выбираем метод “Через git”.
Теперь нам нужно подключиться к существующему репозиторию, для этого копируем команду ниже и вставляем в командную строку.
Необходимо будет ввести логин и пароль от сервиса Amvera. Обратите внимание, что при вводе пароля в терминале он не будет отображаться — это сделано для повышения безопасности.
Затем вводим команды для добавления файлов и создания коммита:
git add .Не забываем про точку в конце, она обязательна!
git commit -m "initial commit"Запушим все файлы и сборка начнется автоматически. Для этого вводим команду:
git push amvera masterЕсли все сделано правильно, после сборки начнётся запуск бота для выкладки постов в Телеграм. В случае ошибки рекомендуется ознакомиться с логами сборки и приложения, а также изучить распространённые ошибки в документации сервиса.