CryptoScan - python библиотека для мониторинга (приёма) крипто платежей

С приветом други!

CryptoScan - библиотека для мониторинга крипто платежей
Этот инструмент позволяет эффективно отслеживать поступления криптовалют в реальном времени через асинхронные запросы.

Особенности CryptoScan:

• Поддержка 5-ти популярных сетей: Bitcoin, Ethereum, Solana, TON, USDT Tron
• Асинхронная архитектура с поддержкой async/await
• Точное определение сумм платежей с автоматической остановкой
• Полная поддержка ******-серверов для всех провайдеров
• Обработка событий в реальном времени с колбэками
• Корпоративный уровень надёжности с механизмами повторных попыток
• Мониторинг нескольких кошельков одновременно
• Гибкая настройка интервалов опроса и таймаутов
• Можно интегрировать куда угодно, например в Aiogram

Установка через pip:

pip install pycryptoscan

Код

pip install pycryptoscan

Базовый пример использования:
​

Простой мониторинг Bitcoin

import asyncio
from cryptoscan import create_monitor

async def main():
# Создаём монитор платежей
monitor = create_monitor(
network="bitcoin",
wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",
expected_amount="0.00611813",
auto_stop=True
)

# Обработчик платежей
@monitor.on_payment
async def handle_payment(event):
payment = event.payment_info
print(f" Платёж получен: {payment.amount} {payment.currency}")
print(f" Транзакция: {payment.transaction_id}")
print(f" От: {payment.from_address}")

# Запускаем мониторинг
await monitor.start()
await asyncio.sleep(60) # код должен продолжать работать, имитируем так

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

Python

import asyncio

from cryptoscan import create_monitor



async def main():

    # Создаём монитор платежей

    monitor = create_monitor(

        network="bitcoin",

        wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",

        expected_amount="0.00611813",

        auto_stop=True

    )



    # Обработчик платежей

    @monitor.on_payment

    async def handle_payment(event):

        payment = event.payment_info

        print(f" Платёж получен: {payment.amount} {payment.currency}")

        print(f" Транзакция: {payment.transaction_id}")

        print(f" От: {payment.from_address}")



    # Запускаем мониторинг

    await monitor.start()

    await asyncio.sleep(60) # код должен продолжать работать, имитируем так



if __name__ == "__main__":

    asyncio.run(main())

Поддерживаемые сети:

Bitcoin (BTC) - `bitcoin`, `btc`
Ethereum (ETH) - `ethereum`, `eth`
Solana (SOL) - `solana`, `sol`
TON - `ton`
USDT Tether (TRC-20) - `usdt_tron`, `usdt`, `trc20`

Мультисетевой мониторинг:
​

Одновременный мониторинг нескольких сетей

import asyncio
from cryptoscan import create_monitor

async def multi_network_example():
# Мониторим несколько сетей одновременно
btc_monitor = create_monitor(
network="bitcoin",
wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",
expected_amount="0.00611813"
)

usdt_monitor = create_monitor(
network="usdt_tron",
wallet_address="TVRzaRqX9soeRpcJVT6zCAZjGtLtQXacCR",
expected_amount="200.0"
)

# Единый обработчик платежей
async def handle_payment(event):
payment = event.payment_info
print(f" {payment.currency} платёж: {payment.amount}")
print(payment.transaction_id)

btc_monitor.on_payment(handle_payment)
usdt_monitor.on_payment(handle_payment)

# Запускаем все мониторы
await asyncio.gather(
btc_monitor.start(),
usdt_monitor.start()
)
await asyncio.sleep(60) # программа должна продолжать работать, тут для имитации

asyncio.run(multi_network_example())

Python

import asyncio

from cryptoscan import create_monitor



async def multi_network_example():

    # Мониторим несколько сетей одновременно

    btc_monitor = create_monitor(

        network="bitcoin",

        wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",

        expected_amount="0.00611813"

    )



    usdt_monitor = create_monitor(

        network="usdt_tron",

        wallet_address="TVRzaRqX9soeRpcJVT6zCAZjGtLtQXacCR",

        expected_amount="200.0"

    )



    # Единый обработчик платежей

    async def handle_payment(event):

        payment = event.payment_info

        print(f" {payment.currency} платёж: {payment.amount}")

        print(payment.transaction_id)



    btc_monitor.on_payment(handle_payment)

    usdt_monitor.on_payment(handle_payment)



    # Запускаем все мониторы

    await asyncio.gather(

        btc_monitor.start(),

        usdt_monitor.start()

    )

    await asyncio.sleep(60) # программа должна продолжать работать, тут для имитации



asyncio.run(multi_network_example())

Продвинутая конфигурация с ******:
​

Настройка ****** и таймаутов

from cryptoscan import create_monitor, create_user_config

# Создаём конфигурацию с ******
user_config = create_user_config(
proxy_url="https://proxy.example.com:8080",
proxy_auth="username:password",
timeout=60,
max_retries=5,
ssl_verify=True
)

monitor = create_monitor(
network="ethereum",
wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",
expected_amount="1.0",
user_config=user_config
)

Python

from cryptoscan import create_monitor, create_user_config



# Создаём конфигурацию с ******

user_config = create_user_config(

    proxy_url="https://proxy.example.com:8080",

    proxy_auth="username:password",

    timeout=60,

    max_retries=5,

    ssl_verify=True

)



monitor = create_monitor(

    network="ethereum",

    wallet_address="0xD45F36545b373585a2213427C12AD9af2bEFCE18",

    expected_amount="1.0",

    user_config=user_config

)

Интеграция с Telegram ботом (Aiogram v3):
​

Telegram бот для мониторинга платежей

from aiogram import Bot, Dispatcher
from aiogram.types import Message
from aiogram.filters import Command
from cryptoscan import create_monitor
import asyncio

bot = Bot(token="YOUR_BOT_TOKEN")
dp = Dispatcher()

@dp.message(Command("start"))
async def start_handler(message: Message):
await message.answer(
" CryptoScan Bot\n\n"
"Мониторинг крипто платежей!\n"
"Использование: /monitor <сеть> <адрес> <сумма>\n\n"
"Поддерживаемые сети: bitcoin, ethereum, solana, usdt_tron, ton"
)

@dp.message(Command("monitor"))
async def monitor_payment(message: Message):
args = message.text.split()[1:]
if len(args) != 3:
await message.answer(" Неверный формат! Пример: /monitor bitcoin 3DVS... 0.001")
return

network, address, amount = args

monitor = create_monitor(
network=network,
wallet_address=address,
expected_amount=amount,
auto_stop=True
)

@monitor.on_payment
async def on_payment(event):
payment = event.payment_info
await message.answer(
f" Платёж получен!\n\n"
f" Сумма: {payment.amount} {payment.currency}\n"
f" Транзакция: {payment.transaction_id[:16]}...\n"
f" От: {payment.from_address[:16]}...\n"
)

await message.answer(f" Мониторинг запущен для {network.upper()}!")
asyncio.create_task(monitor.start())

async def main():
await dp.start_polling(bot)

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

Python

from aiogram import Bot, Dispatcher

from aiogram.types import Message

from aiogram.filters import Command

from cryptoscan import create_monitor

import asyncio



bot = Bot(token="YOUR_BOT_TOKEN")

dp = Dispatcher()



@dp.message(Command("start"))

async def start_handler(message: Message):

    await message.answer(

        " CryptoScan Bot\n\n"

        "Мониторинг крипто платежей!\n"

        "Использование: /monitor <сеть> <адрес> <сумма>\n\n"

        "Поддерживаемые сети: bitcoin, ethereum, solana, usdt_tron, ton"

    )



@dp.message(Command("monitor"))

async def monitor_payment(message: Message):

    args = message.text.split()[1:]

    if len(args) != 3:

        await message.answer(" Неверный формат! Пример: /monitor bitcoin 3DVS... 0.001")

        return



    network, address, amount = args



    monitor = create_monitor(

        network=network,

        wallet_address=address,

        expected_amount=amount,

        auto_stop=True

    )



    @monitor.on_payment

    async def on_payment(event):

        payment = event.payment_info

        await message.answer(

            f" Платёж получен!\n\n"

            f" Сумма: {payment.amount} {payment.currency}\n"

            f" Транзакция: {payment.transaction_id[:16]}...\n"

            f" От: {payment.from_address[:16]}...\n"

        )



    await message.answer(f" Мониторинг запущен для {network.upper()}!")

    asyncio.create_task(monitor.start())



async def main():

    await dp.start_polling(bot)



if __name__ == "__main__":

    asyncio.run(main())

Обработка ошибок и надёжность:
​

Надёжный мониторинг с обработкой ошибок

import asyncio
from cryptoscan import create_monitor, NetworkError

async def reliable_monitoring():
monitor = create_monitor(
network="bitcoin",
wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",
expected_amount="0.001",
max_transactions=20, # Проверяем больше транзакций
poll_interval=30.0
)

@monitor.on_payment
async def on_payment(event):
print(f" Платёж подтверждён: {event.payment_info.amount} BTC")

@monitor.on_error
async def on_error(event):
error = event.error
if isinstance(error, NetworkError):
print(f" Сетевая ошибка: {error.message}")
print(" Автоматический повтор...")
else:
print(f" Неожиданная ошибка: {error}")

try:
await monitor.start()
except Exception as e:
print(f" Монитор упал: {e}")
finally:
await monitor.stop()

asyncio.run(reliable_monitoring())

Python

import asyncio

from cryptoscan import create_monitor, NetworkError



async def reliable_monitoring():

    monitor = create_monitor(

        network="bitcoin",

        wallet_address="3DVSCqZdrNJHyu9Le7Sepdh1KgQTNR8reG",

        expected_amount="0.001",

        max_transactions=20,  # Проверяем больше транзакций

        poll_interval=30.0

    )



    @monitor.on_payment

    async def on_payment(event):

        print(f" Платёж подтверждён: {event.payment_info.amount} BTC")



    @monitor.on_error

    async def on_error(event):

        error = event.error

        if isinstance(error, NetworkError):

            print(f" Сетевая ошибка: {error.message}")

            print(" Автоматический повтор...")

        else:

            print(f" Неожиданная ошибка: {error}")



    try:

        await monitor.start()

    except Exception as e:

        print(f" Монитор упал: {e}")

    finally:

        await monitor.stop()



asyncio.run(reliable_monitoring())

Основные классы и методы:

PaymentMonitor

• `async start()` - Запуск мониторинга платежей
• `async stop()` - Остановка мониторинга
• `on_payment(callback)` - Регистрация обработчика платежей
• `on_error(callback)` - Регистрация обработчика ошибок

PaymentInfo

• `transaction_id` - Хэш транзакции
• `amount` - Сумма платежа в основных единицах
• `currency` - Символ валюты (BTC, ETH, и т.д.)
• `timestamp` - Время транзакции
• `from_address` - Адрес отправителя
• `confirmations` - Количество подтверждений

Полезные ссылки:

• PyPI: pycryptoscan
• Страница на GitHub

CryptoScan делает мониторинг крипто платежей простым, быстрым и надёжным!

​