Add graphQl for chat

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))
 

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()