From 61ff90e51b1e6994d3d287ec6778a67ed403a2b8 Mon Sep 17 00:00:00 2001 From: Yuriy Yuriev Date: Tue, 26 May 2026 18:04:52 +0700 Subject: [PATCH] fix to irc --- services/gql_chat_service.py | 307 +++++++++++++++++++++-------------- 1 file changed, 182 insertions(+), 125 deletions(-) diff --git a/services/gql_chat_service.py b/services/gql_chat_service.py index 455885a..f93618f 100644 --- a/services/gql_chat_service.py +++ b/services/gql_chat_service.py @@ -1,65 +1,115 @@ """ -GraphQL HTTP polling сервис для мониторинга Twitch чата. +Twitch чат через IRC over WebSocket (wss://irc-ws.chat.twitch.tv:443). +Замена raw TCP IRC — тот же протокол, WebSocket транспорт. """ import asyncio import logging import re -from typing import Optional, Callable, Awaitable, List, Set +from typing import Optional, Callable, Awaitable, List from datetime import datetime from urllib.parse import urlparse import aiohttp +from config.settings import settings + logger = logging.getLogger(__name__) -_GQL_URL = "https://gql.twitch.tv/gql" -_HEADERS = { - "Client-ID": "kimne78kx3ncx6brgo4mv6wki5h1ko", - "Content-Type": "application/json", -} -_POLL_INTERVAL = 2.0 # секунд между запросами - -_QUERY = """\ -query ChannelChatMessages($login: String!, $after: ID) { - user(login: $login) { - chatRoom { - chatMessages(first: 50, after: $after) { - pageInfo { - endCursor - hasNextPage - } - nodes { - id - sender { - displayName - login - } - content { - text - } - } - } - } - } -}""" +_IRC_WS_URL = "wss://irc-ws.chat.twitch.tv:443" class TwitchGQLChatClient: """ - Twitch чат клиент через GQL HTTP polling. + Twitch IRC over WebSocket клиент. Интерфейс совместим с TwitchIRCClient. """ - def __init__(self, channel: str, target_username: str): + 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._ws: Optional[aiohttp.ClientWebSocketResponse] = None + self._session: Optional[aiohttp.ClientSession] = None + self._connected = False self._url_pattern = re.compile(r'https?://[^\s<>"]+') - self._seen_ids: Set[str] = set() - self._cursor: Optional[str] = None async def connect(self) -> bool: - return True + try: + self._session = aiohttp.ClientSession() + self._ws = await self._session.ws_connect( + _IRC_WS_URL, + timeout=aiohttp.ClientTimeout(total=15), + ) + + 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"✅ IRC WS подключен к #{self.channel}") + return True + + except Exception as e: + logger.error(f"❌ IRC WS ошибка подключения: {e}") + await self._cleanup() + return False + + async def _send(self, message: str) -> None: + if self._ws and not self._ws.closed: + await self._ws.send_str(f"{message}\r\n") + + async def _cleanup(self) -> None: + self._connected = False + try: + if self._ws and not self._ws.closed: + await self._ws.close() + except Exception: + pass + try: + if self._session and not self._session.closed: + await self._session.close() + except Exception: + pass + self._ws = None + self._session = None + + def _parse_message(self, line: str) -> Optional[dict]: + if not line or "PRIVMSG" not in line: + return None + try: + 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", "") + message_text = line.split(" :", 1)[1] if " :" in line else "" + 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]: urls = [] @@ -83,46 +133,6 @@ class TwitchGQLChatClient: except Exception: return False - async def _fetch_messages(self, session: aiohttp.ClientSession) -> List[dict]: - payload = [{ - "operationName": "ChannelChatMessages", - "variables": {"login": self.channel, "after": self._cursor}, - "query": _QUERY, - }] - try: - async with session.post( - _GQL_URL, - json=payload, - headers=_HEADERS, - timeout=aiohttp.ClientTimeout(total=8), - ) as resp: - data = await resp.json() - - result = data[0] if isinstance(data, list) else data - errors = result.get("errors") - if errors: - logger.error(f"GQL schema errors: {errors}") - return [] - - chat_data = ( - result.get("data", {}) - .get("user", {}) - .get("chatRoom", {}) - .get("chatMessages", {}) - ) - if not chat_data: - return [] - - end_cursor = chat_data.get("pageInfo", {}).get("endCursor") - if end_cursor: - self._cursor = end_cursor - - return chat_data.get("nodes", []) - - except Exception as e: - logger.warning(f"GQL fetch error: {e}") - return [] - async def listen_for_messages( self, on_url_found: Callable[[str, str], Awaitable[None]], @@ -133,70 +143,117 @@ class TwitchGQLChatClient: ) -> dict: stats = {"links_found": 0, "messages": 0, "reconnects": 0} start_time = datetime.now() - fail_streak = 0 + last_message_time = datetime.now() + drain_errors = 0 if is_active and not is_active(): - logger.info(f"⏸️ #{self.channel} ждём resume...") + logger.info(f"⏸️ #{self.channel} ждём resume перед подключением...") while is_active and not is_active(): await asyncio.sleep(2) - logger.info(f"🚀 GQL chat polling: #{self.channel}") + if not await self.connect(): + raise ConnectionError(f"IRC WS connect failed for #{self.channel}") - async with aiohttp.ClientSession() as session: - while duration == 0 or (datetime.now() - start_time).seconds < duration: - if is_active and not is_active(): - logger.info(f"⏸️ #{self.channel} пауза...") - while is_active and not is_active(): - await asyncio.sleep(2) - logger.info(f"▶️ #{self.channel} resume") - - messages = await self._fetch_messages(session) - - if messages is not None and len(messages) == 0 and self._cursor is None: - # Первый запрос — cursor ещё не установлен, это норма - fail_streak = 0 - elif not messages: - fail_streak += 1 - if fail_streak >= 10 and on_drain_fail: + while duration == 0 or (datetime.now() - start_time).seconds < duration: + if is_active and not is_active(): + logger.info(f"⏸️ #{self.channel} пауза...") + await self._cleanup() + while is_active and not is_active(): + await asyncio.sleep(2) + logger.info(f"▶️ #{self.channel} resume, переподключение...") + if not await self.connect(): + drain_errors += 1 + if drain_errors >= 3 and on_drain_fail: await on_drain_fail() return stats else: - fail_streak = 0 + drain_errors = 0 + stats["reconnects"] += 1 + last_message_time = datetime.now() + continue - for msg in messages: - msg_id = msg.get("id") - if not msg_id or msg_id in self._seen_ids: - continue - self._seen_ids.add(msg_id) - if len(self._seen_ids) > 10000: - self._seen_ids = set(list(self._seen_ids)[-5000:]) + if not self._connected or self._ws is None or self._ws.closed: + await asyncio.sleep(5) + if not await self.connect(): + 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 - sender = msg.get("sender", {}) - display_name = sender.get("displayName", "unknown") - text = (msg.get("content") or {}).get("text", "") + # Реконнект при отсутствии активности > 6 минут + if (datetime.now() - last_message_time).total_seconds() > 360: + logger.info("🔄 Нет активности 6 мин, переподключение...") + await self._cleanup() + await asyncio.sleep(1) + if not await self.connect(): + 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 - if not text.strip(): - continue + try: + raw = await asyncio.wait_for(self._ws.receive(), timeout=5.0) + except asyncio.CancelledError: + raise + except asyncio.TimeoutError: + continue + except Exception as e: + logger.warning(f"IRC WS receive error: {e}") + self._connected = False + continue - stats["messages"] += 1 - logger.debug(f"GQL msg from {display_name}: {text[:50]}...") + if raw.type in ( + aiohttp.WSMsgType.CLOSING, + aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.ERROR, + ): + logger.warning(f"WS закрыт: {raw.type}") + self._connected = False + continue - for url in self._extract_urls(text): - if self._is_domain_allowed(url, allowed_domains): - stats["links_found"] += 1 - try: - await on_url_found(url, display_name) - except asyncio.CancelledError: - raise - except Exception: - pass + if raw.type != aiohttp.WSMsgType.TEXT: + continue - try: - await asyncio.sleep(_POLL_INTERVAL) - except asyncio.CancelledError: - raise + line = raw.data.strip() + if not line: + continue + + if line.startswith("PING"): + await self._send("PONG :tmi.twitch.tv") + last_message_time = datetime.now() + logger.debug("🏓 PONG") + continue + + last_message_time = datetime.now() + + msg = self._parse_message(line) + if not msg: + continue + + stats["messages"] += 1 + logger.debug(f"IRC WS msg 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: - pass + await self._cleanup()