Ejemplos del Mundo Real

Esta página muestra patrones de integración completos construidos sobre CryptoBotClient y AsyncCryptoBotClient.

Configuración Compartida

import os

from cryptobot import CryptoBotClient
from cryptobot.errors import CryptoBotError
from cryptobot.models import Asset, ButtonName, CheckStatus, Status

client = CryptoBotClient(api_token=os.environ["CRYPTOBOT_API_TOKEN"])

Escáner de Facturas Asíncrono

Use los ayudantes de paginación asíncrona para procesar grandes conjuntos de facturas pagadas sin cargar todo de una vez.

import asyncio
import os

from cryptobot import AsyncCryptoBotClient
from cryptobot.models import Asset, Status


async def scan_paid_invoices():
    async with AsyncCryptoBotClient(
        api_token=os.environ["CRYPTOBOT_API_TOKEN"],
        max_retries=2,
        retry_backoff=0.5,
    ) as client:
        async for invoice in client.iter_invoices(
            asset=Asset.USDT,
            status=Status.paid,
            page_size=100,
            start_offset=0,
        ):
            print("paid invoice:", invoice.invoice_id, invoice.paid_amount or invoice.amount)


asyncio.run(scan_paid_invoices())

Checkout de Comercio Electrónico

Cree una factura durante el checkout, luego consulte el estado para completar el pedido.

from dataclasses import dataclass
from typing import Dict


@dataclass
class Order:
    order_id: str
    amount_usdt: float
    invoice_id: int
    status: str


class CheckoutService:
    def __init__(self, client: CryptoBotClient):
        self.client = client
        self.orders: Dict[str, Order] = {}

    def create_checkout(self, order_id: str, amount_usdt: float) -> str:
        invoice = self.client.create_invoice(
            asset=Asset.USDT,
            amount=amount_usdt,
            description=f"Order {order_id}",
            payload=order_id,
            expires_in=1800,
            paid_btn_name=ButtonName.callback,
            paid_btn_url=f"https://shop.example/orders/{order_id}",
        )

        self.orders[order_id] = Order(
            order_id=order_id,
            amount_usdt=amount_usdt,
            invoice_id=invoice.invoice_id,
            status="pending",
        )
        return invoice.bot_invoice_url

    def sync_order(self, order_id: str) -> str:
        order = self.orders[order_id]
        invoices = self.client.get_invoices(invoice_ids=str(order.invoice_id))
        if not invoices:
            return order.status

        invoice = invoices[0]
        if invoice.status == Status.paid:
            order.status = "paid"
        elif invoice.status == Status.expired:
            order.status = "expired"

        return order.status


service = CheckoutService(client)
payment_url = service.create_checkout("ORDER-1001", 29.99)
print("Pay here:", payment_url)
print("Current status:", service.sync_order("ORDER-1001"))

Flujo de Trabajo de Pagos

Use valores únicos de spend_id para mantener las transferencias idempotentes.

from datetime import datetime
import uuid


def make_spend_id(prefix: str, user_id: int) -> str:
    stamp = datetime.utcnow().strftime("%Y%m%d%H%M%S")
    return f"{prefix}_{user_id}_{stamp}_{uuid.uuid4().hex[:8]}"


def send_reward(client: CryptoBotClient, user_id: int, amount: float):
    spend_id = make_spend_id("reward", user_id)
    try:
        transfer = client.transfer(
            user_id=user_id,
            asset=Asset.TON,
            amount=amount,
            spend_id=spend_id,
            comment="Referral reward",
        )
        return {"ok": True, "transfer_id": transfer.transfer_id, "spend_id": spend_id}
    except CryptoBotError as exc:
        return {"ok": False, "code": exc.code, "name": exc.name, "spend_id": spend_id}


print(send_reward(client, user_id=123456789, amount=0.5))

Rastreador de Donaciones

Rastree el progreso de donaciones desde facturas pagadas.

::

Cumplimiento de Pedidos por Webhook

Use el listener para marcar pedidos como pagados sin necesidad de polling.

::

Airdrop con Checks Criptográficos

Cree checks para un lote de usuarios y rastree las activaciones.

::

Panel de Estadísticas de la Aplicación

Obtenga estadísticas y muestre un informe resumido.

::

Libro Mayor de Transferencias

Escanee las transferencias salientes y construya un libro mayor local.

::

Prueba Rápida de Testnet

Script de validación rápida para entornos que no son de producción.

import os

from cryptobot import CryptoBotClient
from cryptobot.models import Asset


def smoke_check_testnet() -> bool:
    client = CryptoBotClient(
        api_token=os.environ["CRYPTOBOT_TESTNET_TOKEN"],
        is_mainnet=False,
        timeout=10.0,
    )

    try:
        app = client.get_me()
        balances = client.get_balances()
        rates = client.get_exchange_rates()
        invoice = client.create_invoice(asset=Asset.USDT, amount=1, description="testnet smoke")

        print("App:", app.name)
        print("Balances:", len(balances))
        print("Rates:", len(rates))
        print("Invoice:", invoice.invoice_id)
        return True
    except Exception as exc:
        print("Smoke check failed:", exc)
        return False


if __name__ == "__main__":
    smoke_check_testnet()

Próximos Pasos