Как подключить оплату через xRocket в бота на aiogram 3: Rocket Pay API с кодом

pip install -U aiogram

import aiohttp

API_URL = "https://pay.xrocket.exchange"


class XRocketPay:
    # Минимальный клиент Rocket Pay API (@xRocket) — счета tg-invoices

    def __init__(self, api_key: str, api_url: str = API_URL) -> None:
        self.headers = {"Rocket-Pay-Key": api_key}
        self.api_url = api_url

    async def _request(self, method: str, path: str, body: dict | None = None) -> dict:
        async with aiohttp.ClientSession(headers=self.headers) as session:
            async with session.request(method, f"{self.api_url}{path}", json=body) as resp:
                data = await resp.json()
        if not data.get("success"):
            raise RuntimeError(f"xRocket: {data.get('message') or data.get('errors')}")
        return data["data"]

    async def create_invoice(self, amount: float, currency: str, description: str, payload: str) -> dict:
        return await self._request("POST", "/tg-invoices", {
            "amount": amount,
            "currency": currency,     # монета: USDT, TONCOIN...
            "numPayments": 1,         # счёт можно оплатить один раз
            "description": description,
            "payload": payload,       # ваша метка заказа, вернётся вместе со счётом
            "expiredIn": 3600,        # счёт действует 1 час
        })

    async def get_invoice(self, invoice_id: int) -> dict:
        return await self._request("GET", f"/tg-invoices/{invoice_id}")

import asyncio
import os

from aiogram import Bot, Dispatcher, F, Router
from aiogram.filters import Command
from aiogram.types import CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, Message

from xrocket import XRocketPay

PRICE = 5
CURRENCY = "USDT"
PAID_STATUSES = {"paid", "completed"}  # счёт с numPayments=1 после оплаты получает статус completed

router = Router()
rocket = XRocketPay(os.environ["XROCKET_API_KEY"])
delivered: set[int] = set()  # счета, по которым товар уже выдан (в реальном боте — база данных)


def is_paid(invoice: dict) -> bool:
    return str(invoice.get("status", "")).lower() in PAID_STATUSES


async def deliver(bot: Bot, invoice: dict) -> bool:
    # Выдаёт товар по оплаченному счёту — строго один раз
    if not is_paid(invoice) or invoice["id"] in delivered:
        return False
    if invoice.get("currency") != CURRENCY or float(invoice.get("amount", 0)) != PRICE:
        return False
    delivered.add(invoice["id"])
    user_id = int(invoice["payload"])
    # Здесь выдайте товар: откройте доступ, отправьте файл, начислите баланс
    await bot.send_message(user_id, "Оплата получена! Доступ открыт ✅")
    return True


@router.message(Command("buy"))
async def buy(message: Message) -> None:
    invoice = await rocket.create_invoice(
        amount=PRICE, currency=CURRENCY,
        description="Доступ к курсу",
        payload=str(message.from_user.id),  # кому выдать товар после оплаты
    )
    keyboard = InlineKeyboardMarkup(inline_keyboard=[
        [InlineKeyboardButton(text=f"💳 Оплатить {PRICE} {CURRENCY}", url=invoice["link"])],
        [InlineKeyboardButton(text="✅ Я оплатил", callback_data=f"xr:{invoice['id']}")],
    ])
    await message.answer("Счёт на оплату действует 1 час:", reply_markup=keyboard)


@router.callback_query(F.data.startswith("xr:"))
async def check(callback: CallbackQuery, bot: Bot) -> None:
    invoice_id = int(callback.data.split(":")[1])
    try:
        invoice = await rocket.get_invoice(invoice_id)
    except RuntimeError:
        invoice = None
    # Счёт должен принадлежать тому, кто нажал кнопку
    if not invoice or str(invoice.get("payload")) != str(callback.from_user.id):
        await callback.answer("Счёт не найден", show_alert=True)
        return
    if not is_paid(invoice):
        await callback.answer("Оплата пока не поступила", show_alert=True)
        return
    await deliver(bot, invoice)
    await callback.message.edit_text("Счёт оплачен ✅")
    await callback.answer()


async def main() -> None:
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_router(router)
    await dp.start_polling(bot)


if __name__ == "__main__":
    asyncio.run(main())

export BOT_TOKEN="токен_от_BotFather"
export XROCKET_API_KEY="ключ_Rocket_Pay"
python bot.py

from aiohttp import web


async def xrocket_webhook(request: web.Request) -> web.Response:
    try:
        update = await request.json()
        invoice_id = int(update["data"]["id"])
        # Статус узнаём у xRocket своим ключом, а не из тела уведомления
        invoice = await rocket.get_invoice(invoice_id)
    except (ValueError, KeyError, TypeError, RuntimeError):
        return web.Response(status=400)
    await deliver(request.app["bot"], invoice)
    return web.Response(text="ok")


async def main() -> None:
    bot = Bot(token=os.environ["BOT_TOKEN"])
    dp = Dispatcher()
    dp.include_router(router)

    app = web.Application()
    app["bot"] = bot
    app.router.add_post("/webhooks/xrocket", xrocket_webhook)
    runner = web.AppRunner(app)
    await runner.setup()
    await web.TCPSite(runner, "127.0.0.1", 8080).start()  # снаружи — через nginx с HTTPS

    try:
        await dp.start_polling(bot)
    finally:
        await runner.cleanup()