import asyncio import aiohttp import logging from datetime import datetime from bs4 import BeautifulSoup from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import ( Application, CommandHandler, CallbackQueryHandler, ContextTypes, MessageHandler, filters, ConversationHandler ) # ==== Настройка логирования ==== logging.basicConfig( format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', level=logging.INFO ) logger = logging.getLogger(__name__) # ==== Конфигурация ==== TOKEN = "7682277868:AAFhvQrgu-7ikxU56AhBGoVRpaHTK9olh-Y" ADMIN_ID = 1470225013 HEADERS = {'User-Agent': 'Mozilla/5.0'} UPDATE_INTERVAL = 3600 # Интервал автообновления в секундах (1 час) # ==== Состояния диалога ==== ADDING_DEVICE, EDITING_DEVICE = range(2) # ==== База устройств ==== devices = { # Пример устройства: # "device1": { # "name": "Main Rig", # "status": "online", # "address": "48jF...", # "mining": True, # "hashrate": 1200, # "last_update": "2023-01-01 12:00", # "type": "GPU", # "location": "Home" # } } # ==== Класс для кэширования данных ==== class DataCache: _instance = None xmr_rate = 0.0 last_updated = None def __new__(cls): if cls._instance is None: cls._instance = super(DataCache, cls).__new__(cls) return cls._instance @classmethod async def update_xmr_rate(cls): try: async with aiohttp.ClientSession() as session: async with session.get("https://www.bitget.com/ru/spot/XMRUSDT", headers=HEADERS, timeout=10) as resp: if resp.status == 200: html = await resp.text() soup = BeautifulSoup(html, 'html.parser') price_element = soup.find('div', class_='price') if price_element: cls.xmr_rate = float(price_element.text.strip().replace(',', '')) cls.last_updated = datetime.now().strftime("%Y-%m-%d %H:%M") return True except Exception as e: logger.error(f"Ошибка получения курса XMR: {e}") return False # ==== Утилиты ==== def format_xmr(value): return f"{value:.6f} XMR" def format_rub(value): return f"{value:.2f} RUB" def short_address(address): return f"{address[:6]}...{address[-4:]}" if address else "Не указан" async def fetch_html(url): try: async with aiohttp.ClientSession() as session: async with session.get(url, headers=HEADERS, timeout=10) as resp: if resp.status == 200: return await resp.text() except Exception as e: logger.error(f"Ошибка запроса к {url}: {e}") return None # ==== Парсинг данных ==== async def get_xmr_rate() -> float: if DataCache.xmr_rate == 0 or not DataCache.last_updated or (datetime.now() - datetime.strptime(DataCache.last_updated, "%Y-%m-%d %H:%M")).seconds > 3600: await DataCache.update_xmr_rate() return DataCache.xmr_rate async def get_kryptex_balance(wallet: str) -> dict: try: url = f"https://pool.kryptex.com/api/v1/miner/{wallet}/stats" async with aiohttp.ClientSession() as session: async with session.get(url, headers=HEADERS, timeout=10) as resp: if resp.status == 200: data = await resp.json() return { 'balance': float(data.get('balance', 0)), 'unpaid': float(data.get('unpaid', 0)), 'hashrate': float(data.get('hashrate', 0)), 'workers': int(data.get('workers', 0)) } except Exception as e: logger.error(f"Ошибка получения баланса для {wallet}: {e}") return None # ==== Команды бота ==== async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): if update.effective_user.id != ADMIN_ID: await update.message.reply_text("🔒 Доступ запрещен") return keyboard = [ [InlineKeyboardButton("💰 Баланс", callback_data="balance")], [InlineKeyboardButton("🖥️ Устройства", callback_data="devices")], [InlineKeyboardButton("⛏️ Управление майнингом", callback_data="mining_control")], [InlineKeyboardButton("📊 Статистика", callback_data="stats")] ] text = ( "🏭 *Mining Control Bot*\n\n" "Управляйте вашей майнинг фермой прямо из Telegram!\n\n" "📊 *Доступные команды:*\n" "- /balance - Текущий баланс\n" "- /devices - Управление устройствами\n" "- /mining - Управление майнингом\n" "- /stats - Подробная статистика\n" "- /notify - Настройки уведомлений" ) await update.message.reply_text( text, reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown" ) async def balance_command(update: Update, context: ContextTypes.DEFAULT_TYPE): if update.effective_user.id != ADMIN_ID: await update.message.reply_text("🔒 Доступ запрещен") return await show_balance(update, context) async def show_balance(update: Update, context: ContextTypes.DEFAULT_TYPE): total_balance = 0.0 total_unpaid = 0.0 total_hashrate = 0.0 active_workers = 0 rate = await get_xmr_rate() for device_id, device in devices.items(): if device["status"] == "online": data = await get_kryptex_balance(device["address"]) if data: total_balance += data['balance'] total_unpaid += data['unpaid'] total_hashrate += data['hashrate'] active_workers += data['workers'] devices[device_id]['hashrate'] = data['hashrate'] devices[device_id]['last_update'] = datetime.now().strftime("%Y-%m-%d %H:%M") rub_balance = total_balance * rate rub_unpaid = total_unpaid * rate text = ( "💰 *Текущий баланс*\n\n" f"● Доступно: *{format_xmr(total_balance)}* (~{format_rub(rub_balance)})\n" f"● Невыплачено: *{format_xmr(total_unpaid)}* (~{format_rub(rub_unpaid)})\n" f"● Общий хешрейт: *{total_hashrate:.2f} H/s*\n" f"● Активные воркеры: *{active_workers}*\n\n" f"📊 Курс XMR: *{rate:.2f} RUB*" ) if isinstance(update, Update): await update.message.reply_text(text, parse_mode="Markdown") else: await update.edit_message_text(text, parse_mode="Markdown") async def devices_command(update: Update, context: ContextTypes.DEFAULT_TYPE): if update.effective_user.id != ADMIN_ID: await update.message.reply_text("🔒 Доступ запрещен") return await show_devices_menu(update, context) async def show_devices_menu(update: Update, context: ContextTypes.DEFAULT_TYPE): keyboard = [] for device_id, device in devices.items(): status_icon = "🟢" if device["status"] == "online" else "🔴" mining_icon = "⛏️" if device["mining"] else "💤" btn_text = f"{status_icon} {mining_icon} {device['name']}" keyboard.append([InlineKeyboardButton(btn_text, callback_data=f"device_{device_id}")]) keyboard.append([InlineKeyboardButton("➕ Добавить устройство", callback_data="add_device")]) text = "🖥️ *Список устройств*\nВыберите устройство для управления:" if isinstance(update, Update): await update.message.reply_text(text, reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown") else: await update.edit_message_text(text, reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown") async def handle_device_selection(update: Update, context: ContextTypes.DEFAULT_TYPE): query = update.callback_query await query.answer() if query.data == "add_device": await query.edit_message_text("✏️ Введите адрес кошелька для нового устройства:") return ADDING_DEVICE elif query.data.startswith("device_"): device_id = query.data.split("_")[1] device = devices.get(device_id) if not device: await query.edit_message_text("⚠ Устройство не найдено") return data = await get_kryptex_balance(device["address"]) status_str = "🟢 Онлайн" if device["status"] == "online" else "🔴 Оффлайн" mining_str = "✅ Включен" if device["mining"] else "❌ Выключен" hashrate_str = f"{device.get('hashrate', 0):.2f} H/s" if device.get('hashrate') else "Неизвестно" text = ( f"📌 *{device['name']}*\n\n" f"● Статус: {status_str}\n" f"● Майнинг: {mining_str}\n" f"● Хешрейт: {hashrate_str}\n" f"● Адрес: `{short_address(device['address'])}`\n" f"● Тип: {device.get('type', 'Не указан')}\n" f"● Местоположение: {device.get('location', 'Не указано')}\n\n" ) if data: text += ( f"💰 Баланс: {format_xmr(data['balance'])}\n" f"● Невыплачено: {format_xmr(data['unpaid'])}\n" f"● Воркеры: {data['workers']}\n\n" ) text += f"🔄 Последнее обновление: {device.get('last_update', 'Никогда')}" keyboard = [ [InlineKeyboardButton("🔄 Обновить", callback_data=f"device_{device_id}"), InlineKeyboardButton("✏️ Редактировать", callback_data=f"edit_{device_id}")], [InlineKeyboardButton("▶️ Запустить" if not device["mining"] else "⏹️ Остановить", callback_data=f"toggle_{device_id}")], [InlineKeyboardButton("🗑️ Удалить", callback_data=f"delete_{device_id}")], [InlineKeyboardButton("🔙 Назад", callback_data="devices")] ] await query.edit_message_text( text, reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown" ) async def toggle_mining(update: Update, context: ContextTypes.DEFAULT_TYPE): query = update.callback_query await query.answer() action, device_id = query.data.split("_") device = devices.get(device_id) if not device: await query.answer("⚠ Устройство не найдено", show_alert=True) return # Здесь должна быть логика управления майнингом на реальном устройстве device["mining"] = not device["mining"] status = "запущен" if device["mining"] else "остановлен" await query.answer(f"Майнинг {status} на {device['name']}") await handle_device_selection(update, context) async def add_device_address(update: Update, context: ContextTypes.DEFAULT_TYPE): address = update.message.text.strip() if not address or len(address) < 10: # Простая валидация await update.message.reply_text("❌ Неверный адрес кошелька. Попробуйте еще раз:") return ADDING_DEVICE device_id = f"device_{len(devices)+1}" devices[device_id] = { "name": f"Устройство {len(devices)+1}", "status": "online", "address": address, "mining": False, "hashrate": 0, "last_update": datetime.now().strftime("%Y-%m-%d %H:%M"), "type": "Unknown", "location": "Unknown" } await update.message.reply_text( f"✅ Устройство успешно добавлено!\n" f"Имя: {devices[device_id]['name']}\n" f"Адрес: {short_address(address)}\n\n" f"Используйте команду /devices для редактирования." ) return ConversationHandler.END async def mining_control_command(update: Update, context: ContextTypes.DEFAULT_TYPE): if update.effective_user.id != ADMIN_ID: await update.message.reply_text("🔒 Доступ запрещен") return keyboard = [] for device_id, device in devices.items(): btn_text = ("⏹️ Остановить " if device["mining"] else "▶️ Запустить ") + device["name"] keyboard.append([InlineKeyboardButton(btn_text, callback_data=f"toggle_{device_id}")]) keyboard.append([ InlineKeyboardButton("▶️ Запустить все", callback_data="start_all"), InlineKeyboardButton("⏹️ Остановить все", callback_data="stop_all") ]) keyboard.append([InlineKeyboardButton("🔙 Назад", callback_data="cancel")]) await update.message.reply_text( "⛏️ *Управление майнингом*\nВыберите действие:", reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown" ) async def stats_command(update: Update, context: ContextTypes.DEFAULT_TYPE): if update.effective_user.id != ADMIN_ID: await update.message.reply_text("🔒 Доступ запрещен") return total_hashrate = sum(device.get('hashrate', 0) for device in devices.values()) active_devices = sum(1 for device in devices.values() if device["mining"]) total_workers = sum(device.get('workers', 0) for device in devices.values()) rate = await get_xmr_rate() total_balance = 0.0 total_unpaid = 0.0 for device in devices.values(): data = await get_kryptex_balance(device["address"]) if data: total_balance += data['balance'] total_unpaid += data['unpaid'] rub_total = total_balance * rate rub_unpaid = total_unpaid * rate text = ( "📊 *Статистика майнинг фермы*\n\n" f"● Всего устройств: *{len(devices)}*\n" f"● Активных: *{active_devices}*\n" f"● Воркеров: *{total_workers}*\n" f"● Общий хешрейт: *{total_hashrate:.2f} H/s*\n\n" f"💰 Баланс: *{format_xmr(total_balance)}* (~{format_rub(rub_total)})\n" f"● Невыплачено: *{format_xmr(total_unpaid)}* (~{format_rub(rub_unpaid)})\n\n" f"📈 Курс XMR: *{rate:.2f} RUB*" ) keyboard = [ [InlineKeyboardButton("🔄 Обновить", callback_data="stats")], [InlineKeyboardButton("🔙 Назад", callback_data="cancel")] ] await update.message.reply_text( text, reply_markup=InlineKeyboardMarkup(keyboard), parse_mode="Markdown" ) async def handle_global_actions(update: Update, context: ContextTypes.DEFAULT_TYPE): query = update.callback_query await query.answer() if query.data == "start_all": for device in devices.values(): device["mining"] = True await query.answer("Майнинг запущен на всех устройствах") elif query.data == "stop_all": for device in devices.values(): device["mining"] = False await query.answer("Майнинг остановлен на всех устройствах") elif query.data in ["balance", "devices", "mining_control", "stats"]: handlers = { "balance": show_balance, "devices": show_devices_menu, "mining_control": mining_control_command, "stats": stats_command } await handlers[query.data](update, context) return elif query.data == "cancel": await start(update, context) return # Обновляем текущий экран if query.data in ["start_all", "stop_all"]: await mining_control_command(update, context) elif query.data == "stats": await stats_command(update, context) # ==== Автоматическое обновление данных ==== async def auto_update(context: ContextTypes.DEFAULT_TYPE): logger.info("Выполняю автоматическое обновление данных...") for device_id, device in devices.items(): if device["status"] == "online": data = await get_kryptex_balance(device["address"]) if data: devices[device_id]['hashrate'] = data['hashrate'] devices[device_id]['last_update'] = datetime.now().strftime("%Y-%m-%d %H:%M") await get_xmr_rate() # ==== Обработка ошибок ==== async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE): logger.error(msg="Ошибка в обработчике:", exc_info=context.error) if update and hasattr(update, 'message'): await update.message.reply_text( "⚠ Произошла ошибка. Попробуйте позже или свяжитесь с разработчиком." ) # ==== Основная функция запуска ==== def main(): app = Application.builder().token(TOKEN).build() # Регистрация обработчиков команд app.add_handler(CommandHandler("start", start)) app.add_handler(CommandHandler("balance", balance_command)) app.add_handler(CommandHandler("devices", devices_command)) app.add_handler(CommandHandler("mining", mining_control_command)) app.add_handler(CommandHandler("stats", stats_command)) # Регистрация обработчиков callback-запросов app.add_handler(CallbackQueryHandler(show_balance, pattern="^balance$")) app.add_handler(CallbackQueryHandler(show_devices_menu, pattern="^devices$")) app.add_handler(CallbackQueryHandler(mining_control_command, pattern="^mining_control$")) app.add_handler(CallbackQueryHandler(stats_command, pattern="^stats$")) app.add_handler(CallbackQueryHandler(handle_global_actions, pattern="^(start_all|stop_all|cancel)$")) app.add_handler(CallbackQueryHandler(handle_device_selection, pattern="^device_")) app.add_handler(CallbackQueryHandler(toggle_mining, pattern="^toggle_")) # Диалог добавления устройства conv_handler = ConversationHandler( entry_points=[CallbackQueryHandler(handle_device_selection, pattern="^add_device$")], states={ ADDING_DEVICE: [MessageHandler(filters.TEXT & ~filters.COMMAND, add_device_address)] }, fallbacks=[CommandHandler("cancel", start)] ) app.add_handler(conv_handler) # Обработка ошибок app.add_error_handler(error_handler) # Планировщик для автоматического обновления job_queue = app.job_queue if job_queue: job_queue.run_repeating(auto_update, interval=UPDATE_INTERVAL, first=10) logger.info("Бот запущен и готов к работе!") app.run_polling() if __name__ == "__main__": main()