#!/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())