Цифровой актив

import os import logging from typing import Dict, Optional from decimal import Decimal import requests from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import ( Updater, CommandHandler, MessageHandler, Filters, CallbackContext, CallbackQueryHandler, ) from PIL import Image import pytesseract # Настройки бота TOKEN = os.getenv("TELEGRAM_BOT_TOKEN") EXCHANGE_API_KEY = os.getenv("CRYPTO_EXCHANGE_API_KEY") ADMIN_CHAT_ID = os.getenv("ADMIN_CHAT_ID") # Настройки логгирования logging.basicConfig( format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", level=logging.INFO ) logger = logging.getLogger(__name__) # Поддерживаемые криптообменники EXCHANGES = { "binance": { "name": "Binance", "api_url": "https://api.binance.com/api/v3", "requires_card": True, }, "kucoin": { "name": "KuCoin", "api_url": "https://api.kucoin.com", "requires_card": False, }, } class OrderManager: def __init__(self): self.user_orders = {} def create_order( self, user_id: int, fiat_amount: Decimal, crypto_currency: str, fiat_currency: str, exchange: str, ) -> Dict: """Создает новую заявку""" order_id = f"order_{user_id}_{len(self.user_orders) + 1}" order = { "order_id": order_id, "user_id": user_id, "fiat_amount": fiat_amount, "crypto_currency": crypto_currency, "fiat_currency": fiat_currency, "exchange": exchange, "status": "pending", } self.user_orders[order_id] = order return order def get_order(self, order_id: str) -> Optional[Dict]: """Получает информацию о заявке""" return self.user_orders.get(order_id) def update_order_status(self, order_id: str, status: str) -> bool: """Обновляет статус заявки""" if order_id in self.user_orders: self.user_orders[order_id]["status"] = status return True return False order_manager = OrderManager() def start(update: Update, context: CallbackContext) -> None: """Обработчик команды /start""" user = update.effective_user update.message.reply_text( f"Привет, {user.first_name}!\n\n" "Я бот для размещения закупочных заявок на криптовалюту.\n" "Используйте команду /place_order чтобы начать." ) def place_order(update: Update, context: CallbackContext) -> None: """Начало процесса размещения заявки""" keyboard = [ [InlineKeyboardButton("Разместить заявку", callback_data="start_placement")], ] reply_markup = InlineKeyboardMarkup(keyboard) update.message.reply_text( "Нажмите кнопку ниже, чтобы начать размещение закупочной заявки:", reply_markup=reply_markup, ) def start_placement(update: Update, context: CallbackContext) -> None: """Начало размещения заявки""" query = update.callback_query query.answer() # Запрашиваем изображение с суммой query.edit_message_text( "Пожалуйста, загрузите изображение с указанной суммой в фиате, " "на которую вы хотите приобрести криптовалюту." ) def process_image(update: Update, context: CallbackContext) -> None: """Обработка изображения с суммой""" user = update.effective_user photo_file = update.message.photo[-1].get_file() # Сохраняем изображение временно image_path = f"temp_{user.id}.jpg" photo_file.download(image_path) try: # Используем OCR для распознавания суммы amount = extract_amount_from_image(image_path) if amount: context.user_data["fiat_amount"] = amount ask_for_currency(update, context) else: update.message.reply_text( "Не удалось распознать сумму на изображении. " "Пожалуйста, попробуйте еще раз или введите сумму вручную." ) except Exception as e: logger.error(f"Error processing image: {e}") update.message.reply_text( "Произошла ошибка при обработке изображения. Пожалуйста, попробуйте еще раз." ) finally: # Удаляем временный файл if os.path.exists(image_path): os.remove(image_path) def extract_amount_from_image(image_path: str) -> Optional[Decimal]: """Извлекает сумму из изображения с помощью OCR""" try: # Используем pytesseract для распознавания текста text = pytesseract.image_to_string(Image.open(image_path), lang="eng+rus") # Ищем числовые значения в тексте import re matches = re.findall(r"\d+[\.,]?\d*", text) if matches: # Берем последнее найденное число (предполагая, что это сумма) amount_str = matches[-1].replace(",", ".") return Decimal(amount_str) except Exception as e: logger.error(f"Error in OCR: {e}") return None def ask_for_currency(update: Update, context: CallbackContext) -> None: """Запрашиваем валюту для обмена""" keyboard = [ [ InlineKeyboardButton("BTC", callback_data="crypto_BTC"), InlineKeyboardButton("ETH", callback_data="crypto_ETH"), InlineKeyboardButton("USDT", callback_data="crypto_USDT"), ], [InlineKeyboardButton("Другая", callback_data="other_crypto")], ] update.message.reply_text( "Выберите криптовалюту, которую хотите приобрести:", reply_markup=InlineKeyboardMarkup(keyboard), ) def ask_for_fiat_currency(update: Update, context: CallbackContext) -> None: """Запрашиваем фиатную валюту""" query = update.callback_query query.answer() keyboard = [ [ InlineKeyboardButton("USD", callback_data="fiat_USD"), InlineKeyboardButton("EUR", callback_data="fiat_EUR"), InlineKeyboardButton("RUB", callback_data="fiat_RUB"), ], [InlineKeyboardButton("Другая", callback_data="other_fiat")], ] query.edit_message_text( "Выберите фиатную валюту, которой вы оплачиваете:", reply_markup=InlineKeyboardMarkup(keyboard), ) def ask_for_exchange(update: Update, context: CallbackContext) -> None: """Запрашиваем обменник""" query = update.callback_query query.answer() keyboard = [] for exchange_id, exchange_data in EXCHANGES.items(): keyboard.append( [InlineKeyboardButton(exchange_data["name"], callback_data=f"exchange_{exchange_id}")] ) query.edit_message_text( "Выберите криптообменник для размещения заявки:", reply_markup=InlineKeyboardMarkup(keyboard), ) def confirm_order(update: Update, context: CallbackContext) -> None: """Подтверждение заявки перед размещением""" query = update.callback_query query.answer() user_data = context.user_data exchange_id = query.data.split("_")[1] exchange_data = EXCHANGES[exchange_id] message = ( "Подтвердите детали заявки:\n\n" f"Сумма: {user_data['fiat_amount']} {user_data['fiat_currency']}\n" f"Криптовалюта: {user_data['crypto_currency']}\n" f"Обменник: {exchange_data['name']}\n\n" ) if exchange_data["requires_card"]: message += "⚠️ Для этого обменника требуется привязка банковской карты.\n\n" keyboard = [ [ InlineKeyboardButton("Подтвердить", callback_data=f"confirm_{exchange_id}"), InlineKeyboardButton("Отменить", callback_data="cancel"), ] ] query.edit_message_text( message, reply_markup=InlineKeyboardMarkup(keyboard), ) def execute_order(update: Update, context: CallbackContext) -> None: """Выполнение заявки через API обменника""" query = update.callback_query query.answer() user_data = context.user_data exchange_id = query.data.split("_")[1] exchange_data = EXCHANGES[exchange_id] try: # Здесь должна быть логика вызова API обменника # Это примерная реализация order = place_exchange_order( exchange_id, user_data["fiat_amount"], user_data["fiat_currency"], user_data["crypto_currency"], ) if order and order.get("success"): # Сохраняем заявку order_record = order_manager.create_order( user_id=update.effective_user.id, fiat_amount=user_data["fiat_amount"], crypto_currency=user_data["crypto_currency"], fiat_currency=user_data["fiat_currency"], exchange=exchange_id, ) query.edit_message_text( "✅ Ваша заявка успешно размещена!\n\n" f"ID заявки: {order_record['order_id']}\n" "Вы получите уведомление, когда заявка будет выполнена." ) else: query.edit_message_text( "❌ Не удалось разместить заявку. Пожалуйста, попробуйте позже или выберите другой обменник." ) except Exception as e: logger.error(f"Error placing order: {e}") query.edit_message_text( "❌ Произошла ошибка при размещении заявки. Пожалуйста, попробуйте позже." ) def place_exchange_order( exchange_id: str, fiat_amount: Decimal, fiat_currency: str, crypto_currency: str ) -> Dict: """Размещает заказ через API обменника""" # Это примерная реализация - в реальности нужно использовать конкретное API обменника if exchange_id == "binance": # Пример для Binance P2P API url = f"{EXCHANGES[exchange_id]['api_url']}/order" payload = { "fiat": fiat_currency, "crypto": crypto_currency, "amount": str(fiat_amount), "action": "BUY", } headers = {"X-API-KEY": EXCHANGE_API_KEY} response = requests.post(url, json=payload, headers=headers) return response.json() elif exchange_id == "kucoin": # Пример для KuCoin API url = f"{EXCHANGES[exchange_id]['api_url']}/api/v1/orders" payload = { "clientOid": "unique_order_id", "side": "buy", "symbol": f"{crypto_currency}-{fiat_currency}", "type": "market", "funds": str(fiat_amount), } headers = { "KC-API-KEY": EXCHANGE_API_KEY, "KC-API-PASSPHRASE": "your_passphrase", } response = requests.post(url, json=payload, headers=headers) return response.json() return {"success": False, "error": "Unsupported exchange"} def check_orders_status(context: CallbackContext) -> None: """Проверяет статус заявок и отправляет уведомления""" for order_id, order in order_manager.user_orders.items(): if order["status"] == "pending": try: # Проверяем статус через API обменника status = get_order_status(order["exchange"], order_id) if status == "completed": order_manager.update_order_status(order_id, "completed") notify_user(order["user_id"], order_id, context) except Exception as e: logger.error(f"Error checking order status: {e}") def get_order_status(exchange_id: str, order_id: str) -> str: """Получает статус заявки через API обменника""" if exchange_id == "binance": url = f"{EXCHANGES[exchange_id]['api_url']}/order/{order_id}" headers = {"X-API-KEY": EXCHANGE_API_KEY} response = requests.get(url, headers=headers) data = response.json() return data.get("status", "pending") elif exchange_id == "kucoin": url = f"{EXCHANGES[exchange_id]['api_url']}/api/v1/orders/{order_id}" headers = { "KC-API-KEY": EXCHANGE_API_KEY, "KC-API-PASSPHRASE": "your_passphrase", } response = requests.get(url, headers=headers) data = response.json() return data.get("isActive", False) and "pending" or "completed" return "pending" def notify_user(user_id: int, order_id: str, context: CallbackContext) -> None: """Уведомляет пользователя о выполнении заявки""" order = order_manager.get_order(order_id) if order: message = ( f"✅ Ваша заявка {order_id} выполнена!\n\n" f"Сумма: {order['fiat_amount']} {order['fiat_currency']}\n" f"Криптовалюта: {order['crypto_currency']}\n" f"Обменник: {EXCHANGES[order['exchange']]['name']}\n\n" "Спасибо за использование нашего сервиса!" ) context.bot.send_message(chat_id=user_id, text=message) def handle_callback(update: Update, context: CallbackContext) -> None: """Обработчик callback запросов""" query = update.callback_query data = query.data if data == "start_placement": start_placement(update, context) elif data.startswith("crypto_"): context.user_data["crypto_currency"] = data.split("_")[1] ask_for_fiat_currency(update, context) elif data.startswith("fiat_"): context.user_data["fiat_currency"] = data.split("_")[1] ask_for_exchange(update, context) elif data.startswith("exchange_"): confirm_order(update, context) elif data.startswith("confirm_"): execute_order(update, context) elif data == "cancel": query.edit_message_text("Заявка отменена.") elif data == "other_crypto": query.edit_message_text("Пожалуйста, введите название криптовалюты вручную (например, BTC, ETH):") elif data == "other_fiat": query.edit_message_text("Пожалуйста, введите название фиатной валюты врутяную (например, USD, EUR):") def handle_text(update: Update, context: CallbackContext) -> None: """Обработчик текстовых сообщений""" text = update.message.text # Проверяем, ожидаем ли мы ввода криптовалюты if "crypto_currency" not in context.user_data and not text.isdigit(): context.user_data["crypto_currency"] = text.upper() ask_for_fiat_currency(update, context) # Проверяем, ожидаем ли мы ввода фиатной валюты elif "fiat_currency" not in context.user_data and not text.isdigit(): context.user_data["fiat_currency"] = text.upper() ask_for_exchange(update, context) # Проверяем, является ли текст числом (суммой) elif text.replace(".", "").isdigit(): try: amount = Decimal(text) context.user_data["fiat_amount"] = amount ask_for_currency(update, context) except: update.message.reply_text("Пожалуйста, введите корректную сумму.") def error_handler(update: Update, context: CallbackContext) -> None: """Обработчик ошибок""" logger.error(msg="Exception while handling an update:", exc_info=context.error) if update and update.effective_message: update.effective_message.reply_text( "Произошла ошибка. Пожалуйста, попробуйте еще раз или свяжитесь с поддержкой." ) def main() -> None: """Запуск бота""" updater = Updater(TOKEN) dispatcher = updater.dispatcher # Обработчики команд dispatcher.add_handler(CommandHandler("start", start)) dispatcher.add_handler(CommandHandler("place_order", place_order)) # Обработчики сообщений dispatcher.add_handler(MessageHandler(Filters.photo, process_image)) dispatcher.add_handler(MessageHandler(Filters.text & ~Filters.command, handle_text)) # Обработчики callback запросов dispatcher.add_handler(CallbackQueryHandler(handle_callback)) # Обработчик ошибок dispatcher.add_error_handler(error_handler) # Запускаем проверку статуса заявок каждые 5 минут job_queue = updater.job_queue job_queue.run_repeating(check_orders_status, interval=300, first=0) # Запуск бота updater.start_polling() updater.idle() if __name__ == "__main__": main()