Урок 4.2 — Управление чеками

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

from aiosend import CheckStatus

# Два возможных статуса:
print(CheckStatus.ACTIVE)       # "active"
print(CheckStatus.ACTIVATED)    # "activated"

# Сравнение со строкой тоже работает:
status = "active"
if status == CheckStatus.ACTIVE:
    print("Чек активен")

# Использование в методах API:
checks = await cp.get_checks(status=CheckStatus.ACTIVE)
checks = await cp.get_checks(status=CheckStatus.ACTIVATED)

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

async def get_checks(
    self,
    asset: str | None = None,
    check_ids: list[int] | None = None,
    status: str | None = None,
    offset: int | None = None,
    count: int | None = None,
) -> list[Check]

Все параметры опциональны — если вызвать get_checks() без аргументов, вернутся все чеки приложения (с учётом ограничения API на максимальное количество).

from aiosend import CryptoPay, CheckStatus

cp = CryptoPay(token="YOUR_TOKEN")

# 1. Все чеки без фильтрации
all_checks = await cp.get_checks()
print(f"Всего чеков: {len(all_checks)}")

# 2. Только активные чеки в USDT
active_usdt = await cp.get_checks(
    asset="USDT",
    status=CheckStatus.ACTIVE,
)
for check in active_usdt:
    print(f"Чек #{check.check_id}: {check.amount} {check.asset}")

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

# 4. Только активированные, с пагинацией
activated = await cp.get_checks(
    status=CheckStatus.ACTIVATED,
    offset=0,
    count=50,
)

# 5. Чеки в TON (активные и активированные)
ton_checks = await cp.get_checks(asset="TON")

Когда у тебя сотни или тысячи чеков, API не вернёт их все сразу. Нужно использовать пагинацию с параметрами offset и count.

Максимальное значение count — 1000. Если тебе нужно получить все чеки, когда их больше 1000, используй цикл с увеличением offset на каждой итерации.

from aiosend import CryptoPay

async def get_all_checks(cp: CryptoPay) -> list:
    all_checks = []
    offset = 0
    page_size = 500

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

    print(f"Всего загружено: {len(all_checks)} чеков")
    return all_checks

# Использование:
cp = CryptoPay(token="YOUR_TOKEN")
checks = await get_all_checks(cp)

# Теперь можно анализировать все чеки
active = [c for c in checks if c.status == "active"]
activated = [c for c in checks if c.status == "activated"]
print(f"Активных: {len(active)}, Активированных: {len(activated)}")

Если тебе нужно получить полную информацию об одном конкретном чеке, используй get_check(). Он принимает check_id и возвращает объект Check.

async def get_check(self, check_id: int) -> Check

# Пример использования:
cp = CryptoPay(token="YOUR_TOKEN")

check = await cp.get_check(check_id=42)
print(f"Чек #{check.check_id}")
print(f"Сумма: {check.amount} {check.asset}")
print(f"Статус: {check.status}")
print(f"Создан: {check.created_at}")
if check.activated_at:
    print(f"Активирован: {check.activated_at}")
if check.activated_by:
    print(f"Кем активирован: user #{check.activated_by}")

# У чека также есть метод-сокращение check.update()
# и свойство check.qr (о них позже)

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

Объект Check имеет метод check.update(), который является сокращением для вызова cp.update_check(check_id, ...) на родительском клиенте. Это удобно: тебе не нужно хранить ссылку на клиент отдельно.

Метод check.update() принимает те же параметры, что и update_check() на клиенте — но без check_id (он берётся из самого объекта).

from aiosend import CryptoPay

cp = CryptoPay(token="YOUR_TOKEN")

# Получаем чек
check = await cp.get_check(check_id=42)

# Обновляем чек — новый check_id, та же сумма
# check.update(...) эквивалентно cp.update_check(check.check_id, ...)
updated = await check.update()

print(f"Старый ID: {check.check_id}")
print(f"Новый ID: {updated.check_id}")
print(f"Сумма: {updated.amount} {updated.asset}")
print(f"Статус: {updated.status}")

# Можно явно указать клиент, если чек получен не из него:
# updated = await check.update(cp=another_client)

# То же самое через клиент напрямую
from aiosend import CryptoPay

cp = CryptoPay(token="YOUR_TOKEN")

updated = await cp.update_check(check_id=42)
# или:
updated = await cp.update_check(check_id=42, cp=cp)

# check.update() — это просто syntax sugar для:
# updated = await cp.update_check(check_id=check.check_id)

aiosend предоставляет несколько способов удалить чек. Ты можешь удалить один чек по ID, удалить все чеки сразу или использовать shortcut на объекте Check.

delete_check(check_id) — удаление одного чека

from aiosend import CryptoPay

cp = CryptoPay(token="YOUR_TOKEN")

# Удаляем чек по ID
result = await cp.delete_check(check_id=42)
print(result)  # True — чек удалён

# Удаляем через shortcut на объекте
check = await cp.get_check(check_id=42)
result = await check.delete()
print(result)  # True — чек удалён

# Попытка удалить несуществующий чек
result = await cp.delete_check(check_id=99999)
print(result)  # True — API считает операцию успешной
# (даже если чек уже не существует)

check.delete() — shortcut на объекте

Как и с check.update(), у объекта Check есть метод check.delete(), который вызывает cp.delete_check(check.check_id). Это удобно, когда у тебя есть объект чека, но нет прямого доступа к клиенту.

from aiosend import CryptoPay
from aiosend.exceptions import APIError

async def safe_delete_check(cp: CryptoPay, check_id: int) -> bool:
    try:
        return await cp.delete_check(check_id=check_id)
    except APIError as e:
        print(f"Ошибка при удалении чека #{check_id}: {e}")
        return False

cp = CryptoPay(token="YOUR_TOKEN")
success = await safe_delete_check(cp, 42)
if success:
    print("Чек успешно удалён")

delete_all_checks() — удаление всех чеков

Метод delete_all_checks() удаляет все активные чеки приложения. Это мощная операция — используй её с осторожностью. Активированные чеки этот метод не затрагивает.

from aiosend import CryptoPay

cp = CryptoPay(token="YOUR_TOKEN")

# Получаем количество активных чеков
active_checks = await cp.get_checks(status="active")
print(f"Активных чеков: {len(active_checks)}")

# Удаляем все активные чеки
result = await cp.delete_all_checks()
print(result)  # True — все активные чеки удалены

# Проверяем, что активных чеков больше нет
active_checks = await cp.get_checks(status="active")
print(f"Осталось активных: {len(active_checks)}")

# Этот метод возвращает bool, а не список удалённых чеков

Давай соберём всё вместе и напишем функцию — дашборд для управления чеками. Она будет показывать статистику по всем чекам, позволять удалять просроченные и группировать по активам.

from collections import Counter
from datetime import datetime, timedelta
from aiosend import CryptoPay, CheckStatus
from aiosend.exceptions import APIError

class CheckDashboard:
    """Дашборд для управления крипто-чеками."""

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

    async def show_stats(self):
        """Показывает статистику по всем чекам."""
        all_checks = await self.cp.get_checks()

        active = [c for c in all_checks if c.status == CheckStatus.ACTIVE]
        activated = [c for c in all_checks if c.status == CheckStatus.ACTIVATED]

        print(f"📊 Статистика чеков:")
        print(f"   Всего: {len(all_checks)}")
        print(f"   Активных: {len(active)}")
        print(f"   Активированных: {len(activated)}")

        # Группировка по активам
        asset_counts = Counter(c.asset for c in active)
        print(f"\n   По активам (активные):")
        for asset, count in asset_counts.most_common():
            print(f"     {asset}: {count}")

    async def delete_old_checks(self, days: int = 30):
        """Удаляет активные чеки старше N дней."""
        all_checks = await self.cp.get_checks(status=CheckStatus.ACTIVE)
        threshold = datetime.utcnow() - timedelta(days=days)
        deleted = 0

        for check in all_checks:
            created = datetime.fromisoformat(check.created_at.replace("Z", "+00:00"))
            if created < threshold:
                try:
                    await check.delete()
                    deleted += 1
                    print(f"🗑️ Удалён чек #{check.check_id} от {check.created_at}")
                except APIError as e:
                    print(f"❌ Ошибка удаления #{check.check_id}: {e}")

        print(f"\nУдалено устаревших чеков: {deleted}")
        return deleted

    async def group_by_asset(self) -> dict:
        """Группирует активные чеки по активам."""
        all_checks = await self.cp.get_checks(status=CheckStatus.ACTIVE)
        groups = {}
        for check in all_checks:
            groups.setdefault(check.asset, []).append(check)
        return groups

    async def total_active_amount(self) -> dict[str, float]:
        """Считает сумму активных чеков по каждому активу."""
        all_checks = await self.cp.get_checks(status=CheckStatus.ACTIVE)
        totals = {}
        for check in all_checks:
            amount = float(check.amount)
            totals[check.asset] = totals.get(check.asset, 0) + amount
        return totals


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

    await dashboard.show_stats()

    print("\n--- Группировка по активам ---")
    groups = await dashboard.group_by_asset()
    for asset, checks in groups.items():
        total = sum(float(c.amount) for c in checks)
        print(f"{asset}: {len(checks)} чеков на сумму {total:.2f}")

    print("\n--- Суммы по активам ---")
    totals = await dashboard.total_active_amount()
    for asset, total in totals.items():
        print(f"{asset}: {total:.4f}")

    # Удаляем чеки старше 60 дней
    # await dashboard.delete_old_checks(days=60)


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

Комбинируя get_checks() с delete_check() и update_check(), ты можешь выполнять массовые операции. Например: удалить все активные чеки в TON, обновить все чеки в USDT или получить чеки по конкретному пользователю.

from aiosend import CryptoPay, CheckStatus

async def bulk_update_checks(
    cp: CryptoPay,
    asset: str | None = None,
    max_count: int = 100,
) -> list:
    """Обновляет (перевыпускает) активные чеки по фильтру."""
    checks = await cp.get_checks(
        asset=asset,
        status=CheckStatus.ACTIVE,
        count=max_count,
    )
    updated = []
    for check in checks:
        new_check = await check.update()
        updated.append(new_check)
        print(f"Чек #{check.check_id} → #{new_check.check_id} (обновлён)")
    return updated

async def bulk_delete_checks(
    cp: CryptoPay,
    asset: str | None = None,
    status: str = "active",
) -> int:
    """Удаляет чеки по фильтру."""
    checks = await cp.get_checks(asset=asset, status=status)
    deleted = 0
    for check in checks:
        await check.delete()
        deleted += 1
    return deleted

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

    # Перевыпустить все активные USDT-чеки
    updated = await bulk_update_checks(cp, asset="USDT", max_count=50)
    print(f"Обновлено чеков: {len(updated)}")

    # Удалить все активные TON-чеки
    deleted = await bulk_delete_checks(cp, asset="TON")
    print(f"Удалено чеков: {deleted}")

    # Удалить все активированные чеки (очистка истории)
    # deleted_hist = await bulk_delete_checks(cp, status="activated")
    # print(f"Удалено из истории: {deleted_hist}")


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