Урок 6.2 — Check Polling

Check polling — это механизм, аналогичный invoice polling, но предназначенный для отслеживания статуса крипто-чеков. Чек может быть в одном из трёх состояний:

В aiosend polling для чеков работает через те же механизмы, что и для инвойсов. Разница лишь в названиях декораторов и внутренних методах API (getChecks вместо getInvoices).

Декоратор @cp.check_activated() регистрирует обработчик, который вызывается, когда крипто-чек активирован (получатель забрал средства). Это аналог @cp.invoice_paid() для чеков.

Сигнатура обработчика:

@cp.check_activated()
async def on_check_activated(check: Check, **kwargs) -> None:
    """Обработчик активированного чека.
    
    Аргументы:
        check: объект Check с обновлённым статусом ACTIVATED
        **kwargs: дополнительные аргументы от фильтров
    """
    pass

Объект Check содержит поля:

import asyncio
from aiosend import CryptoPay

cp = CryptoPay(token="YOUR_TOKEN")

@cp.check_activated()
async def on_activated(check, **kwargs):
    print(f"✅ Чек #{check.check_id} активирован!")
    print(f"   Сумма: {check.amount} {check.asset}")
    print(f"   Кто активировал: {check.activated_by_user_id}")
    print(f"   Время: {check.activated_at}")
    print(f"   Payload: {check.payload}")

async def main():
    # Создаём чек
    check = await cp.create_check(
        amount=10,
        asset="USDT",
        payload="bonus_for_user_123",
    )
    print(f"Чек создан: {check.bot_check_url}")
    
    # Запускаем polling
    await cp.start_polling()

asyncio.run(main())

Декоратор @cp.check_expired() вызывается, когда срок действия чека истёк, и он не был активирован получателем. Это важно для возврата средств или пересоздания чека.

@cp.check_expired()
async def on_check_expired(check: Check, **kwargs) -> None:
    """Обработчик истёкшего чека.
    
    Аргументы:
        check: объект Check со статусом EXPIRED
        **kwargs: дополнительные аргументы от фильтров
    """
    pass

Типичные сценарии использования check_expired:

• Возврат средств: если чек не был активирован, вернуть средства отправителю
• Уведомление: отправить сообщение создателю чека об истечении срока
• Пересоздание: автоматически создать новый чек с продлённым сроком
• Логирование: записать статистику неиспользованных чеков

Важно понимать: средства чека не «сгорают» при истечении срока. Они возвращаются на баланс создателя чека. check_expired просто уведомляет вас, что срок истёк.

import asyncio
import logging
from aiosend import CryptoPay
from aiosend.polling import PollingConfig

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

cp = CryptoPay(
    token="YOUR_TOKEN",
    polling_config=PollingConfig(timeout=300, delay=2),
)

# Статистика
stats = {"created": 0, "activated": 0, "expired": 0}

@cp.check_activated()
async def handle_activated(check, **kwargs):
    stats["activated"] += 1
    logger.info(f"🎉 Чек #{check.check_id} активирован!")
    logger.info(f"   Сумма: {check.amount} {check.asset}")
    
    # Бизнес-логика: начисление бонуса, запись в БД
    if check.payload:
        user_id = check.payload.replace("user_", "")
        logger.info(f"   Пользователь #{user_id} получил {check.amount} {check.asset}")
        await award_bonus(int(user_id), check)

@cp.check_expired()
async def handle_expired(check, **kwargs):
    stats["expired"] += 1
    logger.warning(f"⏰ Чек #{check.check_id} истёк!")
    logger.info(f"   Сумма: {check.amount} {check.asset} возвращена на баланс")
    
    if check.payload:
        user_id = check.payload.replace("user_", "")
        logger.info(f"   Уведомление пользователю #{user_id} об истечении чека")
        await notify_expired(int(user_id), check)

async def award_bonus(user_id: int, check):
    await asyncio.sleep(0.3)
    logger.info(f"   ✅ Бонус {check.amount} {check.asset} начислен user #{user_id}")

async def notify_expired(user_id: int, check):
    logger.info(f"   📧 Уведомление отправлено user #{user_id}")

async def main():
    # Создаём несколько чеков
    for i in range(3):
        check = await cp.create_check(
            amount=5 + i * 5,
            asset="USDT",
            payload=f"user_{100 + i}",
        )
        stats["created"] += 1
        logger.info(f"📄 Создан чек #{check.check_id} на {check.amount} {check.asset}")
    
    logger.info(f"\nСоздано чеков: {stats['created']}")
    logger.info("Запуск polling...")
    
    try:
        await cp.start_polling()
    except KeyboardInterrupt:
        logger.info("\nPolling остановлен пользователем")
    
    logger.info(f"Итог: активировано {stats['activated']}, истекло {stats['expired']}")

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

Как и для инвойсов, для чеков есть метод check.poll(), который позволяет дождаться активации конкретного чека вручную.

async def poll(
    self,
    timeout: int | None = None,
    delay: float | None = None,
) -> Check:
    """Ожидает активации или истечения чека.
    
    Аргументы:
        timeout: максимальное время ожидания
        delay: задержка между проверками
    
    Возвращает:
        Check с обновлённым статусом (ACTIVATED или EXPIRED)
    """
    ...

Внутри poll() использует тот же механизм, что и invoice.poll() — циклический вызов get_checks(check_ids=[...]).

import asyncio
from aiosend import CryptoPay

async def main():
    cp = CryptoPay(token="YOUR_TOKEN")
    
    # Создаём чек
    check = await cp.create_check(
        amount=25,
        asset="USDT",
        payload="prize_for_user_456",
    )
    print(f"Чек создан: {check.bot_check_url}")
    print("Ожидаем активации чека...")
    
    # Ждём активации
    result = await check.poll(timeout=300, delay=2)
    
    if result.status == "activated":
        print(f"✅ Чек #{result.check_id} активирован!")
        print(f"   Кем: {result.activated_by_user_id}")
        print(f"   Когда: {result.activated_at}")
    else:
        print(f"❌ Чек просрочен")

asyncio.run(main())

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

           ┌─────────────────────────────────────────────┐
           │        1. Создание чека                       │
           │  cp.create_check(amount, asset, ...)         │
           │         → status = ACTIVE                    │
           └────────────────┬────────────────────────────┘
                            │
                            ▼
           ┌─────────────────────────────────────────────┐
           │        2. Polling (декораторы или poll())    │
           │  cp.start_polling()  или  check.poll()      │
           │  → периодический опрос getChecks()           │
           └────────────────┬────────────────────────────┘
                            │
              ┌─────────────┴─────────────┐
              ▼                           ▼
   ┌─────────────────────┐   ┌─────────────────────┐
   │ 3a. Активирован      │   │ 3b. Истёк срок      │
   │ Получатель забрал    │   │ Срок действия истёк │
   │ → status = ACTIVATED│   │ → status = EXPIRED  │
   │ Вызов on_activated() │   │ Вызов on_expired()  │
   │ check_activated      │   │ check_expired       │
   └─────────────────────┘   └─────────────────────┘
              ▼                           ▼
   ┌─────────────────────┐   ┌─────────────────────┐
   │ 4a. Обработка        │   │ 4b. Обработка       │
   │ Начисление средств   │   │ Возврат на баланс   │
   │ Запись в БД         │   │ Уведомление         │
   │ Благодарность       │   │ Пересоздание чека   │
   └─────────────────────┘   └─────────────────────┘

Важные отличия от invoice polling:

• Чек создаётся вами (вы отдаёте деньги), инвойс создаётся для вас (вам платят)
• При активации чека деньги списываются с вашего баланса и зачисляются получателю
• При истечении чека деньги возвращаются на ваш баланс
• Чек не может быть «оплачен частично» — только полностью активирован или истёк

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

import asyncio
import logging
from dataclasses import dataclass, field
from datetime import datetime
from aiosend import CryptoPay
from aiosend.polling import PollingConfig

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class BonusSystem:
    """Система управления бонусными чеками."""
    cp: CryptoPay
    pending_users: dict[int, str] = field(default_factory=dict)
    
    def register_handlers(self):
        @self.cp.check_activated()
        async def on_activated(check, **kwargs):
            if check.payload and check.payload.startswith("bonus_"):
                user_id = int(check.payload.split("_")[1])
                logger.info(f"🎉 Пользователь #{user_id} активировал бонус!")
                logger.info(f"   Получено: {check.amount} {check.asset}")
                if user_id in self.pending_users:
                    del self.pending_users[user_id]
        
        @self.cp.check_expired()
        async def on_expired(check, **kwargs):
            if check.payload and check.payload.startswith("bonus_"):
                user_id = int(check.payload.split("_")[1])
                logger.warning(f"⏰ Бонус для #{user_id} истёк. Создаём новый...")
                # Автоматически пересоздаём чек
                await self.issue_bonus(user_id, float(check.amount), check.asset)
    
    async def issue_bonus(self, user_id: int, amount: float, asset: str):
        """Создаёт бонусный чек для пользователя."""
        check = await self.cp.create_check(
            amount=amount,
            asset=asset,
            payload=f"bonus_{user_id}",
        )
        self.pending_users[user_id] = check.bot_check_url
        logger.info(f"📄 Бонус #{check.check_id} для user #{user_id}: {check.amount} {check.asset}")
        return check
    
    async def run(self):
        self.register_handlers()
        logger.info("🚀 Запуск check polling...")
        await self.cp.start_polling()

async def main():
    cp = CryptoPay(
        token="YOUR_TOKEN",
        polling_config=PollingConfig(timeout=86400, delay=5),
    )
    
    system = BonusSystem(cp)
    
    # Создаём бонусы для нескольких пользователей
    await system.issue_bonus(101, 10, "USDT")
    await system.issue_bonus(102, 25, "USDT")
    await system.issue_bonus(103, 50, "TON")
    
    print(f"Создано бонусов: {len(system.pending_users)}")
    print("Ожидание активации или истечения...")
    
    try:
        await system.run()
    except KeyboardInterrupt:
        logger.info("Система остановлена")

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

Рассмотрим ещё несколько сценариев использования check polling в реальных проектах.

Пример: Массовая рассылка чеков с отслеживанием

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

import asyncio
import logging
from dataclasses import dataclass, field
from aiosend import CryptoPay
from aiosend.polling import PollingConfig

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class BulkCheckSender:
    """Массовая рассылка чеков с отслеживанием."""
    cp: CryptoPay
    sent: dict[int, str] = field(default_factory=dict)  # check_id -> payload
    activated: set[int] = field(default_factory=set)
    expired: set[int] = field(default_factory=set)
    
    def register_handlers(self):
        @self.cp.check_activated()
        async def on_activated(check, **kwargs):
            self.activated.add(check.check_id)
            logger.info(f"✅ Чек #{check.check_id} активирован!")
            if check.check_id in self.sent:
                payload = self.sent[check.check_id]
                logger.info(f"   Payload: {payload}, сумма: {check.amount} {check.asset}")
        
        @self.cp.check_expired()
        async def on_expired(check, **kwargs):
            self.expired.add(check.check_id)
            logger.warning(f"⏰ Чек #{check.check_id} истёк!")
    
    async def send_bonuses(self, users: list[dict]):
        """Создаёт чеки для списка пользователей."""
        for user in users:
            check = await self.cp.create_check(
                amount=user["amount"],
                asset=user.get("asset", "USDT"),
                payload=f"bulk_bonus_{user['user_id']}",
            )
            self.sent[check.check_id] = f"user_{user['user_id']}"
            logger.info(f"📄 Создан чек #{check.check_id} для user #{user['user_id']}")
            await asyncio.sleep(0.5)  # небольшая пауза между созданиями
    
    def print_stats(self):
        total = len(self.sent)
        act = len(self.activated)
        exp = len(self.expired)
        logger.info(f"\n{'='*40}")
        logger.info(f"Статистика рассылки:")
        logger.info(f"  Отправлено: {total}")
        logger.info(f"  Активировано: {act}")
        logger.info(f"  Истекло: {exp}")
        logger.info(f"  Ожидают: {total - act - exp}")

async def main():
    cp = CryptoPay(
        token="YOUR_TOKEN",
        polling_config=PollingConfig(timeout=86400, delay=3),
    )
    
    sender = BulkCheckSender(cp)
    sender.register_handlers()
    
    # Создаём тестовых пользователей
    test_users = [
        {"user_id": 101, "amount": 10},
        {"user_id": 102, "amount": 25},
        {"user_id": 103, "amount": 50, "asset": "TON"},
    ]
    
    await sender.send_bonuses(test_users)
    logger.info("Запуск polling...")
    
    try:
        await cp.start_polling()
    except KeyboardInterrupt:
        logger.info("Остановлено пользователем")
    
    sender.print_stats()

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

Пример: Создание чека с последующим ожиданием через poll()

Для простых сценариев можно использовать check.poll() в отдельной задаче:

import asyncio
from aiosend import CryptoPay
from aiosend.polling import PollingConfig

async def create_and_wait_check(cp: CryptoPay, amount: float,
                                  asset: str, payload: str):
    """Создаёт чек и ожидает его активации."""
    check = await cp.create_check(
        amount=amount,
        asset=asset,
        payload=payload,
    )
    print(f"Чек создан: {check.bot_check_url}")
    print(f"Ожидание активации (таймаут: 5 минут)...")
    
    try:
        result = await check.poll(timeout=300, delay=2)
        if result.status == "activated":
            print(f"✅ Чек #{result.check_id} активирован!")
            print(f"   Кем: {result.activated_by_user_id}")
            print(f"   Когда: {result.activated_at}")
            return result
        else:
            print(f"❌ Чек истёк (status={result.status})")
    except asyncio.TimeoutError:
        print(f"⏰ Таймаут ожидания активации")
    
    return None

async def main():
    cp = CryptoPay(
        token="YOUR_TOKEN",
        polling_config=PollingConfig(timeout=600, delay=2),
    )
    
    # Создаём и ждём несколько чеков параллельно
    tasks = [
        create_and_wait_check(cp, 10, "USDT", "bonus_user_1"),
        create_and_wait_check(cp, 25, "USDT", "bonus_user_2"),
    ]
    
    results = await asyncio.gather(*tasks)
    activated = sum(1 for r in results if r is not None)
    print(f"\nИтого активировано: {activated} из {len(tasks)}")

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

Пример: Обработка check и invoice в одном polling

Один клиент может одновременно обрабатывать и инвойсы, и чеки. Все обработчики регистрируются на одном cp и запускаются одним start_polling():

import asyncio
from aiosend import CryptoPay
from aiosend.polling import PollingConfig

cp = CryptoPay(
    token="YOUR_TOKEN",
    polling_config=PollingConfig(timeout=600, delay=2),
)

# Регистрируем все обработчики вместе
@cp.invoice_paid()
async def on_invoice_paid(invoice, **kwargs):
    print(f"💵 Инвойс #{invoice.invoice_id} оплачен: {invoice.amount} {invoice.asset}")

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

@cp.check_activated()
async def on_check_activated(check, **kwargs):
    print(f"🎉 Чек #{check.check_id} активирован: {check.amount} {check.asset}")

@cp.check_expired()
async def on_check_expired(check, **kwargs):
    print(f"⏰ Чек #{check.check_id} истёк")

async def main():
    # Создаём и инвойс, и чек
    invoice = await cp.create_invoice(amount=15, asset="USDT")
    check = await cp.create_check(amount=5, asset="USDT")
    print(f"Инвойс: {invoice.bot_invoice_url}")
    print(f"Чек: {check.bot_check_url}")
    
    # Один polling обработает всё
    await cp.start_polling()

asyncio.run(main())