Почему-то перестало работать подключение к прокси через telethon, хотя буквально пару недель назад все было без проблем
main - ERROR - ❌ Connection timeout after 15 seconds - вот что выводит, пробовал и socks5, и http, также попробовал другое приложение Telegram API, ничего не помогло. Раньше с подключением проблем не было, кто-нибудь знает, с чем может быть связано?
from telethon.network.connection.tcpfull import ConnectionTcpFull
from python_socks.async_.asyncio import Proxy
from python_socks import ProxyType
import asyncio
import socket
import traceback
class AsyncioSocksProxyConnection(ConnectionTcpFull):
async def _connect(self, timeout=None, ssl=None):
try:
proxy = self._proxy
proxy_type = proxy.get('proxy_type', 'socks5')
addr = proxy['addr']
port = proxy['port']
username = proxy.get('username')
password = proxy.get('password')
logger.info(f"Creating proxy connection to {addr}:{port} with type {proxy_type}")
proxy_types = {
'socks5': ProxyType.SOCKS5,
'socks4': ProxyType.SOCKS4,
'http': ProxyType.HTTP,
'https': ProxyType.HTTP
}
proxy_type = proxy_types.get(proxy_type.lower(), ProxyType.SOCKS5)
logger.info(f"Creating proxy object with type {proxy_type}")
proxy_obj = Proxy.create(
proxy_type=proxy_type,
host=addr,
port=port,
username=username,
password=password,
rdns=True
)
socket.setdefaulttimeout(timeout or 30)
logger.info(f"Attempting to connect to {self._ip}:{self._port} through proxy")
try:
self._reader, self._writer = await asyncio.wait_for(
proxy_obj.connect(
dest_host=self._ip,
dest_port=self._port
),
timeout=timeout or 30
)
logger.info("Successfully established proxy connection")
return self._reader, self._writer
except asyncio.TimeoutError:
logger.error("Connection timeout while trying to establish proxy connection")
raise
except Exception as e:
logger.error(f"Error during proxy connection: {str(e)}")
raise
except Exception as e:
logger.error(f"Error in proxy connection: {str(e)}")
logger.error(f"Error type: {type(e).__name__}")
logger.error(f"Error details: {traceback.format_exc()}")
raise
finally:
socket.setdefaulttimeout(None)
async def check_proxy(proxy, session_file=None):
try:
logger.info(f"\n? === STARTING PROXY CHECK ===")
logger.info(f"Proxy: {proxy}")
logger.info(f"Session file: {session_file}")
parts = proxy.split(':')
if len(parts) < 4:
raise ValueError("Invalid proxy format. Expected: type:ip:port:username:password")
proxy_type = parts[0].lower()
ip = parts[1]
port = int(parts[2])
username = parts[3]
password = parts[4] if len(parts) > 4 else None
logger.info(f"Parsed proxy parts ({len(parts)} parts): type={proxy_type}, ip={ip}, port={port}")
proxy_types = {
'socks5': ProxyType.SOCKS5,
'socks4': ProxyType.SOCKS4,
'http': ProxyType.HTTP,
'https': ProxyType.HTTP
}
proxy_type = proxy_types.get(proxy_type, ProxyType.SOCKS5)
logger.info(f"Converted proxy type from {parts[0]} to {proxy_type}")
proxy_dict = {
'proxy_type': proxy_type,
'addr': ip,
'port': port,
'username': username,
'password': password,
'rdns': True
}
logger.info(f"Created proxy dictionary: {proxy_dict}")
if session_file:
logger.info(f"Testing proxy with session file: {session_file}")
try:
device_info = generate_device_info()
logger.info(f"Generated device info: {device_info}")
client = TelegramClient(
f'sessions/{session_file}',
api_id=os.getenv('API_ID'),
api_hash=os.getenv('API_HASH'),
device_model=device_info['device_model'],
system_version=device_info['system_version'],
app_version=device_info['app_version'],
lang_code=device_info['lang_code'],
system_lang_code=device_info['system_lang_code'],
proxy=proxy_dict,
connection=AsyncioSocksProxyConnection,
connection_retries=5,
retry_delay=2,
timeout=15,
auto_reconnect=True
)
logger.info("Created TelegramClient instance")
try:
logger.info("Attempting to connect to Telegram...")
await asyncio.wait_for(client.connect(), timeout=15)
logger.info("Successfully connected to Telegram")
me = await client.get_me()
logger.info(f"Successfully made API call. Connected as: {me.first_name} (@{me.username})")
await client.disconnect()
logger.info("Successfully disconnected")
return True
except Exception as e:
logger.error(f"❌ Connection failed: {str(e)}")
logger.error(f"Error type: {type(e).__name__}")
logger.error(f"Error details: {traceback.format_exc()}")
return False
except Exception as e:
logger.error(f"Error during proxy test: {str(e)}")
logger.error(f"Error type: {type(e).__name__}")
logger.error(f"Error details: {traceback.format_exc()}")
return False
else:
logger.info("No session file provided, returning proxy config only")
return proxy_dict
except Exception as e:
logger.error(f"Error in check_proxy: {str(e)}")
logger.error(f"Error type: {type(e).__name__}")
logger.error(f"Error details: {traceback.format_exc()}")
return False