github-sponsors-webhook-bot/payment_sources/binance_alerts.py

#!/usr/bin/env python3
"""
Handles Binance payment alerts.

This module will connect to the Binance API to fetch:
- New cryptocurrency deposits.
- Completed P2P payment receipts.
"""

import os
import logging
# from binance.client import Client # To be added when implementing

logger = logging.getLogger("GitHubSponsorsBot.BinanceAlerts")

# Environment variables for Binance API (ensure these are set)
BINANCE_API_KEY = os.getenv('BINANCE_API_KEY')
BINANCE_API_SECRET = os.getenv('BINANCE_API_SECRET')

class BinanceAlerts:
    def __init__(self, telegram_bot):
        self.telegram_bot = telegram_bot
        # self.client = Client(BINANCE_API_KEY, BINANCE_API_SECRET) # Uncomment when ready
        if not BINANCE_API_KEY or not BINANCE_API_SECRET:
            logger.error("Binance API Key or Secret not configured. Binance alerts will be disabled.")
            self.enabled = False
        else:
            self.enabled = True
            # Initialize API client here
            pass

    def check_for_new_payments(self):
        """
        Checks for new deposits and P2P payments and sends alerts.
        """
        if not self.enabled:
            return

        logger.info("Checking for new Binance payments...")
        # TODO: Implement logic to fetch new crypto deposits
        # Example:
        # deposits = self.client.get_deposit_history()
        # for deposit in deposits:
        #     if self.is_new_deposit(deposit): # Implement is_new_deposit to avoid duplicates
        #         message = self.format_deposit_message(deposit)
        #         self.telegram_bot.send_message(message)
        #         self.mark_deposit_as_processed(deposit) # Mark to avoid re-alerting

        # TODO: Implement logic to fetch completed P2P payments
        # Example:
        # p2p_orders = self.client.get_c2c_trade_history(tradeType='BUY') # or relevant endpoint
        # for order in p2p_orders:
        #     if order['orderStatus'] == 'COMPLETED' and self.is_new_p2p_payment(order):
        #         message = self.format_p2p_message(order)
        #         self.telegram_bot.send_message(message)
        #         self.mark_p2p_as_processed(order)
        logger.info("Finished checking Binance payments.")

    def format_deposit_message(self, deposit_data):
        """
        Formats a cryptocurrency deposit alert message.
        """
        # Example structure, adjust based on actual API response
        # amount = deposit_data.get('amount')
        # currency = deposit_data.get('coin')
        # tx_id = deposit_data.get('txId')
        # timestamp = deposit_data.get('insertTime') # Convert to readable format
        # formatted_time = datetime.fromtimestamp(timestamp/1000).strftime('%Y-%m-%d %H:%M:%S') if timestamp else 'N/A'
        #
        # message = (
        #     f"💰 *New Binance Deposit Received*\n\n"
        #     f"*Amount:* {amount} {currency}\n"
        #     f"*Transaction ID:* `{tx_id}`\n"
        #     f"*Timestamp:* {formatted_time}\n"
        # )
        # return message
        return "Placeholder Binance deposit message" # Replace with actual formatting

    def format_p2p_message(self, p2p_data):
        """
        Formats a P2P payment receipt alert message.
        """
        # Example structure, adjust based on actual API response
        # order_number = p2p_data.get('orderNumber')
        # amount = p2p_data.get('totalPrice')
        # currency = p2p_data.get('fiatUnit')
        # crypto_amount = p2p_data.get('amount')
        # crypto_currency = p2p_data.get('asset')
        # timestamp = p2p_data.get('createTime') # Convert to readable format
        # formatted_time = datetime.fromtimestamp(timestamp/1000).strftime('%Y-%m-%d %H:%M:%S') if timestamp else 'N/A'
        #
        # message = (
        #     f"🤝 *Binance P2P Payment Completed*\n\n"
        #     f"*Order Number:* `{order_number}`\n"
        #     f"*Fiat Amount:* {amount} {currency}\n"
        #     f"*Crypto Amount:* {crypto_amount} {crypto_currency}\n"
        #     f"*Timestamp:* {formatted_time}\n"
        # )
        # return message
        return "Placeholder Binance P2P message" # Replace with actual formatting

    # Helper methods to track processed transactions (to avoid duplicate alerts)
    # These could use a simple in-memory set for recent IDs, or a small DB/file for persistence
    def is_new_deposit(self, deposit_data):
        # TODO: Implement logic to check if this deposit has been alerted already
        return True

    def mark_deposit_as_processed(self, deposit_data):
        # TODO: Implement logic to mark this deposit as processed
        pass

    def is_new_p2p_payment(self, p2p_data):
        # TODO: Implement logic to check if this P2P payment has been alerted already
        return True

    def mark_p2p_as_processed(self, p2p_data):
        # TODO: Implement logic to mark this P2P payment as processed
        pass

if __name__ == '__main__':
    # This section is for testing the module independently
    # You'll need to mock telegram_bot or provide a dummy one
    class MockTelegramBot:
        def send_message(self, message):
            print(f"MockTelegramBot sending: {message}")

    # Ensure environment variables are set for testing
    # load_dotenv() # If you have a .env file for testing
    # BINANCE_API_KEY = os.getenv('BINANCE_API_KEY_TEST')
    # BINANCE_API_SECRET = os.getenv('BINANCE_API_SECRET_TEST')

    if BINANCE_API_KEY and BINANCE_API_SECRET:
        logger.info("Running BinanceAlerts standalone test...")
        mock_bot = MockTelegramBot()
        binance_alerter = BinanceAlerts(mock_bot)
        if binance_alerter.enabled:
            binance_alerter.check_for_new_payments()
        else:
            logger.warning("Binance alerts are disabled due to missing API credentials.")
    else:
        logger.warning("Skipping BinanceAlerts standalone test: API credentials not found.")