diff --git a/handlers/commands.py b/handlers/commands.py index 3a313a4..3bfc99b 100644 --- a/handlers/commands.py +++ b/handlers/commands.py @@ -2226,9 +2226,9 @@ class BotInterface: await self._track_msg(params.user_id, msg.message_id) async def _run_twitch(self, task_id: str, params: TaskParams, message: Message): - from services.irc_service import TwitchIRCClient + from services.gql_chat_service import TwitchGQLChatClient - irc = TwitchIRCClient(params.channel, params.target_username) + irc = TwitchGQLChatClient(params.channel, params.target_username) url_tasks: set = set() async def on_url(url, username): @@ -2248,10 +2248,10 @@ class BotInterface: await self.task_manager.pause_task(task_id) await send_message_safe( message.bot, params.chat_id, - f"⚠️ IRC #{params.channel}: ошибка соединения (3 попытки) — задача поставлена на паузу" + f"⚠️ GQL #{params.channel}: ошибка соединения (3 попытки) — задача поставлена на паузу" ) - logger.info(f"🚀 IRC monitor: {params.channel}") + logger.info(f"🚀 GQL chat monitor: {params.channel}") watcher = asyncio.create_task(self._stream_watcher(task_id, params, message)) diff --git a/services/gql_chat_service.py b/services/gql_chat_service.py new file mode 100644 index 0000000..46e221a --- /dev/null +++ b/services/gql_chat_service.py @@ -0,0 +1,263 @@ +""" +GraphQL WebSocket сервис для мониторинга Twitch чата. +Замена IRC — не требует OAuth токена. +""" + +import asyncio +import json +import logging +import re +from typing import Optional, Callable, Awaitable, List +from datetime import datetime +from urllib.parse import urlparse + +import aiohttp + +logger = logging.getLogger(__name__) + +_GQL_WS_URL = "wss://gql.twitch.tv/connection/websocket" +_CLIENT_ID = "kimne78kx3ncx6brgo4mv6wki5h1ko" + +_CHAT_SUBSCRIPTION = """\ +subscription ChatMessages($channelLogin: String!) { + chatEvents(input: { channelLogin: $channelLogin }) { + ... on ChatMessageEvent { + message { + sender { + displayName + login + } + content { + text + fragments { + ... on TextFragment { + text + } + } + } + } + } + } +}""" + + +class TwitchGQLChatClient: + """ + Twitch чат клиент через GQL WebSocket. + Интерфейс совместим с TwitchIRCClient. + """ + + def __init__(self, channel: str, target_username: str): + self.channel = channel.lower() + self.target_username = target_username.lower() + self._ws: Optional[aiohttp.ClientWebSocketResponse] = None + self._session: Optional[aiohttp.ClientSession] = None + self._connected = False + self._url_pattern = re.compile(r'https?://[^\s<>"]+') + + async def connect(self) -> bool: + try: + self._session = aiohttp.ClientSession() + self._ws = await self._session.ws_connect( + _GQL_WS_URL, + protocols=["graphql-ws"], + headers={"Client-ID": _CLIENT_ID}, + timeout=aiohttp.ClientTimeout(total=15), + ) + + await self._ws.send_str(json.dumps({ + "type": "connection_init", + "payload": { + "Authorization": "undefined", + "Client-ID": _CLIENT_ID, + } + })) + + ack = await asyncio.wait_for(self._ws.receive(), timeout=10) + msg = json.loads(ack.data) + if msg.get("type") != "connection_ack": + logger.error(f"GQL WS: unexpected init response: {msg}") + await self._cleanup() + return False + + await self._ws.send_str(json.dumps({ + "type": "start", + "id": "chat", + "payload": { + "query": _CHAT_SUBSCRIPTION, + "variables": {"channelLogin": self.channel}, + "operationName": "ChatMessages", + } + })) + + self._connected = True + logger.info(f"✅ GQL подключен к #{self.channel}") + return True + + except Exception as e: + logger.error(f"❌ GQL ошибка подключения: {e}") + await self._cleanup() + return False + + 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 _extract_urls(self, text: str) -> List[str]: + 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() + reconnect_errors = 0 + + if is_active and not is_active(): + logger.info(f"⏸️ #{self.channel} ждём resume перед подключением...") + while is_active and not is_active(): + await asyncio.sleep(2) + + if not await self.connect(): + raise ConnectionError(f"GQL 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} пауза...") + 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(): + reconnect_errors += 1 + if reconnect_errors >= 3 and on_drain_fail: + await on_drain_fail() + return stats + else: + reconnect_errors = 0 + stats["reconnects"] += 1 + continue + + if not self._connected or self._ws is None or self._ws.closed: + await asyncio.sleep(5) + if not await self.connect(): + reconnect_errors += 1 + if reconnect_errors >= 3 and on_drain_fail: + await on_drain_fail() + return stats + else: + reconnect_errors = 0 + stats["reconnects"] += 1 + continue + + try: + raw = await asyncio.wait_for(self._ws.receive(), timeout=30.0) + except asyncio.CancelledError: + raise + except asyncio.TimeoutError: + continue + except Exception as e: + logger.warning(f"GQL receive error: {e}") + self._connected = False + continue + + if raw.type in ( + aiohttp.WSMsgType.CLOSING, + aiohttp.WSMsgType.CLOSED, + aiohttp.WSMsgType.ERROR, + ): + logger.warning(f"WS закрыт: {raw.type}") + self._connected = False + continue + + if raw.type != aiohttp.WSMsgType.TEXT: + continue + + try: + msg = json.loads(raw.data) + except Exception: + continue + + msg_type = msg.get("type") + + if msg_type == "ka": # keep-alive от Twitch + continue + + if msg_type == "connection_error": + logger.error(f"GQL connection error: {msg}") + self._connected = False + continue + + if msg_type != "data": + continue + + payload = msg.get("payload", {}) + data = payload.get("data", {}) + chat_event = data.get("chatEvents") + if not chat_event: + continue + + message_obj = chat_event.get("message", {}) + if not message_obj: + continue + + sender = message_obj.get("sender", {}) + display_name = sender.get("displayName", "unknown") + content = message_obj.get("content", {}) + text = content.get("text", "") + + if not text.strip(): + continue + + stats["messages"] += 1 + logger.debug(f"GQL message from {display_name}: {text[:50]}...") + + 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 + + return stats + + async def disconnect(self) -> None: + await self._cleanup()