Приветствую, друзья!
В рамках цикла публикаций о разработке телеграм-ботов с использованием aiogram 3 я давно хотел осветить несколько ключевых тем: получение Telegram ID пользователей, групп и каналов, рассылка сообщений всем пользователям бота, вход в бота только по подписке на определенный канал или каналы, интеграция базы данных SQLite с помощью aiosqlite и деплой (удаленный запуск бота на сервере или хостинге). Сегодня мы закроем все эти вопросы.
Мы создадим бота с использованием вебхуков (о необходимости и способах их настройки я уже рассказывал в одной из предыдущих статей) и в конце статьи осуществим деплой на сервисе Amvera Cloud. Я выбрал этот сервис из-за бесплатного доменного имени с HTTPS-протоколом, которое автоматически выделяется и привязывается к созданному проекту, а также за простоту деплоя: достаточно загрузить файлы через GIT или внутренний интерфейс, сгенерировать простой файл с настройками прямо на сайте Amvera, и проект автоматически соберется и запустится. Подробности далее.
Зачем нам нужен ID пользователя, группы или канала?
Идентфификаторы Telegram-сущностей часто полезны в разработке ботов и user-ботов (подробнее об этом читайте ТУТ). Вот несколько сценариев, когда это может пригодиться:
Персонализированные уведомления: Отправка сообщений конкретным пользователям или группам на основе их уникальных ID позволяет делать уведомления более персонализированными и целенаправленными.
Целевая рассылка: Использование ID для целевой рассылки сообщений в определенные группы или каналы может быть полезным для продвижения контента или проведения опросов.
Интеграция с внешними системами: Идентификаторы могут быть использованы для интеграции бота с внешними системами, например, CRM или системами аналитики, для отслеживания взаимодействий и результатов.
Управление доступом: Если ваш бот предоставляет доступ к различным функциям или контенту, вы можете использовать ID для контроля доступа и управления правами пользователей.
Аналитика и отчеты: ID позволяют собирать данные о том, как пользователи взаимодействуют с вашим ботом, что важно для анализа и оптимизации его работы.
Эти примеры показывают, насколько важны идентификаторы для эффективного управления и настройки функциональности ботов.
О преимуществах рассылки в боте, думаю, отдельно говорить не стоит, так как это и так очевидно.
В коде, который мы сегодня напишем, вы узнаете, как в Aiogram 3 можно легко создать одну функцию, которая закроет рассылку всех типов сообщений (фото, текст, видео и т.д.). Мы создадим универсальный модуль, который сможете использовать в любом телеграм-боте.
Также мы сделаем универсальным блок, который проверяет, подписан ли пользователь на определенные каналы, и только после проверки предоставляет доступ к функционалу бота.
Так как бот получится достаточно объемным и функциональным, сегодня я планирую сосредоточиться только на написании кода с подробными комментариями. Если вам будут непонятны какие-то части кода, пишите в комментариях здесь или в сообществе моего телеграм-канала «Легкий путь в Python». Там мы вместе с участниками разбираем трудные вопросы и просто общаемся.
Для лучшего понимания темы статьи рекомендую ознакомиться с другимив моими статьями по разработке телеграм-ботов на Python с использованием Aiogram 3. Статьи вы найдете в моем профиле на Хабре.
План на сегодня
Теперь кратко подведем итоги плана на сегодня и приступим к написанию кода:
Создать бота на вебхуках, который будет выполнять следующие функции:
Проверка подписки на каналы с последующим открытием доступа к боту.
Получение своего ID, ID‑группы, ID‑канала, ID‑любого бота и т. д. через удобную клавиатуру.
Внутренняя админка с возможностью рассылки любого типа сообщений всем пользователям и отчетом по доставке.
Просмотр списка всех пользователей в админ-панели
Интегрировать в бота базу данных SQLite
Выполнить удаленный запуск бота (на деплой мы потратим около 5 минут, так что обязательно дочитайте до конца).
Подготовка
Для написания кода предлагаю открыть IDE и там создать отдельный проект под нового бота. Я буду использовать Pycharm.
Писать бота мы будем на вебхуках. Так что для удобства его локальной установки установите ngrok-v3. Подробную инструкцию как это сделать и зачем я давал в статье «Telegram Боты на Aiogram 3.x: Простой бот на вебхуках с локальным запуском и деплоем».
Кроме того, вам необходимо будет получить токен бота. Для этого воспользуйтесь телеграмм-ботом BotFather. Там выполните следующие действия:
/newbot– команда для создания нового ботаПишите имя бота
Придумываете ему логин (обязательно на английском языке и в конце bot в любом регистре).
Если все прошло корректно вы должны получить такой результат:
Кроме того, предлагаю сразу следовать моей структуре бота, которую я предложу далее. Она позволит сделать код структурированным и удобным для дальнейшей поддержки и понимания написанного в нем.
Структура проекта
tg_bot_project/: Корневая директория проекта.
keyboards/: Содержит файлы для создания клавиатур бота.
__init__.py: Инициализирует пакет.
kb.py: Определяет клавиатуры.
utils/: Утилитарные файлы и функции.
__init__.py: Инициализирует пакет.
db.py: Работа с базой данных.
utils.py: Вспомогательные функции.
handlers/: Обработчики различных команд и событий.
__init__.py: Инициализирует пакет.
start.py: Обработчик команды пользователя
admin_panel.py: Обработчик административных команд.
requirements.txt: Список зависимостей проекта.
env: Файл с конфиденциальными переменными окружения.
aiogram_run.py: Основной скрипт для запуска бота.
create_bot.py: Скрипт для создания и конфигурации бота.
Файл requirements.txt
aiogram==3.11.0
python-decouple==3.8
aiosqlite==0.20.0Как вы видите, в файле зависимостей всего 3 библиотеки:
Aiogram 3.11.0– модуль при помощи которого мы будем писать нашего асинхронного бота на вебхукахPython-decouple– мой любимый модуль для удобно работы с конфиденциальными переменными окружения.Aiosqlite– Библиотека для асинхронной работы с базой данных Aiosqlite (для нашего бота ее будет более чем достаточно).
Для установки в терминале, с корня проекта, пишем:
pip install -r requirements.txtФайл с конфиденциальными переменными окружения .env
BOT_TOKEN=00000:VVVZtSQ9_ToIKD8PmD83JHMnRuXJc3s
ADMIN_ID=5321351707
PORT=5000
HOST=0.0.0.0
BASE_URL=https://grok-url.grВ примере выше я показал какие переменные окружения необходимы в текущем проекте.
BOT_TOKEN– это токен бота, который вы получили через BotFather.ADMIN_ID– это ваш телеграмм айди, который будет использоваться далее для открытия доступа к админ-панели бота.PORT,HOSTиBASE_URL– это данные для подключения веб-хука бота. Для этапа разработки ссылку получим через NGROK.
О том как работать с NGROK и веб-хуками в телеграмм боте на aiogram 3 я подробно описывал в статье «Telegram Боты на Aiogram 3.x: Простой бот на вебхуках с локальным запуском и деплоем». На эту статью я сегодня часто буду ссылаться, так что настоятельно рекомендую ее прочитать.
Выглядит работающий NGROK на Windows так:
На примере выше вы видите, что порт использовался 5000-й. Порт в файле .env должен соответствовать порту, который вы указывали при запуске NGROK (подробно в статье про вебхуки, на которую уже ссылался ранее).
Опишем логику базы данных
Для работы бота нам будет необходима следующая логика:
Добавления пользователя в базу данных
Получение пользователя / пользователей с базы данных
Изменение статуса пользователя на Открыт доступ к боту (не подписан на каналы) / Закрыт доступ к боту (подписан на каналы)
Конечно, можно было бы создать отдельную логику для добавления каналов в базу данных, изменение и удаление их, но я решил сегодня не перегружать и каналы по подписке захардкодить.
Возможно, если будет запрос от подписчиков моего телеграмм канала «Легкий путь в Python», в качестве эксклюзива, опишу отдельно логику по работе с каналами через базу данных.
Кроме того нам нужно будет создать базу данных SQLITE и создать в ней таблицу под пользователей (в логике сегодняшнего бота будет всего одна таблица).
Писать код я буду в файле utils/db.py
Напишем первую функцию.
import aiosqlite
async def initialize_database():
# Подключаемся к базе данных (если база данных не существует, она будет создана)
async with aiosqlite.connect("bot.db") as db:
# Создаем таблицу users, если она не существует
await db.execute("""
CREATE TABLE IF NOT EXISTS users (
telegram_id INTEGER PRIMARY KEY,
username TEXT,
first_name TEXT,
bot_open BOOLEAN DEFAULT FALSE
)
""")
# Сохраняем изменения
await db.commit()Как вы поняли по комментариям, которые я оставил в коде, данная функция при вызове выполняет 2 действия:
Создает базу данных SQLITE если она еще не была создана
Создает таблицу с пользователями если таблица не существовала
Для самого создания таблицы использовался простой SQL-запрос.
На этом примере, так же, вы можете увидеть как работает библиотека aiosqlite.
Во-первых, использует она асинхронный менеджер async with, создавая клиента для работы с базой данных. Клиента я назвал db.
Далее выполняется SQL запрос и для сохранения изменений вызывается db.commit()
В SQL-запросе для создания таблицы я использовал «IF NOT EXIST», чтоб избежать проблем при создании таблицы, если она существовала.
Такая конструкция кода позволит нам привязать функцию к запуску бота и, тем самым, первый запуск бота – поднимет нашу базу данных. Удобно.
В таблице будет всего четыре колонки:
telegram_id: целое уникальное число, колонка в которой будем хранить телеграмм айди пользователяusername: телеграм логин пользователя если есть (если нет то будет залетать None)first_name: имя пользователя, просто для удобства будущего обращенияbot_open: булево значение. Если False, то доступ к боту закрыт (не подписан пользователь на канал / каналы), если True – доступ открыт.
Надеюсь, что понятно объяснил.
Теперь опишем необходимые нам методы.
Первый метод для добавления пользователя в базу данных:
async def add_user(telegram_id: int, username: str, first_name: str):
async with aiosqlite.connect("bot.db") as db:
await db.execute("""
INSERT INTO users (telegram_id, username, first_name)
VALUES (?, ?, ?)
ON CONFLICT(telegram_id) DO NOTHING
""", (telegram_id, username, first_name))
await db.commit()На вход получаем telegram_id, username и first_name.
Логику описал таким образом, чтоб если пользователь уже существовал в базе, то при конликте ничего не происходило «ON CONFLICT (telegram_id) DO NOTHING».
Обрати внимание на синтаксис. В отличие от psycong / asyncpg (работа с PostgreSQL), тут используется не $, а ?.
В остальном логика как в первом случае: открыли соединение, выполнили запрос и сохранили изменения.
Функция для получения всех пользователей с базы данных
async def get_all_users():
async with aiosqlite.connect("bot.db") as db:
cursor = await db.execute("SELECT * FROM users")
rows = await cursor.fetchall()
# Преобразуем результаты в список словарей
users = [
{
"telegram_id": row[0],
"username": row[1],
"first_name": row[2],
"bot_open": bool(row[3])
}
for row in rows
]
return usersВ данном случае мы сохранили результат запроса на получение всех пользователей с таблицы users в переменной cursor.
После мы получили все строки, а затем строки преобразовали в список питоновских словарей для удобства в дальнейшей работе с данными. Особо ничего примечательного.
Напишем похожую функцию, но для получения одного пользователя по его ID:
async def get_user_by_id(telegram_id: int):
async with aiosqlite.connect("bot.db") as db:
cursor = await db.execute("SELECT * FROM users WHERE telegram_id = ?", (telegram_id,))
row = await cursor.fetchone()
if row is None:
return None
# Преобразуем результат в словарь
user = {
"telegram_id": row[0],
"username": row[1],
"first_name": row[2],
"bot_open": bool(row[3])
}
return userТут я немного изменил SQL запрос под получение конкретного пользователя по его telegram_id – "SELECT * FROM users WHERE telegram_id = ?".
Далее функция возвращает или None, если пользователя нет или словарь с данными по пользователю, если он есть в базе данных.
И напишем последнюю функцию под работу с базой данных для изменения статуса bot_open (подписан пользователь на канал / каналы или нет).
async def update_bot_open_status(telegram_id: int, bot_open: bool):
async with aiosqlite.connect("bot.db") as db:
await db.execute("""
UPDATE users
SET bot_open = ?
WHERE telegram_id = ?
""", (bot_open, telegram_id))
await db.commit()Принимает функция telegram_id пользователя и статус, который необходимо установить (True или False).
Этого кода нам будет достаточно для того, чтоб закрыть всю логику внутренней базы данных бота.
Теперь мы готовы к описанию самого бота.
Файл create_bot.py
В этом файле, я, как обычно, опишу стартовые настройки бота и вынесу переменные окружения в обычные переменные.
Импорты:
from aiogram import Bot, Dispatcher
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from decouple import config
from aiogram.fsm.storage.memory import MemoryStorageКак вы видите, я использовал в качестве FSM MemoryStorage. Честно говоря, хотел еще и RedisStorage подкинуть, но решил что и так слишком много информации.
Подробно про FSM и, в частности, его реализацию через Redis я описывал в статье «Telegram Боты на Aiogram 3.x: Все про FSM простыми словами», там я описывал и почему MemoryStorage лучше не использовать в «боевых» проектах.
Bot и Dispatcher я выношу обычно в своих проекта в файл create_bot.py и тут без исплючений.
ParseMode используем чтоб по умолчанию обучить бота «понимать» HTML теги.
Config из decouple (python-decouple) будем использовать для работы с переменными окружения, что в файле .env.
MemoryStorage, как описал выше, будем использовать в качестве машины состояний на этапе рассылки сообщений пользователям бота.
Достанем значения из переменных окружения.
# переменные для работы
ADMIN_ID = int(config('ADMIN_ID'))
BOT_TOKEN = config("BOT_TOKEN")
HOST = config("HOST")
PORT = int(config("PORT"))
WEBHOOK_PATH = f'/{BOT_TOKEN}'
BASE_URL = config("BASE_URL")Инициализируем бота и диспетчера для работы с ними:
bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
dp = Dispatcher(storage=MemoryStorage())Тут, обратите внимание, что в диспетчер мы передали MemoryStorage() – хранилище будующих состояний админа.
Кроме того, в этот файл я добавил и список каналов (в моем случае свой канал). Конечно, нужно было это дело добавлять в базу данных и описывать дополнительную логику, но я решил сэкономить ваше и своем время.
Список каналов для подписки
kb_list = [{'label': 'Легкий путь в Python', 'url': 'https://t.me/PythonPathMaster'}]Каналы для подписки я представил в виде списка питоновских словарей с двумя ключами:
label – назавние канала (значение ключа будет использовано для надписи на инлайн-кнопке)
url – ссылка на канал
Такая структура позволит добавить неограниченное количество каналов в будующую клавиатуру и, совсем скоро, я покажу как мы это используем в нашем коде.
На этом с файлом create_bot.py – все.
Теперь к немного более сложному файлу – aiogram_run.py (основной файл бота или файл запуска).
Он немного сложнее, так как внутри будет описана логика веб-хуков (в основном большинстве своих публикаций я описывал логику через лонг-поллинг).
Файл aiogram_run.py
import logging
from aiogram.types import BotCommand, BotCommandScopeDefault
from aiohttp import web
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
from create_bot import bot, dp, BASE_URL, WEBHOOK_PATH, HOST, PORT, ADMIN_ID
from utils.db import initialize_database
from handlers.start import router as start_router
from handlers.admin_panel import router as admin_router
# Функция для установки командного меню для бота
async def set_commands():
# Создаем список команд, которые будут доступны пользователям
commands = [BotCommand(command='start', description='Старт')]
# Устанавливаем эти команды как дефолтные для всех пользователей
await bot.set_my_commands(commands, BotCommandScopeDefault())
# Функция, которая будет вызвана при запуске бота
async def on_startup() -> None:
# Устанавливаем командное меню
await set_commands()
# Создаем базу данных и таблицу с пользователями, если таблицы не было
await initialize_database()
# Устанавливаем вебхук для приема сообщений через заданный URL
await bot.set_webhook(f"{BASE_URL}{WEBHOOK_PATH}")
# Отправляем сообщение администратору о том, что бот был запущен
await bot.send_message(chat_id=ADMIN_ID, text='Бот запущен!')
# Функция, которая будет вызвана при остановке бота
async def on_shutdown() -> None:
# Отправляем сообщение администратору о том, что бот был остановлен
await bot.send_message(chat_id=ADMIN_ID, text='Бот остановлен!')
# Удаляем вебхук и, при необходимости, очищаем ожидающие обновления
# await bot.delete_webhook(drop_pending_updates=True)
# Закрываем сессию бота, освобождая ресурсы
await bot.session.close()
# Основная функция, которая запускает приложение
def main() -> None:
# Подключаем маршрутизатор (роутер) для обработки сообщений
dp.include_router(start_router)
dp.include_router(admin_router)
# Регистрируем функцию, которая будет вызвана при старте бота
dp.startup.register(on_startup)
# Регистрируем функцию, которая будет вызвана при остановке бота
dp.shutdown.register(on_shutdown)
# Создаем веб-приложение на базе aiohttp
app = web.Application()
# Настраиваем обработчик запросов для работы с вебхуком
webhook_requests_handler = SimpleRequestHandler(
dispatcher=dp, # Передаем диспетчер
bot=bot # Передаем объект бота
)
# Регистрируем обработчик запросов на определенном пути
webhook_requests_handler.register(app, path=WEBHOOK_PATH)
# Настраиваем приложение и связываем его с диспетчером и ботом
setup_application(app, dp, bot=bot)
# Запускаем веб-сервер на указанном хосте и порте
web.run_app(app, host=HOST, port=PORT)
# Точка входа в программу
if __name__ == "__main__":
# Настраиваем логирование (информация, предупреждения, ошибки) и выводим их в консоль
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__) # Создаем логгер для использования в других частях программы
main() # Запускаем основную функциюПодробно я описывал данную структуру файла в статье «Telegram Боты на Aiogram 3.x: Простой бот на вебхуках с локальным запуском и деплоем» сейчас же приведу только короткое его описание.
Тут суть в том, что запускать нашего бота будет мини-веб сервер aiohttp (встроенный модуль в aiogram 3). Для большего понимания я оставил комментарии в коде.
В остальном, отличий от поллинга нет.
Функция, которую мы будем вызывать сразу после запуска бота.
# Функция, которая будет вызвана при запуске бота
async def on_startup() -> None:
# Устанавливаем командное меню
await set_commands()
# Создаем базу данных и таблицу с пользователями, если таблицы не было
await initialize_database()
# Устанавливаем вебхук для приема сообщений через заданный URL
await bot.set_webhook(f"{BASE_URL}{WEBHOOK_PATH}")
# Отправляем сообщение администратору о том, что бот был запущен
await bot.send_message(chat_id=ADMIN_ID, text='Бот запущен!')Вот тут началась наша магия. Данная функция вызывается сразу после запуска бота и вот что она делает:
Устанавливает команды в командное меню
Создает базу данных с таблицей если их ещё не было
Вешает вебхук, используя переменные из create_bot.py
Отправляет сообщение администратору
Удобно, так как всего парой строк, мы решили много вопросов. Сам код старался писать универсальным, чтоб его можно было переиспользовать и в других проектах.
Функция для запуска бота
def main() -> None:
# Подключаем маршрутизатор (роутер) для обработки сообщений
dp.include_router(start_router)
dp.include_router(admin_router)
# Регистрируем функцию, которая будет вызвана при старте бота
dp.startup.register(on_startup)
# Регистрируем функцию, которая будет вызвана при остановке бота
dp.shutdown.register(on_shutdown)
# Создаем веб-приложение на базе aiohttp
app = web.Application()
# Настраиваем обработчик запросов для работы с вебхуком
webhook_requests_handler = SimpleRequestHandler(
dispatcher=dp, # Передаем диспетчер
bot=bot # Передаем объект бота
)
# Регистрируем обработчик запросов на определенном пути
webhook_requests_handler.register(app, path=WEBHOOK_PATH)
# Настраиваем приложение и связываем его с диспетчером и ботом
setup_application(app, dp, bot=bot)
# Запускаем веб-сервер на указанном хосте и порте
web.run_app(app, host=HOST, port=PORT)В этой функции мы включаем два роутера (админка и пользовательская часть), которые напием совсем скоро.
Затем регистрируем (по факту запускаем) функцию, которая будет вызвана при старте и завершении работы бота.
А далее создаем веб-приложение и, уже при его помощи, выполняем запуск бота. В целом, если читали мою статью про создание бота на вебхуках, проблем в понимании быть не должно.
Пользовательская часть бота
Теперь начнем писать пользовательскую часть.
Напоминаю, что в пользовательской части нам необходимо реализовать следующую логику:
Регистрация пользователя в базе данных, если он ещё не был зарегистрирован
Проверка подписки пользователя на канал / каналы с дальнейшим предоставлением доступа к боту
Логика получения телеграмм айди самого пользователя, других пользователей, группы, канала, любого бота и прочее.
Для осуществления этой логики мы уже выполнили некоторую подготовку, когда создали методы для работы с базой данных. Теперь нам необходимо создать специальную клавиатуру, которая будет возвращать телеграмм айди разных сущностей и метод, который будет проверять подписан ли пользователь на указанный канал. И, прежде чем мы приступим к написанию кода, давайте разберемся как это работает в телеграмм ботах, написанных на aiogram 3.
Как работает проверка подписки на канал
Для обычных ботов общий принцип сводится к тому, что бот должен быть участником того или иного канала. User-ботам не обязательно быть администратором канала, чтоб видеть аодписчиков, но это другая история.
Для того чтоб бот мог видеть всех участников канала – он должен являться администратором канала. То есть, для тестов необходимо сделать бота админом канала.
Теперь у нас появляется возможность использовать метод для проверки наличия пользователя в канале.
Для данного метода необходимо передать телеграмм айди пользователя и логин канала. Напишем код в файл utils/utils.py
from aiogram.enums import ContentType, ChatMemberStatus
from create_bot import bot
async def is_user_subscribed(channel_url: str, telegram_id: int) -> bool:
try:
# Получаем username канала из URL
channel_username = channel_url.split('/')[-1]
# Получаем информацию о пользователе в канале
member = await bot.get_chat_member(chat_id=f"@{channel_username}", user_id=telegram_id)
# Проверяем статус пользователя
if member.status in [ChatMemberStatus.MEMBER, ChatMemberStatus.CREATOR, ChatMemberStatus.ADMINISTRATOR]:
return True
else:
return False
except Exception as e:
# Если возникла ошибка (например, пользователь не найден или бот не имеет доступа к каналу)
print(f"Ошибка при проверке подписки: {e}")
return FalseВ рамках логики этого кода мы получаем логин телеграмм канала из его ссылки. В боевых проектах лучше в переменную chat_id подставлять телеграмм айди канала, тем более, сегодня мы научим нашего бота получать телеграмм айди вообще любого канала, группы и пользователя.
Для удобства проверки я импортировал из aiogram.enums объект ChatMemberStatus. Там, по сути, хранятся в строках возможные статусы пользователя (администратор, участник и так далее).
Для получения информации по пользователю в канале я использовал:
member = await bot.get_chat_member(chat_id=f"@{channel_username}", user_id=telegram_id)Далее выполнил простую проверку, попадает ли статус пользователя в подходящие для проверки статусы: создатель канала, участник или администратор. Если попадает, то функция вернет True иначе False.
Как видите, логика достаточно простая.
Теперь в файле keyboards/kb.py напишем клавиатуру, которая будет возвращать каналы для подписки и кнопку «Проверить подписку».
def channels_kb(kb_list: list):
inline_keyboard = []
for channel_data in kb_list:
label = channel_data.get('label')
url = channel_data.get('url')
# Проверка на наличие необходимых ключей
if label and url:
kb = [InlineKeyboardButton(text=label, url=url)]
inline_keyboard.append(kb)
# Добавление кнопки "Проверить подписку"
inline_keyboard.append([InlineKeyboardButton(text="Проверить подписку", callback_data="check_subscription")])
return InlineKeyboardMarkup(inline_keyboard=inline_keyboard)Принимает клавиатура список из каналов (описывали этот список в файле create_bot.py).
Тут я использовал обычный цикл FOR, чтоб пробежаться по списку. Благодаря этому нам не важно сколько каналов в списке – все они попадут в финальную клавиатуру.
В конце, независимо от количества каналов, добавим кнопку «Проверить подписку». Она уже будет содержать, не ссылку на канал, а callback_data «check_subscription». Далее мы напишем обработчик этих данных.
На этом этапе с подготовкой все и мы можем начать писать логику пользовательской части бота.
Файл handlers/start.py
Тут нас будет интересовать функция, которая будет обрабатывать команду /start. Она должна будет выполнить следующую логику:
Проверит есть ли пользователь в базе данных (если нет, то выполнит регистрацию)
Получим статус пользователя с базы данных по подпискам (у новых пользователей статус False). Тут смысл в том, чтоб не давать доступа к основному функционалу бота, если пользователь не подписан на все каналы. У нового пользователя этот статус автоматически False
Далее мы или даем доступ к основному функционалу бота (чуть позже напишем) или отправляем клавиатуру с подпиской на каналы.
from aiogram import Router, F
from aiogram.filters import CommandStart
from aiogram.types import Message, CallbackQuery
from create_bot import kb_list
from utils.db import get_user_by_id, add_user, update_bot_open_status
from keyboards.kb import main_contact_kb, channels_kb
from utils.utils import is_user_subscribed
router = Router()@router.message(CommandStart())
async def start(message: Message):
telegram_id = message.from_user.id
user_data = await get_user_by_id(telegram_id)
if user_data is None:
# Если пользователь не найден, добавляем его в базу данных
await add_user(
telegram_id=telegram_id,
username=message.from_user.username,
first_name=message.from_user.first_name
)
bot_open = False
else:
# Получаем статус bot_open для пользователя
bot_open = user_data.get('bot_open', False) # Второй параметр по умолчанию False
if bot_open:
# Если пользователь подписался на все каналы
await message.answer("Тут основная логика бота, чуть позже напишем")
else:
# Иначе показываем клавиатуру с каналами для подписки
await message.answer(
"Для пользования ботом необходимо подписаться на следующие каналы:",
reply_markup=channels_kb(kb_list)Тут обратите внимание на импорты. В файл start.py необходимо импортировать функции для работы с базой данных, утилиты и функции, которые генерируют наши клавиатуры.
Теперь выполним первый запуск бота.
Нас встретила инлайн клавиатура с одним каналом и кнопкой «Проверить подписку». Опишем логику проверки подписки на канал / каналы.
@router.callback_query(F.data == 'check_subscription')
async def check_subs_func(call: CallbackQuery):
await call.answer('Запускаю проверку подписок на каналы')
for channel in kb_list:
label = channel.get('label')
channel_url = channel.get('url')
telegram_id = call.from_user.id
check = await is_user_subscribed(channel_url, telegram_id)
if check is False:
await call.message.answer(f"❌ вы не подписались на канал Интересные статьи
Интересные статьи
