Урок 3.2 — Получение и фильтрация инвойсов

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

async def get_invoices(
    self,
    asset: str | None = None,
    fiat: str | None = None,
    invoice_ids: list[int] | None = None,
    status: str | None = None,
    offset: int = 0,
    count: int = 100,
) -> list[Invoice]

Все параметры опциональны. Если вызвать get_invoices() без параметров, будут возвращены последние 100 инвойсов (начиная с offset=0). Максимальное значение count — 1000.

import asyncio
from aiosend import CryptoPay

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

    # 1. Все счета (последние 100)
    all_invoices = await cp.get_invoices()
    print(f"Всего счетов (последние 100): {len(all_invoices)}")

    # 2. Первые 50 счетов
    first_50 = await cp.get_invoices(offset=0, count=50)
    print(f"Первые 50: {len(first_50)}")

    # 3. Пропустить 100, взять 50
    next_50 = await cp.get_invoices(offset=100, count=50)
    print(f"Следующие 50: {len(next_50)}")

    # 4. Вывод ID всех счетов
    for inv in all_invoices:
        print(f"  #{inv.invoice_id}: {inv.amount} {inv.asset or inv.fiat}")

asyncio.run(basic_example())

Параметры asset и fiat работают аналогично соответствующим параметрам в create_invoice(), но теперь — для фильтрации существующих счетов. Это позволяет быстро находить все счета в определённой валюте.

Особенности фильтрации:

• Если указать asset, будут возвращены только счета, созданные с этим активом
• Если указать fiat, будут возвращены только fiat-счета с этой валютой
• Нельзя указать одновременно asset и fiat — API вернёт ошибку
• Если не указывать ни один — будут возвращены все счета (с учётом других фильтров)

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

    # Все счета в USDT
    usdt_invoices = await cp.get_invoices(asset="USDT")
    print(f"Найдено USDT счетов: {len(usdt_invoices)}")

    # Все счета в TON
    ton_invoices = await cp.get_invoices(asset="TON")
    print(f"Найдено TON счетов: {len(ton_invoices)}")

    # Все fiat-счета в USD
    usd_invoices = await cp.get_invoices(fiat="USD")
    print(f"Найдено USD счетов: {len(usd_invoices)}")

    # Комбинированный фильтр: USDT + оплаченные
    paid_usdt = await cp.get_invoices(asset="USDT", status="paid")
    print(f"Оплаченных USDT: {len(paid_usdt)}")

Фильтрация по статусу — одна из самых полезных возможностей. Ты можешь получить все активные счета (ожидающие оплаты), все оплаченные или все просроченные. Это основа для любого мониторинга платежей.

Значения параметра status:

from aiosend import InvoiceStatus

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

    # Активные счета (ждут оплаты)
    active = await cp.get_invoices(status=InvoiceStatus.ACTIVE)
    print(f"Активных счетов: {len(active)}")
    for inv in active:
        print(f"  #{inv.invoice_id}: {inv.amount} "
              f"{inv.asset or inv.fiat} — истекает: {inv.expires_at}")

    # Оплаченные счета
    paid = await cp.get_invoices(status=InvoiceStatus.PAID)
    print(f"Оплаченных счетов: {len(paid)}")
    for inv in paid:
        print(f"  #{inv.invoice_id}: {inv.amount} "
              f"{inv.asset or inv.fiat} — оплачено: {inv.paid_at}")

    # Просроченные счета
    expired = await cp.get_invoices(status=InvoiceStatus.EXPIRED)
    print(f"Просроченных счетов: {len(expired)}")
    for inv in expired:
        print(f"  #{inv.invoice_id}: {inv.amount} "
              f"{inv.asset or inv.fiat} — создан: {inv.created_at}")

    # Комбинированный фильтр: оплаченные счета в USDT
    paid_usdt = await cp.get_invoices(
        asset="USDT",
        status=InvoiceStatus.PAID,
    )
    print(f"Оплаченных USDT: {len(paid_usdt)}")
    for inv in paid_usdt:
        print(f"  #{inv.invoice_id}: {inv.amount} USDT — "
              f"получено {inv.paid_amount} {inv.paid_asset}")

Сравнение фильтров статуса

Параметр invoice_ids принимает список ID инвойсов и возвращает только те, которые есть в этом списке. Это удобно, когда ты сохранил ID счетов в своей базе данных и хочешь проверить их актуальный статус одним запросом.

Важные ограничения:

• Максимум 1000 ID в одном запросе
• Если ID не существует, он просто будет проигнорирован (без ошибки)
• Порядок элементов в ответе не гарантируется — сортируй самостоятельно
• Другие фильтры (asset, status, fiat) игнорируются, если передан invoice_ids
• Если передать пустой список — вернётся пустой результат

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

    # Получаем несколько конкретных счетов по их ID
    invoice_ids = [12345, 12346, 12347, 12350]
    invoices = await cp.get_invoices(invoice_ids=invoice_ids)

    for inv in invoices:
        print(f"#{inv.invoice_id}: {inv.amount} "
              f"{inv.asset or inv.fiat} — {inv.status}")

    # Если ID не существует — он просто пропускается
    result = await cp.get_invoices(invoice_ids=[99999999])
    print(f"Найдено с несуществующим ID: {len(result)}")  # → 0

    # Пакетная проверка статусов из базы данных
    db_ids = [12345, 12346, 12347, 12348, 12349, 12350]
    batch_size = 1000
    all_results = []

    for i in range(0, len(db_ids), batch_size):
        batch = db_ids[i:i + batch_size]
        batch_result = await cp.get_invoices(invoice_ids=batch)
        all_results.extend(batch_result)

    print(f"Проверено {len(db_ids)} ID, найдено {len(all_results)}")

Когда у тебя тысячи инвойсов, получать их все сразу неэффективно и невозможно (лимит 1000). offset и count позволяют проходить по списку страницами (пагинация).

async def paginate_all_invoices():
    cp = CryptoPay(token="YOUR_TOKEN")
    page_size = 100
    offset = 0
    all_invoices = []

    while True:
        invoices = await cp.get_invoices(
            offset=offset,
            count=page_size,
        )

        if not invoices:
            print("Больше записей нет")
            break

        page_num = offset // page_size + 1
        print(f"Страница #{page_num}: получено {len(invoices)} инвойсов")

        all_invoices.extend(invoices)
        offset += page_size

        if len(invoices) < page_size:
            print("Это последняя страница")
            break

    print(f"Всего получено: {len(all_invoices)} инвойсов")

Также можно комбинировать пагинацию с фильтрацией:

async def paginate_paid_usdt():
    cp = CryptoPay(token="YOUR_TOKEN")
    page_size = 500
    offset = 0
    all_paid = []

    while True:
        invoices = await cp.get_invoices(
            asset="USDT",
            status="paid",
            offset=offset,
            count=page_size,
        )

        if not invoices:
            break

        all_paid.extend(invoices)
        offset += page_size

        if len(invoices) < page_size:
            break

    total = sum(float(i.paid_amount or i.amount) for i in all_paid)
    print(f"Всего оплаченных USDT счетов: {len(all_paid)}")
    print(f"Общая сумма: {total:.2f} USDT")

    return all_paid

Если тебе нужно получить только один конкретный счёт по его ID, используй get_invoice(). Это проще и быстрее, чем фильтрация через get_invoices(invoice_ids=[...]), и возвращает сразу объект Invoice, а не список.

async def get_invoice(self, invoice_id: int) -> Invoice

from aiosend.exceptions import APIError

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

    # Получаем счёт по ID
    try:
        invoice = await cp.get_invoice(invoice_id=12345)
        print(f"Счёт #{invoice.invoice_id}")
        print(f"Сумма: {invoice.amount} {invoice.asset or invoice.fiat}")
        print(f"Статус: {invoice.status}")
        print(f"Создан: {invoice.created_at}")
        print(f"Истекает: {invoice.expires_at}")
        print(f"Ссылка: {invoice.bot_invoice_url}")

        if invoice.status == "paid":
            print(f"Оплачен: {invoice.paid_at}")
            print(f"Сумма: {invoice.paid_amount} {invoice.paid_asset}")
            print(f"Курс: {invoice.paid_usd_rate}")
            if invoice.comment:
                print(f"Комментарий: {invoice.comment}")

    except APIError as e:
        print(f"Счёт не найден: {e}")
    except Exception as e:
        print(f"Ошибка: {e}")

У объекта Invoice есть метод update(), который обновляет его данные из API. Это удобный shortcut вместо вызова cp.get_invoice(invoice_id). Особенно полезно для проверки статуса: не изменился ли он на "paid".

async def update(self) -> Invoice

Метод запрашивает актуальные данные с API и обновляет все поля текущего объекта. Возвращает сам объект для удобства (chain call). Метод изменяет объект "на месте" — тебе не нужно присваивать результат обратно.

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

    # Создаём счёт
    invoice = await cp.create_invoice(amount=10, asset="USDT")
    print(f"Статус после создания: {invoice.status}")  # active

    # Ждём 30 секунд (пользователь должен оплатить)
    await asyncio.sleep(30)

    # Обновляем статус — метод обновляет объект на месте
    await invoice.update()
    print(f"Статус после обновления: {invoice.status}")

    # Если оплачен — выводим детали
    if invoice.status == "paid":
        print(f"Оплачено: {invoice.paid_amount} {invoice.paid_asset}")
        print(f"Время: {invoice.paid_at}")
        print(f"Курс: {invoice.paid_usd_rate}")

    # Chain call — обновить и сразу использовать
    status = (await invoice.update()).status
    print(f"Статус (chain): {status}")

    # Альтернатива: обновить и получить новое значение
    await invoice.update()
    if invoice.status == "paid":
        print("✅ Счёт оплачен!")
    else:
        print(f"⏳ Статус: {invoice.status}")

Метод update() доступен у обоих основных объектов: Invoice и Check. Это часть общего паттерна shortcut-методов в aiosend (урок 7.2).

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

from aiosend.exceptions import (
    APIError,
    APITimeoutError,
    MethodValuesError,
)

async def safe_get_invoice(invoice_id: int) -> Invoice | None:
    cp = CryptoPay(token="YOUR_TOKEN")
    try:
        return await cp.get_invoice(invoice_id=invoice_id)
    except APIError as e:
        print(f"Счёт #{invoice_id} не найден: {e}")
    except APITimeoutError:
        print(f"Таймаут при запросе счёта #{invoice_id}")
    except Exception as e:
        print(f"Неизвестная ошибка: {e}")
    return None

async def safe_get_invoices(**kwargs) -> list[Invoice]:
    cp = CryptoPay(token="YOUR_TOKEN")
    try:
        return await cp.get_invoices(**kwargs)
    except MethodValuesError as e:
        print(f"Неверные параметры фильтрации: {e}")
    except APITimeoutError:
        print("Таймаут при запросе списка счетов")
    except Exception as e:
        print(f"Неизвестная ошибка: {e}")
    return []

Ретраи (повторные попытки)

При сетевых ошибках полезно использовать повторные попытки с задержкой:

import asyncio
from aiosend.exceptions import APITimeoutError

async def get_invoice_with_retry(
    invoice_id: int,
    max_retries: int = 3,
) -> Invoice | None:
    cp = CryptoPay(token="YOUR_TOKEN")

    for attempt in range(max_retries):
        try:
            return await cp.get_invoice(invoice_id=invoice_id)
        except APITimeoutError as e:
            if attempt == max_retries - 1:
                print(f"Не удалось получить счёт после {max_retries} попыток")
                return None
            wait = 2 ** attempt  # Экспоненциальная задержка: 1, 2, 4 сек
            print(f"Попытка {attempt + 1} не удалась, ждём {wait}с...")
            await asyncio.sleep(wait)
        except APIError:
            # Счёт не найден — ретрай не поможет
            return None

    return None

Пример 1: Список всех оплаченных счетов со статистикой

Получаем все оплаченные счета и выводим статистику по суммам и валютам.

from aiosend import InvoiceStatus
from collections import defaultdict

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

    # Собираем все оплаченные счета с пагинацией
    all_paid = []
    offset = 0
    while True:
        batch = await cp.get_invoices(
            status=InvoiceStatus.PAID,
            offset=offset,
            count=1000,
        )
        if not batch:
            break
        all_paid.extend(batch)
        if len(batch) < 1000:
            break
        offset += 1000

    # Статистика по активам
    by_asset = defaultdict(lambda: {"count": 0, "total": 0.0})

    for inv in all_paid:
        asset = inv.paid_asset or inv.asset or "UNKNOWN"
        amount = float(inv.paid_amount or inv.amount)
        by_asset[asset]["count"] += 1
        by_asset[asset]["total"] += amount

    print(f"Всего оплаченных счетов: {len(all_paid)}")
    print(f"\nСтатистика по активам:")
    print(f"{'Актив':<10} {'Кол-во':<10} {'Сумма':<15}")
    print("-" * 35)
    for asset, stats in sorted(by_asset.items()):
        print(f"{asset:<10} {stats['count']:<10} {stats['total']:<15.2f}")

    total = sum(float(i.paid_amount or i.amount) for i in all_paid)
    print("-" * 35)
    print(f"{'ИТОГО':<10} {len(all_paid):<10} {total:<15.2f}")

    return all_paid

Пример 2: Мониторинг активных счетов с автообновлением

Проверяем все активные счета и обновляем их статус. Если счёт уже оплачен — уведомляем.

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

    while True:
        active = await cp.get_invoices(
            status=InvoiceStatus.ACTIVE,
            count=100,
        )

        print(f"[{datetime.now()}] Активных счетов: {len(active)}")

        for invoice in active:
            await invoice.update()
            if invoice.status == "paid":
                print(
                    f"  ✅ Счёт #{invoice.invoice_id} оплачен! "
                    f"{invoice.paid_amount} {invoice.paid_asset}"
                )
                # Здесь можно вызвать callback, отправить уведомление и т.д.

        if len(active) < 100:
            print("Все активные счета проверены")
            break

        await asyncio.sleep(5)  # Пауза между проверками

Пример 3: Поиск счёта по payload

API не поддерживает фильтрацию по payload напрямую, но можно получить все счета и отфильтровать на своей стороне.

async def find_invoices_by_payload(search_term: str):
    cp = CryptoPay(token="YOUR_TOKEN")
    offset = 0
    found = []

    while True:
        invoices = await cp.get_invoices(offset=offset, count=1000)
        if not invoices:
            break

        for inv in invoices:
            if inv.payload and search_term in inv.payload:
                found.append(inv)

        if len(invoices) < 1000:
            break
        offset += 1000

    return found

# Использование
async def search_example():
    found = await find_invoices_by_payload("user_12345")
    print(f"Найдено счетов с user_12345: {len(found)}")
    for inv in found:
        print(f"  #{inv.invoice_id}: {inv.status} — "
              f"{inv.amount} {inv.asset or inv.fiat} — "
              f"payload: {inv.payload}")

Пример 4: Полная статистика по всем счетам

Собираем статистику по всем счетам: количество, суммы, распределение по статусам.

from collections import Counter

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

    # Собираем все счета
    all_invoices = []
    offset = 0
    while True:
        batch = await cp.get_invoices(offset=offset, count=1000)
        if not batch:
            break
        all_invoices.extend(batch)
        if len(batch) < 1000:
            break
        offset += 1000

    # Статистика по статусам
    status_counts = Counter(i.status for i in all_invoices)
    paid_invoices = [i for i in all_invoices if i.status == "paid"]
    active_invoices = [i for i in all_invoices if i.status == "active"]
    expired_invoices = [i for i in all_invoices if i.status == "expired"]

    # Суммы
    total_requested = sum(float(i.amount) for i in all_invoices)
    total_received = sum(
        float(i.paid_amount or i.amount) for i in paid_invoices
    )

    # Средний чек
    avg_check = total_received / len(paid_invoices) if paid_invoices else 0

    print("=" * 50)
    print("📊 СТАТИСТИКА ИНВОЙСОВ")
    print("=" * 50)
    print(f"Всего счетов:        {len(all_invoices)}")
    print(f"🟡 Активных:         {len(active_invoices)}")
    print(f"🟢 Оплаченных:       {len(paid_invoices)}")
    print(f"🔴 Просроченных:     {len(expired_invoices)}")
    print(f"💰 Запрошено:        {total_requested:.2f}")
    print(f"💵 Получено:         {total_received:.2f}")
    print(f"📊 Средний чек:      {avg_check:.2f}")
    print(f"📈 Конверсия:        "
          f"{len(paid_invoices)/len(all_invoices)*100:.1f}%"
          if all_invoices else "N/A")
    print("=" * 50)

Пример 5: Пакетное обновление статусов по списку ID

Обновляем статусы нескольких счетов, получая их одним запросом.

async def batch_check_statuses(ids: list[int]):
    cp = CryptoPay(token="YOUR_TOKEN")
    results = {}

    # Разбиваем на батчи по 1000
    for i in range(0, len(ids), 1000):
        batch_ids = ids[i:i + 1000]
        invoices = await cp.get_invoices(invoice_ids=batch_ids)

        for inv in invoices:
            old_status = results.get(inv.invoice_id, {}).get("status")
            results[inv.invoice_id] = {
                "status": inv.status,
                "amount": inv.amount,
                "asset": inv.asset or inv.fiat,
                "paid_amount": inv.paid_amount,
                "paid_asset": inv.paid_asset,
            }
            if old_status and old_status != inv.status:
                print(f"#{inv.invoice_id}: {old_status} → {inv.status}")

    return results

# Использование
ids_to_check = [12345, 12346, 12347]
statuses = await batch_check_statuses(ids_to_check)
for inv_id, data in statuses.items():
    print(f"#{inv_id}: {data['status']} — "
          f"{data['amount']} {data['asset']}")

Пример 6: Поиск просроченных счетов для очистки

Находим все просроченные счета и подготавливаем их к удалению (см. урок 3.3).

async def find_expired_for_cleanup(days_old: int = 7):
    cp = CryptoPay(token="YOUR_TOKEN")
    from datetime import datetime, timedelta, timezone

    cutoff = datetime.now(timezone.utc) - timedelta(days=days_old)
    expired_for_deletion = []

    offset = 0
    while True:
        expired = await cp.get_invoices(
            status=InvoiceStatus.EXPIRED,
            offset=offset,
            count=1000,
        )
        if not expired:
            break

        for inv in expired:
            if inv.expires_at:
                expires_dt = datetime.fromisoformat(
                    inv.expires_at.replace("Z", "+00:00")
                )
                if expires_dt < cutoff:
                    expired_for_deletion.append(inv)

        if len(expired) < 1000:
            break
        offset += 1000

    print(f"Найдено просроченных старше {days_old} дней: "
          f"{len(expired_for_deletion)}")
    return expired_for_deletion

Все методы получения счетов доступны и в синхронной версии через SyncCryptoPay:

from aiosend import SyncCryptoPay, InvoiceStatus

cp = SyncCryptoPay(token="YOUR_TOKEN")

# get_invoices (синхронно)
all_invoices = cp.get_invoices()
paid = cp.get_invoices(status=InvoiceStatus.PAID)
active = cp.get_invoices(status=InvoiceStatus.ACTIVE, asset="USDT")

# get_invoice (синхронно)
invoice = cp.get_invoice(invoice_id=12345)
print(f"Статус: {invoice.status}")

# invoice.update() (синхронно)
invoice.update()
print(f"Статус после обновления: {invoice.status}")

# Пагинация (синхронно)
offset = 0
while True:
    batch = cp.get_invoices(offset=offset, count=1000)
    if not batch:
        break
    for inv in batch:
        print(f"#{inv.invoice_id}: {inv.amount} {inv.asset}")
    if len(batch) < 1000:
        break
    offset += 1000

Как и create_invoice(), методы получения счетов доступны напрямую через Network:

from aiosend.network import Network

network = Network(token="YOUR_TOKEN")
invoices = await network.get_invoices(status="paid", count=50)
invoice = await network.get_invoice(invoice_id=12345)

Поля created_at, paid_at и expires_at в объекте Invoice представлены в формате ISO 8601 (строка). Для работы с ними удобно использовать модуль datetime.

from datetime import datetime, timezone

def parse_invoice_date(date_str: str | None) -> datetime | None:
    if not date_str:
        return None
    # Убираем 'Z' если есть и парсим ISO формат
    return datetime.fromisoformat(
        date_str.replace("Z", "+00:00")
    )

# Использование
invoice = await cp.get_invoice(invoice_id=12345)

created = parse_invoice_date(invoice.created_at)
paid = parse_invoice_date(invoice.paid_at)
expires = parse_invoice_date(invoice.expires_at)

if created and paid:
    delta = paid - created
    print(f"Время до оплаты: {delta.total_seconds():.0f} секунд")

if expires and created:
    remaining = expires - datetime.now(timezone.utc)
    print(f"Осталось до истечения: {remaining}")
    if remaining.total_seconds() < 0:
        print("Счёт уже просрочен")

Фильтрация по дате на стороне клиента

API не поддерживает фильтрацию по дате напрямую, но ты можешь получить все счета и отфильтровать на своей стороне:

from datetime import datetime, timedelta, timezone

async def get_invoices_last_24h():
    cp = CryptoPay(token="YOUR_TOKEN")
    cutoff = datetime.now(timezone.utc) - timedelta(hours=24)
    recent = []

    offset = 0
    while True:
        invoices = await cp.get_invoices(offset=offset, count=1000)
        if not invoices:
            break

        for inv in invoices:
            created = parse_invoice_date(inv.created_at)
            if created and created >= cutoff:
                recent.append(inv)

        if len(invoices) < 1000:
            break
        offset += 1000

    print(f"Счетов за последние 24 часа: {len(recent)}")
    return recent