Урок 5.2 — Получение и управление переводами

Метод get_transfers() возвращает список переводов с возможностью фильтрации. Это основной способ получить историю всех исходящих переводов твоего приложения.

async def get_transfers(
    self,
    asset: str | None = None,
    transfer_ids: list[int] | None = None,
    spend_id: str | None = None,
    offset: int | None = None,
    count: int | None = None,
) -> list[Transfer]

Все параметры опциональны. Если вызвать без аргументов — вернутся все переводы приложения.

from aiosend import CryptoPay

cp = CryptoPay(token="YOUR_TOKEN")

# 1. Все переводы без фильтрации
all_transfers = await cp.get_transfers()
print(f"Всего переводов: {len(all_transfers)}")

# 2. Только USDT-переводы
usdt_transfers = await cp.get_transfers(asset="USDT")
for t in usdt_transfers:
    print(f"#{t.transfer_id}: {t.amount} {t.asset} → user {t.user_id}")

# 3. Конкретные переводы по ID
specific = await cp.get_transfers(transfer_ids=[1, 2, 3])
print(f"Найдено: {len(specific)}")

# 4. Поиск по spend_id
transfer = await cp.get_transfers(spend_id="my-unique-spend-id-123")
# Вернёт список с одним элементом (или пустой)

# 5. С пагинацией
page = await cp.get_transfers(offset=0, count=50)
print(f"Первая страница: {len(page)} переводов")

Параметр spend_id в get_transfers() позволяет найти перевод по его ID идемпотентности. Это особенно полезно, когда:

• Ты хочешь проверить, был ли выполнен перевод с определённым spend_id
• Нужно найти перевод, связанный с заказом/транзакцией в твоей системе
• Ты обрабатываешь вебхук и хочешь получить полную информацию о переводе

from aiosend import CryptoPay

async def find_transfer_by_spend_id(cp: CryptoPay, spend_id: str):
    """Находит перевод по spend_id."""
    transfers = await cp.get_transfers(spend_id=spend_id)

    if not transfers:
        print(f"Перевод с spend_id '{spend_id}' не найден")
        return None

    transfer = transfers[0]
    print(f"✅ Найден перевод #{transfer.transfer_id}")
    print(f"   Получатель: {transfer.user_id}")
    print(f"   Сумма: {transfer.amount} {transfer.asset}")
    print(f"   Статус: {transfer.status}")
    print(f"   Дата: {transfer.completed_at}")
    return transfer

async def check_transfer_status(cp: CryptoPay, order_id: str):
    """Проверяет статус перевода по ID заказа."""
    spend_id = f"refund_{order_id}"
    transfers = await cp.get_transfers(spend_id=spend_id)

    if transfers:
        t = transfers[0]
        print(f"Возврат по заказу {order_id}: #{t.transfer_id}, {t.status}")
        return t
    else:
        print(f"Возврат по заказу {order_id}: не найден")
        return None

Как и с чеками, для переводов работает пагинация через offset и count. Это необходимо, когда переводов много (сотни и тысячи).

from aiosend import CryptoPay

async def get_all_transfers(cp: CryptoPay) -> list:
    """Получает все переводы приложения через пагинацию."""
    all_transfers = []
    offset = 0
    page_size = 500

    while True:
        page = await cp.get_transfers(offset=offset, count=page_size)
        if not page:
            break
        all_transfers.extend(page)
        offset += len(page)
        print(f"Загружено {len(all_transfers)} переводов...")

    print(f"Всего загружено: {len(all_transfers)} переводов")
    return all_transfers

async def get_transfers_by_asset(cp: CryptoPay, asset: str) -> list:
    """Получает все переводы в указанном активе."""
    transfers = []
    offset = 0
    page_size = 500

    while True:
        page = await cp.get_transfers(
            asset=asset,
            offset=offset,
            count=page_size,
        )
        if not page:
            break
        transfers.extend(page)
        offset += len(page)

    print(f"Переводов в {asset}: {len(transfers)}")
    return transfers

При работе с переводами могут возникать различные ошибки. Рассмотрим основные и способы их обработки.

Недостаточный баланс (Insufficient Balance)

Самая частая ошибка при отправке переводов — недостаточно средств на балансе приложения. Всегда проверяй баланс перед отправкой.

from aiosend import CryptoPay
from aiosend.exceptions import APIError

async def transfer_with_balance_check(
    cp: CryptoPay,
    user_id: int,
    asset: str,
    amount: float,
    spend_id: str,
):
    """Безопасный перевод с проверкой баланса."""
    # 1. Проверяем баланс
    balances = await cp.get_balances()
    balance_map = {b.code: float(b.available) for b in balances}

    if asset not in balance_map:
        print(f"❌ Актив {asset} не найден на балансе")
        return None

    available = balance_map[asset]
    if available < amount:
        print(f"❌ Недостаточно {asset}: есть {available}, нужно {amount}")
        return None

    # 2. Выполняем перевод
    try:
        transfer = await cp.transfer(
            user_id=user_id,
            asset=asset,
            amount=amount,
            spend_id=spend_id,
        )
        print(f"✅ Перевод #{transfer.transfer_id} выполнен")
        return transfer

    except APIError as e:
        print(f"❌ Ошибка API: {e}")
        # Анализируем ошибку
        if "insufficient" in str(e).lower():
            print("   Причина: недостаточно средств")
        elif "not found" in str(e).lower():
            print("   Причина: пользователь не в @CryptoBot")
        elif "rate limit" in str(e).lower():
            print("   Причина: слишком много запросов")
        return None

Rate Limiting

Crypto Pay API имеет ограничение на количество запросов в минуту. При превышении возвращается ошибка TooManyRequestsError.

import asyncio
from aiosend.exceptions import TooManyRequestsError

async def transfer_with_retry(cp, user_id, asset, amount, spend_id, max_retries=3):
    """Перевод с повторными попытками при rate limit."""
    for attempt in range(max_retries):
        try:
            return await cp.transfer(
                user_id=user_id,
                asset=asset,
                amount=amount,
                spend_id=spend_id,
            )
        except TooManyRequestsError:
            if attempt < max_retries - 1:
                wait = 2 ** (attempt + 1)  # 2, 4, 8 секунд
                print(f"Rate limit, ждём {wait}с...")
                await asyncio.sleep(wait)
            else:
                print("❌ Превышен лимит запросов")
                raise
    return None

Пользователь не найден

Если пользователь никогда не запускал @CryptoBot, API вернёт ошибку. Это нужно учитывать при проектировании логики выплат.

import uuid
from aiosend import CryptoPay
from aiosend.exceptions import (
    APIError,
    TooManyRequestsError,
    MethodValuesError,
)

class TransferError(Exception):
    """Базовое исключение для ошибок перевода."""

class InsufficientBalanceError(TransferError):
    """Недостаточно средств."""

class UserNotFoundError(TransferError):
    """Пользователь не найден в @CryptoBot."""

async def robust_transfer(cp: CryptoPay, user_id: int, asset: str, amount: float):
    """Надёжный перевод с полной обработкой ошибок."""
    try:
        return await cp.transfer(
            user_id=user_id,
            asset=asset,
            amount=amount,
            spend_id=str(uuid.uuid4()),
        )

    except MethodValuesError as e:
        raise TransferError(f"Неверные параметры: {e}") from e

    except TooManyRequestsError as e:
        raise TransferError("Слишком много запросов. Попробуйте позже.") from e

    except APIError as e:
        error_text = str(e).lower()
        if "insufficient" in error_text:
            raise InsufficientBalanceError(
                f"Недостаточно {asset} на балансе"
            ) from e
        elif "user" in error_text and "not found" in error_text:
            raise UserNotFoundError(
                f"Пользователь {user_id} не найден в @CryptoBot"
            ) from e
        else:
            raise TransferError(f"Ошибка API: {e}") from e

Давай создадим полноценный инструмент для логирования и анализа переводов — TransferLogger.

from collections import Counter, defaultdict
from datetime import datetime
from aiosend import CryptoPay

class TransferLogger:
    """Логирование и анализ переводов."""

    def __init__(self, cp: CryptoPay):
        self.cp = cp

    async def show_all(self, asset: str | None = None):
        """Выводит все переводы в читаемом формате."""
        transfers = await self.cp.get_transfers(asset=asset)

        if not transfers:
            print("Нет переводов.")
            return

        print(f"{'ID':<8} {'Пользователь':<12} {'Сумма':<16} {'Актив':<8} {'Дата':<24} {'Комментарий'}")
        print("-" * 100)

        for t in transfers:
            comment = (t.comment[:30] + "..") if t.comment and len(t.comment) > 30 else (t.comment or "")
            print(f"{t.transfer_id:<8} {t.user_id:<12} {t.amount:<16} {t.asset:<8} {t.completed_at or '':<24} {comment}")

        print(f"\nВсего: {len(transfers)} переводов")

    async def stats(self):
        """Показывает статистику по переводам."""
        all_t = await self.cp.get_transfers()

        print("📊 Статистика переводов")
        print(f"   Всего переводов: {len(all_t)}")

        # По активам
        by_asset = Counter(t.asset for t in all_t)
        print("\n   По активам:")
        for asset, count in by_asset.most_common():
            total = sum(float(t.amount) for t in all_t if t.asset == asset)
            print(f"     {asset}: {count} переводов, сумма {total:.4f}")

        # По пользователям (топ-10)
        by_user = Counter(t.user_id for t in all_t)
        print("\n   Топ-10 получателей:")
        for user_id, count in by_user.most_common(10):
            total = sum(float(t.amount) for t in all_t if t.user_id == user_id)
            print(f"     user #{user_id}: {count} раз, всего {total:.4f}")

    async def export_csv(self, filename: str = "transfers.csv"):
        """Экспортирует переводы в CSV."""
        import csv

        transfers = await self.cp.get_transfers()
        with open(filename, "w", newline="") as f:
            writer = csv.writer(f)
            writer.writerow(["ID", "User ID", "Asset", "Amount", "Status",
                           "Completed At", "Comment", "Spend ID"])
            for t in transfers:
                writer.writerow([
                    t.transfer_id, t.user_id, t.asset, t.amount,
                    t.status, t.completed_at, t.comment, t.spend_id,
                ])
        print(f"✅ Экспортировано {len(transfers)} переводов в {filename}")


async def main():
    cp = CryptoPay(token="YOUR_TOKEN")
    logger = TransferLogger(cp)

    await logger.show_all()
    await logger.stats()
    # await logger.export_csv("my_transfers.csv")


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

Используя get_transfers(), ты можешь построить систему мониторинга, которая отслеживает все переводы и оповещает о проблемах.

from aiosend import CryptoPay

class TransferMonitor:
    """Мониторинг переводов с оповещениями."""

    def __init__(self, cp: CryptoPay):
        self.cp = cp
        self.last_id = 0

    async def check_new(self) -> list:
        """Проверяет новые переводы (последние 10)."""
        transfers = await self.cp.get_transfers(count=10)
        new_ones = [t for t in transfers if t.transfer_id > self.last_id]

        if new_ones:
            self.last_id = max(t.transfer_id for t in new_ones)

        return new_ones

    async def check_large_transfers(self, threshold: float = 1000):
        """Находит переводы больше указанной суммы."""
        transfers = await self.cp.get_transfers()
        large = [t for t in transfers if float(t.amount) >= threshold]

        if large:
            print(f"⚠️ Крупные переводы (>{threshold} USD):")
            for t in large:
                print(f"   #{t.transfer_id}: {t.amount} {t.asset} → {t.user_id}")
        return large

    async def summary_report(self) -> str:
        """Генерирует текстовый отчёт."""
        transfers = await self.cp.get_transfers()
        total = len(transfers)
        total_usdt = sum(float(t.amount) for t in transfers if t.asset == "USDT")
        unique_users = len(set(t.user_id for t in transfers))

        return (
            f"📋 Отчёт по переводам:\n"
            f"   Всего переводов: {total}\n"
            f"   Сумма USDT: {total_usdt:.2f}\n"
            f"   Уникальных получателей: {unique_users}\n"
            f"   Последний ID: {max((t.transfer_id for t in transfers), default=0)}"
        )


async def main():
    cp = CryptoPay(token="YOUR_TOKEN")
    monitor = TransferMonitor(cp)

    # Проверяем новые переводы
    new = await monitor.check_new()
    if new:
        print(f"Новых переводов: {len(new)}")

    # Проверяем крупные переводы
    await monitor.check_large_transfers(threshold=500)

    # Отчёт
    print(await monitor.summary_report())


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

Комбинируя get_transfers() с другими методами, можно выполнять массовые операции: проверять статус всех переводов, группировать, экспортировать.

from aiosend import CryptoPay

async def verify_transfers(cp: CryptoPay, spend_ids: list[str]) -> dict:
    """Проверяет статус нескольких переводов по spend_id."""
    results = {}
    for sid in spend_ids:
        transfers = await cp.get_transfers(spend_id=sid)
        results[sid] = transfers[0] if transfers else None
    return results

async def get_user_transfer_history(
    cp: CryptoPay,
    user_id: int,
) -> list:
    """Получает все переводы конкретного пользователя."""
    all_t = await cp.get_transfers()
    return [t for t in all_t if t.user_id == user_id]

async def total_spent_on_user(
    cp: CryptoPay,
    user_id: int,
    asset: str = "USDT",
) -> float:
    """Считает общую сумму выплат пользователю."""
    user_transfers = await get_user_transfer_history(cp, user_id)
    return sum(
        float(t.amount)
        for t in user_transfers
        if t.asset == asset
    )

async def main():
    cp = CryptoPay(token="YOUR_TOKEN")

    # Проверяем несколько переводов
    statuses = await verify_transfers(cp, ["spend_1", "spend_2", "spend_3"])
    for sid, transfer in statuses.items():
        if transfer:
            print(f"{sid}: #{transfer.transfer_id} — {transfer.status}")
        else:
            print(f"{sid}: не найден")

    # История пользователя
    history = await get_user_transfer_history(cp, user_id=123456789)
    print(f"Переводов пользователю: {len(history)}")

    # Общая сумма
    total = await total_spent_on_user(cp, user_id=123456789)
    print(f"Всего отправлено: {total:.2f} USDT")


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