From 015b22264cc2f02f8683b1c4023dbe6238d18712 Mon Sep 17 00:00:00 2001 From: Kavalar Date: Sun, 3 May 2026 23:33:47 +0300 Subject: [PATCH] fix --- bot.py | 339 ++++++++++++------------------------- debug_bot.py | 175 +++++++++++++++++++ requirements.txt | 6 +- test_socks5_longpolling.py | 247 +++++++++++++++++++++++++++ 4 files changed, 536 insertions(+), 231 deletions(-) create mode 100644 debug_bot.py create mode 100644 test_socks5_longpolling.py diff --git a/bot.py b/bot.py index ab92075..d6693ea 100644 --- a/bot.py +++ b/bot.py @@ -14,9 +14,6 @@ from telegram.ext import ( filters, ContextTypes ) -from telegram.request import HTTPXRequest -import httpx -from httpx_socks import AsyncProxyTransport from config import TELEGRAM_BOT_TOKEN, PIApi_API_KEY, PIApi_BASE_URL, PROXY_URL, PROXY_TYPE from database import init_db, save_user, save_image_record, update_image_record, get_user_images @@ -39,97 +36,6 @@ os.makedirs(UPLOAD_DIR, exist_ok=True) os.makedirs(PROCESSED_DIR, exist_ok=True) -def get_telegram_request(): - """Настройка SOCKS5 прокси для Telegram с использованием httpx_socks""" - if not PROXY_URL: - return None - - logger.info(f"🔌 Настройка SOCKS5 прокси для Telegram") - - try: - # Парсим прокси URL - import re - match = re.search(r'socks5://([^:]+):([^@]+)@([^:]+):(\d+)', PROXY_URL) - - if match: - username = match.group(1) - password = match.group(2) - host = match.group(3) - port = int(match.group(4)) - - logger.info(f"🔌 Прокси: {host}:{port}") - - # Создаем транспорт с SOCKS5 прокси - transport = AsyncProxyTransport.from_url( - f"socks5://{username}:{password}@{host}:{port}" - ) - request = HTTPXRequest(transport=transport) - logger.info("✅ SOCKS5 прокси настроен успешно") - return request - else: - # Пробуем без авторизации - from urllib.parse import urlparse - parsed = urlparse(PROXY_URL) - host = parsed.hostname - port = parsed.port - - transport = AsyncProxyTransport.from_url(f"socks5://{host}:{port}") - request = HTTPXRequest(transport=transport) - logger.info("✅ SOCKS5 прокси настроен (без авторизации)") - return request - - except Exception as e: - logger.error(f"Ошибка настройки: {e}") - return None - - -async def test_telegram_connection(): - """Тест подключения к Telegram через прокси""" - if not PROXY_URL: - logger.error("Прокси не настроен") - return False - - logger.info("Тестируем подключение к Telegram через SOCKS5 прокси...") - - try: - # Используем aiohttp с socks прокси - from aiohttp_socks import ProxyConnector - - # Парсим прокси - import re - match = re.search(r'socks5://([^:]+):([^@]+)@([^:]+):(\d+)', PROXY_URL) - - if match: - username = match.group(1) - password = match.group(2) - host = match.group(3) - port = int(match.group(4)) - - connector = ProxyConnector.from_url(f"socks5://{username}:{password}@{host}:{port}") - else: - from urllib.parse import urlparse - parsed = urlparse(PROXY_URL) - connector = ProxyConnector.from_url(f"socks5://{parsed.hostname}:{parsed.port}") - - async with aiohttp.ClientSession(connector=connector) as session: - url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/getMe" - async with session.get(url, timeout=15) as response: - if response.status == 200: - data = await response.json() - if data.get('ok'): - logger.info(f"✅ Подключение работает! Бот: @{data['result']['username']}") - return True - else: - logger.error(f"Ошибка API: {data}") - return False - else: - logger.error(f"HTTP {response.status}") - return False - except Exception as e: - logger.error(f"Ошибка: {e}") - return False - - async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): """Обработчик /start""" user = update.effective_user @@ -164,7 +70,7 @@ async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE): await update.message.reply_text( "📖 Инструкция:\n\n" "1. Отправь фото\n" - "2. Напиши промпт (например: 'сделай фон пляжем')\n" + "2. Напиши промпт\n" "3. Получи результат!\n\n" "Команды: /start, /test, /history" ) @@ -205,6 +111,7 @@ async def handle_photo(update: Update, context: ContextTypes.DEFAULT_TYPE): async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE): """Отмена""" + logger.info(f"Cancel от {update.effective_user.id}") if 'image_path' in context.user_data: path = context.user_data['image_path'] if os.path.exists(path): @@ -215,6 +122,98 @@ async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE): return ConversationHandler.END +async def call_piapi_api(image_path: str, prompt: str) -> str: + """Вызов PiAPI через SOCKS5 прокси""" + if not PIApi_API_KEY: + raise Exception("PIApi_API_KEY не настроен") + + with open(image_path, "rb") as f: + image_base64 = base64.b64encode(f.read()).decode('utf-8') + + headers = { + "x-api-key": PIApi_API_KEY, + "Content-Type": "application/json" + } + + payload = { + "model": "black-forest-labs/FLUX.1-dev", + "task_type": "image-to-image", + "input": { + "image": f"data:image/jpeg;base64,{image_base64}", + "prompt": prompt, + "num_inference_steps": 28, + "guidance_scale": 7.5, + "strength": 0.8 + } + } + + # Используем aiohttp-socks для запросов + from aiohttp_socks import ProxyConnector + import re + + # Парсим прокси + match = re.search(r'socks5://([^:]+):([^@]+)@([^:]+):(\d+)', PROXY_URL) + if match: + username = match.group(1) + password = match.group(2) + host = match.group(3) + port = int(match.group(4)) + connector = ProxyConnector.from_url(f"socks5://{username}:{password}@{host}:{port}") + else: + from urllib.parse import urlparse + parsed = urlparse(PROXY_URL) + connector = ProxyConnector.from_url(f"socks5://{parsed.hostname}:{parsed.port}") + + async with aiohttp.ClientSession(connector=connector) as session: + async with session.post(PIApi_BASE_URL, headers=headers, json=payload, timeout=30) as response: + if response.status != 200: + text = await response.text() + raise Exception(f"HTTP {response.status}: {text[:100]}") + + data = await response.json() + if data.get('code') != 200: + raise Exception(f"API Error: {data.get('message')}") + + task_id = data['data']['task_id'] + logger.info(f"Создана задача: {task_id}") + + return await poll_task(task_id, connector) + + +async def poll_task(task_id: str, connector, max_attempts: int = 60) -> str: + """Ожидание завершения задачи""" + get_url = f"https://api.piapi.ai/api/v1/task/{task_id}" + headers = {"x-api-key": PIApi_API_KEY} + + async with aiohttp.ClientSession(connector=connector) as session: + for attempt in range(max_attempts): + try: + async with session.get(get_url, headers=headers) as response: + if response.status != 200: + await asyncio.sleep(2) + continue + + data = await response.json() + status = data.get('data', {}).get('status') + logger.info(f"Задача {task_id}: {status} (попытка {attempt + 1})") + + if status == 'completed': + output = data.get('data', {}).get('output', {}) + result_url = output.get('url') or output.get('image') + if result_url: + logger.info(f"Задача {task_id} выполнена") + return result_url + elif status == 'failed': + raise Exception("Задача не выполнена") + + await asyncio.sleep(2) + except Exception as e: + logger.error(f"Ошибка опроса: {e}") + await asyncio.sleep(2) + + raise Exception("Превышено время ожидания") + + async def process_image(update: Update, context: ContextTypes.DEFAULT_TYPE): """Обработка через PiAPI""" prompt = update.message.text @@ -239,7 +238,7 @@ async def process_image(update: Update, context: ContextTypes.DEFAULT_TYPE): await db.execute("UPDATE images SET prompt = ? WHERE id = ?", (prompt, image_id)) await db.commit() - result_url = await call_piapi_with_proxy(image_path, prompt) + result_url = await call_piapi_api(image_path, prompt) if result_url: result_path = os.path.join( @@ -247,7 +246,7 @@ async def process_image(update: Update, context: ContextTypes.DEFAULT_TYPE): f"user_{user.id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}_result.jpg" ) - # Скачиваем результат через SOCKS5 + # Скачиваем результат from aiohttp_socks import ProxyConnector import re @@ -276,10 +275,9 @@ async def process_image(update: Update, context: ContextTypes.DEFAULT_TYPE): caption=f"✅ Готово!\n\n✨ {prompt}" ) - msg = await update.message.reply_photo(photo=open(result_path, "rb")) await update_image_record( image_id=image_id, - processed_file_id=msg.photo[-1].file_id, + processed_file_id=result_path, processed_url=result_path, status='completed' ) @@ -295,6 +293,7 @@ async def process_image(update: Update, context: ContextTypes.DEFAULT_TYPE): await status_msg.edit_text(f"❌ Ошибка: {str(e)}\n\nПопробуйте еще раз.") await update_image_record(image_id=image_id, status='failed') + # Очистка if os.path.exists(image_path): os.remove(image_path) @@ -302,123 +301,13 @@ async def process_image(update: Update, context: ContextTypes.DEFAULT_TYPE): return ConversationHandler.END -async def call_piapi_with_proxy(image_path: str, prompt: str) -> str: - """Вызов PiAPI через SOCKS5 прокси""" - if not PIApi_API_KEY: - raise Exception("PIApi_API_KEY не настроен") - - with open(image_path, "rb") as f: - image_base64 = base64.b64encode(f.read()).decode('utf-8') - - headers = { - "x-api-key": PIApi_API_KEY, - "Content-Type": "application/json" - } - - payload = { - "model": "black-forest-labs/FLUX.1-dev", - "task_type": "image-to-image", - "input": { - "image": f"data:image/jpeg;base64,{image_base64}", - "prompt": prompt, - "num_inference_steps": 28, - "guidance_scale": 7.5, - "strength": 0.8 - } - } - - # Создаем SOCKS5 коннектор - from aiohttp_socks import ProxyConnector - import re - - match = re.search(r'socks5://([^:]+):([^@]+)@([^:]+):(\d+)', PROXY_URL) - if match: - username = match.group(1) - password = match.group(2) - host = match.group(3) - port = int(match.group(4)) - connector = ProxyConnector.from_url(f"socks5://{username}:{password}@{host}:{port}") - else: - from urllib.parse import urlparse - parsed = urlparse(PROXY_URL) - connector = ProxyConnector.from_url(f"socks5://{parsed.hostname}:{parsed.port}") - - async with aiohttp.ClientSession(connector=connector) as session: - async with session.post( - PIApi_BASE_URL, - headers=headers, - json=payload, - timeout=30 - ) as response: - if response.status != 200: - text = await response.text() - raise Exception(f"HTTP {response.status}: {text[:100]}") - - data = await response.json() - if data.get('code') != 200: - raise Exception(f"API Error: {data.get('message')}") - - task_id = data['data']['task_id'] - logger.info(f"Создана задача: {task_id}") - - return await poll_task_with_proxy(task_id) - - -async def poll_task_with_proxy(task_id: str, max_attempts: int = 60) -> str: - """Ожидание завершения задачи через SOCKS5 прокси""" - get_url = f"https://api.piapi.ai/api/v1/task/{task_id}" - headers = {"x-api-key": PIApi_API_KEY} - - from aiohttp_socks import ProxyConnector - import re - - match = re.search(r'socks5://([^:]+):([^@]+)@([^:]+):(\d+)', PROXY_URL) - if match: - username = match.group(1) - password = match.group(2) - host = match.group(3) - port = int(match.group(4)) - connector = ProxyConnector.from_url(f"socks5://{username}:{password}@{host}:{port}") - else: - from urllib.parse import urlparse - parsed = urlparse(PROXY_URL) - connector = ProxyConnector.from_url(f"socks5://{parsed.hostname}:{parsed.port}") - - async with aiohttp.ClientSession(connector=connector) as session: - for attempt in range(max_attempts): - try: - async with session.get(get_url, headers=headers) as response: - if response.status != 200: - await asyncio.sleep(2) - continue - - data = await response.json() - status = data.get('data', {}).get('status') - - if status == 'completed': - output = data.get('data', {}).get('output', {}) - result_url = output.get('url') or output.get('image') - if result_url: - logger.info(f"Задача {task_id} выполнена") - return result_url - elif status == 'failed': - raise Exception("Задача не выполнена") - - await asyncio.sleep(2) - except Exception as e: - logger.error(f"Ошибка опроса: {e}") - await asyncio.sleep(2) - - raise Exception("Превышено время ожидания") - - async def history_command(update: Update, context: ContextTypes.DEFAULT_TYPE): """История обработок""" user = update.effective_user images = await get_user_images(user.id, limit=5) if not images: - await update.message.reply_text("📭 Нет обработанных изображений\n\nОтправьте фото для начала работы!") + await update.message.reply_text("📭 Нет обработанных изображений") return text = "🖼 **История обработок:**\n\n" @@ -442,32 +331,28 @@ async def error_handler(update: Update, context: ContextTypes.DEFAULT_TYPE): async def main(): """Запуск бота""" print("\n" + "=" * 50) - print("🤖 ЗАПУСК TELEGRAM БОТА ЧЕРЕЗ SOCKS5 ПРОКСИ") + print("🤖 ЗАПУСК TELEGRAM БОТА") print("=" * 50) if not PROXY_URL: - print("❌ Прокси не настроен в .env файле!") + print("❌ Прокси не настроен!") return - print(f"🔌 Тип: {PROXY_TYPE.upper()}") + print(f"🔌 Прокси: {PROXY_URL.split('@')[-1] if '@' in PROXY_URL else PROXY_URL}") print("=" * 50) - # Проверяем подключение - if not await test_telegram_connection(): - print("\n❌ НЕТ ПОДКЛЮЧЕНИЯ К TELEGRAM ЧЕРЕЗ ПРОКСИ!") - print("Проверьте настройки прокси в файле .env") - return - - # Инициализация + # Инициализация БД await init_db() - # Создаем приложение - request = get_telegram_request() - if request is None: - print("\n❌ Не удалось настроить прокси для Telegram!") - return + # Создаем приложение с поддержкой SOCKS5 + # python-telegram-bot[socks] автоматически поддерживает SOCKS5 + application = Application.builder().token(TELEGRAM_BOT_TOKEN).build() - application = Application.builder().token(TELEGRAM_BOT_TOKEN).request(request).build() + # Для исходящих запросов бота используем прокси + # Устанавливаем прокси через переменные окружения + os.environ['ALL_PROXY'] = PROXY_URL + os.environ['HTTP_PROXY'] = PROXY_URL + os.environ['HTTPS_PROXY'] = PROXY_URL # Регистрируем обработчики application.add_handler(CommandHandler("start", start)) @@ -491,7 +376,7 @@ async def main(): await application.updater.start_polling() print("\n" + "=" * 50) - print("🤖 БОТ УСПЕШНО ЗАПУЩЕН ЧЕРЕЗ SOCKS5 ПРОКСИ!") + print("🤖 БОТ УСПЕШНО ЗАПУЩЕН!") print("=" * 50) print("\n📱 Отправь команду /test в Telegram\n") diff --git a/debug_bot.py b/debug_bot.py new file mode 100644 index 0000000..ecc7d5e --- /dev/null +++ b/debug_bot.py @@ -0,0 +1,175 @@ +import asyncio +import aiohttp +import sys +from aiohttp_socks import ProxyConnector +import re + +PROXY_URL = "socks5://FYqj9n:n4br4L@185.168.251.163:8000" +TOKEN = "8460953229:AAFZ_1iYLsCTRWR-glNumHxoHXMxTe2-A84" + + +async def check_updates(): + """Проверка получения обновлений""" + print("\n" + "=" * 60) + print("🔍 ДИАГНОСТИКА ПОЛУЧЕНИЯ ОБНОВЛЕНИЙ") + print("=" * 60) + + # Парсим прокси + match = re.search(r'socks5://([^:]+):([^@]+)@([^:]+):(\d+)', PROXY_URL) + if not match: + print("❌ Не удалось распарсить прокси") + return + + username = match.group(1) + password = match.group(2) + host = match.group(3) + port = int(match.group(4)) + + connector = ProxyConnector.from_url(f"socks5://{username}:{password}@{host}:{port}") + + async with aiohttp.ClientSession(connector=connector) as session: + # 1. Проверка базового подключения + print("\n1️⃣ Проверка базового подключения...") + url = f"https://api.telegram.org/bot{TOKEN}/getMe" + async with session.get(url, timeout=10) as resp: + data = await resp.json() + if data.get('ok'): + print(f" ✅ Бот найден: @{data['result']['username']}") + else: + print(f" ❌ Ошибка: {data}") + return + + # 2. Сброс offset (чтобы получить все новые сообщения) + print("\n2️⃣ Сброс offset...") + url = f"https://api.telegram.org/bot{TOKEN}/getUpdates?offset=-1" + async with session.get(url, timeout=10) as resp: + await resp.json() + print(" ✅ Offset сброшен") + + # 3. Проверка получения сообщений (короткий опрос) + print("\n3️⃣ Проверка получения сообщений (короткий опрос)...") + url = f"https://api.telegram.org/bot{TOKEN}/getUpdates?timeout=1" + async with session.get(url, timeout=5) as resp: + data = await resp.json() + if data.get('result'): + print(f" ✅ Найдено {len(data['result'])} сообщений!") + for update in data['result']: + msg = update.get('message', {}) + text = msg.get('text', 'нет текста') + print(f" 📨 Сообщение: {text}") + else: + print(" ⚠️ Нет новых сообщений") + + # 4. ТЕСТ LONG POLLING (30 секунд) + print("\n4️⃣ ТЕСТ LONG POLLING (будет ждать 30 секунд)...") + print(" ⏰ Отправьте команду /test боту в Telegram прямо СЕЙЧАС!") + print(" Ожидание...") + + url = f"https://api.telegram.org/bot{TOKEN}/getUpdates?timeout=30&offset=-1" + try: + async with session.get(url, timeout=35) as resp: + data = await resp.json() + if data.get('result'): + print(f" ✅✅✅ ПОЛУЧЕНО СООБЩЕНИЕ! ✅✅✅") + for update in data['result']: + msg = update.get('message', {}) + text = msg.get('text', 'нет текста') + print(f" 📨 Текст: {text}") + return True + else: + print(" ❌ Нет сообщений за 30 секунд") + return False + except asyncio.TimeoutError: + print(" ❌ ТАЙМАУТ! Long polling не работает через этот прокси") + return False + except Exception as e: + print(f" ❌ Ошибка: {e}") + return False + + +async def test_different_methods(): + """Тест разных методов подключения""" + print("\n" + "=" * 60) + print("🔧 ТЕСТ РАЗНЫХ МЕТОДОВ ПОДКЛЮЧЕНИЯ") + print("=" * 60) + + match = re.search(r'socks5://([^:]+):([^@]+)@([^:]+):(\d+)', PROXY_URL) + if not match: + return + + username = match.group(1) + password = match.group(2) + host = match.group(3) + port = int(match.group(4)) + + methods = [ + ("SOCKS5 с авторизацией", f"socks5://{username}:{password}@{host}:{port}"), + ("SOCKS5 без авторизации", f"socks5://{host}:{port}"), + ("SOCKS5h с авторизацией", f"socks5h://{username}:{password}@{host}:{port}"), + ] + + for name, proxy_url in methods: + print(f"\n📡 Тестируем: {name}") + try: + connector = ProxyConnector.from_url(proxy_url) + async with aiohttp.ClientSession(connector=connector) as session: + url = f"https://api.telegram.org/bot{TOKEN}/getMe" + async with session.get(url, timeout=10) as resp: + if resp.status == 200: + data = await resp.json() + if data.get('ok'): + print(f" ✅ РАБОТАЕТ! Бот: @{data['result']['username']}") + + # Проверяем long polling + print(f" 🔄 Проверяем long polling...") + get_url = f"https://api.telegram.org/bot{TOKEN}/getUpdates?timeout=5" + async with session.get(get_url, timeout=8) as resp2: + print(f" ✅ Long polling тоже работает") + return True + else: + print(f" ❌ Ошибка: статус {resp.status}") + except Exception as e: + print(f" ❌ Ошибка: {str(e)[:50]}") + + return False + + +async def main(): + print("\n" + "=" * 60) + print("🐛 ДИАГНОСТИКА ПРОБЛЕМЫ С БОТОМ") + print("=" * 60) + + # Тест разных методов + method_works = await test_different_methods() + + # Тест получения обновлений + updates_work = await check_updates() + + print("\n" + "=" * 60) + print("📊 РЕЗУЛЬТАТЫ ДИАГНОСТИКИ") + print("=" * 60) + + if not updates_work: + print("\n❌ ПРОБЛЕМА: Long polling НЕ РАБОТАЕТ через этот прокси!") + print("\n🔍 Это означает, что:") + print(" - Бот может отправлять сообщения (исходящие запросы работают)") + print(" - Но НЕ МОЖЕТ получать команды (входящие запросы не работают)") + print("\n💡 РЕШЕНИЯ:") + print(" 1. Использовать Webhook вместо Polling (нужен HTTPS сервер)") + print(" 2. Найти другой прокси, который поддерживает длительные соединения") + print(" 3. Использовать VPN вместо прокси") + print(" 4. Настроить свой собственный прокси (3proxy, Squid)") + + print("\n📌 Быстрое решение для теста:") + print(" Временно отключите прокси и запустите бота без него") + print(" (если в вашем регионе Telegram не заблокирован)") + + elif method_works: + print("\n✅ Найден рабочий метод подключения!") + print(" Используйте соответствующий PROXY_URL в .env") + + print("\n" + "=" * 60) + + +if __name__ == "__main__": + asyncio.run(main()) \ No newline at end of file diff --git a/requirements.txt b/requirements.txt index e306e59..4ec5e4b 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,8 +1,6 @@ -python-telegram-bot[socks]==20.7 +python-telegram-bot==20.7 aiohttp==3.9.1 aiosqlite==0.19.0 python-dotenv==1.0.0 Pillow==10.1.0 -httpx==0.25.2 -aiohttp-socks==0.8.4 -httpx-socks==0.8.0 \ No newline at end of file +aiohttp-socks==0.8.4 \ No newline at end of file diff --git a/test_socks5_longpolling.py b/test_socks5_longpolling.py new file mode 100644 index 0000000..c33c3a4 --- /dev/null +++ b/test_socks5_longpolling.py @@ -0,0 +1,247 @@ +#!/usr/bin/env python3 +""" +Тестирование long polling через SOCKS5 прокси +Проверяет, может ли прокси поддерживать длительные соединения +""" + +import asyncio +import aiohttp +import sys +import logging +from datetime import datetime + +# Настройка логирования +logging.basicConfig( + format='%(asctime)s - %(levelname)s - %(message)s', + level=logging.INFO +) +logger = logging.getLogger(__name__) + +# Конфигурация прокси +PROXY_URL = "socks5://FYqj9n:n4br4L@185.168.251.163:8000" +TELEGRAM_BOT_TOKEN = "8460953229:AAFZ_1iYLsCTRWR-glNumHxoHXMxTe2-A84" + + +async def test_socks5_connection(): + """Тест базового подключения через SOCKS5""" + print("\n" + "=" * 60) + print("ТЕСТ 1: Базовое подключение через SOCKS5") + print("=" * 60) + + try: + from aiohttp_socks import ProxyConnector + + # Парсим прокси + import re + match = re.search(r'socks5://([^:]+):([^@]+)@([^:]+):(\d+)', PROXY_URL) + + if match: + username = match.group(1) + password = match.group(2) + host = match.group(3) + port = int(match.group(4)) + + connector = ProxyConnector.from_url(f"socks5://{username}:{password}@{host}:{port}") + logger.info(f"✅ SOCKS5 прокси настроен: {host}:{port}") + else: + logger.error("❌ Не удалось распарсить прокси URL") + return False + + async with aiohttp.ClientSession(connector=connector) as session: + # Тест 1: Проверка IP через прокси + async with session.get("https://api.ipify.org?format=json", timeout=10) as resp: + data = await resp.json() + logger.info(f"✅ IP через прокси: {data.get('ip')}") + + # Тест 2: Проверка подключения к Telegram API + url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/getMe" + async with session.get(url, timeout=10) as resp: + if resp.status == 200: + data = await resp.json() + if data.get('ok'): + logger.info(f"✅ Подключение к Telegram работает! Бот: @{data['result']['username']}") + return True + else: + logger.error(f"❌ Ошибка API: {data}") + return False + else: + logger.error(f"❌ HTTP {resp.status}") + return False + + except ImportError: + logger.error("❌ Установите aiohttp-socks: pip install aiohttp-socks") + return False + except Exception as e: + logger.error(f"❌ Ошибка: {e}") + return False + + +async def test_long_polling(): + """Тест long polling через SOCKS5""" + print("\n" + "=" * 60) + print("ТЕСТ 2: Long polling через SOCKS5") + print("=" * 60) + print("⚠️ Этот тест будет ждать 30 секунд...") + print(" Отправьте команду /test боту в Telegram в течение 30 секунд\n") + + try: + from aiohttp_socks import ProxyConnector + import re + + match = re.search(r'socks5://([^:]+):([^@]+)@([^:]+):(\d+)', PROXY_URL) + if not match: + logger.error("❌ Не удалось распарсить прокси") + return False + + username = match.group(1) + password = match.group(2) + host = match.group(3) + port = int(match.group(4)) + + connector = ProxyConnector.from_url(f"socks5://{username}:{password}@{host}:{port}") + + async with aiohttp.ClientSession(connector=connector) as session: + # Получаем последнее обновление + get_updates_url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/getUpdates" + + # Сначала сбрасываем offset + async with session.get(f"{get_updates_url}?offset=-1", timeout=10) as resp: + await resp.json() + + # Тестируем long polling с timeout=30 секунд + start_time = datetime.now() + logger.info("🔄 Начинаем long polling (ожидание 30 секунд)...") + + async with session.get(f"{get_updates_url}?timeout=30&offset=-1", timeout=35) as resp: + if resp.status == 200: + data = await resp.json() + elapsed = (datetime.now() - start_time).total_seconds() + + if data.get('result'): + logger.info(f"✅ Получено сообщение через {elapsed:.1f} секунд!") + logger.info(f"📨 Сообщение: {data['result'][0].get('message', {}).get('text')}") + return True + else: + logger.warning(f"⚠️ Long polling завершился через {elapsed:.1f} секунд без сообщений") + logger.warning(" Прокси работает, но сообщений не было получено") + return True # Это НЕ ошибка, просто не было сообщений + else: + logger.error(f"❌ Ошибка HTTP {resp.status}") + return False + + except asyncio.TimeoutError: + logger.error("❌ Таймаут long polling!") + logger.error(" Прокси НЕ поддерживает длительные соединения") + return False + except Exception as e: + logger.error(f"❌ Ошибка: {e}") + return False + + +async def test_alternative_methods(): + """Тест альтернативных методов (телеграф, вебхук)""" + print("\n" + "=" * 60) + print("ТЕСТ 3: Альтернативные методы") + print("=" * 60) + + try: + from aiohttp_socks import ProxyConnector + import re + + match = re.search(r'socks5://([^:]+):([^@]+)@([^:]+):(\d+)', PROXY_URL) + if not match: + return False + + username = match.group(1) + password = match.group(2) + host = match.group(3) + port = int(match.group(4)) + + connector = ProxyConnector.from_url(f"socks5://{username}:{password}@{host}:{port}") + + async with aiohttp.ClientSession(connector=connector) as session: + # Тест: Проверка скорости ответа + urls = [ + ("Telegram API (getMe)", f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/getMe"), + ("Google", "https://google.com"), + ("Cloudflare", "https://cloudflare.com"), + ] + + for name, url in urls: + try: + start = datetime.now() + async with session.get(url, timeout=10) as resp: + elapsed = (datetime.now() - start).total_seconds() + if resp.status == 200: + logger.info(f"✅ {name}: {elapsed:.2f} сек (статус {resp.status})") + else: + logger.warning(f"⚠️ {name}: {elapsed:.2f} сек (статус {resp.status})") + except Exception as e: + logger.error(f"❌ {name}: ошибка - {e}") + + return True + + except Exception as e: + logger.error(f"❌ Ошибка: {e}") + return False + + +async def main(): + """Главная функция""" + print("\n" + "=" * 60) + print("🧪 ТЕСТИРОВАНИЕ SOCKS5 ПРОКСИ ДЛЯ LONG POLLING") + print("=" * 60) + + print(f"\n📡 Прокси: {PROXY_URL.split('@')[-1] if '@' in PROXY_URL else PROXY_URL}") + print(f"🤖 Бот: @ten_may_bot") + + # Тест 1: Базовое подключение + basic_ok = await test_socks5_connection() + + if not basic_ok: + print("\n❌ Базовое подключение не работает!") + print("Проверьте настройки прокси") + return + + # Тест 2: Long polling + long_polling_ok = await test_long_polling() + + # Тест 3: Альтернативные методы + await test_alternative_methods() + + # Итоги + print("\n" + "=" * 60) + print("📊 РЕЗУЛЬТАТЫ ТЕСТИРОВАНИЯ") + print("=" * 60) + + if long_polling_ok: + print("\n✅ SOCKS5 прокси ПОДДЕРЖИВАЕТ long polling!") + print(" Бот должен работать нормально.") + print("\n💡 Запустите бота:") + print(" python bot.py") + else: + print("\n❌ SOCKS5 прокси НЕ ПОДДЕРЖИВАЕТ long polling!") + print("\n⚠️ Это означает, что вы НЕ сможете получать обновления") + print(" через этот прокси в реальном времени.") + print("\n📌 Рекомендации:") + print(" 1. Используйте VPN вместо прокси") + print(" 2. Найдите другой прокси, который поддерживает длительные соединения") + print(" 3. Настройте свой собственный прокси (например, 3proxy)") + print(" 4. Используйте webhook (требуется HTTPS сервер)") + + print("\n" + "=" * 60) + + +if __name__ == "__main__": + # Установка зависимостей если нет + try: + import aiohttp_socks + except ImportError: + print("Устанавливаю aiohttp-socks...") + import subprocess + + subprocess.check_call([sys.executable, "-m", "pip", "install", "aiohttp-socks"]) + print("Установка завершена. Перезапустите тест.\n") + sys.exit(1) + + asyncio.run(main()) \ No newline at end of file