Урок 3.3 — Удаление и срок действия инвойсов

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

Важное правило: оплаченный инвойс нельзя удалить. API Crypto Pay возвращает ошибку при попытке удалить PAID инвойс. Это защита от случайного удаления финансовых записей. Удалять можно только счета со статусом ACTIVE или EXPIRED.

Метод delete_invoice() удаляет инвойс из системы Crypto Pay. После удаления счёт становится недоступен для оплаты, а его статус меняется на удалённый.

async def delete_invoice(
    self,
    invoice_id: int,
) -> bool

Метод принимает ID инвойса и возвращает True в случае успешного удаления. Если инвойс уже оплачен, выбрасывается исключение. Если инвойс с таким ID не существует — также ошибка.

import asyncio
from aiosend import CryptoPay

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

    # Создаём инвойс
    invoice = await cp.create_invoice(amount=10, asset="USDT")
    print(f"Создан инвойс #{invoice.invoice_id}, статус: {invoice.status}")

    # Удаляем его
    success = await cp.delete_invoice(invoice.invoice_id)
    print(f"Удалён: {success}")

    # Проверяем — инвойс больше не существует
    try:
        deleted = await cp.get_invoice(invoice.invoice_id)
        print(f"Статус после удаления: {deleted.status}")
    except Exception as e:
        print(f"Инвойс не найден: {e}")

asyncio.run(main())

Shortcut: invoice.delete()

В aiosend есть удобный shortcut — метод invoice.delete() прямо на объекте Invoice. Он вызывает cp.delete_invoice(invoice_id) автоматически. Это избавляет от необходимости хранить ссылку на клиента.

import asyncio
from aiosend import CryptoPay

async def main():
    cp = CryptoPay(token="YOUR_TOKEN")
    invoice = await cp.create_invoice(amount=25, asset="USDT")

    # Shortcut — удаляем через сам объект инвойса
    success = await invoice.delete()
    # Эквивалентно: await cp.delete_invoice(invoice.invoice_id)

    print(f"Инвойс #{invoice.invoice_id} удалён: {success}")

asyncio.run(main())

Shortcut особенно удобен, когда у тебя нет прямого доступа к экземпляру CryptoPay — например, в обработчиках событий или колбэках, куда приходит только объект Invoice.

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

from aiosend import CryptoPay
from aiosend.exceptions import APIError, APITimeoutError

async def safe_delete_invoice(cp: CryptoPay, invoice_id: int) -> bool:
    """Безопасное удаление инвойса с обработкой ошибок."""
    try:
        # Проверяем статус перед удалением
        invoice = await cp.get_invoice(invoice_id)

        if invoice.status == "paid":
            print(f"Инвойс #{invoice_id} уже оплачен — удаление невозможно")
            return False

        if invoice.status in ("expired", "active"):
            success = await cp.delete_invoice(invoice_id)
            print(f"Инвойс #{invoice_id} удалён: {success}")
            return success

    except APIError as e:
        if "not found" in str(e).lower():
            print(f"Инвойс #{invoice_id} не существует")
        elif "paid" in str(e).lower():
            print(f"Инвойс #{invoice_id} уже оплачен")
        else:
            print(f"Ошибка API при удалении #{invoice_id}: {e}")
        return False
    except APITimeoutError:
        print(f"Таймаут при удалении #{invoice_id}, повторите позже")
        return False

    return False

Метод delete_all_invoices() позволяет удалить несколько инвойсов по их ID за один запрос. Это эффективнее, чем вызывать delete_invoice() в цикле.

async def delete_all_invoices(
    self,
    *invoice_ids: int,
) -> bool

Принимает любое количество ID через *args (вариадический параметр). Возвращает True, если все инвойсы успешно удалены. Если хотя бы один не удался (например, оплачен), метод выбрасывает исключение.

import asyncio
from aiosend import CryptoPay

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

    # Создаём несколько инвойсов
    invoices = []
    for i in range(5):
        inv = await cp.create_invoice(
            amount=10 + i,
            asset="USDT",
            description=f"Тестовый счёт #{i+1}",
        )
        invoices.append(inv)
        print(f"Создан инвойс #{inv.invoice_id}")

    # Удаляем все разом
    ids = [inv.invoice_id for inv in invoices]
    success = await cp.delete_all_invoices(*ids)
    print(f"Все {len(ids)} инвойсов удалены: {success}")

asyncio.run(main())

Обрати внимание на звёздочку *ids — метод ожидает отдельные аргументы, а не список. Если у тебя список ID, используй распаковку (*ids).

Оптимизация: удаление с предварительной фильтрацией

На практике удобно комбинировать get_invoices() с delete_all_invoices() — сначала получить список просроченных счетов, а затем массово их удалить.

import asyncio
from aiosend import CryptoPay, InvoiceStatus

async def clean_expired_invoices(cp: CryptoPay) -> int:
    """Удаляет все просроченные инвойсы. Возвращает количество удалённых."""
    expired = await cp.get_invoices(status=InvoiceStatus.EXPIRED)
    active = await cp.get_invoices(status=InvoiceStatus.ACTIVE)

    all_ids = [inv.invoice_id for inv in expired + active]
    if not all_ids:
        print("Нет счетов для удаления")
        return 0

    # Удаляем пачками по 1000
    deleted_count = 0
    for i in range(0, len(all_ids), 1000):
        batch = all_ids[i:i + 1000]
        try:
            await cp.delete_all_invoices(*batch)
            deleted_count += len(batch)
            print(f"Удалена пачка {i//1000 + 1}: {len(batch)} инвойсов")
        except Exception as e:
            print(f"Ошибка при удалении пачки: {e}")

    print(f"Всего удалено: {deleted_count}")
    return deleted_count

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

asyncio.run(main())

expires_in — это параметр метода create_invoice(), который задаёт время жизни счёта в секундах. По истечении этого времени статус инвойса автоматически меняется на EXPIRED.

import asyncio
from aiosend import CryptoPay

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

    # Быстрый счёт на 5 минут (для очереди оплаты)
    fast = await cp.create_invoice(
        amount=5,
        asset="USDT",
        expires_in=300,
        description="Быстрая оплата",
    )
    print(f"Быстрый счёт #{fast.invoice_id}, истекает: {fast.expires_at}")

    # Стандартный счёт на 1 час
    standard = await cp.create_invoice(
        amount=100,
        fiat="USD",
        expires_in=3600,
        description="Товар из корзины",
    )
    print(f"Стандартный счёт #{standard.invoice_id}")

    # Долгосрочный счёт на 7 дней
    weekly = await cp.create_invoice(
        amount=500,
        asset="USDT",
        expires_in=604800,
        description="Недельный абонемент",
    )
    print(f"Долгий счёт #{weekly.invoice_id}")

    # Без expires_in — системный срок по умолчанию
    default = await cp.create_invoice(
        amount=10,
        asset="TON",
    )
    print(f"Обычный счёт #{default.invoice_id} (срок по умолчанию)")

asyncio.run(main())

Когда срок жизни инвойса истекает, он автоматически переходит в статус EXPIRED. Это необратимый процесс — просроченный счёт нельзя «реанимировать» или продлить. Единственная операция, доступная для EXPIRED инвойса — удаление.

Как работает механизм истечения:

• При создании инвойса с expires_in сервер устанавливает expires_at — точную дату истечения
• Когда текущее время превышает expires_at, статус автоматически меняется на EXPIRED
• Попытка оплатить просроченный счёт приводит к ошибке — Crypto Pay отклоняет платёж
• EXPIRED инвойсы участвуют в лимитах активных счетов, поэтому их рекомендуется удалять

import asyncio
from datetime import datetime
from aiosend import CryptoPay, InvoiceStatus

async def check_expired_invoices(cp: CryptoPay):
    """Проверяет просроченные инвойсы и выводит информацию."""
    # Получаем все просроченные счета
    expired = await cp.get_invoices(status=InvoiceStatus.EXPIRED)
    print(f"Найдено просроченных счетов: {len(expired)}")

    for inv in expired:
        print(f"  #{inv.invoice_id}: {inv.amount} {inv.asset}")
        print(f"    Создан: {inv.created_at}")
        print(f"    Истёк: {inv.expires_at}")

        # Рассчитываем, как давно истёк
        if inv.expires_at:
            expires_dt = datetime.fromisoformat(inv.expires_at)
            now = datetime.now()
            diff = now - expires_dt
            hours = diff.total_seconds() / 3600
            print(f"    Просрочен на {hours:.1f} часов")

    return expired

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

asyncio.run(main())

Проверка даты истечения

Дата истечения хранится в поле expires_at объекта Invoice. Это строка в формате ISO 8601. Ты можешь анализировать эту дату для превентивных действий — например, уведомлять пользователя, что счёт скоро истечёт.

from datetime import datetime, timedelta
from aiosend import CryptoPay, InvoiceStatus

async def find_almost_expired(cp: CryptoPay, threshold_minutes: int = 10):
    """
    Находит активные инвойсы, которые скоро истекают.
    threshold_minutes — за сколько минут до истечения считать «скоро».
    """
    active = await cp.get_invoices(status=InvoiceStatus.ACTIVE)
    now = datetime.now()
    threshold = timedelta(minutes=threshold_minutes)
    almost_expired = []

    for inv in active:
        if inv.expires_at:
            expires_dt = datetime.fromisoformat(inv.expires_at)
            time_left = expires_dt - now
            if timedelta(0) < time_left <= threshold:
                almost_expired.append(inv)
                print(
                    f"Счёт #{inv.invoice_id} истекает через "
                    f"{time_left.seconds // 60} минут"
                )

    return almost_expired

В реальном приложении удобно запускать фоновую задачу, которая регулярно чистит просроченные и старые активные инвойсы. Рассмотрим два подхода: простой asyncio-цикл и интеграцию с APScheduler.

Вариант 1: Asyncio-цикл

Самый простой способ — запустить фоновую корутину с бесконечным циклом.

import asyncio
from aiosend import CryptoPay, InvoiceStatus

class InvoiceCleaner:
    """Автоматическая очистка просроченных инвойсов."""

    def __init__(self, cp: CryptoPay, interval: int = 3600):
        self.cp = cp
        self.interval = interval  # секунд между очистками
        self._task: asyncio.Task | None = None

    async def clean_once(self) -> int:
        """Один цикл очистки. Возвращает количество удалённых."""
        expired = await self.cp.get_invoices(status=InvoiceStatus.EXPIRED)
        active = await self.cp.get_invoices(status=InvoiceStatus.ACTIVE)

        # Удаляем только просроченные (active оставляем)
        ids = [inv.invoice_id for inv in expired]
        if not ids:
            return 0

        deleted = 0
        for i in range(0, len(ids), 1000):
            batch = ids[i:i + 1000]
            try:
                await self.cp.delete_all_invoices(*batch)
                deleted += len(batch)
            except Exception as e:
                print(f"Ошибка при очистке пачки: {e}")

        return deleted

    async def run_forever(self):
        """Запускает бесконечный цикл очистки."""
        while True:
            try:
                count = await self.clean_once()
                if count:
                    print(f"Очистка: удалено {count} просроченных инвойсов")
            except Exception as e:
                print(f"Ошибка в цикле очистки: {e}")
            await asyncio.sleep(self.interval)

    def start(self):
        """Запускает фоновую задачу."""
        self._task = asyncio.create_task(self.run_forever())

    async def stop(self):
        """Останавливает фоновую задачу."""
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass

# Использование:
async def main():
    cp = CryptoPay(token="YOUR_TOKEN")
    cleaner = InvoiceCleaner(cp, interval=1800)  # каждые 30 минут
    cleaner.start()

    # Основная логика приложения...
    await asyncio.sleep(7200)  # работаем 2 часа

    await cleaner.stop()

asyncio.run(main())

Вариант 2: APScheduler (для больших проектов)

Если ты используешь APScheduler, интеграция с aiosend тоже проста.

# pip install apscheduler
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from aiosend import CryptoPay, InvoiceStatus

async def cleanup_job():
    """Задача для APScheduler."""
    cp = CryptoPay(token="YOUR_TOKEN")
    expired = await cp.get_invoices(status=InvoiceStatus.EXPIRED)
    ids = [inv.invoice_id for inv in expired]

    if not ids:
        return

    for i in range(0, len(ids), 1000):
        batch = ids[i:i + 1000]
        try:
            await cp.delete_all_invoices(*batch)
            print(f"APScheduler: удалено {len(batch)} просроченных инвойсов")
        except Exception as e:
            print(f"APScheduler: ошибка {e}")

# Настройка планировщика
scheduler = AsyncIOScheduler()

# Запуск каждый день в 03:00
scheduler.add_job(
    cleanup_job,
    trigger="cron",
    hour=3,
    minute=0,
)

scheduler.start()

# Основная логика приложения...

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

import asyncio
import sys
from aiosend import CryptoPay, InvoiceStatus
from aiosend.exceptions import APIError

class InvoiceManager:
    """Менеджер для управления жизненным циклом инвойсов."""

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

    async def create_with_expiry(
        self,
        amount: float,
        asset: str,
        expires_in: int,
        description: str = "",
    ):
        """Создаёт инвойс с указанным сроком жизни."""
        invoice = await self.cp.create_invoice(
            amount=amount,
            asset=asset,
            expires_in=expires_in,
            description=description,
        )
        print(f"✅ Создан инвойс #{invoice.invoice_id}")
        print(f"   Сумма: {invoice.amount} {invoice.asset}")
        print(f"   Статус: {invoice.status}")
        print(f"   Истекает: {invoice.expires_at}")
        print(f"   Ссылка: {invoice.bot_invoice_url}")
        return invoice

    async def get_status(self, invoice_id: int):
        """Получает актуальный статус инвойса."""
        try:
            invoice = await self.cp.get_invoice(invoice_id)
            print(f"Инвойс #{invoice_id}:")
            print(f"  Статус: {invoice.status}")
            print(f"  Создан: {invoice.created_at}")
            print(f"  Истекает: {invoice.expires_at or 'N/A'}")
            if invoice.status == "paid":
                print(f"  Оплачено: {invoice.paid_amount} {invoice.paid_asset}")
                print(f"  Время оплаты: {invoice.paid_at}")
            return invoice
        except APIError as e:
            print(f"❌ Инвойс #{invoice_id} не найден: {e}")
            return None

    async def delete_if_expired(self, invoice_id: int) -> bool:
        """Удаляет инвойс, только если он просрочен."""
        try:
            invoice = await self.cp.get_invoice(invoice_id)
            if invoice.status == "expired":
                await invoice.delete()
                print(f"🗑️ Просроченный инвойс #{invoice_id} удалён")
                return True
            elif invoice.status == "paid":
                print(f"❌ Инвойс #{invoice_id} оплачен — удалить нельзя")
                return False
            else:
                print(f"⏳ Инвойс #{invoice_id} ещё активен (статус: {invoice.status})")
                return False
        except APIError as e:
            print(f"❌ Ошибка: {e}")
            return False

    async def clean_all_expired(self) -> int:
        """Массово удаляет все просроченные инвойсы."""
        expired = await self.cp.get_invoices(status=InvoiceStatus.EXPIRED)
        if not expired:
            print("✅ Нет просроченных инвойсов")
            return 0

        ids = [inv.invoice_id for inv in expired]
        for i in range(0, len(ids), 1000):
            batch = ids[i:i + 1000]
            try:
                await self.cp.delete_all_invoices(*batch)
            except APIError as e:
                print(f"Ошибка при удалении пачки: {e}")

        print(f"🗑️ Удалено просроченных инвойсов: {len(ids)}")
        return len(ids)

    async def summary(self):
        """Показывает сводку по всем инвойсам."""
        active = await self.cp.get_invoices(status=InvoiceStatus.ACTIVE)
        paid = await self.cp.get_invoices(status=InvoiceStatus.PAID)
        expired = await self.cp.get_invoices(status=InvoiceStatus.EXPIRED)

        print("\n📊 Сводка по инвойсам:")
        print(f"   🟡 Активных: {len(active)}")
        print(f"   🟢 Оплаченных: {len(paid)}")
        print(f"   🔴 Просроченных: {len(expired)}")
        print(f"   📦 Всего: {len(active) + len(paid) + len(expired)}")

# Демонстрация
async def main():
    manager = InvoiceManager(token="YOUR_TOKEN")

    # 1. Создаём инвойсы с разным сроком
    await manager.create_with_expiry(10, "USDT", 60, "Тест 1 минута")
    await manager.create_with_expiry(25, "TON", 3600, "Тест 1 час")

    # 2. Показываем сводку
    await manager.summary()

    # 3. Удаляем просроченные
    await manager.clean_all_expired()

    # 4. Повторная сводка
    await manager.summary()

asyncio.run(main())

Если ты работаешь в синхронном коде (Flask, простые скрипты), используй SyncCryptoPay. Все методы удаления доступны и в синхронной версии:

from aiosend import SyncCryptoPay

cp = SyncCryptoPay(token="YOUR_TOKEN")

# Создаём инвойс
invoice = cp.create_invoice(amount=10, asset="USDT")
print(f"Создан счёт #{invoice.invoice_id}")

# Удаляем
success = cp.delete_invoice(invoice.invoice_id)
print(f"Удалён: {success}")

# Или через shortcut
another = cp.create_invoice(amount=5, asset="TON")
another.delete()  # синхронно

# Массовое удаление
cp.delete_all_invoices(1, 2, 3)

В Production-среде полезно вести статистику удалений: сколько, когда и каких инвойсов было удалено. Это помогает отслеживать состояние системы и выявлять аномалии.

import asyncio
import logging
from datetime import datetime
from aiosend import CryptoPay, InvoiceStatus

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("invoice_cleaner")

class LoggingInvoiceCleaner:
    """Очистка инвойсов с подробным логированием."""

    def __init__(self, cp: CryptoPay):
        self.cp = cp
        self.stats = {
            "total_deleted": 0,
            "total_attempts": 0,
            "total_errors": 0,
            "last_run": None,
        }

    async def clean_and_log(self) -> dict:
        """Удаляет просроченные инвойсы и логирует результат."""
        self.stats["total_attempts"] += 1
        self.stats["last_run"] = datetime.now().isoformat()

        expired = await self.cp.get_invoices(status=InvoiceStatus.EXPIRED)
        logger.info(f"Найдено просроченных инвойсов: {len(expired)}")

        deleted_ids = []
        error_ids = []

        for inv in expired:
            try:
                await inv.delete()
                deleted_ids.append(inv.invoice_id)
                self.stats["total_deleted"] += 1
                logger.debug(f"Удалён инвойс #{inv.invoice_id}")
            except Exception as e:
                error_ids.append(inv.invoice_id)
                self.stats["total_errors"] += 1
                logger.error(f"Ошибка при удалении #{inv.invoice_id}: {e}")

        run_stats = {
            "found": len(expired),
            "deleted": len(deleted_ids),
            "errors": len(error_ids),
            "deleted_ids": deleted_ids,
            "error_ids": error_ids,
            "timestamp": self.stats["last_run"],
        }

        logger.info(
            f"Очистка завершена: удалено {len(deleted_ids)}, "
            f"ошибок {len(error_ids)}"
        )
        return run_stats

    def get_summary(self) -> dict:
        """Возвращает сводную статистику."""
        return self.stats