From 493c67cb1d46308025842714c2a4d2d3fbb6f2f3 Mon Sep 17 00:00:00 2001 From: Yuriy Yuriev Date: Tue, 26 May 2026 18:01:07 +0700 Subject: [PATCH] fix gql --- services/gql_chat_service.py | 293 ++++++++++++++--------------------- 1 file changed, 116 insertions(+), 177 deletions(-) diff --git a/services/gql_chat_service.py b/services/gql_chat_service.py index e06e2ac..455885a 100644 --- a/services/gql_chat_service.py +++ b/services/gql_chat_service.py @@ -1,13 +1,11 @@ """ -GraphQL WebSocket сервис для мониторинга Twitch чата. -Замена IRC — не требует OAuth токена. +GraphQL HTTP polling сервис для мониторинга Twitch чата. """ import asyncio -import json import logging import re -from typing import Optional, Callable, Awaitable, List +from typing import Optional, Callable, Awaitable, List, Set from datetime import datetime from urllib.parse import urlparse @@ -15,24 +13,30 @@ import aiohttp logger = logging.getLogger(__name__) -_GQL_WS_URL = "wss://gql.twitch.tv/gql" -_CLIENT_ID = "kimne78kx3ncx6brgo4mv6wki5h1ko" +_GQL_URL = "https://gql.twitch.tv/gql" +_HEADERS = { + "Client-ID": "kimne78kx3ncx6brgo4mv6wki5h1ko", + "Content-Type": "application/json", +} +_POLL_INTERVAL = 2.0 # секунд между запросами -_CHAT_SUBSCRIPTION = """\ -subscription ChatMessages($channelLogin: String!) { - chatEvents(input: { channelLogin: $channelLogin }) { - ... on ChatMessageEvent { - message { - sender { - displayName - login +_QUERY = """\ +query ChannelChatMessages($login: String!, $after: ID) { + user(login: $login) { + chatRoom { + chatMessages(first: 50, after: $after) { + pageInfo { + endCursor + hasNextPage } - content { - text - fragments { - ... on TextFragment { - text - } + nodes { + id + sender { + displayName + login + } + content { + text } } } @@ -43,76 +47,19 @@ subscription ChatMessages($channelLogin: String!) { class TwitchGQLChatClient: """ - Twitch чат клиент через GQL WebSocket. + Twitch чат клиент через GQL HTTP polling. Интерфейс совместим с 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<>"]+') + self._seen_ids: Set[str] = set() + self._cursor: Optional[str] = None 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 + return True def _extract_urls(self, text: str) -> List[str]: urls = [] @@ -136,6 +83,46 @@ 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]], @@ -146,118 +133,70 @@ class TwitchGQLChatClient: ) -> dict: stats = {"links_found": 0, "messages": 0, "reconnects": 0} start_time = datetime.now() - reconnect_errors = 0 + fail_streak = 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) - if not await self.connect(): - raise ConnectionError(f"GQL connect failed for #{self.channel}") + logger.info(f"🚀 GQL chat polling: #{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: + 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: await on_drain_fail() return stats else: - reconnect_errors = 0 - stats["reconnects"] += 1 - continue + fail_streak = 0 - 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 + 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:]) - 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 + sender = msg.get("sender", {}) + display_name = sender.get("displayName", "unknown") + text = (msg.get("content") or {}).get("text", "") - if raw.type in ( - aiohttp.WSMsgType.CLOSING, - aiohttp.WSMsgType.CLOSED, - aiohttp.WSMsgType.ERROR, - ): - logger.warning(f"WS закрыт: {raw.type}") - self._connected = False - continue + if not text.strip(): + continue - if raw.type != aiohttp.WSMsgType.TEXT: - continue + stats["messages"] += 1 + logger.debug(f"GQL msg from {display_name}: {text[:50]}...") - try: - msg = json.loads(raw.data) - except Exception: - 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 - 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 + try: + await asyncio.sleep(_POLL_INTERVAL) + except asyncio.CancelledError: + raise return stats async def disconnect(self) -> None: - await self._cleanup() + pass