Модуль 8 · Интеграции с Telegram-ботами

Урок 8.3 — Интеграция с pyTelegramBotAPI

Учимся интегрировать aiosend с pyTelegramBotAPI (telebot) — синхронной библиотекой для создания Telegram-ботов.

pyTelegramBotAPI (telebot) — популярная синхронная библиотека для Telegram-ботов. Так как aiosend — асинхронная библиотека, интеграция требует запуска aiosend в отдельном потоке или использования SyncCryptoPay. В этом уроке мы рассмотрим оба подхода.

🧵

Подход 1: aiosend в отдельном потоке

Самый надёжный способ — запустить aiosend polling в отдельном потоке, так как pyTelegramBotAPI блокирует основной поток своим bot.infinity_polling().

Python · pyTelegramBotAPI + aiosend в потоке

import asyncio
import threading
import telebot
from aiosend import CryptoPay

BOT_TOKEN = "123456:ABCdefGHIjklmNOP"
CP_TOKEN = "1234:ABCDefGHIJklmnOPQRST"

bot = telebot.TeleBot(BOT_TOKEN)
cp = CryptoPay(token=CP_TOKEN)

@bot.message_handler(commands=["start"])
def start(message):
    bot.reply_to(message, "Привет! Используй /pay 10 USDT")

@bot.message_handler(commands=["pay"])
def pay(message):
    args = message.text.split()[1:]
    if len(args) < 2:
        bot.reply_to(message, "Пример: /pay 10 USDT")
        return

    amount, asset = float(args[0]), args[1].upper()

    # Создаём инвойс (синхронный вызов)
    invoice = cp.create_invoice(
        amount=amount,
        asset=asset,
        payload=str(message.from_user.id),
    )

    bot.reply_to(
        message,
        f"🧾 Счёт на {amount} {asset}:\n"
        f"{invoice.bot_invoice_url}",
    )

@cp.invoice_paid()
async def on_paid(invoice):
    """Асинхронный хендлер оплаты."""
    user_id = int(invoice.payload)
    bot.send_message(
        user_id,
        f"✅ Оплата {invoice.amount} {invoice.asset} получена!",
    )

def run_aiosend():
    """Запускает aiosend polling в отдельном потоке."""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(cp.start_polling())

if __name__ == "__main__":
    # Запускаем aiosend в фоновом потоке
    threading.Thread(target=run_aiosend, daemon=True).start()

    # Запускаем бота (блокирует основной поток)
    print("Бот запущен...")
    bot.infinity_polling()

💡 Как это работает?

Создаётся отдельный поток, в котором запускается event loop с aiosend polling. pyTelegramBotAPI работает в основном потоке. Оба потока не блокируют друг друга. Обрати внимание: cp.create_invoice() вызывается без await, так как aiosend поддерживает синхронный режим вызовов.

⏳

Ожидание оплаты через invoice.poll()

В синхронном контексте invoice.poll() тоже можно вызвать синхронно. Он заблокирует поток до оплаты или таймаута.

Python · Синхронный invoice.poll с pyTelegramBotAPI

@bot.message_handler(commands=["buy"])
def buy(message):
    invoice = cp.create_invoice(
        amount=15,
        asset="USDT",
        description="Премиум",
        payload=str(message.from_user.id),
    )

    msg = bot.send_message(
        message.chat.id,
        f"💳 Счёт на 15 USDT\n{invoice.bot_invoice_url}\n\n⏳ Ожидаем...",
    )

    try:
        # Синхронный poll — блокирует поток до оплаты
        paid = invoice.poll()
        bot.edit_message_text(
            f"✅ Оплачено: {paid.paid_amount} {paid.paid_asset}",
            message.chat.id,
            msg.message_id,
        )
    except Exception as e:
        bot.edit_message_text(
            f"❌ Ошибка: {e}",
            message.chat.id,
            msg.message_id,
        )

⚠️ Важно: блокировка потока

Синхронный invoice.poll() блокирует поток до оплаты или таймаута. В это время бот не сможет отвечать на другие сообщения. Используй этот подход только если оплата ожидается быстро (1-2 минуты). Для длительного ожидания лучше использовать глобальный хендлер @cp.invoice_paid().

🚀

Полный пример: магазин на pyTelegramBotAPI

Python · Полный бот с платежами (pyTelegramBotAPI)

import asyncio
import threading
import telebot
from telebot import types
from aiosend import CryptoPay

BOT_TOKEN = "123456:ABCdefGHIjklmNOP"
CP_TOKEN = "1234:ABCDefGHIJklmnOPQRST"

bot = telebot.TeleBot(BOT_TOKEN)
cp = CryptoPay(token=CP_TOKEN)

PRODUCTS = {
    "premium": {"name": "Премиум", "price": 15, "asset": "USDT"},
    "vip": {"name": "VIP", "price": 50, "asset": "USDT"},
}

@bot.message_handler(commands=["start"])
def start(message):
    markup = types.InlineKeyboardMarkup()
    markup.add(
        types.InlineKeyboardButton(
            "💎 Премиум", callback_data="buy_premium"
        ),
        types.InlineKeyboardButton(
            "⭐ VIP", callback_data="buy_vip"
        ),
    )
    bot.send_message(
        message.chat.id,
        "🛒 Добро пожаловать в магазин!",
        reply_markup=markup,
    )

@bot.callback_query_handler(func=lambda c: c.data.startswith("buy_"))
def handle_buy(callback):
    product_key = callback.data.replace("buy_", "")
    product = PRODUCTS.get(product_key)
    if not product:
        bot.answer_callback_query(callback.id, "Товар не найден")
        return

    invoice = cp.create_invoice(
        amount=product["price"],
        asset=product["asset"],
        description=product["name"],
        payload=f"{callback.from_user.id}:{product_key}",
    )

    bot.answer_callback_query(callback.id, "Счёт создан!")
    bot.send_message(
        callback.message.chat.id,
        f"💳 {product['name']}\n"
        f"Сумма: {product['price']} {product['asset']}\n"
        f"{invoice.bot_invoice_url}\n\n"
        f"Ожидаем оплату...",
    )

@bot.message_handler(commands=["buy"])
def buy_direct(message):
    """Быстрая покупка с ожиданием."""
    invoice = cp.create_invoice(
        amount=10, asset="USDT",
        payload=str(message.from_user.id),
    )
    msg = bot.send_message(
        message.chat.id,
        f"🧾 Счёт: {invoice.bot_invoice_url}"
    )

    try:
        paid = invoice.poll()
        bot.edit_message_text(
            f"✅ Оплачено: {paid.paid_amount} {paid.paid_asset}",
            message.chat.id,
            msg.message_id,
        )
    except Exception as e:
        bot.edit_message_text(
            f"❌ {e}", message.chat.id, msg.message_id,
        )

@cp.invoice_paid()
async def on_paid(invoice):
    """Глобальный хендлер оплаты."""
    user_id, product_key = (invoice.payload or ":").split(":")
    product = PRODUCTS.get(product_key, {"name": "товар"})
    bot.send_message(
        int(user_id),
        f"✅ {product['name']} оплачен и активирован!",
    )
    print(f"Платёж: user={user_id} product={product['name']}")

def run_aiosend():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(cp.start_polling())

if __name__ == "__main__":
    threading.Thread(target=run_aiosend, daemon=True).start()
    print("Бот запущен...")
    bot.infinity_polling()

🔍 Ключевые моменты

aiosend запускается в отдельном потоке через threading.Thread
Синхронные вызовы aiosend (cp.create_invoice) работают без await
Глобальный хендлер @cp.invoice_paid() асинхронный, но может вызывать синхронный bot.send_message
bot.edit_message_text использует синтаксис с chat_id и message_id
pyTelegramBotAPI использует bot.infinity_polling() для запуска

📌

Что важно запомнить

1️⃣

threading.Thread — запускай aiosend в отдельном потоке.

2️⃣

Синхронные вызовы aiosend — работают без await даже внутри синхронного кода.

3️⃣

invoice.poll() синхронно — блокирует поток. Используй только для быстрых платежей.

4️⃣

@cp.invoice_paid() — лучший способ для асинхронной обработки платежей.

🎯

Практическое задание

Задание: Donate-бот на pyTelegramBotAPI

Создай бота на pyTelegramBotAPI + aiosend, который:

По команде /donate создаёт инвойс на 5 USDT
Отправляет ссылку пользователю
Обрабатывает оплату через @cp.invoice_paid()
Благодарит пользователя
Запускает aiosend в отдельном потоке

Подсказка:

Python · Шаблон решения

import asyncio, threading
import telebot
from aiosend import CryptoPay

bot = telebot.TeleBot("BOT_TOKEN")
cp = CryptoPay(token="CP_TOKEN")

@bot.message_handler(commands=["donate"])
def donate(message):
    invoice = cp.create_invoice(
        amount=5, asset="USDT",
        payload=str(message.from_user.id),
    )
    bot.send_message(
        message.chat.id,
        f"💛 Счёт на 5 USDT\n{invoice.bot_invoice_url}"
    )

@cp.invoice_paid()
async def on_paid(invoice):
    bot.send_message(
        int(invoice.payload),
        "❤️ Спасибо за донат!"
    )

def run_aiosend():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loop.run_until_complete(cp.start_polling())

threading.Thread(target=run_aiosend, daemon=True).start()
bot.infinity_polling()

Урок 8.3: Интеграция с pyTelegramBotAPI

5 вопросов