diff --git a/handlers/commands.py b/handlers/commands.py index 96ceded..fbfcdc2 100644 --- a/handlers/commands.py +++ b/handlers/commands.py @@ -266,6 +266,20 @@ class BotInterface: await interface._show_status(callback.message, edit=True) await callback.answer(f"✅ Прокси перезагружены: {count} шт.") + @dp.callback_query(F.data == "check_proxies") + async def cb_check_proxies(callback: CallbackQuery): + if not await require_admin(callback): + return + total = interface.proxy_manager.count + if total == 0: + await callback.answer("❌ Нет загруженных прокси", show_alert=True) + return + await callback.message.edit_text( + f"⏳ Проверка {total} прокси через туннель...\n\nЭто займёт несколько секунд." + ) + await callback.answer() + asyncio.create_task(interface._check_all_proxies(callback.message)) + @dp.callback_query(F.data == "bot_restart_confirm") async def cb_restart_confirm(callback: CallbackQuery): if not await require_admin(callback): @@ -1687,6 +1701,7 @@ class BotInterface: builder = InlineKeyboardBuilder() builder.row(InlineKeyboardButton(text="🔄 Обновить", callback_data="menu_status")) builder.row(InlineKeyboardButton(text="🔃 Перезагрузить прокси", callback_data="reload_proxies")) + builder.row(InlineKeyboardButton(text="🔍 Проверить прокси", callback_data="check_proxies")) builder.row(InlineKeyboardButton(text="🔁 Перезапустить бота", callback_data="bot_restart_confirm")) builder.row(InlineKeyboardButton(text="🔙 В меню", callback_data="menu_main")) await self._edit_or_send(message, text, builder.as_markup(), edit) @@ -1703,6 +1718,87 @@ class BotInterface: finally: await asyncio.sleep(0.5) os.execv(sys.executable, [sys.executable] + sys.argv) + + async def _check_all_proxies(self, message) -> None: + """Проверяет все загруженные прокси через туннель и редактирует сообщение с результатами.""" + from services.socks5_to_http_proxy import Socks5ToHttpProxy + from managers.proxy_manager import ProxyType + + proxies = list(self.proxy_manager._proxies) + if not proxies: + builder = InlineKeyboardBuilder() + builder.row(InlineKeyboardButton(text="🔙 К статусу", callback_data="menu_status")) + try: + await message.edit_text("❌ Нет загруженных прокси", reply_markup=builder.as_markup()) + except Exception: + pass + return + + sem = asyncio.Semaphore(5) + results: dict[str, bool] = {} + + async def _check_one(proxy): + async with sem: + try: + if proxy.proxy_type == ProxyType.SOCKS5: + tunnel = Socks5ToHttpProxy( + socks5_host=proxy.ip, + socks5_port=int(proxy.port), + username=proxy.login, + password=proxy.password, + ) + await tunnel.start() + try: + ok = await tunnel.check_connection(timeout=8.0) + finally: + await tunnel.stop() + else: + from aiohttp import ClientSession, ClientTimeout + proxy_url = proxy.server + if proxy.login and proxy.password: + from urllib.parse import urlparse + parsed = urlparse(proxy_url) + proxy_url = f"{parsed.scheme}://{proxy.login}:{proxy.password}@{parsed.netloc}" + async with ClientSession(timeout=ClientTimeout(total=8)) as s: + async with s.get( + "https://www.google.com/generate_204", + proxy=proxy_url, + allow_redirects=False, + ) as resp: + ok = resp.status in (200, 204) + except Exception: + ok = False + results[proxy.id] = ok + + await asyncio.gather(*[_check_one(p) for p in proxies], return_exceptions=True) + + ok_ids = [pid for pid, ok in results.items() if ok] + fail_ids = [pid for pid, ok in results.items() if not ok] + + lines = [ + f"🔍 Проверка прокси завершена\n", + f"✅ Рабочих: {len(ok_ids)} / {len(proxies)}", + f"❌ Нерабочих: {len(fail_ids)}\n", + ] + if ok_ids: + lines.append("✅ Работают:") + for pid in ok_ids[:15]: + lines.append(f" • {pid}") + if len(ok_ids) > 15: + lines.append(f" ...ещё {len(ok_ids) - 15}") + if fail_ids: + lines.append("\n❌ Не работают:") + for pid in fail_ids[:15]: + lines.append(f" • {pid}") + if len(fail_ids) > 15: + lines.append(f" ...ещё {len(fail_ids) - 15}") + + builder = InlineKeyboardBuilder() + builder.row(InlineKeyboardButton(text="🔙 К статусу", callback_data="menu_status")) + try: + await message.edit_text("\n".join(lines), reply_markup=builder.as_markup()) + except Exception as e: + logger.error(f"Proxy check result edit failed: {e}") # ========================================================================= # УПРАВЛЕНИЕ ЗАДАЧЕЙ diff --git a/services/irc_service.py b/services/irc_service.py index fefd7a0..4bf20b0 100644 --- a/services/irc_service.py +++ b/services/irc_service.py @@ -85,16 +85,15 @@ class TwitchIRCClient: decoded = line.decode('utf-8', errors='ignore').strip() - # PING от Twitch — отвечаем СРАЗУ, возвращаем None чтобы не путать с нашим PONG + # PING от Twitch — отвечаем СРАЗУ и возвращаем строку (обновит last_message_time) if decoded.startswith('PING'): - pong_msg = f"PONG {decoded.split()[1]}" - self._writer.write(f"{pong_msg}\r\n".encode()) + self._writer.write(b"PONG :tmi.twitch.tv\r\n") try: await self._writer.drain() except (ConnectionResetError, BrokenPipeError, OSError) as e: logger.warning(f"Failed to send PONG: {e}") - logger.debug(f"🏓 {pong_msg}") - return None # не пробрасываем в основной цикл + logger.debug("🏓 PONG :tmi.twitch.tv") + return decoded # возвращаем чтобы основной цикл обновил last_message_time return decoded @@ -176,11 +175,6 @@ class TwitchIRCClient: start_time = datetime.now() last_message_time = datetime.now() - last_ping_time = datetime.now() - ping_interval = 120 # пинговать каждые 2 минуты - pong_timeout = 30 # ждать PONG 30 сек, иначе реконнект - waiting_pong = False - ping_sent_at: Optional[float] = None drain_errors = 0 # Если задача стартует на паузе — ждём resume до подключения к IRC @@ -205,40 +199,19 @@ class TwitchIRCClient: logger.info(f"▶️ #{self.channel} resumed, reconnecting") await self.connect() last_message_time = datetime.now() - last_ping_time = datetime.now() - waiting_pong = False continue # Нет соединения — ждём перед повтором if not self._connected: await asyncio.sleep(5) await self.connect() - last_ping_time = datetime.now() - waiting_pong = False + last_message_time = datetime.now() continue now = datetime.now() - # === PONG не пришёл вовремя — соединение мёртвое, реконнект === - if waiting_pong and ping_sent_at and (now.timestamp() - ping_sent_at) > pong_timeout: - logger.warning(f"🔌 #{self.channel} PONG timeout, reconnecting...") - waiting_pong = False - self._connected = False - continue - - # === Отправляем PING каждые ping_interval секунд === - if not waiting_pong and (now - last_ping_time).seconds >= ping_interval: - try: - await self._send("PING :tmi.twitch.tv") - ping_sent_at = now.timestamp() - waiting_pong = True - last_ping_time = now - except Exception: - self._connected = False - continue - - # === Реконнект если вообще нет активности > 5 минут (failsafe) === - if (now - last_message_time).seconds > 300: + # === Реконнект если нет активности > 6 минут (Twitch PING каждые ~5 мин) === + if (now - last_message_time).total_seconds() > 360: logger.info("🔄 No activity 5min, reconnecting...") try: await self.disconnect() @@ -272,10 +245,8 @@ class TwitchIRCClient: await asyncio.sleep(0.5) continue - # Twitch отвечает: ":tmi.twitch.tv PONG tmi.twitch.tv :..." — не startswith - if 'PONG' in line: + if 'PONG' in line or line.startswith('PING'): last_message_time = datetime.now() - waiting_pong = False continue msg = self._parse_message(line) diff --git a/services/socks5_to_http_proxy.py b/services/socks5_to_http_proxy.py index 2170f35..fc109ab 100644 --- a/services/socks5_to_http_proxy.py +++ b/services/socks5_to_http_proxy.py @@ -44,7 +44,31 @@ class Socks5ToHttpProxy: @property def proxy_config_for_browser(self) -> dict: return {'server': self.local_url} - + + async def check_connection( + self, + test_url: str = "https://www.google.com/generate_204", + timeout: float = 10.0, + ) -> bool: + """Проверяет связность туннеля: делает запрос через локальный HTTP прокси.""" + if not self._server or not self._port: + return False + try: + from aiohttp import ClientSession, ClientTimeout + from aiohttp import TCPConnector + proxy_url = self.local_url + async with ClientSession( + connector=TCPConnector(), + timeout=ClientTimeout(total=timeout), + ) as session: + async with session.get(test_url, proxy=proxy_url, allow_redirects=False) as resp: + ok = resp.status in (200, 204) + logger.debug(f"Tunnel check {self.socks5_host}:{self.socks5_port} → {resp.status} ok={ok}") + return ok + except Exception as e: + logger.debug(f"Tunnel check failed {self.socks5_host}:{self.socks5_port}: {e}") + return False + async def start(self) -> str: """Запускает прокси сервер.""" self._port = self._find_free_port() @@ -404,22 +428,36 @@ class Socks5ProxyPool: if key in self._last_used: self._last_used[key] = 0 + async def check_proxy( + self, + socks5_proxy, + test_url: str = "https://www.google.com/generate_204", + timeout: float = 10.0, + ) -> bool: + """Запускает туннель для прокси (если не запущен) и проверяет связность.""" + await self.get_proxy_config(socks5_proxy) + async with self._lock: + tunnel = self._proxies.get(socks5_proxy.id) + if not tunnel: + return False + return await tunnel.check_connection(test_url=test_url, timeout=timeout) + async def stop_all(self): """Останавливает все прокси с таймаутом.""" if self._cleanup_task: self._cleanup_task.cancel() - + async with self._lock: proxies = list(self._proxies.values()) self._proxies.clear() self._last_used.clear() - + for proxy in proxies: try: await asyncio.wait_for(proxy.stop(), timeout=3) except (asyncio.TimeoutError, Exception): pass - + logger.info("All SOCKS5 proxies stopped") @property