twitch-telegram-notify/bot.py

import logging
import os

import aiohttp
from dotenv import load_dotenv
from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
from telegram.constants import ParseMode
from telegram.ext import Application, CommandHandler, ContextTypes

load_dotenv()

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
log = logging.getLogger("twitch-notify")

# --- Конфигурация из .env ---
TELEGRAM_TOKEN = os.environ["TELEGRAM_TOKEN"]
TWITCH_LOGIN = os.getenv("TWITCH_LOGIN", "theraiwy")  # ник стримера на твиче
# ID чата/канала для оповещений: @username канала или числовой id (напр. -1001234567890)
NOTIFY_CHAT_ID = os.environ["NOTIFY_CHAT_ID"]
CHECK_INTERVAL = int(os.getenv("CHECK_INTERVAL", "60"))  # секунды между проверками

# --- Ссылки для кнопок ---
TWITCH_URL = f"https://twitch.tv/{TWITCH_LOGIN}"
LINKS_URL = "https://links.theraiwy.top"


def _chat_id(raw: str) -> int | str:
    """Числовой id приводим к int, @username оставляем строкой."""
    raw = raw.strip()
    try:
        return int(raw)
    except ValueError:
        return raw


class TwitchStatus:
    """Определение статуса стрима через публичный decapi.me — БЕЗ токенов."""

    BASE = "https://decapi.me/twitch"

    def __init__(self, login: str):
        self.login = login
        self._session: aiohttp.ClientSession | None = None

    async def _ensure_session(self) -> aiohttp.ClientSession:
        if self._session is None or self._session.closed:
            self._session = aiohttp.ClientSession()
        return self._session

    async def _get_text(self, endpoint: str) -> str:
        session = await self._ensure_session()
        url = f"{self.BASE}/{endpoint}/{self.login}"
        async with session.get(url) as resp:
            resp.raise_for_status()
            return (await resp.text()).strip()

    async def get_stream(self) -> dict | None:
        """Возвращает данные стрима, если стример в эфире, иначе None."""
        uptime = await self._get_text("uptime")
        # decapi возвращает "... is offline" когда стрим не идёт
        if "offline" in uptime.lower():
            return None

        title = await self._get_text("title")
        game = await self._get_text("game")
        thumb = (
            f"https://static-cdn.jtvnw.net/previews-ttv/"
            f"live_user_{self.login.lower()}-1280x720.jpg"
        )
        return {
            "user_name": self.login,
            "title": title or "Без названия",
            "game_name": game or "—",
            "thumbnail_url": thumb,
            "uptime": uptime,
        }

    async def close(self):
        if self._session and not self._session.closed:
            await self._session.close()


def build_keyboard() -> InlineKeyboardMarkup:
    """Inline-клавиатура с кнопками-ссылками."""
    return InlineKeyboardMarkup(
        [
            [
                InlineKeyboardButton("🟣 Смотреть на Twitch", url=TWITCH_URL),
                InlineKeyboardButton("🔗 Все ссылки", url=LINKS_URL),
            ]
        ]
    )


def build_caption(stream: dict, test: bool = False) -> str:
    """Собирает текст-подпись (HTML) для оповещения о стриме."""
    title = stream.get("title", "Без названия")
    game = stream.get("game_name", "—")
    user_name = stream.get("user_name", TWITCH_LOGIN)

    prefix = "🧪 [ТЕСТ] " if test else ""
    lines = [
        f"{prefix}🔴 <b>{_esc(user_name)}</b> сейчас в эфире!",
        "",
        f"<a href=\"{TWITCH_URL}\">{_esc(title)}</a>",
        f"🎮 <b>{_esc(game)}</b>",
    ]
    return "\n".join(lines)


def _esc(text: str) -> str:
    """Экранирование под HTML parse mode Telegram."""
    return text.replace("&", "&amp;").replace("<", "&lt;").replace(">", "&gt;")


async def send_announce(bot, stream: dict, test: bool = False):
    """Отправляет оповещение о стриме в целевой чат."""
    caption = build_caption(stream, test=test)
    thumb = stream.get("thumbnail_url", "")
    chat_id = _chat_id(NOTIFY_CHAT_ID)

    try:
        if thumb:
            await bot.send_photo(
                chat_id=chat_id,
                photo=thumb,
                caption=caption,
                parse_mode=ParseMode.HTML,
                reply_markup=build_keyboard(),
            )
        else:
            raise ValueError("no thumbnail")
    except Exception:
        # Превью может быть недоступно/невалидно — падаем в обычное сообщение
        await bot.send_message(
            chat_id=chat_id,
            text=caption,
            parse_mode=ParseMode.HTML,
            reply_markup=build_keyboard(),
            disable_web_page_preview=False,
        )
    log.info("Отправлено оповещение о стриме: %s", stream.get("title"))


async def check_stream(context: ContextTypes.DEFAULT_TYPE):
    """Периодическая задача: опрос статуса и оповещение при выходе в эфир."""
    twitch: TwitchStatus = context.bot_data["twitch"]
    try:
        stream = await twitch.get_stream()
    except Exception:
        log.exception("Ошибка при запросе статуса стрима")
        return

    is_live = stream is not None
    was_live = context.bot_data.get("was_live", False)

    # Оповещаем только при переходе оффлайн -> онлайн
    if is_live and not was_live:
        await send_announce(context.bot, stream)

    context.bot_data["was_live"] = is_live


async def stream_test(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Отправить тестовое оповещение о стриме."""
    fake_stream = {
        "user_name": TWITCH_LOGIN,
        "title": "🧪 Тестовое оповещение — проверка бота",
        "game_name": "Just Chatting",
        "thumbnail_url": (
            f"https://static-cdn.jtvnw.net/previews-ttv/"
            f"live_user_{TWITCH_LOGIN.lower()}-1280x720.jpg"
        ),
    }
    await send_announce(context.bot, fake_stream, test=True)
    log.info("Отправлено тестовое оповещение по команде /stream_test")


async def on_startup(app: Application):
    log.info("Бот запущен, слежу за стримером: %s", TWITCH_LOGIN)


async def on_shutdown(app: Application):
    twitch: TwitchStatus = app.bot_data.get("twitch")
    if twitch:
        await twitch.close()


def main():
    app = Application.builder().token(TELEGRAM_TOKEN).post_init(on_startup).post_shutdown(on_shutdown).build()

    app.bot_data["twitch"] = TwitchStatus(TWITCH_LOGIN)
    app.bot_data["was_live"] = False

    app.add_handler(CommandHandler("stream_test", stream_test))
    app.job_queue.run_repeating(check_stream, interval=CHECK_INTERVAL, first=5)

    app.run_polling(allowed_updates=Update.ALL_TYPES)


if __name__ == "__main__":
    main()