""" IRC сервис для мониторинга Twitch чата. """ import asyncio import re import logging from typing import Optional, Callable, Awaitable, List from datetime import datetime from urllib.parse import urlparse from config.settings import settings logger = logging.getLogger(__name__) class TwitchIRCClient: """ Twitch IRC клиент для мониторинга чата. """ def __init__( self, channel: str, target_username: str, irc_username: str = None, irc_oauth: str = None ): self.channel = channel.lower() self.target_username = target_username.lower() self.irc_username = irc_username or settings.IRC_USERNAME self.irc_oauth = irc_oauth or settings.IRC_OAUTH self._reader: Optional[asyncio.StreamReader] = None self._writer: Optional[asyncio.StreamWriter] = None self._connected = False self._url_pattern = re.compile(r'https?://[^\s<>"]+') async def connect(self) -> bool: """Подключение к IRC серверу.""" try: import ssl ssl_context = ssl.create_default_context() self._reader, self._writer = await asyncio.wait_for( asyncio.open_connection( settings.IRC_SERVER, settings.IRC_PORT, ssl=ssl_context ), timeout=15.0 ) await self._send(f"PASS {self.irc_oauth}") await self._send(f"NICK {self.irc_username}") await self._send("CAP REQ :twitch.tv/tags") await self._send("CAP REQ :twitch.tv/commands") await self._send(f"JOIN #{self.channel}") await asyncio.sleep(2) self._connected = True logger.info(f"✅ Подключен к #{self.channel}") return True except Exception as e: logger.error(f"❌ Ошибка подключения: {e}") return False async def _send(self, message: str) -> None: """Отправка сообщения.""" if self._writer: self._writer.write(f"{message}\r\n".encode()) await self._writer.drain() async def _read_line(self) -> Optional[str]: """Чтение строки с обработкой PING.""" if not self._reader: return None try: line = await self._reader.readline() if not line: logger.warning("Empty line - connection closed") self._connected = False return None decoded = line.decode('utf-8', errors='ignore').strip() # PING от Twitch — отвечаем СРАЗУ, возвращаем None чтобы не путать с нашим PONG if decoded.startswith('PING'): pong_msg = f"PONG {decoded.split()[1]}" self._writer.write(f"{pong_msg}\r\n".encode()) 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 # не пробрасываем в основной цикл return decoded except Exception as e: logger.error(f"Read error: {e}") self._connected = False return None def _parse_message(self, line: str) -> Optional[dict]: if not line or 'PRIVMSG' not in line: return None try: # Парсим display-name из тегов tags = {} if line.startswith('@'): tags_part = line.split(' ', 1)[0] for tag in tags_part.lstrip('@').split(';'): if '=' in tag: k, v = tag.split('=', 1) tags[k] = v display_name = tags.get('display-name', '') # Извлекаем сообщение if ' :' in line: message_text = line.split(' :', 1)[1] else: message_text = '' # Пропускаем пустые сообщения if not message_text.strip(): return None return { 'display_name': display_name or 'unknown', 'message': message_text, 'tags': tags } except Exception as e: logger.error(f"Parse error: {e}") return None def _extract_urls(self, text: str) -> List[str]: """Извлечение URL из текста.""" urls = [] for raw_url in self._url_pattern.findall(text): try: parsed = urlparse(raw_url) clean = f"{parsed.scheme}://{parsed.netloc}{parsed.path}" if parsed.query: clean += f"?{parsed.query}" urls.append(clean) except Exception: urls.append(raw_url.strip()) return urls def _is_domain_allowed(self, url: str, allowed_domains: List[str]) -> bool: """Проверка домена.""" if not allowed_domains: return True try: domain = urlparse(url).netloc.lower() return any(d.lower() in domain for d in allowed_domains) except Exception: return False async def listen_for_messages( self, on_url_found: Callable[[str, str], Awaitable[None]], duration: int, allowed_domains: Optional[List[str]] = None, is_active: Callable[[], bool] = None, on_drain_fail: Callable[[], Awaitable[None]] = None, ) -> dict: stats = {"links_found": 0, "messages": 0, "reconnects": 0} 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 if is_active and not is_active(): logger.info(f"⏸️ #{self.channel} waiting for resume before connecting...") while is_active and not is_active(): await asyncio.sleep(2) if not is_active and not is_active(): return stats logger.info(f"▶️ #{self.channel} resumed, connecting to IRC") if not await self.connect(): raise ConnectionError(f"IRC connect failed for #{self.channel}") while duration == 0 or (datetime.now() - start_time).seconds < duration: # === ПРОВЕРКА АКТИВНОСТИ === if is_active and not is_active(): logger.info(f"⏸️ #{self.channel} paused, waiting for resume...") await self.disconnect() while is_active and not is_active(): await asyncio.sleep(2) 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 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: logger.info("🔄 No activity 5min, reconnecting...") try: await self.disconnect() except Exception: pass await asyncio.sleep(1) connected = await self.connect() if not connected: drain_errors += 1 if drain_errors >= 3 and on_drain_fail: await on_drain_fail() return stats else: drain_errors = 0 stats["reconnects"] += 1 last_message_time = datetime.now() continue # Читаем строку try: line = await asyncio.wait_for(self._read_line(), timeout=5.0) except asyncio.CancelledError: raise except asyncio.TimeoutError: continue except Exception as e: logger.warning(f"IRC read error: {e}") continue if not line: await asyncio.sleep(0.5) continue # Twitch отвечает: ":tmi.twitch.tv PONG tmi.twitch.tv :..." — не startswith if 'PONG' in line: last_message_time = datetime.now() waiting_pong = False continue msg = self._parse_message(line) if not msg: continue last_message_time = datetime.now() # Обновляем таймер stats["messages"] += 1 logger.debug(f"IRC message from {msg['display_name']}: {msg['message'][:50]}...") for url in self._extract_urls(msg['message']): if self._is_domain_allowed(url, allowed_domains): stats["links_found"] += 1 try: await on_url_found(url, msg['display_name']) except asyncio.CancelledError: raise except Exception: pass return stats async def disconnect(self) -> None: """Закрытие соединения.""" self._connected = False if self._writer: try: self._writer.close() await self._writer.wait_closed() except (ConnectionResetError, BrokenPipeError, OSError) as e: logger.debug(f"Disconnect warning: {e}") self._writer = None self._reader = None