Как сделать оплату в Telegram боте-магазине на вебхуках

Друзья, приветствую!

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

Ретроспектива прошлой статьи

Напоминаю, в прошлый раз мы шаг за шагом разобрали процесс создания Telegram-бота на Python с использованием SQLAlchemy 2 и Aiogram 3. Этот бот включал:

• Админ-панель для управления товарами и просмотра статистики.

• Каталог товаров с возможностью покупки через ЮКасса без выхода на сторонние ресурсы.

• Личный профиль пользователя, где отображались купленные товары.

Вот как бот выглядит в работе:

Ссылка на работающего бота первой версии: https://t.me/DigitalMarketAiogramBot (решил не отключать)

Как я говорил в прошлой статье, теперь для Telegram-ботов, продающих цифровые товары, обязательно использование системы оплаты Telegram Stars. Хотя блокировок за использование классических систем оплаты лично я пока не встречал, внедрение Telegram Stars становится необходимым для соответствия требованиям платформы.

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

Сегодня мы:

1. Внедрим Telegram Stars для цифровых товаров.

2. Добавим поддержку Robokassa с применением вебхуков для обработки платежей.

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

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

Теперь подробно о плане работы:

1. Переписываем проект с лонг-поллинга на вебхуки

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

2. Интеграция Telegram Stars

Реализуем систему оплаты, соответствующую новым требованиям Telegram.

3. Интеграция Robokassa

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

4. Небольшие правки в функционале

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

5. Деплой проекта

Разберем, как развернуть проект с использованием технологии вебхуков. Особое внимание уделим процессу деплоя, так как для работы вебхуков необходим домен с поддержкой HTTPS.

В процессе деплоя таких ботов можно выделить 2 основных подхода:

Вариант 1: VPS-сервер

1. Покупка и настройка VPS.

2. Приобретение доменного имени.

3. Настройка сервера (NGINX/APACHE).

4. Установка HTTPS.

5. Перенос файлов бота на сервер.

6. Настройка и запуск проекта.

Вариант 2: Хостинг (на примере Amvera Cloud)

1. Загрузка файлов бота на сервис через сайт или Git.

2. Активация бесплатного доменного имени.

3. Запуск проекта.

Так как второй способ проще, я буду демонстрировать деплой именно с использованием Amvera Cloud.

Что такое вебхуки и в чем отличие от поллинга?

Понятие вебхуков и поллинга присутствует не только в контексте Telegram-ботов, но и в мире программирования в целом.

Поллинг

Суть подхода поллинга в том, что мы самостоятельно и с определенными интервалами опрашиваем сервисы на предмет появления новых обновлений. Вот два примера:

• Телеграмм-бот через поллинг: бот постоянно опрашивает сервера Telegram о новых событиях. Если обновления есть (например, клиент нажал кнопку «Мои покупки»), сервер сообщает боту: «Есть событие, пользователь с ID 123456 нажал на кнопку». После этого бот выполняет нужное действие. Если обновлений нет, запрос повторяется через определенный интервал времени. Это происходит даже в периоды бездействия пользователей.

• Платежная система через поллинг: после выставления счета клиенту, вы начинаете регулярно отправлять запросы на сервер платежной системы, чтобы узнать статус платежа. Как только система возвращает статус «оплачено», процесс завершается.

Вебхуки

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

Примеры:

• Телеграмм-бот через вебхуки: как только сервер Telegram фиксирует событие (например, пользователь нажал на кнопку), он отправляет уведомление на указанный вебхук. Бот моментально обрабатывает событие без дополнительных запросов.

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

Преимущества вебхуков

• Экономия ресурсов: минимизация количества запросов.

• Мгновенная обработка событий: данные поступают сразу после изменения.

• Подходит для проектов с высокой нагрузкой: снижение нагрузки на сервер.

Вот пример из реальной жизни. Представим, что вы узнаете у своего друга новости двумя способами:

• Поллинг: вы каждые 5 секунд спрашиваете друга: «Что нового?».

• Вебхук: друг сам сообщает вам новости, когда они появляются.

Как вы понимаете, вебхуки — это не только удобнее, но и значительно эффективнее.

Подготовка к началу работы

К этому моменту я предполагаю, что вы уже ознакомились с первой статьей и у вас есть:

1. Токен Telegram-бота, который вы получили через @BotFather.

2. Привязанная система оплаты ЮКасса к боту, настроенная через @BotFather (Provider token).

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

Теперь приступим к подготовке нового проекта.

Установка и настройка веб-сервера

Сегодня мы будем работать с вебхуками, что требует запуска веб-сервера и обеспечения доступа к нему через доменное имя с HTTPS-протоколом.

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

Для корректной работы вебхуков необходимо доменное имя с HTTPS-протоколом. На этапе разработки мы будем использовать NGROK или любой другой туннель, с которым вы знакомы.

Связка туннеля и веб-сервера

1. Запускаем веб-сервер на локальном порту (например, 8000).

2. Подключаем туннель (например, NGROK) к этому порту.

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

Таким образом, ваш локальный веб-сервер становится доступным для внешних запросов. Например, сервер Telegram или Robokassa сможет отправлять запросы на ваш веб-сервер, а вам останется их обработать. В статье «Бесплатный домен с HTTPS для локальных приложений: универсальное руководство с примером на Flask» я более подробно рассматривал работу тунелей на более простом примере.

После деплоя мы перейдем на постоянное доменное имя. В моем случае это бесплатное доменное имя от Amvera Cloud, предоставляемое вместе с HTTPS.

Установка NGROK

1. Перейдите на официальный сайт NGROK.

2. Зарегистрируйтесь и войдите в профиль.

3. Перейдите на вкладку установки и выберите вашу операционную систему (например, Windows).

4. Скачайте установочный файл и выполните установку.

5. Скопируйте строку с токеном авторизации из вашего профиля — она понадобится для настройки.

6. Запустите NGROK командой:

   ngrok http 8000

   Здесь 8000 — порт, на котором будет работать ваш веб-сервер. Можно указать свой.

После успешного запуска NGROK выдаст ссылку вида https://<ваш-домен>.ngrok.io. На этапе разработки используйте её как временное доменное имя.

Подготовка к интеграции оплаты в боте

Для интеграции платежной системы Robokassa выполните следующие шаги:

1. Зарегистрируйтесь на сайте Robokassa.

2. Создайте новый «Магазин» в личном кабинете.

3. Заполните данные о магазине. Для тестовых платежей важно настроить:

   • Два пароля для тестовых платежей.

   • Алгоритм расчета хэша — MD5 (рекомендуется для начального этапа).

Эти пароли будут использоваться как токены для взаимодействия с Robokassa (подробнее разберем на этапе разбора хука).

Установка проекта

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

1. Клонируйте репозиторий:

   git clone https://githuйb.com/Yakvenalex/DigitalMarketBot.git

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

2. Перейдите в директорию проекта:

   cd DigitalMarketBot

3. Создайте и активируйте виртуальное окружение:

   python -m venv venv source venv/bin/activate  # Для Windows: venv\Scripts\activate

4. Установите зависимости:

   pip install -r requirements.txt

5. Создайте файл .env в корне проекта и заполните его следующими данными:

   BOT_TOKEN=ВАШ_ТОКЕН ADMIN_IDS=[ADMIN_TG1, ADMIN_TG2, ADMIN_TG3] PROVIDER_TOKEN=ТОКЕН_Ю_КАССЫ SITE_URL=ВЕБ_САЙТ_ДЛЯ_ХУКОВ SITE_HOST=0.0.0.0 SITE_PORT=8000 MRH_LOGIN=ЛОГИН_С_РОБОКАССЫ MRH_PASS_1=ПАРОЛЬ_1_С_РОБОКАССЫ MRH_PASS_2=ПАРОЛЬ_2_С_РОБОКАССЫ IN_TEST=1

Вот комментарии по переменным в .env:

1. BOT_TOKEN: Токен вашего Telegram-бота, который вы получили через @BotFather. Обеспечивает авторизацию бота на платформе Telegram.

2. ADMIN_IDS: Список ID администраторов, которые будут управлять ботом. Указывается в формате списка: [ID1, ID2, ID3].

3. PROVIDER_TOKEN: Токен платежной системы ЮКасса, привязанный к вашему боту через @BotFather. Используется для обработки платежей.

4. SITE_URL: Полный URL вашего веб-сайта, который будет принимать вебхуки. На этапе разработки это может быть ссылка от NGROK.

5. SITE_HOST: Хост для вашего локального сервера. Обычно оставляют значение 0.0.0.0, чтобы сервер принимал запросы на всех интерфейсах.

6. SITE_PORT: Порт, на котором запускается веб-сервер. В примере указан порт 8000.

7. MRH_LOGIN: Логин от Robokassa, используемый для идентификации вашего магазина в системе.

8. MRH_PASS_1: Первый пароль для взаимодействия с Robokassa. Используется для формирования хэша при валидации запросов.

9. MRH_PASS_2: Второй пароль для взаимодействия с Robokassa. Используется для дополнительных проверок безопасности.

10. IN_TEST: Флаг тестового режима. Устанавливается в 1 для тестовых платежей и в 0 для боевого режима.

Теперь проект готов к дальнейшей разработке!

Переводим бота на веб-хуки

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

Начнем с изменений в конфигурации. Откройте файл bot/config.py и добавьте новые параметры в класс Settings:

class Settings(BaseSettings):     BOT_TOKEN: str     ADMIN_IDS: List[int]     PROVIDER_TOKEN: str     FORMAT_LOG: str = "{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}"     LOG_ROTATION: str = "10 MB"     DB_URL: str = 'sqlite+aiosqlite:///data/db.sqlite3'     SITE_URL: str     SITE_HOST: str     SITE_PORT: int     MRH_LOGIN: str     MRH_PASS_1: str     MRH_PASS_2: str     IN_TEST: int     model_config = SettingsConfigDict(         env_file=os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", ".env")     )      @property     def get_webhook_url(self) -> str:         """Динамически формирует путь для вебхука на основе токена и URL сайта."""         return f"{self.SITE_URL}/{self.BOT_TOKEN}"      @property     def get_provider_hook_url(self) -> str:         """Динамически формирует путь для вебхука на основе токена и URL сайта."""         return f"{self.SITE_URL}/robokassa"

Новые параметры:

• SITE_URL, SITE_HOST, SITE_PORT — для настройки веб-сервера.

• MRH_LOGIN, MRH_PASS_1, MRH_PASS_2 — для интеграции с RoboKassa (позже будет подробно разобрано).

• IN_TEST — режим тестирования.

Свойство get_webhook_url автоматически формирует URL для вебхука, упрощая управление настройками.

Создаем структуру веб-приложения

В папке bot создайте пакет app с файлом __init__.py. Внутри этой папки будет следующая структура:

app/     __init__.py     app.py    # Обработчики для Aiohttp     utils.py  # Утилиты для работы веб-сервера

Реализация обработчика вебхуков

В файле app.py реализуем основной обработчик вебхуков. Начнем с импортов:

from aiohttp import web from aiogram.types import Update from loguru import logger from bot.config import bot, dp, settings

Добавим обработчик:

async def handle_webhook(request: web.Request):     try:         update = Update(**await request.json())         await dp.feed_update(bot, update)         return web.Response(status=200)     except Exception as e:         logger.error(f"Ошибка при обработке вебхука: {e}")         return web.Response(status=500)

Как это работает:

• Запрос от Telegram приходит на сервер и преобразуется в объект Update.

• Диспетчер (dp) обрабатывает обновление через метод feed_update.

• В случае успеха возвращается статус 200 (успешно), а при ошибке — статус 500.

Метод dp.feed_update в контексте Telegram-ботов выполняет ключевую задачу — он принимает объект обновления (Update) и передает его обработчикам, зарегистрированным в диспетчере (Dispatcher), для выполнения соответствующих действий.

Дальше нам останется зарегистрировать этот обработчик, чем мы займемся совсем скоро.

Пример рендеринга HTML страницы

Для демонстрации возможностей Aiohttp добавим простой рендер HTML-страницы:

# Функция для обработки запроса на эндпоинт главной страницы веб-сервера async def home_page(request: web.Request) -> web.Response:     """     Обработчик для отображения главной страницы с информацией о сервисе.     """     current_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")      html_content = f"""     <!DOCTYPE html>     <html lang="ru">     <head>         <meta charset="UTF-8">         <meta name="viewport" content="width=device-width, initial-scale=1.0">         <title>aiohttp Демонстрация</title>         <style>             body {{ font-family: Arial, sans-serif; line-height: 1.6; padding: 20px; max-width: 800px; margin: 0 auto; }}             h1 {{ color: #333; }}             .info {{ background-color: #f4f4f4; padding: 15px; border-radius: 5px; margin-top: 20px; }}         </style>     </head>     <body>         <h1>Привет, это демонстрация Aiohttp</h1>         <p>Этот сервер обрабатывает:</p>         <ul>             <li>Хуки от Telegram-бота</li>             <li>Хуки от RoboKassa</li>         </ul>         <p>Текущее время сервера: {current_time}</p>     </body>     </html>     """     return web.Response(text=html_content, content_type='text/html')

Этот пример демонстрирует, что Aiohttp может обрабатывать запросы и рендерить HTML-страницы, сравнимо с Flask.

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

Настройка главного файла (bot/main.py)

Теперь нам остается внести правки в main-файл (файл bot/main.py) с которого будет происходить запуск бота. Для удобства чтения кода я немного реорганизовал некоторые старые функции и добавил новые. Вот что получилось.

Импорты:

from aiogram.webhook.aiohttp_server import setup_application from aiohttp import web from aiogram.types import BotCommand, BotCommandScopeDefault from loguru import logger from bot.app.app import handle_webhook, robokassa_result, robokassa_fail, home_page from bot.config import bot, admins, dp, settings from bot.dao.database_middleware import DatabaseMiddlewareWithoutCommit, DatabaseMiddlewareWithCommit from bot.admin.admin import admin_router from bot.user.user_router import user_router from bot.user.catalog_router import catalog_router 

Из того что не разбирали в прошлой статье — это импорт setup_application и импорт from aiohttp import web с этим разберемся подробнее.

Разбор импортов

1. from aiogram.webhook.aiohttp_server import setup_application:

   • Этот импорт предоставляет функцию setup_application, которая интегрирует диспетчер (Dispatcher) и бота (Bot) с приложением Aiohttp.

   • Она связывает сервер Aiohttp с обновлениями Telegram, позволяя обработчикам из диспетчера реагировать на входящие вебхуки.

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

     setup_application(app, dp, bot=bot)

     После вызова этой функции сервер Aiohttp начинает обрабатывать запросы, отправляемые Telegram, и передавать их в диспетчер для дальнейшей обработки.

2. from aiohttp import web:

   • Этот импорт предоставляет все необходимые инструменты для работы с библиотекой Aiohttp, включая:

     • web.Application — создание экземпляра веб-приложения.

     • web.Request — объект для работы с входящими запросами.

     • web.Response — объект для формирования ответов сервера.

   • Aiohttp используется для реализации полноценного веб-сервера, способного обрабатывать как вебхуки Telegram, так и другие типы HTTP-запросов.

В совокупности, эти импорты обеспечивают связь между сервером Telegram, веб-сервером Aiohttp и ботом, реализованным через Aiogram.

Кроме того есть ряд импортов из папкета app. Метод handle_webhook и home_page мы уже реализовали. Остальные методы реализуем чуть позже.

Теперь приступаем к коду. Там где все просто — комментарии будут в коде. Места которые будут требовать дополнительных обсуждений обсудим подробнее:

# Функция для установки команд по умолчанию для бота async def set_default_commands():     """     Устанавливает команды по умолчанию для бота.     """     commands = [BotCommand(command='start', description='Запустить бота')]     await bot.set_my_commands(commands, BotCommandScopeDefault())

# Функции для запуска и остановки бота async def on_startup(app):     """     Выполняется при запуске приложения.     """     await set_default_commands()     await bot.set_webhook(settings.get_webhook_url)     for admin_id in admins:         try:             await bot.send_message(admin_id, 'Бот запущен 🥳.')         except Exception as e:             logger.error(f"Не удалось отправить сообщение админу {admin_id}: {e}")     logger.info("Бот успешно запущен.")

Тут вы видите видоизмененную логику функции start_bot с первой части. Из нового тут добавилась строка await bot.set_webhook(settings.get_webhook_url).

Строка await bot.set_webhook(settings.get_webhook_url) регистрирует URL вебхука для бота в Telegram. Это сообщает Telegram, куда отправлять обновления (например, сообщения или события) — на указанный адрес (settings.get_webhook_url).

Таким образом, Telegram начинает передавать данные напрямую на ваш сервер через POST-запросы.

async def on_shutdown(app):     """     Выполняется при остановке приложения.     """     for admin_id in admins:         try:             await bot.send_message(admin_id, 'Бот остановлен. Почему? 😔')         except Exception as e:             logger.error(f"Не удалось отправить сообщение админу {admin_id}: {e}")     await bot.delete_webhook(drop_pending_updates=True)     await bot.session.close()     logger.error("Бот остановлен!")

Это видоизмененная функция stop_bot из первой части. Из нового — я вынес закрытие сессии прямо в функцию, а не в блок Try Except, как было в первой версии.

# Регистрация мидлварей и роутеров def register_middlewares():     """     Регистрирует мидлвари для диспетчера.     """     dp.update.middleware.register(DatabaseMiddlewareWithoutCommit())     dp.update.middleware.register(DatabaseMiddlewareWithCommit())       def register_routers():     """     Регистрирует маршруты для диспетчера.     """     dp.include_router(catalog_router)     dp.include_router(user_router)     dp.include_router(admin_router)

Регистрацию мидлварей и роутеров (маршрутов) бота вынес в отдельные функции, просто для удобства поддержки кода.

Теперь опишем функцию для создания приложения Aiohttp (веб-сервера)

# Функция для создания приложения aiohttp def create_app():     """     Создает и настраивает приложение aiohttp.     """     # Создаем приложение     app = web.Application()      # Регистрация обработчиков маршрутов     app.router.add_post(f"/{settings.BOT_TOKEN}", handle_webhook)     app.router.add_post("/robokassa/result/", robokassa_result)     app.router.add_get("/robokassa/fail/", robokassa_fail)     app.router.add_get("/", home_page)      # Настройка приложения с диспетчером и ботом     setup_application(app, dp, bot=bot)      # Регистрация функций запуска и остановки     app.on_startup.append(on_startup)     app.on_shutdown.append(on_shutdown)      return app

Как работает create_app:

• Создает экземпляр приложения Aiohttp (web.Application).

• Регистрирует маршруты, которые обрабатывают запросы от Telegram, Robokassa и отображают главную страницу.

• Настраивает интеграцию диспетчера и бота через setup_application.

• Добавляет функции, выполняемые при запуске (on_startup) и остановке (on_shutdown).

Теперь нам остается лишь объединить все функции в единую структуру и запустить веб-сервер. Именно он обеспечит работу нашего бота. Для этого создаем основную функцию:

# Главная функция def main():     """     Главная функция для запуска приложения.     """     # Регистрация мидлварей и роутеров     register_middlewares()     register_routers()      # Создаем приложение и запускаем его     app = create_app()     web.run_app(app, host=settings.SITE_HOST, port=settings.SITE_PORT)

На этом этапе мы подключаем мидлвари и маршруты, создаем веб-приложение на базе Aiohttp и запускаем его. Сервер будет работать на заданном хосте (в нашем случае — 0.0.0.0) и порте, соответствующем настройкам NGINX.

Для выполнения приложения достаточно указать:

if __name__ == "__main__":     main()

Благодаря этой оптимизации и добавлению нескольких строк кода, бот теперь функционирует через вебхуки, избавляясь от необходимости использования метода поллинга. На переход к новой технологии уходит не более 10–15 минут.

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

python -m bot.main

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

Для дополнительного тестирования вы можете перейти по текущему доменному имени, на главную страницу сайта. Там вы должны увидеть похожий результат:

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

Подключаем в боте оплату Telegram Stars (звезды) и адаптируем старый код

Наш текущий функционал веб-хука полностью покрывает потребности бота для корректной обработки внутренних платежей. Это включает интеграцию с Юкассой (реализованной через BotFather) и Telegram Stars — платежной системы, встроенной в Telegram по умолчанию.

Однако для добавления поддержки Robocassa нам потребуется написать дополнительные методы и организовать ещё один веб-хук на нашем сервере, работающем на Aiohttp.

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

@catalog_router.message(F.content_type == ContentType.SUCCESSFUL_PAYMENT) 

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

Для этого создадим в папке bot/user файл utils.py и заполним его следующим образом:

from aiogram import Bot from loguru import logger from sqlalchemy.ext.asyncio import AsyncSession from bot.config import settings from bot.dao.dao import PurchaseDao, ProductDao from bot.user.kbs import main_user_kb from bot.user.schemas import PaymentData   async def successful_payment_logic(session: AsyncSession, payment_data, currency, user_tg_id, bot: Bot):     product_id = int(payment_data.get("product_id"))     price = payment_data.get("price")     payment_type = payment_data.get("payment_type")     payment_id = payment_data.get("payment_id")     user_id = payment_data.get("user_id")     await PurchaseDao.add(session=session, values=PaymentData(**payment_data))     product_data = await ProductDao.find_one_or_none_by_id(session=session,                                                            data_id=product_id)      # Отправка уведомлений администраторам     for admin_id in settings.ADMIN_IDS:         try:             await bot.send_message(                 chat_id=admin_id,                 text=(                     f"💲 Пользователь c ID {user_id} купил товар <b>{product_data.name}</b> (ID: {product_id}) "                     f"за <b>{price} {currency}</b>."                 )             )         except Exception as e:             logger.error(f"Ошибка при отправке уведомления администраторам: {e}")      # Отправка информации пользователю     file_text = "📦 <b>Товар включает файл:</b>" if product_data.file_id else "📄 <b>Товар не включает файлы:</b>"     product_text = (         f"🎉 <b>Спасибо за покупку!</b>\n\n"         f"🛒 <b>Информация о вашем товаре:</b>\n"         f"━━━━━━━━━━━━━━━━━━\n"         f"🔹 <b>Название:</b> <b>{product_data.name}</b>\n"         f"🔹 <b>Описание:</b>\n<i>{product_data.description}</i>\n"         f"🔹 <b>Цена:</b> <b>{price} {currency}</b>\n"         f"🔹 <b>Закрытое описание:</b>\n<i>{product_data.hidden_content}</i>\n"         f"━━━━━━━━━━━━━━━━━━\n"         f"{file_text}\n\n"         f"ℹ️ <b>Информацию о всех ваших покупках вы можете найти в личном профиле.</b>"     )      if product_data.file_id:         await bot.send_document(document=product_data.file_id,                                 chat_id=user_tg_id,                                 caption=product_text, reply_markup=main_user_kb(user_tg_id))      else:         await bot.send_message(chat_id=user_tg_id, text=product_text, reply_markup=main_user_kb(user_tg_id))      # автоматический возврат звезд за покупку     if payment_type == 'stars':         await bot.refund_star_payment(user_id=user_tg_id, telegram_payment_charge_id=payment_id)

Логика здесь практически идентична предыдущему подходу, поэтому, если вы читали прошлую статью, вопросов у вас возникнуть не должно. Единственное отличие — я заменил прямую обработку объекта Message на использование отдельных переменных. Этот шаг может показаться избыточным сейчас, но станет очевидным и обоснованным, когда мы перейдём к созданию обработчика для Robocassa. Такой подход обеспечивает гибкость и упрощает последующую реализацию.

Из абсолютно нового подхода — это эти строки кода:

if payment_type == 'stars':         await bot.refund_star_payment(user_id=user_tg_id, telegram_payment_charge_id=payment_id)

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

Сам метод достаточно простой. В него необходимо передать телеграмм айди пользователя и telegram_payment_charge_id. В прошлой статье я уже показал как извлекать этот параметр.

Что касается системы звёзд, то я установил для всех товаров цену в 10 звёзд при выборе оплаты в этой системе. Это сделано для экономии вашего и моего времени, так как проект учебный. В реальном проекте можно было бы решить эту задачу двумя способами:

1. В админке, при добавлении товара, можно было бы запрашивать цену товаров не только в рублях, но и в звёздах.

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

Я решил не тратить время на реализацию этих функций и установить для всех товаров цену в 10 звёзд.

Теперь поработаем с клавиатурами (файл user/kbs.py).

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

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

def product_kb(product_id, price, stars_price) -> InlineKeyboardMarkup:     kb = InlineKeyboardBuilder()     kb.button(text="💳 Оплатить ЮКасса", callback_data=f"buy_yukassa_{product_id}_{price}")     kb.button(text="💳 Оплатить Robocassa", callback_data=f"buy_robocassa_{product_id}_{price}")     kb.button(text="⭐ Оплатить звездами", callback_data=f"buy_stars_{product_id}_{stars_price}")     kb.button(text="🛍 Назад", callback_data="catalog")     kb.button(text="🏠 На главную", callback_data="home")     kb.adjust(2)     return kb.as_markup()   

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

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

def get_product_buy_youkassa(price) -> InlineKeyboardMarkup:     return InlineKeyboardMarkup(inline_keyboard=[         [InlineKeyboardButton(text=f'Оплатить {price}₽', pay=True)],         [InlineKeyboardButton(text='Отменить', callback_data='home')]     ])   def get_product_buy_robocassa(price: int, payment_link: str) -> InlineKeyboardMarkup:     return InlineKeyboardMarkup(inline_keyboard=[         [InlineKeyboardButton(             text=f'Оплатить {price}₽',             web_app=WebAppInfo(url=payment_link)         )],         [InlineKeyboardButton(text='Отменить', callback_data='home')]     ])   def get_product_buy_stars(price) -> InlineKeyboardMarkup:     return InlineKeyboardMarkup(inline_keyboard=[         [InlineKeyboardButton(text=f"Оплатить {price} ⭐", pay=True)],         [InlineKeyboardButton(text='Отменить', callback_data='home')]     ]) 

Клавиатуру под Robokassa обсудим подробнее далее, пока не заостряем на ней внимание.

Заметьте, что клавиатура для оплаты Telegram Stars не отличается по формату от клавиатуры для оплаты через Юкассу. Это не просто совпадение.

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

Теперь нам остается внести правки в файл с каталогом (продажами). Работаем с файлом: bot/user/catalog_router.py.

Первые изменения начинаются в обработчике callback-даты, которая начинается на buy_. В новой версии обработчик выглядит так:

@catalog_router.callback_query(F.data.startswith('buy_')) async def process_about(call: CallbackQuery, session_without_commit: AsyncSession):     user_info = await UserDAO.find_one_or_none(         session=session_without_commit,         filters=TelegramIDModel(telegram_id=call.from_user.id)     )     _, payment_type, product_id, price = call.data.split('_')      if payment_type == 'yukassa':         await send_yukassa_invoice(call, user_info, product_id, price)     elif payment_type == 'stars':         await send_stars_invoice(call, user_info, product_id, 10)     elif payment_type == 'robocassa':         await send_robocassa_invoice(call, user_info, product_id, price, session_without_commit)      await call.message.delete() 

Таким образом, теперь в зависимости от выбранного пользователем способа оплаты, мы применяем различный алгоритм для выставления счета (инвойса).

Для Юкассы все будет аналогично тому, что было в первой части:

async def send_yukassa_invoice(call, user_info, product_id, price):     await bot.send_invoice(         chat_id=call.from_user.id,         title=f'Оплата 👉 {price}₽',         description=f'Пожалуйста, завершите оплату в размере {price}₽, чтобы открыть доступ к выбранному товару.',         payload=f"yukassa_{user_info.id}_{product_id}",         provider_token=settings.PROVIDER_TOKEN,         currency='RUB’         prices=[LabeledPrice(             label=f'Оплата {price}',             amount=int(price) * 100         )],         reply_markup=get_product_buy_youkassa(price)     ) 

Процесс выставления счёта со звёздами почти полностью повторяет процесс оплаты через Юкассу, и вы уже знаете почему.

async def send_stars_invoice(call, user_info, product_id, stars_price):     await bot.send_invoice(         chat_id=call.from_user.id,         title=f'Оплата 👉 {stars_price} ⭐',         description=f'Пожалуйста, завершите оплату в размере {stars_price} звезд, '                     f'чтобы открыть доступ к выбранному товару.',         payload=f"stars_{user_info.id}_{product_id}",         provider_token="",         currency='XTR',         prices=[LabeledPrice(             label=f'Оплата {stars_price} ⭐',             amount=int(stars_price)         )],         reply_markup=get_product_buy_stars(stars_price)     ) 

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

И самое главное — это изменение валюты, в которой мы выставляем счета. Теперь она будет обозначаться как «XTR», в отличие от привычного «RUB».

Логику обработки выставления счета на Робокассу рассмотрим позже.

Радует, что как оплата через Юкассу, так и оплата звездами относятся к внутренним платежам Telegram. Это означает, что для их обработки достаточно воспользоваться стандартными методами Aiogram 3. Такой подход позволяет нам сначала проверить возможность проведения платежа, а затем успешно обработать его.

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

@catalog_router.pre_checkout_query(lambda query: True) async def pre_checkout_query(pre_checkout_q: PreCheckoutQuery):     await bot.answer_pre_checkout_query(pre_checkout_q.id, ok=True)   @catalog_router.message(F.content_type == ContentType.SUCCESSFUL_PAYMENT) async def successful_payment(message: Message, session_with_commit: AsyncSession):     payment_info = message.successful_payment     payment_type, user_id, product_id = payment_info.invoice_payload.split('_')      if payment_type == 'stars':         price = payment_info.total_amount         currency = '⭐'     else:         price = payment_info.total_amount / 100         currency = '₽'      payment_data = {         'user_id': int(user_id),         'payment_id': payment_info.telegram_payment_charge_id,         'price': price,         'product_id': int(product_id),         'payment_type': payment_type     }      await successful_payment_logic(session=session_with_commit,                                    payment_data=payment_data, currency=currency,                                    user_tg_id=message.from_user.id, bot=bot) 

В блоке с обработкой успешного платежа я выполнил небольшую адаптацию под обновленный обработчик успешного платежа (successful_payment_logic).

Перезапустим бота и проверим оплату звездами.

После оплаты нас встречает салют (в конце будет видео-демонстрация) и бот сообщает, что сумма в размере 10-ти звезд возвращена:

Кстати, звёзды можно легко приобрести и в России. Вчера я купил их, используя стандартную карту МИР. Чтобы купить звёзды, достаточно зайти в настройки своего профиля в Telegram и перейти в раздел «Мои звёзды».

Кликнув на эту кнопку у вас появится возможность выполнить покупку, посмотреть текущее количество звезд и можно посмотреть расходы по звездам:

Итак, блок с интеграцией платежей через Telegram Stars в нашего бота успешно завершён. Теперь мы готовы перейти к следующему этапу — интеграции оплаты через Robocassa. Этот процесс будет немного сложнее, но вполне выполним.

Подключаем оплату Robocassa c веб-хуком

Для интеграции Robokassa мы напишем набор утилит, которые позволяют обрабатывать запросы от этого сервиса, включая проверку оплаты и генерацию ссылок. Все функции написаны в файле bot/app/utils.py. Рассмотрим их по порядку.

Импорты

import hashlib from urllib import parse from urllib.parse import urlparse from bot.config import settings 

Мы импортируем модули для работы с хэшированием (hashlib) и URL (urllib), а также настройки нашего приложения из файла settings.

Далее пойдет достаточно много кода. Поэтому сейчас, под спойлером, я представлю его как есть, а далее, отдельно, дам комментарии по функциям. Там все достаточно просто.

Полный код утилит:

def calculate_signature(login, cost, inv_id, password, user_id, user_telegram_id, product_id, is_result=False):     if is_result:         base_string = f"{cost}:{inv_id}:{password}"  # Для Result URL     else:         base_string = f"{login}:{cost}:{inv_id}:{password}"  # Для initital URL и Success URL      additional_params = {         'Shp_user_id': user_id,         'Shp_user_telegram_id': user_telegram_id,         'Shp_product_id': product_id     }     for key, value in sorted(additional_params.items()):         base_string += f":{key}={value}"      return hashlib.md5(base_string.encode('utf-8')).hexdigest()   def generate_payment_link(cost: float, number: int, description: str,                           user_id: int, user_telegram_id: int, product_id: int,                           is_test=1, robokassa_payment_url='https://auth.robokassa.ru/Merchant/Index.aspx') -> str:     """     Генерирует ссылку для оплаты через Robokassa с обязательными параметрами.      :param cost: Стоимость товара     :param number: Номер заказа     :param description: Описание заказа     :param user_id: ID пользователя     :param user_telegram_id: Telegram ID пользователя     :param product_id: ID товара     :param is_test: Флаг тестового режима (1 - тест, 0 - боевой режим)     :param robokassa_payment_url: URL для оплаты Robokassa     :return: Ссылка на страницу оплаты     """     signature = calculate_signature(         settings.MRH_LOGIN,         cost,         number,         settings.MRH_PASS_1,         user_id,         user_telegram_id,         product_id     )      data = {         'MerchantLogin': settings.MRH_LOGIN,         'OutSum': cost,         'InvId': number,         'Description': description,         'SignatureValue': signature,         'IsTest': is_test,         'Shp_user_id': user_id,         'Shp_user_telegram_id': user_telegram_id,         'Shp_product_id': product_id     }      return f'{robokassa_payment_url}?{parse.urlencode(data)}'   def parse_response(request: str) -> dict:     """     Разбирает строку запроса на параметры.      :param request: Строка запроса     :return: Словарь с параметрами     """     return dict(parse.parse_qsl(urlparse(request).query))   def check_signature_result(out_sum, inv_id, received_signature, password, user_id, user_telegram_id, product_id) -> bool:     signature = calculate_signature(         settings.MRH_LOGIN,         out_sum,         inv_id,         password,         user_id,         user_telegram_id,         product_id,         is_result=True  # Важный флаг для Result URL     )     return signature.lower() == received_signature.lower()   def result_payment(request: str) -> str:     """     Обрабатывает результат оплаты (ResultURL).      :param request: Строка запроса с параметрами оплаты     :return: 'OK' + номер заказа, если оплата прошла успешно, иначе 'bad sign'     """     params = parse_response(request)     out_sum = params['OutSum']     inv_id = params['InvId']     signature = params['SignatureValue']     user_id = params['Shp_user_id']     user_telegram_id = params['Shp_user_telegram_id']     product_id = params['Shp_product_id']      if check_signature_result(out_sum, inv_id, signature, settings.MRH_PASS_2, user_id, user_telegram_id, product_id):         return f'OK{inv_id}'     return "bad sign"   def check_success_payment(request: str) -> str:     """     Проверяет успешность оплаты (SuccessURL).      :param request: Строка запроса с параметрами оплаты     :return: Сообщение об успешной оплате или 'bad sign' при неверной подписи     """     params = parse_response(request)     out_sum = params['OutSum']     inv_id = params['InvId']     signature = params['SignatureValue']     user_id = params['Shp_user_id']     user_telegram_id = params['Shp_user_telegram_id']     product_id = params['Shp_product_id']      if check_signature_result(out_sum, inv_id, signature, settings.MRH_PASS_1, user_id, user_telegram_id, product_id):         return "Thank you for using our service"     return "bad sign" 

Вычисление подписи

def calculate_signature(login, cost, inv_id, password, user_id, user_telegram_id, product_id, is_result=False): 

Эта функция формирует цифровую подпись, которая подтверждает подлинность запроса. Подпись создается на основе переданных параметров (login, cost, inv_id и т.д.) и хэшируется с помощью алгоритма MD5. Это ключевой механизм для защиты данных.

• Зачем? Чтобы убедиться, что запрос пришел от Robokassa и не был подделан.

• Почему MD5? Это требование API Robokassa и мы сами выбрали этот алгоритм хэширования на этапе настройки магазина.

Генерация ссылки для оплаты

def generate_payment_link(cost, number, description, user_id, user_telegram_id, product_id, is_test=1, robokassa_payment_url='https://auth.robokassa.ru/Merchant/Index.aspx'): 

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

• Зачем? Чтобы перенаправить пользователя на Robokassa для совершения платежа.

• Почему с подписью? Подпись подтверждает целостность данных в URL.

Разбор строки запроса

def parse_response(request): 

Функция разбирает строку запроса (например, URL из веб-хука) и извлекает параметры в виде словаря.

• Зачем? Чтобы работать с параметрами из запроса Robokassa.

• Почему словарь? Удобно для последующей проверки данных.

Проверка подписи результата

def check_signature_result(out_sum, inv_id, received_signature, password, user_id, user_telegram_id, product_id): 

Здесь мы проверяем подпись, полученную в ответе от Robokassa, сравнивая ее с нашей расчетной подписью.

• Зачем? Чтобы удостовериться в подлинности запроса.

• Почему важно? Если подпись не совпадает, запрос считается подделанным.

Обработка результата оплаты (ResultURL)

def result_payment(request): 

Эта функция отвечает за обработку уведомления о результате платежа. Она проверяет подпись и возвращает OK с номером заказа при успешной оплате, иначе сообщает о проблеме.

• Зачем? Чтобы получать уведомления от Robokassa о статусе платежей.

• Почему важно? Так мы знаем, что пользователь действительно оплатил товар или услугу.

Проверка успешности оплаты (SuccessURL)

def check_success_payment(request): 

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

• Зачем? Чтобы отобразить пользователю сообщение об успешной оплате.

• Почему подпись? Это обязательное требование для безопасности.

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

Обработка запросов от Robokassa в веб-сервере

Теперь нам осталось включить эти утилиты в обработчик веб-сервера. Напишем его в файле bot/app/app.py

Полный код импортов:

import datetime  from aiohttp import web from aiogram.types import Update from loguru import logger from bot.app.utils import check_signature_result from bot.config import bot, dp, settings from bot.dao.database import async_session_maker from bot.user.utils import successful_payment_logic 

• aiohttp.web: Для работы с веб-сервером.

• loguru: Для удобного логирования.

• check_signature_result: Утилита для проверки цифровой подписи.

• async_session_maker: Фабрика для работы с базой данных.

• successful_payment_logic: Логика обработки успешного платежа.

Полный код обработчика:

async def robokassa_result(request: web.Request) -> web.Response:     """     Обрабатывает запрос от Робокассы на ResultURL.      :param request: HTTP-запрос     :return: Текстовый ответ с результатами проверки     """     logger.success("Получен ответ от Робокассы!")     data = await request.post()      # Извлекаем параметры из запроса     signature = data.get('SignatureValue')     out_sum = data.get('OutSum')     inv_id = data.get('InvId')     user_id = data.get('Shp_user_id')     user_telegram_id = data.get('Shp_user_telegram_id')     product_id = data.get('Shp_product_id')      # Проверяем подпись     if check_signature_result(         out_sum=out_sum,         inv_id=inv_id,         received_signature=signature,         password=settings.MRH_PASS_2,         user_id=user_id,         user_telegram_id=user_telegram_id,         product_id=product_id     ):         result = f"OK{inv_id}"         logger.info(f"Успешная проверка подписи для InvId: {inv_id}")          payment_data = {             'user_id': int(user_id),             'payment_id': signature,             'price': int(out_sum),             'product_id': int(product_id),             'payment_type': "robocassa"         }         async with async_session_maker() as session:             await successful_payment_logic(                 session=session,                 payment_data=payment_data,                 currency="₽",                 user_tg_id=int(user_telegram_id),                 bot=bot             )             await session.commit()     else:         result = "bad sign"         logger.warning(f"Неверная подпись для InvId: {inv_id}")      logger.info(f"Ответ: {result}")     return web.Response(text=result) 

Разбор основного обработчика результата: robokassa_result

async def robokassa_result(request: web.Request) -> web.Response: 

Этот обработчик принимает HTTP-запросы от Robokassa на указанный нами ResultURL.

Ссылку на этот обработчик необходимо указать на сайте Robocassa в настройках магазина:

В файле bot/main.py необходимо выполнить регистрацию этого эндпоинта:

app.router.add_post("/robokassa/result/", robokassa_result) 

Этого достаточно для того, чтоб связать сервис Robocassa с нашим веб-сервером.

Рассмотрим работу этого обработчика подробнее.

1. Логирование получения запроса

logger.success("Получен ответ от Робокассы!") data = await request.post() 

Сначала логируется факт получения запроса, затем извлекаются данные из HTTP-запроса методом POST.

2. Извлечение параметров

signature = data.get('SignatureValue') out_sum = data.get('OutSum') inv_id = data.get('InvId') user_id = data.get('Shp_user_id') user_telegram_id = data.get('Shp_user_telegram_id') product_id = data.get('Shp_product_id') 

Далее из данных запроса извлекаются ключевые параметры:

• SignatureValue: Цифровая подпись запроса.

• OutSum: Сумма оплаты.

• InvId: Номер заказа.

• Shp_user_id, Shp_user_telegram_id, Shp_product_id: Дополнительные параметры, переданные нами при создании платежа.

3. Проверка подписи

if check_signature_result(     out_sum=out_sum,     inv_id=inv_id,     received_signature=signature,     password=settings.MRH_PASS_2,     user_id=user_id,     user_telegram_id=user_telegram_id,     product_id=product_id ): 

С помощью утилиты check_signature_result выполняется проверка цифровой подписи. Это критически важно для подтверждения, что запрос действительно пришел от Robokassa.

• Успех: Если подпись верна, фиксируем успешную транзакцию.

• Ошибка: Если подпись не совпадает, отклоняем запрос.

4. Обработка успешной оплаты

result = f"OK{inv_id}" logger.info(f"Успешная проверка подписи для InvId: {inv_id}")  payment_data = {     'user_id': int(user_id),     'payment_id': signature,     'price': int(out_sum),     'product_id': int(product_id),     'payment_type': "robocassa" } 

Обратите внимание. В таблице purchases появилсь колонка «payment_type». Не забудьте ее включить в модель таблицы и выполните миграцию через Alembic.

При успешной проверке формируется ответ OK с номером заказа. Параллельно собираются данные об оплате для сохранения в базе данных.

4.1 Сохранение данных в базу

async with async_session_maker() as session:     await successful_payment_logic(         session=session,         payment_data=payment_data,         currency="₽",         user_tg_id=int(user_telegram_id),         bot=bot     )     await session.commit() 

Функция successful_payment_logic обрабатывает данные:

• Сохраняет оплату в базу данных.

• Выполняет дальнейшие действия, которые предусмотрены данным обработчиком

4.2 Ошибка подписи

else:     result = "bad sign"     logger.warning(f"Неверная подпись для InvId: {inv_id}") 

Если подпись неверна, фиксируется ошибка в логах и отправляется ответ bad sign.

5. Возврат результата

logger.info(f"Ответ: {result}") return web.Response(text=result) 

В конце отправляется HTTP-ответ:

• OK{inv_id}: При успешной проверке.

• bad sign: При ошибке подписи.

Почему это важно?

• Защита данных: Проверка подписи гарантирует, что данные не были изменены.

• Информирование пользователя: Успешная обработка позволяет уведомить пользователя о платеже.

• Автоматизация: Все действия выполняются автоматически — от проверки оплаты до сохранения данных.

Итог

Этот обработчик — ключевая часть интеграции Robokassa. Он обрабатывает запросы на ResultURL, защищает данные, сохраняет результаты платежей в базе и информирует пользователей. Это позволяет организовать безопасный и удобный процесс приема оплат.

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

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

Для того, чтоб понять было проще — можете в моем телеграмм канале «Легкий путь в Python» найти полный исходник кода и задать вопросы если что-то было не понятно. Сейчас в сообществе без малого 2000 человек, так что даже если не я, то участники смогут вам помочь разобраться с этой темой. А мы продолжим.

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

Интегрируем платежку Robocassa на стороне клиента

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

def get_product_buy_robocassa(price: int, payment_link: str) -> InlineKeyboardMarkup:     return InlineKeyboardMarkup(inline_keyboard=[         [InlineKeyboardButton(             text=f'Оплатить {price}₽',             web_app=WebAppInfo(url=payment_link)         )],         [InlineKeyboardButton(text='Отменить', callback_data='home')]     ]) 

В этом случае мы отправляем ссылку в формате Telegram Mini App. Хотя мы не будем иметь никакого отношения к созданию посадочной страницы для платежа, такой подход позволяет имитировать действие MiniApp, что очень удобно.

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

В текущей реализации внешний вид страницы с оплатой будет таким:

Теперь в файле bot/user/catalog_router.py напишем обработчик, при помощи которого будем выставлять счет на оплату через Robocassa:

async def send_robocassa_invoice(call, user_info, product_id, price, session: AsyncSession):     pay_id = await PurchaseDao.get_next_id(session=session)     text = f'Пожалуйста, завершите оплату в размере {price}₽, чтобы открыть доступ к выбранному товару.'     description = f"Оплата за товар: ID {user_info.id} ({price}₽)"     payment_link = generate_payment_link(cost=float(price), number=pay_id, description=description,                                          user_id=user_info.id, user_telegram_id=call.from_user.id,                                          product_id=product_id)     kb = get_product_buy_robocassa(price, payment_link)     await call.message.answer(text, reply_markup=kb)      

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

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

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

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

Далее логика успешной оплаты будет сводиться к следующему:

1. Клиент нажал на «Оплатить»

2. Robocassa проверила постепление денег на счет

3. Robocassa отправляет на наш хук сообщение с результатом платежа

4. Наш хук обрабатывает результат и если все хорошо, инициирует процесс «Платеж успешен»: регистрирует платеж, отправляет уведомление админам о том, что платеж успешен и отправляет клиенту купленный товар.

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

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

class PurchaseDao(BaseDAO[Purchase]):     model = Purchase      # другие методы      @classmethod     async def get_next_id(cls, session: AsyncSession) -> int:         """         Возвращает следующий свободный ID для новой записи.          :param session: Асинхронная сессия базы данных         :return: Следующий свободный ID         """         query = select(func.coalesce(func.max(cls.model.id) + 1, 1))         result = await session.execute(query)         return result.scalar() 

Прием платежей в боте

Видео-презентация всех трех типов оплаты в данном проекте:

Деплой проекта в Amvera Cloud

Если вы клонировали репозиторий с первой части, то у вас уже есть все для деплоя, так как в корне проекта вы сможете найти файл amvera.yml со следующим содержимым:

meta:   environment: python   toolchain:     name: pip     version: 3.12 build:   requirementsPath: requirements.txt run:   persistenceMount: /data   containerPort: 8000   command: python3 -m bot.main 

Настройки можно не менять. Единственное убедитесь, что значение containerPort (8000) соответствует значению вашего .env файла (переменная SITE_PORT).

Теперь нам остается выполнить деплой. Шаги не будут принципиально отличаться. Делаем следующее:

1. Регистрируемся в Amvera Cloud (если не было регистрации)

2. Кликаем на «Создать проект»

3. Даем имя проекту и выбираем тарифный план. После жмем на «Далее».

4. Выбираем способ доставки файлов бота. Доставляем или командами GIT (на втором экране вы увидите инструкцию) или выбрав «Через интерфейс» простым перетягиванием файлов бота прямо в Amvera. Я, как и в прошлый раз, выбираю «Через интерфейс». После жмем на далее.

5. Проверяем корректно ли подгрузились настройки. Если да, то нажимаем на «Завершить».

Подключаем бесплатный домен к проекту:

1. Заходим в созданный проект

2. Перемещаемся на вкладку «Настройки»

3. Кликаем на «Добавить доменное имя» и выбираем «бесплатное доменное имя»

4. Копируем доменное имя и вставляем его на локальной машине в файл .env

У меня получилась такая строка:

SITE_URL=https://aiogrambotpayhook-yakvenalex.amvera.io

5. Перезаписываем .env файл в проекте Amvera Cloud на файл с актуальным доменом

6. Кликаем на «Пересобрать проект»

Если все пройдет корректно, то бот уведомит вас о том, что он запущен, а по адресу выделенного домена будет доступна наша главная страница:

Проект запущен. Если интересно поклацать работающего бота, то вот ссылка на него: https://t.me/DigitalMarketAiogramHookBot. Полный исходный код этого проекта, как и материал, который я не публикую на Хабре вы знаете где найти.

Заключение

Проект завершен, и мы проделали огромный путь! Теперь вы знаете, как интегрировать разные платежные системы в ботов, используя различные подходы — от простых настроек через BotFather до полноценного самописного веб-хука.

Мы также познакомились с новой и перспективной системой оплаты в Telegram, которая использует валюту Telegram Stars. Эта технология, скорее всего, станет популярной в ближайшем будущем. Разобравшись с материалами сегодняшней статьи, вы получите важный навык, который позволит вам уверенно реализовывать такие интеграции в ваших проектах и стать востребованным специалистом.

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

И напоследок — немного тепла. Поздравляю вас с Наступающим Новым годом и всеми зимними праздниками! Пусть 2025 год принесет вам новые достижения, вдохновение и успех — не только в программировании, но и в жизни.

Спасибо за ваше внимание и интерес. До встречи в новом, 2025 году! 🎄✨