""" Интерактивная авторизация UserBot через интерфейс бота """ import os import logging from telegram import Update, InlineKeyboardMarkup, InlineKeyboardButton from telegram.ext import ContextTypes, ConversationHandler from telethon import TelegramClient from telethon.errors import SessionPasswordNeededError, PhoneNumberInvalidError from app.database import AsyncSessionLocal from app.utils.keyboards import CallbackType import asyncio logger = logging.getLogger(__name__) # Состояния для авторизации AUTH_START = 1 AUTH_PHONE = 2 AUTH_CODE = 3 AUTH_PASSWORD = 4 AUTH_WAITING = 5 # Хранилище для временных данных авторизации auth_sessions = {} async def auth_menu(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: """Меню авторизации UserBot""" query = update.callback_query if query: await query.answer() user_id = update.effective_user.id # Проверяем статус session_file = f"app/sessions/userbot_{user_id}.session" is_authorized = os.path.exists(session_file) text = """🔐 Авторизация UserBot UserBot использует ваш личный аккаунт Telegram для сбора информации о группах и участниках. Важно: • Авторизация безопасна и происходит локально • Ваши данные не передаются никому • UserBot будет работать 24/7 на сервере """ if is_authorized: text += "✅ Статус: Авторизован\n\nВы можете использовать все функции UserBot." keyboard = [ [InlineKeyboardButton("🔄 Переавторизоваться", callback_data="auth_start_phone")], [InlineKeyboardButton("ℹ️ Информация", callback_data="auth_info")], [InlineKeyboardButton("⬅️ Назад", callback_data="userbot_menu")], ] else: text += "❌ Статус: Не авторизован\n\nНажмите кнопку ниже, чтобы начать авторизацию." keyboard = [ [InlineKeyboardButton("🚀 Начать авторизацию", callback_data="auth_start_phone")], [InlineKeyboardButton("ℹ️ Информация", callback_data="auth_info")], [InlineKeyboardButton("⬅️ Назад", callback_data="userbot_menu")], ] if query: await query.edit_message_text( text, parse_mode='HTML', reply_markup=InlineKeyboardMarkup(keyboard) ) else: await update.message.reply_text( text, parse_mode='HTML', reply_markup=InlineKeyboardMarkup(keyboard) ) return AUTH_START async def auth_info(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: """Информация об авторизации""" query = update.callback_query await query.answer() text = """ℹ️ Информация об авторизации UserBot Как это работает: 1. Введите номер телефона вашего аккаунта Telegram 2. Получите SMS-код подтверждения (5 цифр) 3. Введите код в бот 4. Если включена 2FA - введите пароль 5. Готово! UserBot авторизован Что такое SMS-код? Это 5-значный код, который Telegram отправляет на ваш номер. Действителен ~5 минут. Нужно ввести быстро. Что такое 2FA пароль? Это пароль, который ВЫ установили в Telegram на случай потери телефона. НЕ SMS-код, НЕ пароль от почты! 📍 Где его найти? Telegram → Настройки → Приватность и безопасность → Двухфакторная аутентификация Безопасность: ✓ Авторизация происходит локально на сервере ✓ Пароли НЕ сохраняются в базе (обработаны Telethon) ✓ SMS-коды НЕ логируются ✓ Сессия сохраняется в зашифрованном виде ✓ Доступ имеет только этот бот Что может делать UserBot: ✓ Собирать информацию о ваших группах ✓ Получать список участников группы ✓ Сохранять данные в базу данных ✓ Работать 24/7 без вашего участия Что НЕ может делать: ✗ Отправлять сообщения от вашего имени ✗ Удалять или изменять ваши сообщения ✗ Изменять настройки групп ✗ Получать доступ к приватным чатам других пользователей Справка по ошибкам: • "Неверный номер" - проверьте формат (+XX...) • "Код истек" - повторите авторизацию • "Требуется пароль" - введите пароль 2FA • "Неверный пароль" - проверьте регистр и опечатки 📖 Полное руководство: смотрите 2FA_GUIDE.md """ keyboard = [ [InlineKeyboardButton("⬅️ Назад", callback_data="auth_menu")], ] await query.edit_message_text( text, parse_mode='HTML', reply_markup=InlineKeyboardMarkup(keyboard) ) return AUTH_START async def start_phone_input(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: """Начало ввода номера телефона""" query = update.callback_query if query: await query.answer() await query.delete_message() user_id = update.effective_user.id text = """📱 Введите номер телефона Введите номер телефона вашего аккаунта Telegram в формате: +7 (XXX) XXX-XX-XX Примеры: • +79991234567 (Россия) • +82101234567 (Южная Корея) • +11234567890 (США) После ввода номера вам будет отправлен SMS-код """ await update.effective_message.reply_text( text, parse_mode='HTML' ) # Инициализируем хранилище для этого пользователя if user_id not in auth_sessions: auth_sessions[user_id] = {} return AUTH_PHONE async def handle_phone(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: """Обработка номера телефона""" user_id = update.effective_user.id phone = update.message.text.strip() # Очищаем от спецсимволов phone_clean = ''.join(c for c in phone if c.isdigit()) if not phone_clean or len(phone_clean) < 10: await update.message.reply_text( "❌ Некорректный номер телефона. Пожалуйста, введите номер в международном формате (с кодом страны).", parse_mode='HTML' ) return AUTH_PHONE # Восстанавливаем номер с + спереди если его нет if not phone_clean.startswith('1') and not phone_clean[0].isdigit(): phone_clean = phone_clean if not phone.startswith('+'): phone = '+' + phone_clean else: phone = '+' + phone_clean # Сохраняем номер auth_sessions[user_id]['phone'] = phone text = f"""📤 Отправляем код подтверждения... Номер: {phone} Пожалуйста, подождите. Отправляем SMS на ваш номер... """ message = await update.message.reply_text(text, parse_mode='HTML') try: # Создаем Telethon клиент api_id = os.getenv('TELETHON_API_ID') api_hash = os.getenv('TELETHON_API_HASH') if not api_id or not api_hash: await message.edit_text( "❌ Ошибка: API ID или API Hash не установлены", parse_mode='HTML' ) return AUTH_PHONE session_file = f"app/sessions/userbot_auth_{user_id}" client = TelegramClient(session_file, int(api_id), api_hash) # Подключаемся и запрашиваем код await client.connect() try: result = await client.send_code_request(phone) auth_sessions[user_id]['client'] = client auth_sessions[user_id]['phone_code_hash'] = result.phone_code_hash await message.edit_text( f"""✅ Код отправлен! SMS с кодом подтверждения отправлен на номер: {phone} Введите полученный код: """, parse_mode='HTML' ) return AUTH_CODE except PhoneNumberInvalidError: await message.edit_text( f"""❌ Неверный номер телефона Номер {phone} не является корректным номером Telegram. Пожалуйста, попробуйте еще раз с корректным номером. """, parse_mode='HTML' ) await client.disconnect() return AUTH_PHONE except Exception as e: logger.error(f"Auth error: {e}") await message.edit_text( f"""❌ Ошибка при отправке кода {str(e)} Пожалуйста, попробуйте еще раз. """, parse_mode='HTML' ) return AUTH_PHONE async def handle_code(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: """Обработка SMS-кода""" user_id = update.effective_user.id code = update.message.text.strip() if not code.isdigit() or len(code) != 5: await update.message.reply_text( "❌ Код должен состоять из 5 цифр. Пожалуйста, попробуйте еще раз.", parse_mode='HTML' ) return AUTH_CODE message = await update.message.reply_text( "🔄 Проверяем код...", parse_mode='HTML' ) try: if user_id not in auth_sessions or 'client' not in auth_sessions[user_id]: await message.edit_text( "❌ Сессия потеряна. Пожалуйста, начните авторизацию заново.", parse_mode='HTML' ) return AUTH_PHONE client = auth_sessions[user_id]['client'] phone_code_hash = auth_sessions[user_id]['phone_code_hash'] try: # Пытаемся войти с кодом await client.sign_in( phone=auth_sessions[user_id]['phone'], code=code, phone_code_hash=phone_code_hash ) # Успешная авторизация await message.edit_text( """✅ Авторизация успешна! Ваш UserBot авторизован и готов к работе. Сессия сохранена безопасно на сервере. """, parse_mode='HTML' ) # Сохраняем правильное имя сессии correct_session = f"app/sessions/userbot_{user_id}" if os.path.exists(f"app/sessions/userbot_auth_{user_id}.session"): os.rename( f"app/sessions/userbot_auth_{user_id}.session", f"{correct_session}.session" ) await client.disconnect() # Очищаем временные данные if user_id in auth_sessions: del auth_sessions[user_id] # Возвращаемся в меню keyboard = [ [InlineKeyboardButton("✅ Готово", callback_data="userbot_menu")], ] await message.reply_text( "Нажмите кнопку, чтобы вернуться в меню UserBot.", reply_markup=InlineKeyboardMarkup(keyboard) ) return ConversationHandler.END except SessionPasswordNeededError: # Нужна двухфакторная аутентификация (облачный пароль) await message.edit_text( """🔐 Требуется облачный пароль (Cloud Password) Ваш аккаунт Telegram защищен паролем 2FA. Что вводить: Введите облачный пароль, который вы установили в Telegram. ⚠️ Это НЕ: • НЕ SMS-код (он был выше) • НЕ пароль от электронной почты • НЕ пароль от другого аккаунта 📍 Где его найти: На мобильном телефоне откройте Telegram: Настройки → Приватность и безопасность → Двухфакторная аутентификация Там будет переключатель и пароль, который вы установили. 💡 Подсказки: • Пароль чувствителен к регистру (A ≠ a) • Пароль НЕ содержит SMS-код • Это пароль, который только вы знаете • Если забыли - используйте восстановительный код Введите облачный пароль: """, parse_mode='HTML' ) return AUTH_PASSWORD except Exception as e: error_str = str(e).lower() logger.error(f"Code verification error: {e}") # Обработка различных типов ошибок if "expired" in error_str or "code has expired" in error_str: await message.edit_text( """❌ Код истёк SMS-коды действуют примерно 5 минут. К сожалению, этот код уже недействителен. Что делать: Вариант 1 (рекомендуется): 1. Нажмите кнопку ниже "🔄 Получить новый код" 2. Введите полученный код сразу же Вариант 2 (если ошибка повторяется): 1. Подождите 1-2 часа 2. Попробуйте авторизацию снова 3. (Telegram может блокировать при частых попытках) Вариант 3 (если стоит 2FA): 1. Временно отключите 2FA в Telegram 2. Попробуйте авторизацию здесь 3. Снова включите 2FA """, reply_markup=InlineKeyboardMarkup([ [InlineKeyboardButton("🔄 Получить новый код", callback_data="auth_start_phone")], [InlineKeyboardButton("⬅️ Назад", callback_data="auth_menu")], ]), parse_mode='HTML' ) return AUTH_CODE elif "unregistered" in error_str or "phone" in error_str: await message.edit_text( f"""❌ Ошибка номера телефона Номер не зарегистрирован в Telegram или некорректен. Проверьте: • Правильность номера • Наличие кода страны (например +82 для Кореи) • Что это ваш номер (на котором зарегистрирован Telegram) Ошибка: {str(e)} """, reply_markup=InlineKeyboardMarkup([ [InlineKeyboardButton("🔄 Попробовать снова", callback_data="auth_start_phone")], [InlineKeyboardButton("⬅️ Назад", callback_data="auth_menu")], ]), parse_mode='HTML' ) return AUTH_CODE else: await message.edit_text( f"""❌ Ошибка при проверке кода {str(e)} Что попробовать: 1. 🔄 Получить новый код (кнопка ниже) 2. ⏱️ Ввести код быстрее (коды действуют ~5 минут) 3. ⏲️ Подождать час перед следующей попыткой Если ошибка повторяется: • Проверьте интернет соединение • Попробуйте авторизацию через мобильное приложение Telegram • Обратитесь в поддержку Telegram """, reply_markup=InlineKeyboardMarkup([ [InlineKeyboardButton("🔄 Получить новый код", callback_data="auth_start_phone")], [InlineKeyboardButton("⬅️ Назад", callback_data="auth_menu")], ]), parse_mode='HTML' ) return AUTH_CODE async def handle_password(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: """Обработка пароля 2FA""" user_id = update.effective_user.id password = update.message.text message = await update.message.reply_text( "🔄 Проверяем пароль...", parse_mode='HTML' ) try: if user_id not in auth_sessions or 'client' not in auth_sessions[user_id]: await message.edit_text( "❌ Сессия потеряна. Пожалуйста, начните авторизацию заново.", parse_mode='HTML' ) return AUTH_PASSWORD client = auth_sessions[user_id]['client'] try: # Пытаемся войти с пароль await client.sign_in(password=password) # Успешная авторизация await message.edit_text( """✅ Двухфакторная аутентификация успешна! Ваш UserBot авторизован и готов к работе. Сессия сохранена на сервере. Переавторизация не требуется. """, parse_mode='HTML' ) # Сохраняем правильное имя сессии correct_session = f"app/sessions/userbot_{user_id}" if os.path.exists(f"app/sessions/userbot_auth_{user_id}.session"): os.rename( f"app/sessions/userbot_auth_{user_id}.session", f"{correct_session}.session" ) await client.disconnect() # Очищаем временные данные if user_id in auth_sessions: del auth_sessions[user_id] keyboard = [ [InlineKeyboardButton("✅ Готово", callback_data="userbot_menu")], ] await message.reply_text( "Нажмите кнопку, чтобы вернуться в меню UserBot.", reply_markup=InlineKeyboardMarkup(keyboard) ) return ConversationHandler.END except Exception as password_error: error_msg = str(password_error).lower() # Проверяем тип ошибки if "password" in error_msg or "invalid" in error_msg: await message.edit_text( """❌ Неверный облачный пароль Пароль, который вы ввели, неправильный. Проверьте: • Копируете ли вы пароль правильно (без пробелов) • Совпадает ли пароль с тем, что вы установили в Telegram • Учитывается регистр букв (A ≠ a) • Не содержит ли пароль SMS-код Если забыли пароль: 1. На мобильном телефоне откройте Telegram 2. Настройки → Приватность и безопасность → Двухфакторная аутентификация 3. Нажмите "Восстановительный код" 4. Используйте этот код вместо пароля Введите пароль еще раз: """, parse_mode='HTML' ) else: await message.edit_text( f"""❌ Ошибка при проверке облачного пароля {error_msg[:100]} Пожалуйста, попробуйте еще раз или начните авторизацию заново. """, parse_mode='HTML' ) return AUTH_PASSWORD except Exception as e: logger.error(f"Password verification error: {e}") await message.edit_text( f"""❌ Критическая ошибка {str(e)[:150]} Пожалуйста, начните авторизацию заново. """, parse_mode='HTML' ) return AUTH_PASSWORD async def cancel_auth(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int: """Отмена авторизации""" user_id = update.effective_user.id # Отключаем клиент если он есть if user_id in auth_sessions and 'client' in auth_sessions[user_id]: try: await auth_sessions[user_id]['client'].disconnect() except: pass del auth_sessions[user_id] query = update.callback_query if query: await query.answer() await query.delete_message() await update.effective_message.reply_text( "Авторизация отменена.", parse_mode='HTML' ) return ConversationHandler.END def get_auth_conversation_handler(): """Возвращает ConversationHandler для авторизации""" return ConversationHandler( entry_points=[ # Когда пользователь нажимает кнопку авторизации ], states={ AUTH_START: [ # Информация об авторизации ], AUTH_PHONE: [ # Обработка номера телефона ], AUTH_CODE: [ # Обработка SMS кода ], AUTH_PASSWORD: [ # Обработка пароля 2FA ], }, fallbacks=[], name="userbot_auth", persistent=True )