Yuriy Yuriev
48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
"""Получение числа зрителей Twitch через публичный GQL эндпоинт."""
|
|
|
|
import logging
|
|
import aiohttp
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_GQL_URL = "https://gql.twitch.tv/gql"
|
|
_HEADERS = {
|
|
"Client-ID": "kimne78kx3ncx6brgo4mv6wki5h1ko",
|
|
"Content-Type": "application/json",
|
|
}
|
|
_QUERY = """
|
|
query ($login: String!) {
|
|
user(login: $login) {
|
|
stream {
|
|
viewersCount
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
async def get_viewer_count(channel: str) -> int | None:
|
|
"""
|
|
Возвращает число зрителей (0 = оффлайн, >0 = онлайн).
|
|
Возвращает None при ошибке запроса — статус неизвестен, не менять состояние.
|
|
"""
|
|
payload = {"query": _QUERY, "variables": {"login": channel.lower()}}
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.post(
|
|
_GQL_URL,
|
|
json=payload,
|
|
headers=_HEADERS,
|
|
timeout=aiohttp.ClientTimeout(total=8),
|
|
) as resp:
|
|
data = await resp.json()
|
|
|
|
stream = data.get("data", {}).get("user", {}).get("stream")
|
|
count = stream.get("viewersCount", 0) if stream else 0
|
|
logger.debug(f"Viewers on {channel}: {count}")
|
|
return count
|
|
|
|
except Exception as e:
|
|
logger.warning(f"GQL request failed for {channel}: {e}")
|
|
return None # ошибка сети — не трогаем состояние задачи
|