Приветствую, друзья!

В рамках цикла публикаций о разработке телеграм-ботов с использованием aiogram 3 я давно хотел осветить несколько ключевых тем: получение Telegram ID пользователей, групп и каналов, рассылка сообщений всем пользователям бота, вход в бота только по подписке на определенный канал или каналы, интеграция базы данных SQLite с помощью aiosqlite и деплой (удаленный запуск бота на сервере или хостинге). Сегодня мы закроем все эти вопросы.

Мы создадим бота с использованием вебхуков (о необходимости и способах их настройки я уже рассказывал в одной из предыдущих статей) и в конце статьи осуществим деплой на сервисе Amvera Cloud. Я выбрал этот сервис из-за бесплатного доменного имени с HTTPS-протоколом, которое автоматически выделяется и привязывается к созданному проекту, а также за простоту деплоя: достаточно загрузить файлы через GIT или внутренний интерфейс, сгенерировать простой файл с настройками прямо на сайте Amvera, и проект автоматически соберется и запустится. Подробности далее.

Зачем нам нужен ID пользователя, группы или канала?

Идентфификаторы Telegram-сущностей часто полезны в разработке ботов и user-ботов (подробнее об этом читайте ТУТ). Вот несколько сценариев, когда это может пригодиться:

  1. Персонализированные уведомления: Отправка сообщений конкретным пользователям или группам на основе их уникальных ID позволяет делать уведомления более персонализированными и целенаправленными.

  2. Целевая рассылка: Использование ID для целевой рассылки сообщений в определенные группы или каналы может быть полезным для продвижения контента или проведения опросов.

  3. Интеграция с внешними системами: Идентификаторы могут быть использованы для интеграции бота с внешними системами, например, CRM или системами аналитики, для отслеживания взаимодействий и результатов.

  4. Управление доступом: Если ваш бот предоставляет доступ к различным функциям или контенту, вы можете использовать ID для контроля доступа и управления правами пользователей.

  5. Аналитика и отчеты: ID позволяют собирать данные о том, как пользователи взаимодействуют с вашим ботом, что важно для анализа и оптимизации его работы.

Эти примеры показывают, насколько важны идентификаторы для эффективного управления и настройки функциональности ботов.

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

В коде, который мы сегодня напишем, вы узнаете, как в Aiogram 3 можно легко создать одну функцию, которая закроет рассылку всех типов сообщений (фото, текст, видео и т.д.). Мы создадим универсальный модуль, который сможете использовать в любом телеграм-боте.

Также мы сделаем универсальным блок, который проверяет, подписан ли пользователь на определенные каналы, и только после проверки предоставляет доступ к функционалу бота.

Так как бот получится достаточно объемным и функциональным, сегодня я планирую сосредоточиться только на написании кода с подробными комментариями. Если вам будут непонятны какие-то части кода, пишите в комментариях здесь или в сообществе моего телеграм-канала «Легкий путь в Python». Там мы вместе с участниками разбираем трудные вопросы и просто общаемся.

Для лучшего понимания темы статьи рекомендую ознакомиться с другимив моими статьями по разработке телеграм-ботов на Python с использованием Aiogram 3. Статьи вы найдете в моем профиле на Хабре.

План на сегодня

Теперь кратко подведем итоги плана на сегодня и приступим к написанию кода:

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

    • Проверка подписки на каналы с последующим открытием доступа к боту.

    • Получение своего ID, ID‑группы, ID‑канала, ID‑любого бота и т. д. через удобную клавиатуру.

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

    • Просмотр списка всех пользователей в админ-панели

  2. Интегрировать в бота базу данных SQLite

  3. Выполнить удаленный запуск бота (на деплой мы потратим около 5 минут, так что обязательно дочитайте до конца).

Подготовка

Для написания кода предлагаю открыть IDE и там создать отдельный проект под нового бота. Я буду использовать Pycharm.

Писать бота мы будем на вебхуках. Так что для удобства его локальной установки установите ngrok-v3. Подробную инструкцию как это сделать и зачем я давал в статье «Telegram Боты на Aiogram 3.x: Простой бот на вебхуках с локальным запуском и деплоем».

Кроме того, вам необходимо будет получить токен бота. Для этого воспользуйтесь телеграмм-ботом BotFather. Там выполните следующие действия:

  1. /newbot – команда для создания нового бота

  2. Пишите имя бота

  3. Придумываете ему логин (обязательно на английском языке и в конце 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 действия:

  1. Создает базу данных SQLITE если она еще не была создана

  2. Создает таблицу с пользователями если таблица не существовала

Для самого создания таблицы использовался простой 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. Она должна будет выполнить следующую логику:

  1. Проверит есть ли пользователь в базе данных (если нет, то выполнит регистрацию)

  2. Получим статус пользователя с базы данных по подпискам (у новых пользователей статус 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"❌ вы не подписались на канал ? {label}",
                                      reply_markup=channels_kb(kb_list))
            return False

    await update_bot_open_status(telegram_id=call.from_user.id, bot_open=True)
    await call.message.answer("Спасибо за подписки на все каналы! Теперь можете воспользоваться функционалом бота",
                              reply_markup=main_contact_kb(call.from_user.id))

Функция срабатывает на call_data == «check_subscription». Логика тут следующая:

  • Пробегаемся по списку с каналом и проверяем подписан ли пользователь на каждый из них. Если мы пройдемся по всем каналам и не впоймаем выход из функции (return False) значит пользователь подписан на все каналы. Иначе мы сообщим пользвателю на какой канал тот не подписан.

  • В случае успеха мы обновим статус пользователя на True и он больше не будет получать клавиатуру с каналами для подписки.

В коде я специально оставил клавиатуру main_contact_kb и мы ее совсем скоро опишем. А пока выполним тесты. Для начала я попробую получить функционал бота без подписки на канал.

Бот меня не пустил
Бот меня не пустил

Теперь выполню подписку на канал и повторю попытку.

Как видите, проверка прошла успешно и я получил шикарную клавиатуру, которую мы напишем совсем скоро. Кстати, давайте заглянем в базу данных (для просмотра я использовал программу DB Browser For SQLite).

Обратите внимание, в отличие от PostgreSQL в SQLite булевы значения обозначены 0 и 1.

Тут мы видим, что данные обновлены и у меня значение bot_open теперь равно True (1).

Теперь опишем логику для получения telegram ID для разных сущностей.

Для этого мы воспользуемся функционалом aiogram 3, а именно специальными текстовыми кнопками, которые будут использоваться для предоставления боту данных по пользователю, группе и так далее.

Для решения этой задачи мы напишем специальную функцию в файле keyboards/kb.py. В эту же функцию мы включим логику для вывода кнопки «Админка», которую будет получать только администратор бота.

Сейчас напишем полный код функции, а после его подробно разберем.

def main_contact_kb(user_id: int):
    buttons = [
        [
            KeyboardButton(
                text="? USER ID",
                request_user=KeyboardButtonRequestUser(
                    request_id=1,
                    user_is_bot=False
                )
            ),
            KeyboardButton(
                text="? BOT ID",
                request_user=KeyboardButtonRequestUser(
                    request_id=4,
                    user_is_bot=True
                )
            )
        ],
        [
            KeyboardButton(
                text="? GROUP ID",
                request_chat=KeyboardButtonRequestChat(
                    request_id=2,
                    chat_is_channel=False,  # Включает только обычные группы (не каналы)
                    chat_has_username=True
                )
            ),
            KeyboardButton(
                text="? CHANNEL ID",
                request_chat=KeyboardButtonRequestChat(
                    request_id=3,
                    chat_is_channel=True  # Включает только каналы
                )
            )
        ],
        [
            KeyboardButton(
                text="? MY INFO",
            )
        ]
    ]

    if user_id == ADMIN_ID:
        buttons.append([
            KeyboardButton(
                text="⚙️ АДМИНКА",
            )
        ])

    keyboard = ReplyKeyboardMarkup(
        keyboard=buttons,
        resize_keyboard=True,
        one_time_keyboard=False,
        input_field_placeholder="По ком получим ID?"
    )

    return keyboard

Структура такой клавиатуры, практически, не отличается от структуры обычной текстовой клавиатуры, но тут появляется два новых параметра в описании кнопок: request_chat и request_user. С помощью этих параметров мы и получим наше волшебство.

Функция main_contact_kb создаёт и возвращает клавиатуру для бота, которая позволяет пользователю получать различные идентификаторы Telegram-сущностей. Вот как она работает:

  1. Определение кнопок:

    • Кнопка «? USER ID»: Запрашивает идентификатор пользователя. Используется параметр request_user, который открывает запрос на получение ID пользователя, если тот не является ботом (user_is_bot=False).

    • Кнопка «? BOT ID»: Запрашивает идентификатор бота. Здесь также используется параметр request_user, но с установкой user_is_bot=True, чтобы получить ID бота.

    • Кнопка «? GROUP ID»: Запрашивает идентификатор группы. Используется параметр request_chat с chat_is_channel=False, чтобы запрос был направлен только на обычные группы (не каналы), и chat_has_username=True, чтобы запрос был на группы с юзернеймом.

    • Кнопка «? CHANNEL ID»: Запрашивает идентификатор канала. Параметр request_chat с chat_is_channel=True указывает, что запрос будет только на каналы.

    • Кнопка «? MY INFO»: Предназначена для запроса информации о текущем пользовател.

  2. Дополнительная кнопка для администраторов:

    • Если user_id совпадает с идентификатором администратора (ADMIN_ID), в клавиатуру добавляется кнопка "⚙️ АДМИНКА". Эта кнопка предназначена для доступа к административной панели бота.

  3. Конфигурация клавиатуры:

    • ReplyKeyboardMarkup используется для создания разметки клавиатуры.

    • Параметр resize_keyboard=True позволяет автоматически подстраивать размер клавиатуры под экран устройства.

    • one_time_keyboard=False указывает, что клавиатура останется видимой после использования (не будет скрыта).

    • input_field_placeholder = "По ком получим ID?" добавляет текст‑подсказку в поле ввода для улучшения пользовательского опыта.

Функция возвращает созданную клавиатуру в виде объекта ReplyKeyboardMarkup, который можно использовать в дальнейшем в обработке сообщений бота.

Request_id для таких кнопок являются обязательным параметром и это значение мы с вами будем использовать далее.

Думаю, что вам уже не терпится написать логику для обработки этих кнопок. Так давайте сделаем это!

await call.message.answer("Спасибо за подписки на все каналы! Теперь можете воспользоваться функционалом бота", reply_markup=main_contact_kb(call.from_user.id))

Принцип сводится к тому, чтоб пользователь получил нашу «волшебную» клавиатуру.

Далее нам необходимо обработать клики по тем или иным кнопкам.

Начнем с простого и понятного.

@router.message(F.text == '? MY INFO')
async def handle_my_id(message: Message):
    user_id = message.from_user.id
    first_name = message.from_user.first_name
    last_name = message.from_user.last_name or "Не указано"
    username = message.from_user.username or "Не указано"

    await message.answer(
        f"? Ваши данные:\n\n"
        f"? ID: <code>{user_id}</code>\n"
        f"? Имя: {first_name} {last_name}\n"
        f"? Username: @{username}"
    )

Это простой обработчик, который будет реагировать на текст «? MY INFO» при помощи магических фильтров Aiogram 3 (F-фильтры).

Все данные о пользователе мы получаем через объект message и возвращаем сообщение с данными о пользователе.

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

@router.message(F.user_shared)
async def handle_user(message: Message):
    user_id = message.user_shared.user_id
    request_id = message.user_shared.request_id

    if request_id == 1:
        await message.answer(f"? Вы выбрали пользователя с ID: <code>{user_id}</code>")
    elif request_id == 4:
        await message.answer(f"? Вы выбрали бота с ID: <code>{user_id}</code>")

Для запуска обработчика я воспользовался магическим фильтром F.user_shared. Далее, при помощи установленных мною же идентификаторов кнопок я проверил были ли запрос на пользователя или на бота.

Получился такой результат для ботов:

Для пользователей:

Попробуем получить данные.

Отработало как я и планировал.

Каналы и группы в телеграмм ботах объединены в «чаты». Этой особенностью мы воспользуемся и напишем один универсальный обработчик, как для каналов, так и для группы.

@router.message(F.chat_shared)
async def handle_chat_or_channel(message: Message):
    chat_id = message.chat_shared.chat_id
    request_id = message.chat_shared.request_id

    if request_id == 2:
        await message.answer(f"? Вы выбрали группу с ID: <code>{chat_id}</code>")
    elif request_id == 3:
        await message.answer(f"? Вы выбрали канал с ID: <code>{chat_id}</code>")

Для запуска этого обработчика я использовал магический фильтр F.chat_shared.

Далее логика похожа на обработчик user_shared.

Получился такой результат для каналов:

Для групп:

Проверим.

Идеально!

Таким образом мы полностью закрыли пользовательский блок в нашем боте. Теперь нам остается написать админ-панель бота.

Админ-панель

В функционал админ-панели бота я решил включить: универсальную рассылку любого типа сообщений всем участникам бота и просмотр списка участников бота.

Для рассылки нам необходимо выполнить некую подготовку и разобраться с тем как рассылка работает.

Как работает рассылка сообщений в телеграмм ботах?

Для рассылки сообщений мы будем использовать объект bot, который мы описали в файле create_bot.py.

Тут у нас будут следующие части:

  • user_id – телеграмм айди пользователя, которому нужно отправить сообщение

  • text – текст в текстовых сообщениях

  • file_id / photo_id – для отправки медиа сообщений

  • caption – для подписи к медиа, если она будет необходима (по умолчанию None)

Теперь напишем специальную утилиту для отправки сообщений (utils/utils.py).

async def broadcast_message(users_data: list, text: str = None, photo_id: int = None, document_id: int = None,
                            video_id: int = None, audio_id: int = None, caption: str = None, content_type: str = None):
    good_send = 0
    bad_send = 0
    for user in users_data:
        try:
            chat_id = user.get('telegram_id')
            if content_type == ContentType.TEXT:
                await bot.send_message(chat_id=chat_id, text=text, reply_markup=main_contact_kb(chat_id))
            elif content_type == ContentType.PHOTO:
                await bot.send_photo(chat_id=chat_id, photo=photo_id, caption=caption,
                                     reply_markup=main_contact_kb(chat_id))
            elif content_type == ContentType.DOCUMENT:
                await bot.send_document(chat_id=chat_id, document=document_id, caption=caption,
                                        reply_markup=main_contact_kb(chat_id))
            elif content_type == ContentType.VIDEO:
                await bot.send_video(chat_id=chat_id, video=video_id, caption=caption,
                                     reply_markup=main_contact_kb(chat_id))
            elif content_type == ContentType.AUDIO:
                await bot.send_audio(chat_id=chat_id, audio=audio_id, caption=caption,
                                     reply_markup=main_contact_kb(chat_id))
            good_send += 1
        except Exception as e:
            print(e)
            bad_send += 1
        finally:
            await asyncio.sleep(1)
    return good_send, bad_send

Давайте разбираться.

Утилита broadcast_message предназначена для отправки сообщений нескольким пользователям в Telegram. Вот как она работает:

  1. Параметры функции:

    • users_data — список словарей с данными пользователей, включая их Telegram ID. Получать этот список мы будем с базы данных (недавно написали для этой цели специальную функцию)

    • text — текст сообщения (если отправляется текстовое сообщение).

    • photo_id, document_id, video_id, audio_id — ID медиафайлов (фото, документы, видео, аудио).

    • caption — подпись для медиафайлов, если она требуется.

    • content_type — тип отправляемого контента (текст, фото, документ, видео, аудио).

  2. Работа функции:

    • Функция проходит по списку пользователей и пытается отправить им сообщение в зависимости от указанного типа контента (content_type).

    • Для текстовых сообщений используется метод send_message.

    • Для медиафайлов (фото, документы, видео, аудио) используются соответствующие методы (send_photo, send_document, send_video, send_audio).

    • К каждому сообщению добавляется клавиатура, возвращаемая функцией main_contact_kb.

  3. Учёт успешных и неудачных отправок:

    • good_send отслеживает количество успешно отправленных сообщений.

    • bad_send отслеживает количество сообщений, отправка которых не удалась.

  4. Задержка и обработка ошибок:

    • После попытки отправки сообщения функция ждёт 1 секунду перед следующей попыткой.

    • Если отправка не удалась, ошибка выводится в консоль.

Функция возвращает кортеж с количеством успешных и неудачных отправок сообщений.

Для проверки типа контента я использовал ContentType из aiogram.enums.

В целом, эта лаконичная функция возьмет на себя всю логику рассылки, а нам останется только грамотно ее вмонтировать в админ-панель.

Код админ-панели (файл handlers/admin_panel.py)

Теперь можем начать писать код админ-панели.

Выполним импорты:

from aiogram import Router, F
from aiogram.enums import ContentType
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State
from aiogram.types import Message, CallbackQuery
from create_bot import ADMIN_ID
from utils.db import get_all_users
from keyboards.kb import admin_kb, cancel_btn
from utils.utils import broadcast_message

Назначим роутер

router = Router()

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

from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import StatesGroup, State

Опишем логику класса, для хранения состояний. Оно у нас одно будет.

class Form(StatesGroup):
    start_broadcast = State()

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

В файле keyboards/kb.py опишем клавиатуры для администратора.

Основная клавиатура:

def admin_kb():
    keyboard = InlineKeyboardMarkup(
        inline_keyboard=[
            [InlineKeyboardButton(text="? Пользователи", callback_data="admin_users")],
            [InlineKeyboardButton(text="? Рассылка", callback_data="admin_broadcast")]
        ]
    )
    return keyboard

Реализовал ее в формате инлайн-клавиатуры с двумя опциями: запуск рассылки и просмотр пользователей.

Напишем клавиатуру для отмены рассылки.

def cancel_btn():
    return ReplyKeyboardMarkup(
        keyboard=[[KeyboardButton(text="❌ Отмена")]],
        resize_keyboard=True,
        one_time_keyboard=False,
        input_field_placeholder="Или нажмите на 'ОТМЕНА' для отмены"
    )

При клике на кнопку «❌ Отмена» будет остановлен сценарий рассылки.

Теперь в файле handlers/admin_panel.py опишем стартовую функцию для входа в админ панель. Тут смысл в том чтоб проверить, что пользователь является администратором бота и, если это так, отправить ему инлайн-клавиатуру администратора.

@router.message((F.from_user.id == ADMIN_ID) & (F.text == '⚙️ АДМИНКА'))
async def admin_handler(message: Message):
    await message.answer('Вам открыт доступ в админку! Выберите действие?', reply_markup=admin_kb())

Для проверки я воспользовался магическим фильтром ((F.from_user.id == ADMIN_ID), объединённый с фильтром, срабатывающим на текст '⚙️ АДМИНКА'.

Далее, если условия фильтра пройдены – пользователь получит админ-клавиатуру.

Начнем с логики «Пользователи».

@router.callback_query((F.from_user.id == ADMIN_ID) & (F.data == 'admin_users'))
async def admin_users_handler(call: CallbackQuery):
    await call.answer('Готовлю список пользователей')
    users_data = await get_all_users()

    text = f'В базе данных {len(users_data)}. Вот информация по ним:\n\n'

    for user in users_data:
        text += f'<code>{user["telegram_id"]} - {user["first_name"]}</code>\n'

    await call.message.answer(text, reply_markup=admin_kb())

В данном обработчике я использовал похожий фильтр, но уже на F.data.

Внутри функции мы получаем список всех пользователей с базы данных.

users_data = await get_all_users()

Затем формируем сообщение через F-строку с циклом FOR. В результате это выглядит так:

Бот работает уже пару дней, а его презентацию с исходным кодом я сделал в своем телеграмм канале «Легкий путь в Python» ещё  27 августа, так что подписывайтесь, если хотите получать доступ к материалам первым.

Отлично. Теперь опишем логику рассылки.

Для старта рассылки опишем следующий обработчик:

@router.callback_query((F.from_user.id == ADMIN_ID) & (F.data == 'admin_broadcast'))
async def admin_broadcast_handler(call: CallbackQuery, state: FSMContext):
    await call.answer()
    await call.message.answer(
        'Отправьте любое сообщение, а я его перехвачу и перешлю всем пользователям с базы данных',
        reply_markup=cancel_btn()
    )
    await state.set_state(Form.start_broadcast)

Тут использовал уже знакомую вам конструкцию с фильтрами.

Затем бот отправляет сообщение 'Отправьте любое сообщение, а я его перехвачу и перешлю всем пользователям с базы данных' с текстовой кнопкой «Отмена» для рассылки.

Сам сценарий ожидания сообщения от админа запускает эта конструкция:

await state.set_state(Form.start_broadcast)

Теперь нам остается обработать 2 ситуации:

  1. Отмена рассылки

  2. Рассылка сообщений всем пользователям

Эти 2 случая мы обработаем в одной функции. Сначала покажу полный код, а после разберемся с ним.

@router.message(F.content_type.in_({'text', 'photo', 'document', 'video', 'audio'}), Form.start_broadcast)
async def universe_broadcast(message: Message, state: FSMContext):
    users_data = await get_all_users()

    # Определяем параметры для рассылки в зависимости от типа сообщения
    content_type = message.content_type

    if content_type == ContentType.TEXT and message.text == '❌ Отмена':
        await state.clear()
        await message.answer('Рассылка отменена!', reply_markup=admin_kb())
        return

    await message.answer(f'Начинаю рассылку на {len(users_data)} пользователей.')

    good_send, bad_send = await broadcast_message(
        users_data=users_data,
        text=message.text if content_type == ContentType.TEXT else None,
        photo_id=message.photo[-1].file_id if content_type == ContentType.PHOTO else None,
        document_id=message.document.file_id if content_type == ContentType.DOCUMENT else None,
        video_id=message.video.file_id if content_type == ContentType.VIDEO else None,
        audio_id=message.audio.file_id if content_type == ContentType.AUDIO else None,
        caption=message.caption,
        content_type=content_type
    )

    await state.clear()
    await message.answer(f'Рассылка завершена. Сообщение получило <b>{good_send}</b>, '
                         f'НЕ получило <b>{bad_send}</b> пользователей.', reply_markup=admin_kb())

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

F.content_type.in_({'text', 'photo', 'document', 'video', 'audio'}

Таким образом мы указали, что данный код должен срабатывать на текстовые сообщения, сообщения с фото, видео и аудио.

Запускаться этот код должен в состоянии Form.start_broadcast.

Теперь по внутренней логике.

users_data = await get_all_users()

Таким образом мы получили список из всех пользователей. В данном списке нас будет интересовать только телеграмм айди пользователя, но, бывают случаи, когда логично прописать обращение к человеку. Поэтому я решил вывести все данные.

После мы извлекаем тип сообщения из объекта message

content_type = message.content_type

Данная переменная нам необходима будет для формирования рассылки.

Далее, если тип сообщения Text и этот текст равен «Отмена» мы прерываем сценарий рассылки. Выглядит это так:

Блок с сообщением «Начинаю рассылку на … пользователей» уже после написания кода и деплоя перенес под проверку на отмену. На скрине старый вариант.

Далее, если сообщение не было равно «Отмена» бот уведомляет админа о том на какое кол-во пользователей тот выполнит рассылку. А после бот скинет отчет о том, какое кол-во пользователей получило сообщение, а какое нет.

Поздороваемся с участниками бота.

Как вы видите, я решил отправить фото с подписью.

Теперь ждем завершения рассылки.

Так как несколько моих аккаунтов подключено к боту – я получил несколько сообщений.

И посмотрим, что там по отчету.

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

Бот готов и остается последний штрих – запустить его удаленно на хостинге Amvera Cloud.

Деплой бота на Amvera Cloud

Как я писал в начале статьи, сервис я выбрал из-за простоты в использовании и бесплатного домена, который не просто дают под проект, но и настраивают за вас https сертификат.

Кроме того, за простую регистрацию в системе вы получите 111 рублей, чего будет достаточно чтоб несколько дней плотно тестировать данный сервис со своими проектами на Python и не только.

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

Подвязать, так-же, можно свое доменное имя. Тоже удобно и просто.

Для запуска своего проекта там есть 2 основных подхода:

  1. Файл настроек проекта, который можно сгенерировать прямо на сайте Amvera (этот подход я буду использовать сегодня)

  2. Создание собственного Dockerfile

Далее вам достаточно доставить файлы вашего проекта с файлом настроек / Dockerfile в созданный проект в Amvera и сервис сам соберет и запустит проект. Работает схожим образом с Heroku, но все ещё проще.

Для доставки можно использовать или GIT (сегодня продемонстрирую) или просто файлы вашего приложения закинуть через внутренний интерфейс прямо на сайте Amvera.

Я буду использовать файл настроек, который сгенерирую на сайте Amvera и GIT для доставки файлов.

Приступим к деплою.

  • Регистрируемся на сайте Amvera Cloud,  если ещё не было регистрации

  • Переходим в проекты

  • Кликаем на «Создать проект». На этом этапе нужно дать имя проекту и выбрать тариф. Я выберу начальный

  • Второй экран (репозиторий) мы пока пропустим и вернемся к нему чуть позже (жмем на «далее»)

  • На появившемся экране остановимся. Так как тут мы будем генерировать файл настроек приложения. Должно получится нечто похожее.

В графе containerPort пропишите необходимый порт. Например 5000
В графе containerPort пропишите необходимый порт. Например 5000
  • После кликаем на «Завершить» и заходим в созданный проект. Там нас будет интересовать вкладка «Настройки». Задача текущего шага – активация домена, который уже выделен под наш проект

  • Копируем домен, он нам пригодится совсем скоро. В моем случае ссылка имеет следующий вид: https://getidbot-yakvenalex.amvera.io

  • Переходим на вкладку «Репозиторий». Там нас будет интересовать ссылка на проект.

Теперь мы возвращаемся в свой локальный проект и сейчас займемся доставкой файлов бота в проект в Amvera, но для этого выполним небольшую подготовку.

Для начала вам необходимо остановить бота и заменить BASE_URL в файле .env

В моем случае запись выглядит так:

BASE_URL=https://getidbot-yakvenalex.amvera.io

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

Выполняем инициализацию локального GIT-репозитория

git init

Привязываем удаленный репозиторий (подставьте свою ссылку):

git remote add amvera https://git.amvera.ru/yakvenalex/getidbot

Получаем файл настроек из удаленного репозитория

git pull amvera master

После этой команды вы получите файл настроек. Проверьте его, чтоб все данные были заполнены корректно. У меня настройки (файл amvera.yml) имеет следующий вид:

---
meta:
  environment: python
  toolchain:
    name: pip
    version: 3.12
build:
  requirementsPath: requirements.txt
run:
  persistenceMount: /data
  containerPort: 5000
  scriptName: aiogram_run.py

Тут обратите внимание, чтоб порт соответствовал порту из файла .env. Кроме того, важно наличие файла requirements.txt – без него проект не сможет собраться и проверьте правильность написания для файла запуска бота (в моем случае это aiogram_run.py).

Теперь отправим файлы в Amvera через GIT

git add .
git commit -m "init commit"
git push amvera master

Проверим доставлены ли файлы.

Все файлы на месте, а это значит, что буквально через 2-3 минуты наш проект соберется, а бот сообщит о том, что он запущен. Пока ждем, мы можем заглянуть во вкладку «Лог сборки» и «Лог приложения».

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

На вкладке «Лог приложения» мы можем увидеть консоль бота.

Для подгрузки новых логов просто кликаем на «Загрузить лог».

Готового бота можно поклацать тут: IDBot Finder Pro

Таким образом мы не только написали бота, но и выполнили его деплой, а это значит, что мы можем переходить к заключению.

Заключение

Если вы дочитали до этого места, поздравляю вас! Сегодня вы освоили создание ботов с админ-панелью, рассылкой и доступом к функционалу после подписки на каналы. Вы узнали, как интегрировать простую базу данных SQLite и использовать функционал клавиатур в aiogram 3 для отправки ID бота, пользователя, канала и многого другого.

Кода сегодня было много, и он не всегда был простым, несмотря на мои старания объяснить все максимально подробно. Поэтому, если у вас остались вопросы по функционалу бота, не стесняйтесь задавать их в комментариях к этой статье или в сообществе моего канала «Легкий путь в Python».

На канале вы найдете эксклюзивный контент, который я не публикую на Хабре, исходники кода всех моих статей и сообщество единомышленников — программистов разного уровня, которые всегда готовы помочь и поделиться опытом.

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

До скорого!

Комментарии (15)


  1. pulivilizator
    30.08.2024 18:08
    +1

    Рассылка в боте через фор? А ты харош мужик, харош


    1. yakvenalex Автор
      30.08.2024 18:08

      А ты как бы делал, через while?)


      1. pulivilizator
        30.08.2024 18:08
        +2

        Вот тебе план так чисто накидал, это конечно не в фор все сувать, но вроде тоже ничего

        1. Админ создал рассылку с датой начала

        2. Рассылка отправляется в базу с этой датой и статусом created напрямую (без консьюмера БД)

        3. Сразу в этом же хэндлере шедулеру шлем сигнал запланировать рассылку id такой-то, время такое-то

        4. Шедулер принимает сообщение, планирует старт, шлет сигнал консьюмеру БД на изменение статуса на scheduled

        5. Из этого же хэндлера шлем загрузчику сигнал "грузи"

        6. Загрузчик берет всех активных юзеров из базы и id-шником рассылки и помещает их в стрим, шлет консьюмеру БД сигнал изменить статус рассылки на "uploaded"

        7. Шедулер в нужное время шлет сигнал рассыльщику "пора"

        8. Рассыльщик получает сигнал (пуш-модель), шлет сигнал консьюмеру БД на изменение статуса на active, получает из стрима (пулл-модель) id юзера и id рассылки, рассылает.

        9. Рассыльщик смотрит метаданные, чтобы понять на каком этапе рассылка (тут мне пока непонятно, я почитаю)

        10. Рассылка заканчивается, рассыльщик шлет сигнал консьюмеру БД на изменение статуса completed


        1. yakvenalex Автор
          30.08.2024 18:08

          Да. В этом плане согласен. Но так вышло, что целевая аудитория моего контента - новички. В своих статьях тему отложенных и фоновых задач я не поднимал, а сам текст вышел на 39 минут и так. Слишком много кода. Для тех кто в теме разбирается - можно брать на вооружение. Благодарю


          1. pulivilizator
            30.08.2024 18:08

            Не лучше ли тогда вообще не показывать рассылки, чем показывать так? Почему бы не начать добавлять в статьи вместо них новые инструменты: очереди, шедулер, и постепенно прийти к статье про нормальную систему рассылок?


            1. yakvenalex Автор
              30.08.2024 18:08

              Я показал максимально простой вариант. Считаю, что для учебных проектов и для небольших подойдет и FOR) В остальном да. Очереди, шуделер и прочее - пока не раскрыл. Тут вроде как цикл был и тема эта выпала. На выходе это 12 или 13-я, если не ошибаюсь)


        1. yakvenalex Автор
          30.08.2024 18:08

          В статье про фоновые задачи в ТГ-ботах возьму твой пример как раз и реализую)


  1. anonymous
    30.08.2024 18:08

    НЛО прилетело и опубликовало эту надпись здесь


    1. yakvenalex Автор
      30.08.2024 18:08

      Спасибо за обратную связь) Все зависит от предпологаемой нагрузки на бота. Если она будет небльшой или средней, то, вполне, подойдет Amvera Cloud (описывал деплой в этой статье). Для более нагруженных ботов достаточно выбрать любой VPS-сервер. Настраивается чуть сложнее, но, на выходе, будет выдерживать большие нагрузки.


  1. Zabrell
    30.08.2024 18:08

    Хотел начать с первой статьи, но не смог(


    1. Dominux
      30.08.2024 18:08

      Вы видели, автор опять налил воды на аж 39 минут скроллинга? Миллион скринов и примеров кода, на который стыдно смотреть. Хотите, чтобы он ещё и это вам попытался раскрыть и облажался, как всегда?


      1. yakvenalex Автор
        30.08.2024 18:08
        +1

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

        Если не нравится — не читайте, а если можете лучше, покажите, как. Написали одну кривую статью, от прочтения которой уже с первых абзацев кровь из глаз течет, и теперь бегаете тут и "облачаете" авторов. В статье нет воды, нет нейронок, и я раскрыл тему так, как посчитал нужным. Не больше и не меньше. Синтаксические, стилистические и орфографические ошибки исправьте лучше в своей заветной статье.

        Для тех, кто не в курсе: Кирилл (Dominux) — 25-летний парень, фанат Гарри Поттера, который еще с моих первых статей приходил под каждую мою статью и рассказывал, какая она отвратительная. То про нейронки говорил, что это они мне "воду" делают, то хвалился, что его единственная 12-минутная статья больше всех моих статей в несколько раз, потом говорил, что я пару недель в программировании, затем просил подробнее писать про ботов. В общем, шизофрения.

        Молодой человек, пожалуйста, займитесь своими делами. Если уж не кодингом, то еще пару фоток в женском платье сделайте.

        Всем мира)


        1. yakvenalex Автор
          30.08.2024 18:08

          "Буквально в последние пару лет модели машинного и, в частности, глубокого обучения, приобрели огромную популярность. Процессы, которые ранее являлись сугубо прерогативой человека и были подвластны только ему, относительно резко получили свою программную имплементацию, и теперь множество таких процессов может быть автоматизировано, ускорено и даже в каком-то роде улучшено (всё же возможности человека в, например, генерации текстов, изображений, видео, аудио и прочего контента также ограничены его способностями и абстрактностью мышления каждого)." - один из абзацев замечательной статьи этого автора. Это кто мне тут про "воду" решил расказать) Да если воду с вашей 12-ти минутной статьи выкинуть, то она сократится да 5 минут))


  1. Verstov
    30.08.2024 18:08

    • все команды админки стоит ловить через специальный @admin_router,для того, чтобы проверка на то, админ ли юзер, была в одном месте - в фильтре роутера

    • выводить список юзеров в сообщении? А если юзеров 20000? В админку выводить количество, и возможность выгрузить csv

    • Если юзеров много, то при рассылке надо ОБЯЗАТЕЛЬНО делать тротлинг. Для этого стоит обрабатывать ошибку TelegramRetryAfter . Можно навесить через middleware сразу на весь проект

    • Деплоить лучше на vps - большему научитесь. Один сервер на 200р/месяц может послужить базой для десятка мини проектов.


    1. yakvenalex Автор
      30.08.2024 18:08

      Вы видели, что текущая статья в прочтении занимает 39 минут. Вам кажется, что тут необходимо было дополнительно описать настройке VPS сервера (за спиной у меня больше 100 ботов размещенных на VPS, интересно чему я должен научиться)? По поводу списка юзеров в сообщении тут согласен. Под admin_router. А у меня как? Вы в код смотрели?