Урок 6.3 — Webhook обработка

Вебхук — это обратный вызов (callback) от Crypto Pay API на ваш сервер. Когда происходит событие (например, оплата инвойса), Crypto Pay отправляет POST-запрос на заранее указанный URL вашего сервера с данными о событии.

Преимущества вебхуков перед polling:

Когда использовать вебхуки? Если у вас есть публичный сервер с HTTPS и вы хотите получать уведомления об оплате мгновенно. Если сервера нет (простой скрипт, бот) — используйте polling.

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

Важные требования к вебхуку:

• URL должен начинаться с https:// (Crypto Pay не отправляет на HTTP)
• Сервер должен быть доступен из интернета
• Время ответа сервера не должно превышать 5 секунд (иначе Crypto Pay считает, что доставка не удалась)
• Crypto Pay будет повторять отправку при ошибках (до 24 часов)

В коде aiosend вы также можете программно установить/удалить вебхук через API:

from aiosend import CryptoPay

cp = CryptoPay(token="YOUR_TOKEN")

# Установить вебхук
result = await cp.set_webhook(url="https://example.com/webhook")
print(f"Вебхук установлен: {result}")

# Удалить вебхук (для переключения на polling)
result = await cp.delete_webhook()
print(f"Вебхук удалён: {result}")

# Получить информацию о текущем вебхуке
info = await cp.get_webhook_info()
print(f"Текущий вебхук: {info.url if info else 'не установлен'}")

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

Основные методы WebhookManager:

aiosend предоставляет три встроенные реализации:

AiohttpManager — менеджер вебхуков для фреймворка aiohttp. Он не требует дополнительных зависимостей, так как aiohttp уже используется aiosend для HTTP-запросов.

Конструктор принимает:

• app: экземпляр aiohttp.web.Application
• path: путь, на котором будет приниматься вебхук (по умолчанию "/webhook")
• **kwargs: дополнительные аргументы для post маршрута

import asyncio
from aiohttp import web
from aiosend import CryptoPay
from aiosend.webhook import AiohttpManager

# Создаём aiohttp приложение
app = web.Application()

# Создаём менеджер вебхуков
manager = AiohttpManager(app, path="/webhook")

# Создаём клиента с вебхук-менеджером
cp = CryptoPay(
    token="YOUR_TOKEN",
    webhook_manager=manager,
)

# Регистрируем обработчики (те же декораторы)
@cp.invoice_paid()
async def on_paid(invoice, **kwargs):
    print(f"✅ Инвойс #{invoice.invoice_id} оплачен!")
    print(f"   Сумма: {invoice.amount} {invoice.asset}")
    # Здесь бизнес-логика

@cp.invoice_expired()
async def on_expired(invoice, **kwargs):
    print(f"❌ Инвойс #{invoice.invoice_id} истёк")

async def start_webhook():
    # Устанавливаем вебхук в Crypto Pay
    await cp.set_webhook(url="https://example.com/webhook")
    print("Вебхук установлен!")

app.on_startup.append(lambda _: asyncio.create_task(start_webhook()))

if __name__ == "__main__":
    # Запускаем aiohttp сервер
    web.run_app(app, host="0.0.0.0", port=8080)

Когда Crypto Pay отправляет вебхук на https://example.com/webhook, AiohttpManager принимает запрос, парсит данные и вызывает соответствующий обработчик (on_paid или on_expired) в зависимости от типа события.

FastAPIManager — менеджер вебхуков для FastAPI. Требует установки aiosend[fastapi].

Конструктор принимает:

• app: экземпляр FastAPI
• path: путь для вебхука (по умолчанию "/webhook")

# Установка: pip install aiosend[fastapi]

import uvicorn
from fastapi import FastAPI
from aiosend import CryptoPay
from aiosend.webhook import FastAPIManager

# Создаём FastAPI приложение
app = FastAPI()

# Создаём менеджер вебхуков
manager = FastAPIManager(app, path="/webhook")

# Создаём клиента
cp = CryptoPay(
    token="YOUR_TOKEN",
    webhook_manager=manager,
)

@cp.invoice_paid()
async def on_paid(invoice, **kwargs):
    print(f"✅ Получена оплата #{invoice.invoice_id}")
    print(f"   Сумма: {invoice.amount} {invoice.asset}")

@cp.invoice_expired()
async def on_expired(invoice, **kwargs):
    print(f"❌ Инвойс #{invoice.invoice_id} просрочен")

@app.on_event("startup")
async def startup():
    await cp.set_webhook(url="https://example.com/webhook")
    print("Вебхук FastAPI установлен!")

@app.get("/")
async def root():
    return {"status": "ok", "message": "Webhook server running"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

FlaskManager — менеджер вебхуков для Flask. Требует установки aiosend[flask].

# Установка: pip install aiosend[flask]

from flask import Flask
from aiosend import SyncCryptoPay
from aiosend.webhook import FlaskManager

# Создаём Flask приложение
app = Flask(__name__)

# Создаём менеджер вебхуков
manager = FlaskManager(app, path="/webhook")

# Создаём синхронного клиента (Flask не async)
cp = SyncCryptoPay(
    token="YOUR_TOKEN",
    webhook_manager=manager,
)

@cp.invoice_paid()
def on_paid(invoice, **kwargs):
    print(f"✅ Инвойс #{invoice.invoice_id} оплачен!")
    print(f"   Сумма: {invoice.amount} {invoice.asset}")

@cp.invoice_expired()
def on_expired(invoice, **kwargs):
    print(f"❌ Инвойс #{invoice.invoice_id} истёк")

@app.route("/")
def index():
    return "Webhook server is running!"

if __name__ == "__main__":
    # Устанавливаем вебхук (синхронно)
    cp.set_webhook(url="https://example.com/webhook")
    print("Вебхук Flask установлен!")
    
    # Запускаем Flask сервер
    app.run(host="0.0.0.0", port=5000, debug=True)

WebhookRouter — это класс, который управляет маршрутизацией входящих вебхуков к соответствующим обработчикам. Он анализирует тип события (invoice_paid, invoice_expired, check_activated, check_expired) и вызывает нужную функцию.

Вы можете создать свой собственный WebhookManager, если хотите интегрировать aiosend с другим веб-фреймворком или добавить кастомную логику:

from aiosend.webhook import WebhookManager, WebhookRouter

class CustomManager(WebhookManager):
    """Кастомный менеджер вебхуков для любого фреймворка."""
    
    def __init__(self, path: str = "/webhook"):
        self.path = path
        self.router = WebhookRouter()
    
    def register(self, router: WebhookRouter) -> None:
        """Регистрирует роутер с обработчиками."""
        self.router = router
    
    def unregister(self) -> None:
        """Отменяет регистрацию."""
        self.router = WebhookRouter()
    
    async def handle_webhook(self, request_data: dict) -> dict:
        """Обрабатывает входящий вебхук.
        
        Вызовите этот метод из вашего веб-обработчика,
        передав тело запроса как словарь.
        """
        update_type = request_data.get("type")
        payload = request_data.get("payload", {})
        
        if update_type == "invoice_paid":
            await self.router.process_invoice_paid(payload)
        elif update_type == "invoice_expired":
            await self.router.process_invoice_expired(payload)
        elif update_type == "check_activated":
            await self.router.process_check_activated(payload)
        elif update_type == "check_expired":
            await self.router.process_check_expired(payload)
        
        return {"ok": True}

# Использование с произвольным веб-фреймворком:
# manager = CustomManager(path="/webhook")
# cp = CryptoPay(token="TOKEN", webhook_manager=manager)
# 
# В вашем веб-обработчике:
# @app.post("/webhook")
# async def webhook_handler(request):
#     data = await request.json()
#     result = await manager.handle_webhook(data)
#     return result

Рассмотрим три способа запустить один и тот же вебхук-сервер. Логика обработки одинаковая — меняется только фреймворк.

# shared.py — переиспользуемые обработчики
import logging
from aiosend import CryptoPay

logger = logging.getLogger(__name__)

def register_handlers(cp: CryptoPay):
    """Регистрирует обработчики событий."""
    
    @cp.invoice_paid()
    async def on_paid(invoice, **kwargs):
        logger.info(f"💰 Оплачен инвойс #{invoice.invoice_id}")
        logger.info(f"   Сумма: {invoice.amount} {invoice.asset}")
        logger.info(f"   Payload: {invoice.payload}")
    
    @cp.invoice_expired()
    async def on_expired(invoice, **kwargs):
        logger.warning(f"⏰ Просрочен инвойс #{invoice.invoice_id}")
    
    @cp.check_activated()
    async def on_check_activated(check, **kwargs):
        logger.info(f"🎉 Активирован чек #{check.check_id}")
    
    @cp.check_expired()
    async def on_check_expired(check, **kwargs):
        logger.warning(f"⏰ Истёк чек #{check.check_id}")

import asyncio
from aiohttp import web
from aiosend import CryptoPay
from aiosend.webhook import AiohttpManager
from shared import register_handlers

app = web.Application()
manager = AiohttpManager(app, path="/webhook")

cp = CryptoPay(token="YOUR_TOKEN", webhook_manager=manager)
register_handlers(cp)

async def on_startup(app):
    await cp.set_webhook(url="https://example.com/webhook")
    print("Вебхук aiohttp установлен!")

app.on_startup.append(on_startup)

if __name__ == "__main__":
    web.run_app(app, host="0.0.0.0", port=8080)

import uvicorn
from fastapi import FastAPI
from aiosend import CryptoPay
from aiosend.webhook import FastAPIManager
from shared import register_handlers

app = FastAPI()
manager = FastAPIManager(app, path="/webhook")

cp = CryptoPay(token="YOUR_TOKEN", webhook_manager=manager)
register_handlers(cp)

@app.on_event("startup")
async def startup():
    await cp.set_webhook(url="https://example.com/webhook")
    print("Вебхук FastAPI установлен!")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

from flask import Flask
from aiosend import SyncCryptoPay
from aiosend.webhook import FlaskManager

app = Flask(__name__)
manager = FlaskManager(app, path="/webhook")

cp = SyncCryptoPay(token="YOUR_TOKEN", webhook_manager=manager)

# Регистрируем обработчики (синхронные)
@cp.invoice_paid()
def on_paid(invoice, **kwargs):
    print(f"💰 Оплачен инвойс #{invoice.invoice_id}")

@cp.invoice_expired()
def on_expired(invoice, **kwargs):
    print(f"⏰ Просрочен инвойс #{invoice.invoice_id}")

if __name__ == "__main__":
    cp.set_webhook(url="https://example.com/webhook")
    print("Вебхук Flask установлен!")
    app.run(host="0.0.0.0", port=5000)

Рассмотрим продвинутые сценарии использования вебхуков в aiosend: комбинирование с polling, верификация запросов, работа с несколькими клиентами.

Пример: Верификация входящих вебхуков

Для безопасности стоит проверять, что вебхук действительно пришёл от Crypto Pay. Хотя aiosend делает это автоматически, вы можете добавить дополнительную проверку:

import hmac
import hashlib
from aiosend.webhook import WebhookManager, WebhookRouter

class VerifiedWebhookManager(WebhookManager):
    """Менеджер вебхуков с проверкой подписи."""
    
    def __init__(self, app, path: str, secret: str):
        self.app = app
        self.path = path
        self.secret = secret.encode()
        self.router = WebhookRouter()
    
    def _verify_signature(self, body: bytes, signature: str) -> bool:
        """Проверяет HMAC-подпись запроса."""
        expected = hmac.new(
            self.secret,
            body,
            hashlib.sha256,
        ).hexdigest()
        return hmac.compare_digest(expected, signature)
    
    def register(self, router: WebhookRouter) -> None:
        self.router = router
    
    def unregister(self) -> None:
        self.router = WebhookRouter()
    
    async def handle_webhook(self, request):
        body = await request.read()
        signature = request.headers.get("X-Crypto-Pay-Signature", "")
        
        if not self._verify_signature(body, signature):
            return {"ok": False, "error": "Invalid signature"}
        
        import json
        data = json.loads(body)
        update_type = data.get("type")
        payload = data.get("payload", {})
        
        # Обработка событий
        if update_type == "invoice_paid":
            await self.router.process_invoice_paid(payload)
        elif update_type == "invoice_expired":
            await self.router.process_invoice_expired(payload)
        
        return {"ok": True}
    
    def setup(self):
        from aiohttp import web
        self.app.router.add_post(
            self.path,
            self.handle_webhook,
        )
    
    def shutdown(self):
        pass

Пример: Несколько клиентов с вебхуками

Вы можете запустить несколько клиентов на одном сервере, каждый со своим вебхуком:

import asyncio
from aiohttp import web
from aiosend import CryptoPay
from aiosend.webhook import AiohttpManager

app = web.Application()

# Первый клиент — mainnet
manager1 = AiohttpManager(app, path="/webhook/mainnet")
cp1 = CryptoPay(
    token="MAINNET_TOKEN",
    webhook_manager=manager1,
)

@cp1.invoice_paid()
async def on_paid_mainnet(invoice, **kwargs):
    print(f"[MAINNET] Оплата: {invoice.amount} {invoice.asset}")

# Второй клиент — testnet
manager2 = AiohttpManager(app, path="/webhook/testnet")
cp2 = CryptoPay(
    token="TESTNET_TOKEN",
    webhook_manager=manager2,
)

@cp2.invoice_paid()
async def on_paid_testnet(invoice, **kwargs):
    print(f"[TESTNET] Оплата: {invoice.amount} {invoice.asset}")

async def start():
    await cp1.set_webhook(url="https://example.com/webhook/mainnet")
    await cp2.set_webhook(url="https://example.com/webhook/testnet")
    print("Оба вебхука установлены!")

app.on_startup.append(lambda _: asyncio.create_task(start()))

if __name__ == "__main__":
    web.run_app(app, host="0.0.0.0", port=8080)

Пример: Комбинирование вебхуков и polling

Можно использовать вебхуки как основной механизм, а polling — как запасной (fallback):

import asyncio
from aiohttp import web
from aiosend import CryptoPay
from aiosend.webhook import AiohttpManager
from aiosend.polling import PollingConfig

app = web.Application()
manager = AiohttpManager(app, path="/webhook")

cp = CryptoPay(
    token="YOUR_TOKEN",
    webhook_manager=manager,
    polling_config=PollingConfig(timeout=300, delay=5),
)

processed_ids = set()

@cp.invoice_paid()
async def on_paid(invoice, **kwargs):
    if invoice.invoice_id in processed_ids:
        return  # уже обработано через webhook
    processed_ids.add(invoice.invoice_id)
    print(f"Обработано: {invoice.invoice_id}")

async def start_webhook_and_polling():
    # Устанавливаем вебхук
    await cp.set_webhook(url="https://example.com/webhook")
    
    # Запускаем polling как fallback (с меньшей частотой)
    # polling будет обрабатывать события, которые
    # по какой-то причине не пришли через webhook
    await cp.start_polling(reset_webhook=False)

app.on_startup.append(
    lambda _: asyncio.create_task(start_webhook_and_polling())
)

if __name__ == "__main__":
    web.run_app(app, host="0.0.0.0", port=8080)