Twitch stream notify bot for Telegram with Docker

Telegram port of the Discord notifier: tokenless decapi.me status checks, photo+caption announce with inline link buttons, /stream_test command. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-13 11:39:33 +03:00
commit c63b51ff79
8 changed files with 356 additions and 0 deletions
@@ -0,0 +1,8 @@
+.env
+.venv/
+venv/
+__pycache__/
+*.pyc
+.git/
+.gitignore
+README.md
@@ -0,0 +1,12 @@
+# Токен Telegram-бота (@BotFather -> /newbot) — единственный обязательный секрет
+TELEGRAM_TOKEN=your_telegram_bot_token
+
+# Ник стримера на Twitch (то, что в URL: twitch.tv/ЭТО). По умолчанию theraiwy
+TWITCH_LOGIN=theraiwy
+
+# Куда слать оповещения: @username канала (напр. @my_channel)
+# или числовой id чата/группы (напр. -1001234567890)
+NOTIFY_CHAT_ID=@your_channel
+
+# Интервал проверки в секундах (по умолчанию 60)
+CHECK_INTERVAL=60
@@ -0,0 +1,13 @@
+# Секреты — НИКОГДА не коммитить
+.env
+
+# Python
+__pycache__/
+*.py[cod]
+.venv/
+venv/
+*.egg-info/
+
+# IDE
+.vscode/
+.idea/
@@ -0,0 +1,20 @@
+FROM python:3.12-slim
+
+# Не создавать .pyc и не буферизовать stdout (чтобы логи шли в реальном времени)
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+# Сначала зависимости — для кэша слоёв
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Затем код
+COPY bot.py .
+
+# Запуск от непривилегированного пользователя
+RUN useradd --create-home appuser
+USER appuser
+
+CMD ["python", "bot.py"]
@@ -0,0 +1,84 @@
+# Twitch → Telegram Stream Notify Bot
+
+Бот проверяет, вышел ли стример в эфир на Twitch, и при старте стрима
+отправляет оповещение в указанный чат/канал Telegram.
+Определение статуса работает через публичный [decapi.me](https://decapi.me) — **без Twitch-токенов**.
+
+Это Telegram-версия [Discord-бота](../twitch_discord).
+
+## Возможности
+
+- Определение статуса стрима **без токенов Twitch** (публичный decapi.me)
+- Оповещение **один раз** при переходе оффлайн → онлайн (без спама, пока стрим идёт)
+- Сообщение с превью эфира, названием стрима и игрой
+- Inline-кнопки под сообщением: **Twitch** и **Все ссылки**
+- Команда `/stream_test` — тестовое оповещение
+
+## Переменные окружения
+
+| Переменная | Обязательна | Описание |
+|---|---|---|
+| `TELEGRAM_TOKEN` | ✅ | Токен бота из [@BotFather](https://t.me/BotFather) → `/newbot` |
+| `NOTIFY_CHAT_ID` | ✅ | Куда слать: `@username` канала или числовой id чата/группы (напр. `-1001234567890`) |
+| `TWITCH_LOGIN` | нет | Ник стримера из URL `twitch.tv/ЭТО`. По умолчанию `theraiwy` |
+| `CHECK_INTERVAL` | нет | Интервал проверки в секундах. По умолчанию `60` |
+
+Единственный обязательный секрет — `TELEGRAM_TOKEN`.
+
+## Настройка Telegram-бота
+
+1. Создайте бота у [@BotFather](https://t.me/BotFather) командой `/newbot`, получите токен.
+2. **Добавьте бота в канал/группу как администратора** с правом отправлять сообщения —
+   иначе он не сможет писать в `NOTIFY_CHAT_ID`.
+3. Укажите `NOTIFY_CHAT_ID`:
+   - для публичного канала — его `@username` (напр. `@theraiwy_streams`);
+   - для приватной группы/канала — числовой id (напр. через `@userinfobot` или
+     `@getidsbot`), для супергрупп/каналов он вида `-100...`.
+
+### Как узнать числовой id чата
+
+Добавьте бота в чат, напишите там любое сообщение и загляните в логи бота, либо
+воспользуйтесь ботом `@getidsbot`.
+
+## Запуск через Docker (рекомендуется)
+
+```bash
+cp .env.example .env    # заполнить TELEGRAM_TOKEN и NOTIFY_CHAT_ID
+docker compose up -d --build
+```
+
+Логи: `docker compose logs -f`
+
+### Деплой в DockHand
+
+`docker-compose.yml` читает значения из окружения через `${...}`, поэтому `.env`
+в репозиторий не коммитится. В панели DockHand при создании стека задайте переменные
+в разделе **Environment**:
+
+```
+TELEGRAM_TOKEN=твой_токен
+NOTIFY_CHAT_ID=@your_channel
+TWITCH_LOGIN=theraiwy
+CHECK_INTERVAL=60
+```
+
+Обязательны только `TELEGRAM_TOKEN` и `NOTIFY_CHAT_ID` — у остальных есть значения по умолчанию.
+
+## Запуск без Docker
+
+```bash
+pip install -r requirements.txt
+cp .env.example .env    # заполнить значения
+python bot.py
+```
+
+## Команды
+
+- `/stream_test` — отправить тестовое оповещение в целевой чат (помечается `🧪 [ТЕСТ]`).
+
+## Как определяется стрим
+
+Каждые `CHECK_INTERVAL` секунд бот опрашивает decapi.me (`uptime`, `title`, `game`).
+При переходе оффлайн → онлайн отправляется оповещение. Публичный сервис может изредка
+быть недоступен — в этом случае проверка пропускается и повторяется в следующем цикле,
+бот не падает. Рекомендуемый интервал — **30–60 секунд**.
@@ -0,0 +1,205 @@
+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()
@@ -0,0 +1,11 @@
+services:
+  twitch-notify-tg:
+    build: .
+    container_name: twitch-notify-tg
+    restart: unless-stopped
+    environment:
+      # Значения подставляются из окружения стека (панель DockHand / .env рядом с compose)
+      TELEGRAM_TOKEN: ${TELEGRAM_TOKEN}
+      NOTIFY_CHAT_ID: ${NOTIFY_CHAT_ID}
+      TWITCH_LOGIN: ${TWITCH_LOGIN:-theraiwy}
+      CHECK_INTERVAL: ${CHECK_INTERVAL:-60}
@@ -0,0 +1,3 @@
+python-telegram-bot[job-queue]>=21.0
+aiohttp>=3.9.0
+python-dotenv>=1.0.0