Урок 6.4 — Фильтры событий

Фильтры — это механизм, который позволяет декораторам @cp.invoice_paid(), @cp.invoice_expired(), @cp.check_activated() и @cp.check_expired() принимать только те события, которые соответствуют определённым условиям.

Фильтры передаются как аргументы в декоратор:

@cp.invoice_paid(filter1, filter2, ...)
async def handler(invoice, **kwargs):
    # Вызовется только если все фильтры вернули True
    pass

aiosend поддерживает четыре типа фильтров:

Несколько фильтров, переданных в один декоратор, работают как логическое И — обработчик вызовется только если все фильтры вернули True.

MagicFilter — это мощный механизм, позволяющий создавать фильтры через точечную нотацию, сравнивая поля объекта Invoice/Check. Он импортируется из magic_filter (библиотека magic-filter) или из aiogram.

# Если установлен aiogram
from aiogram.types import F

# Или напрямую из magic-filter
from magic_filter import F

# Или из aiosend (если экспортируется)
from aiosend.magic_filter import F

MagicFilter поддерживает множество операторов:

from magic_filter import F
from aiosend import CryptoPay

cp = CryptoPay(token="YOUR_TOKEN")

# Только USDT платежи суммой от 10 до 100
@cp.invoice_paid(F.asset == "USDT", F.amount >= 10, F.amount <= 100)
async def on_usdt_payment(invoice, **kwargs):
    print(f"USDT платёж: {invoice.amount}")

# Платежи в TON или BTC
@cp.invoice_paid(F.asset.in_(["TON", "BTC"]))
async def on_ton_or_btc(invoice, **kwargs):
    print(f"Платёж в {invoice.asset}: {invoice.amount}")

# Платежи НЕ в BTC
@cp.invoice_paid(F.asset != "BTC")
async def on_not_btc(invoice, **kwargs):
    print(f"Платёж не в BTC: {invoice.amount}")

# Чеки с определённым payload
@cp.check_activated(F.payload == "bonus_welcome")
async def on_welcome_bonus(check, **kwargs):
    print("Приветственный бонус активирован!")

Если MagicFilter недостаточно, вы можете написать свою функцию-фильтр. Она получает объект (Invoice или Check) и должна вернуть bool.

Синхронные функции

def is_large_payment(invoice) -> bool:
    """Фильтр: сумма больше 1000 USDT."""
    return float(invoice.amount) > 1000

def is_specific_payload(invoice) -> bool:
    """Фильтр: payload начинается с 'premium_'."""
    return invoice.payload and invoice.payload.startswith("premium_")

@cp.invoice_paid(is_large_payment, is_specific_payload)
async def on_premium_payment(invoice, **kwargs):
    print(f"💎 Премиум-платёж: {invoice.amount} {invoice.asset}")

Асинхронные функции

Если фильтру нужно сделать запрос к БД или внешнему API, используйте async def:

import asyncpg  # гипотетически

async def user_has_active_subscription(invoice) -> bool:
    """Проверяет, есть ли у пользователя активная подписка."""
    if not invoice.payload:
        return False
    user_id = invoice.payload.split("_")[1]
    # Асинхронный запрос к БД
    result = await db.fetchval(
        "SELECT COUNT(*) FROM subscriptions WHERE user_id=$1 AND active=true",
        int(user_id),
    )
    return result > 0

@cp.invoice_paid(user_has_active_subscription)
async def on_renewal(invoice, **kwargs):
    print(f"🔄 Продление подписки: {invoice.amount}")

Lambda-фильтры

Для самых простых условий можно использовать lambda:

# Фильтр: только платежи с hidden_message
@cp.invoice_paid(lambda inv: inv.hidden_message is not None)
async def on_with_message(invoice, **kwargs):
    print(f"Счёт с секретным сообщением: {invoice.hidden_message}")

# Фильтр: только платежи до 50 USDT
@cp.invoice_paid(lambda inv: float(inv.amount) <= 50)
async def on_small_payment(invoice, **kwargs):
    print(f"Мелкий платёж: {invoice.amount}")

# Фильтр: чек с payload содержащим "bonus"
@cp.check_activated(lambda ch: ch.payload and "bonus" in ch.payload)
async def on_bonus_check(check, **kwargs):
    print("Бонусный чек активирован!")

Самый гибкий способ — создать класс с методом __call__. Это позволяет иметь фильтры с параметрами, состоянием и сложной логикой.

class AmountRange:
    """Фильтр: сумма в заданном диапазоне."""
    
    def __init__(self, min_amount: float, max_amount: float):
        self.min_amount = min_amount
        self.max_amount = max_amount
    
    def __call__(self, invoice) -> bool:
        amount = float(invoice.amount)
        return self.min_amount <= amount <= self.max_amount

class PayloadFilter:
    """Фильтр: payload соответствует шаблону."""
    
    def __init__(self, prefix: str):
        self.prefix = prefix
    
    def __call__(self, invoice) -> bool:
        return invoice.payload and invoice.payload.startswith(self.prefix)

# Использование:
@cp.invoice_paid(AmountRange(50, 200), PayloadFilter("premium_"))
async def on_premium_mid(invoice, **kwargs):
    print(f"Премиум средний: {invoice.amount}")

@cp.invoice_paid(AmountRange(200, 1000))
async def on_large(invoice, **kwargs):
    print(f"Крупный платёж: {invoice.amount}")

@cp.check_activated(PayloadFilter("bonus_"))
async def on_bonus(check, **kwargs):
    print(f"Бонус активирован: {check.amount} {check.asset}")

Класс-фильтры особенно полезны, когда:

• У фильтра есть параметры (диапазон, префикс, список)
• Фильтру нужно состояние (счётчик вызовов, временные метки)
• Вы хотите переиспользовать один фильтр в нескольких обработчиках с разными настройками

class RateLimiter:
    """Фильтр: не более N срабатываний в минуту."""
    
    def __init__(self, max_calls: int = 5, period: int = 60):
        self.max_calls = max_calls
        self.period = period
        self.timestamps: list[float] = []
    
    def __call__(self, obj) -> bool:
        import time
        now = time.time()
        # Удаляем старые записи
        self.timestamps = [t for t in self.timestamps if now - t < self.period]
        
        if len(self.timestamps) >= self.max_calls:
            return False  # превышен лимит
        
        self.timestamps.append(now)
        return True

@cp.invoice_paid(RateLimiter(max_calls=3, period=60))
async def on_paid_limited(invoice, **kwargs):
    print(f"Оплата #{invoice.invoice_id} (не более 3 в минуту)")

Метод .as("param_name") позволяет передать результат работы фильтра в обработчик через **kwargs. Фильтр должен вернуть не bool, а словарь или значение, которое будет передано в обработчик.

Если фильтр возвращает dict, его элементы распаковываются в **kwargs. Если возвращает другое значение — оно передаётся под указанным именем.

from magic_filter import F

class ParsePayload:
    """Фильтр, который парсит payload и возвращает данные."""
    
    async def __call__(self, invoice) -> dict | None:
        if not invoice.payload:
            return None
        try:
            import json
            data = json.loads(invoice.payload)
            return {
                "user_id": data["user_id"],
                "product": data["product"],
                "quantity": data.get("quantity", 1),
            }
        except (json.JSONDecodeError, KeyError):
            return None

# .as() передаёт возвращённый dict в обработчик
@cp.invoice_paid(ParsePayload().as("order_data"))
async def on_parsed_payment(invoice, **kwargs):
    order_data = kwargs.get("order_data")
    if order_data:
        print(f"Заказ от user #{order_data['user_id']}: "
              f"{order_data['product']} x{order_data['quantity']}")
        print(f"Сумма: {invoice.amount} {invoice.asset}")

# MagicFilter тоже поддерживает .as()
# F.amount.as("amount") вернёт значение amount в kwargs
@cp.invoice_paid(F.asset.as("asset_name"), F.amount.as("amount_value"))
async def on_with_kwargs(invoice, **kwargs):
    print(f"Актив: {kwargs.get('asset_name')}")
    print(f"Сумма: {kwargs.get('amount_value')}")

from magic_filter import F

class UserResolver:
    """Фильтр: извлекает user_id из payload и загружает данные."""
    
    async def __call__(self, invoice) -> dict | None:
        if not invoice.payload:
            return None
        try:
            user_id = int(invoice.payload.split("_")[1])
        except (IndexError, ValueError):
            return None
        
        # Асинхронная загрузка данных пользователя
        user_data = await load_user_from_db(user_id)
        if user_data:
            return {"user": user_data, "user_id": user_id}
        return None

@cp.invoice_paid(UserResolver().as("user_info"))
async def on_user_payment(invoice, **kwargs):
    user_info = kwargs.get("user_info")
    if user_info:
        user = user_info["user"]
        print(f"Пользователь {user['name']} оплатил {invoice.amount} USDT")
        # Отправляем уведомление пользователю
        await notify_user(user["telegram_id"], invoice)

async def load_user_from_db(user_id: int) -> dict | None:
    # Имитация асинхронного запроса к БД
    await asyncio.sleep(0.1)
    return {"id": user_id, "name": f"User_{user_id}", "telegram_id": 12345}

async def notify_user(telegram_id: int, invoice):
    # Имитация отправки уведомления
    await asyncio.sleep(0.2)
    print(f"Уведомление отправлено пользователю @{telegram_id}")

Фильтры можно комбинировать разными способами. Все переданные в декоратор фильтры работают как логическое И. Но вы также можете использовать операторы & (И), | (ИЛИ) и ~ (НЕ) для MagicFilter.

from magic_filter import F

# Логическое ИЛИ: USDT или TON платежи
@cp.invoice_paid(
    (F.asset == "USDT") | (F.asset == "TON")
)
async def on_usdt_or_ton(invoice, **kwargs):
    print(f"Платёж в USDT или TON: {invoice.amount}")

# Логическое И: asset=USDT и amount > 50
@cp.invoice_paid(
    (F.asset == "USDT") & (F.amount > 50)
)
async def on_large_usdt(invoice, **kwargs):
    print(f"Крупный USDT: {invoice.amount}")

# Логическое НЕ: всё, кроме BTC
@cp.invoice_paid(~(F.asset == "BTC"))
async def on_not_btc(invoice, **kwargs):
    print(f"Платёж не в BTC: {invoice.asset} {invoice.amount}")

# Сложная комбинация:
# (USDT или TON) И сумма > 10 И payload не пустой
@cp.invoice_paid(
    ((F.asset == "USDT") | (F.asset == "TON")) &
    (F.amount > 10) &
    (F.payload.is_not(None))
)
async def on_filtered(invoice, **kwargs):
    print(f"Сложный фильтр: {invoice.amount} {invoice.asset}")

Комбинирование работает только для MagicFilter. Для function-фильтров и class-фильтров используйте передачу нескольких аргументов в декоратор (логическое И) или создайте комбинирующий фильтр вручную.

# Все фильтры переданные в декоратор работают как И
@cp.invoice_paid(
    F.asset == "USDT",           # MagicFilter
    lambda inv: float(inv.amount) > 10,  # Lambda
    AmountRange(10, 500),        # Class-фильтр
)
async def on_combined(invoice, **kwargs):
    print(f"Все фильтры прошли: {invoice.amount} USDT")

Рассмотрим более сложные и специфические сценарии применения фильтров в реальных проектах.

Пример: Фильтр для временных диапазонов

Создадим фильтр, который пропускает события только в определённое время суток:

import datetime

class BusinessHoursFilter:
    """Фильтр: только рабочее время (9:00 - 18:00 UTC)."""
    
    def __init__(self, start_hour: int = 9, end_hour: int = 18):
        self.start_hour = start_hour
        self.end_hour = end_hour
    
    def __call__(self, obj) -> bool:
        now = datetime.datetime.now(datetime.timezone.utc)
        return self.start_hour <= now.hour < self.end_hour

class WeekendFilter:
    """Фильтр: только выходные дни."""
    
    def __call__(self, obj) -> bool:
        now = datetime.datetime.now(datetime.timezone.utc)
        return now.weekday() >= 5  # суббота (5) или воскресенье (6)

# Применение: обрабатывать платежи только в рабочее время
@cp.invoice_paid(BusinessHoursFilter(9, 18))
async def on_business_hours_payment(invoice, **kwargs):
    print(f"💼 Рабочий платёж: {invoice.amount} USDT")

# Применение: обрабатывать платежи только на выходных
@cp.invoice_paid(WeekendFilter())
async def on_weekend_payment(invoice, **kwargs):
    print(f"🎉 Платёж на выходных: {invoice.amount} USDT")

# Комбинирование: будни до 12:00
class MorningWeekdayFilter:
    def __call__(self, obj) -> bool:
        now = datetime.datetime.now(datetime.timezone.utc)
        return now.weekday() < 5 and now.hour < 12

@cp.invoice_paid(MorningWeekdayFilter())
async def on_morning_weekday(invoice, **kwargs):
    print(f"🌅 Утренний будний платёж: {invoice.amount} USDT")

Пример: Фильтр с проверкой через внешний API

Асинхронные фильтры позволяют проверять данные через внешние сервисы:

import aiohttp

class FraudCheckFilter:
    """Фильтр: проверяет платёж на фрод через внешний API."""
    
    def __init__(self, api_url: str, api_key: str):
        self.api_url = api_url
        self.api_key = api_key
    
    async def __call__(self, invoice) -> bool:
        """Асинхронная проверка платежа на мошенничество."""
        if not invoice.payload:
            return True  # нет данных — пропускаем
        
        async with aiohttp.ClientSession() as session:
            try:
                async with session.post(
                    self.api_url,
                    json={
                        "invoice_id": invoice.invoice_id,
                        "amount": invoice.amount,
                        "asset": invoice.asset,
                        "payload": invoice.payload,
                    },
                    headers={"Authorization": f"Bearer {self.api_key}"},
                    timeout=aiohttp.ClientTimeout(total=5),
                ) as resp:
                    if resp.status == 200:
                        data = await resp.json()
                        return data.get("is_legit", True)
            except Exception:
                # При ошибке API — пропускаем (лучше пропустить, чем
                # заблокировать легитимный платёж)
                return True
        
        return True

# Использование:
fraud_filter = FraudCheckFilter(
    api_url="https://fraud-api.example.com/check",
    api_key="your_api_key",
)

@cp.invoice_paid(fraud_filter)
async def on_checked_payment(invoice, **kwargs):
    print(f"✅ Платёж #{invoice.invoice_id} прошёл проверку на фрод")

Пример: Фильтр для A/B тестирования

Используйте фильтры для разделения трафика на группы A/B:

import hashlib

class ABTestFilter:
    """Фильтр для A/B тестирования на основе user_id."""
    
    def __init__(self, variant: str, weight: float = 0.5):
        """
        variant: "A" или "B"
        weight: доля пользователей в группе A (0.0 - 1.0)
        """
        self.variant = variant
        self.weight = weight
    
    def __call__(self, invoice) -> bool:
        if not invoice.payload:
            return False
        
        # Извлекаем user_id из payload
        try:
            user_id = int(invoice.payload.split("_")[-1])
        except (ValueError, IndexError):
            return False
        
        # Детерминированное распределение на основе хеша
        hash_val = int(hashlib.md5(str(user_id).encode()).hexdigest(), 16)
        group = "A" if (hash_val % 100) / 100 < self.weight else "B"
        
        return group == self.variant

# Группа A — старая цена
@cp.invoice_paid(ABTestFilter("A", weight=0.7))
async def on_group_a(invoice, **kwargs):
    print(f"🔵 [Группа A] Стандартная обработка: {invoice.amount} USDT")

# Группа B — новая цена со скидкой
@cp.invoice_paid(ABTestFilter("B", weight=0.3))
async def on_group_b(invoice, **kwargs):
    print(f"🟢 [Группа B] Обработка со скидкой: {invoice.amount} USDT")

Пример: Цепочка фильтров с накоплением данных

Фильтры можно объединять в цепочку, где каждый следующий фильтр получает данные из предыдущего через .as():

class ExtractUser:
    """Извлекает user_id из payload."""
    
    async def __call__(self, invoice) -> dict | None:
        if not invoice.payload:
            return None
        try:
            user_id = int(invoice.payload.split("_")[-1])
            return {"user_id": user_id}
        except (ValueError, IndexError):
            return None

class CheckUserTier:
    """Проверяет уровень пользователя (использует user_id)."""
    
    async def __call__(self, invoice, user_id: int = 0) -> dict | None:
        # Имитация запроса к БД
        tiers = {1: "vip", 2: "basic", 3: "vip"}
        tier = tiers.get(user_id, "guest")
        return {"user_id": user_id, "tier": tier}

# Примечание: цепочка фильтров с .as() — продвинутая техника.
# Обычно проще сделать один фильтр, который делает всё.
# Но для модульности можно использовать такой подход:

@cp.invoice_paid(ExtractUser().as("user_data"))
async def on_payment_with_user(invoice, **kwargs):
    user_data = kwargs.get("user_data", {})
    user_id = user_data.get("user_id", "unknown")
    print(f"Платёж от user #{user_id}: {invoice.amount} USDT")

import asyncio
import json
from magic_filter import F
from aiosend import CryptoPay
from aiosend.polling import PollingConfig

cp = CryptoPay(
    token="YOUR_TOKEN",
    polling_config=PollingConfig(timeout=600, delay=2),
)

# ─── MagicFilter ────────────────────────────────────────
@cp.invoice_paid(F.asset == "USDT")
async def on_usdt(invoice, **kwargs):
    print(f"1️⃣ USDT платёж: {invoice.amount}")

@cp.invoice_paid(F.asset.in_(["TON", "BTC"]))
async def on_ton_or_btc(invoice, **kwargs):
    print(f"2️⃣ TON/BTC платёж: {invoice.amount} {invoice.asset}")

# ─── Lambda ──────────────────────────────────────────────
@cp.invoice_paid(lambda i: float(i.amount) > 100)
async def on_large(invoice, **kwargs):
    print(f"3️⃣ Крупный платёж: {invoice.amount}")

# ─── Function ────────────────────────────────────────────
def has_payload(invoice):
    return invoice.payload is not None

@cp.invoice_paid(has_payload)
async def on_with_payload(invoice, **kwargs):
    print(f"4️⃣ С payload: {invoice.payload}")

# ─── Async function ──────────────────────────────────────
async def check_user_tier(invoice):
    if not invoice.payload:
        return False
    try:
        data = json.loads(invoice.payload)
        return data.get("tier") == "vip"
    except (json.JSONDecodeError, KeyError):
        return False

@cp.invoice_paid(check_user_tier)
async def on_vip(invoice, **kwargs):
    print(f"5️⃣ VIP-пользователь оплатил {invoice.amount} USDT")

# ─── Class with __call__ ─────────────────────────────────
class AssetAmountFilter:
    def __init__(self, asset: str, min_amount: float):
        self.asset = asset
        self.min_amount = min_amount
    
    def __call__(self, invoice):
        return (invoice.asset == self.asset and
                float(invoice.amount) >= self.min_amount)

@cp.invoice_paid(AssetAmountFilter("USDT", 500))
async def on_bulk_usdt(invoice, **kwargs):
    print(f"6️⃣ Оптовый USDT: {invoice.amount}")

# ─── .as() — передача данных ─────────────────────────────
def extract_product(invoice) -> dict | None:
    if not invoice.payload:
        return None
    try:
        data = json.loads(invoice.payload)
        return {"product_id": data["product_id"], "name": data.get("name")}
    except (json.JSONDecodeError, KeyError):
        return None

@cp.invoice_paid(extract_product.as("product"))
async def on_product_purchase(invoice, **kwargs):
    product = kwargs.get("product")
    if product:
        print(f"7️⃣ Куплен товар #{product['product_id']}: {product['name']}")

# ─── Комбинированный ─────────────────────────────────────
@cp.invoice_paid(
    (F.asset == "USDT") & (F.amount >= 10) & (F.amount <= 100),
    lambda i: i.payload and "subscription" in i.payload,
)
async def on_subscription(invoice, **kwargs):
    print(f"8️⃣ Подписка: {invoice.amount} USDT")

async def main():
    for i in range(3):
        await cp.create_invoice(
            amount=25 + i * 10,
            asset="USDT",
            payload=json.dumps({"tier": "vip", "product_id": i}),
        )
    await cp.start_polling()

asyncio.run(main())