Click/services/gql_chat_service.py

"""
GraphQL HTTP polling сервис для мониторинга Twitch чата.
"""

import asyncio
import logging
import re
from typing import Optional, Callable, Awaitable, List, Set
from datetime import datetime
from urllib.parse import urlparse

import aiohttp

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
          }
        }
      }
    }
  }
}"""


class TwitchGQLChatClient:
    """
    Twitch чат клиент через GQL HTTP polling.
    Интерфейс совместим с TwitchIRCClient.
    """

    def __init__(self, channel: str, target_username: str):
        self.channel = channel.lower()
        self.target_username = target_username.lower()
        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

    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 _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]],
        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()
        fail_streak = 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)

        logger.info(f"🚀 GQL chat polling: #{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:
                        await on_drain_fail()
                        return stats
                else:
                    fail_streak = 0

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

                    sender = msg.get("sender", {})
                    display_name = sender.get("displayName", "unknown")
                    text = (msg.get("content") or {}).get("text", "")

                    if not text.strip():
                        continue

                    stats["messages"] += 1
                    logger.debug(f"GQL msg 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:
        pass