Как создать Telegram-бот для мониторинга YouTube-каналов

Отслеживание контента на YouTube и раньше работало через раз — «колокольчики» то и дело слетали, и даже ручная переподписка на уведомления не спасала ситуацию. После так называемого замедления же узнать о выходе нового ролика стало и вовсе невозможно: нужно либо постоянно проверять канал вручную — что априори бред, либо следить за другими соцсетями интересующего вас канала. Если они вообще есть.

Но даже если есть и активно ведутся — это все равно неудобно, ведь когда каналов десятки, они сливаются в один сплошной фон. Как итог, актуальная в моменте инфа проходит мимо нас. Когда же мы ее наконец-то замечаем, момент уже часто бывает упущен. К счастью, эту проблему можно легко решить, запустив бот для мониторинга YouTube-каналов — рассказываем ниже.

Какие задачи выполняет бот для мониторинга YouTube-каналов

Как понятно из названия — он мониторит YouTube-каналы, что позволяет:

• Не ходить с вечно включенным РКН в надежде получить заветный увед.

• Вовремя получать актуальную в моменте инфу.

• Не пропускать любимый развлекательный контент.

Впрочем, все это актуально для пользователя/зрителя.

Но ведь никто не мешает использовать такой бот и для повышения вовлеченности в собственный канал. Одно дело — подписка в Телеге (никто не следит за вашей Телегой только ради того, чтобы узнать, вышел ли ролик, — смиритесь), и совсем другое — увед от бота. По сути, это старый добрый пуш — только в виде ЛС от Telegram-бота. Только такой «пуш» еще и не бесит, ведь он добровольный.

Принцип работы бота для мониторинга YouTube-каналов

Алгоритмически бот крайне простой — так что разобраться или даже навешать поверх имеющегося что-то свое сможет даже новичок. Принцип его работы сводится к следующему:

1. Прослушивание эфира и реакция на команды.
2. Приветствие и краткое описание возможностей при вводе /start
3. Добавления канала в список отслеживаемых при вводе /add <ссылка>
4. Удаление канала из списка отслеживаемых при вводе /remove <ссылка>
5. Вывод списка отслеживаемых каналов при вводе /list
6. Проверка здесь и сейчас при вводе /force
7. Установка времени автоматической проверки при вводе /interval <минут>
8. Непосредственно сама проверка путем анализа RSS целевых каналов. С указанной при вводе /interval <минут> периодичностью (по умолчанию в код зашит интервал 5 минут) или же в моменте при вводе /force — соответственно.
9. Запись id последнего видео для каждого отслеживаемого канала в json.
10. Ну и разные сервисные сообщения на случай каких-либо ошибок. Их перечислять нет смысла.

Пошаговая инструкция, как сделать бот мониторинга YouTube-каналов

Прежде всего следует получить в специальном боте Telegram @BotFather токен для управления нашим ботом. Сделать это несложно:

1. Пишем в ЛС @BotFather.
2. Делаем то, что он напишет.
3. Сохраняем токен.

После — настраиваем свой сервак. Да-да, подразумевается, что он у вас есть и на нем уже развернут Python. Если нет — можете поднять сервер прямо на своем компе по этой инструкции. Это бесплатно. Ну или просто арендуйте подходящую машину — дело ваше.

1. Открываем консоль машины и пишем:
   pip install python-telegram-bot==20.7 apscheduler requests feedparser nest_asyncio
   

   

2. Ждем окончания настройки.
3. Создаем в корне файл bot.py.
4. Вставляем в него следующий код:
   import json
   import reimport os
   import logging
   import requests
   import feedparser
   import asyncio
   from apscheduler.schedulers.background import BackgroundScheduler
   from telegram import Update
   from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes
   TOKEN = "ТОКЕН СЮДА"
   DATA_FILE = 'storage.json'
   INTERVAL_FILE = 'interval.json'
   DEFAULT_INTERVAL_MINUTES = 5
   logging.basicConfig(level=logging.INFO)
   def ensure_file_exists(path, default_content):
   if not os.path.exists(path):
   with open(path, 'w') as f:
   json.dump(default_content, f, indent=2)
   def load_data():
   ensure_file_exists(DATA_FILE, {})
   with open(DATA_FILE, 'r') as f:
   return json.load(f)
   def save_data(data):
   with open(DATA_FILE, 'w') as f:
   json.dump(data, f, indent=2)
   def load_interval():
   ensure_file_exists(INTERVAL_FILE, {"interval": DEFAULT_INTERVAL_MINUTES})
   with open(INTERVAL_FILE, 'r') as f:
   return json.load(f).get("interval", DEFAULT_INTERVAL_MINUTES)
   def save_interval(minutes):
   with open(INTERVAL_FILE, 'w') as f:
   json.dump({"interval": minutes}, f)
   def get_rss_url(channel_url: str) -> str:
   try:
   response = requests.get(channel_url, timeout=10)
   html = response.text
   match = re.search(r'<link rel="canonical" href="https://www\.youtube\.com/channel/(UC[\w-]{22})"', html)
   if match:
   channel_id = match.group(1)
   return f'https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}'
   else:
   raise ValueError("❌ Не удалось найти channel_id в canonical-ссылке.")
   except Exception as e:
   raise ValueError(f"Ошибка при получении channel_id: {e}")
   async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
   interval = load_interval()
   await update.message.reply_text(
   f"👋 Привет! Я бот, который уведомляет о новых видео на YouTube-каналах.\n"
   f"Текущий интервал проверки: каждые {interval} минут.\n\n"
   "Команды:\n"
   "/add <ссылка> — подписаться на канал\n"
   "/remove <ссылка> — удалить канал\n"
   "/list — список твоих каналов\n"
   "/force — проверить сейчас\n"
   "/interval <минут> — изменить интервал проверки"
   )
   async def add_channel(update: Update, context: ContextTypes.DEFAULT_TYPE):
   user_id = str(update.effective_chat.id)
   if not context.args:
   await update.message.reply_text("⚠️ Использование: /add <ссылка_на_канал>")
   return
   url = context.args[0]
   try:
   rss = get_rss_url(url)
   except ValueError as e:
   await update.message.reply_text(str(e))
   return
   data = load_data()
   if user_id not in data:
   data[user_id] = {}
   if rss in data[user_id]:
   await update.message.reply_text("✅ Этот канал уже добавлен.")
   return
   feed = feedparser.parse(rss)
   if not feed.entries:
   await update.message.reply_text("❌ Не удалось получить видео с канала.")
   return
   latest_video = feed.entries[0]['id']
   data[user_id][rss] = latest_video
   save_data(data)
   await update.message.reply_text(f"🆗 Канал добавлен: {feed.feed.title}")
   async def list_channels(update: Update, context: ContextTypes.DEFAULT_TYPE):
   user_id = str(update.effective_chat.id)
   data = load_data()
   if user_id not in data or not data[user_id]:
   await update.message.reply_text("📭 У вас пока нет добавленных каналов.")
   return
   text = "📺 Ваши каналы:\n\n"
   for rss in data[user_id]:
   feed = feedparser.parse(rss)
   text += f"- {feed.feed.title}\n"
   await update.message.reply_text(text)
   async def remove_channel(update: Update, context: ContextTypes.DEFAULT_TYPE):
   user_id = str(update.effective_chat.id)
   if not context.args:
   await update.message.reply_text("⚠️ Использование: /remove <ссылка_на_канал>")
   return
   url = context.args[0]
   try:
   rss = get_rss_url(url)
   except ValueError as e:
   await update.message.reply_text(str(e))
   return
   data = load_data()
   if user_id in data and rss in data[user_id]:
   del data[user_id][rss]
   save_data(data)
   await update.message.reply_text("✅ Канал удалён.")
   else:
   await update.message.reply_text("⚠️ Этот канал не найден в вашем списке.")
   async def check_new_videos(application, silent: bool = False):
   print("🔄 Автоматическая проверка сработала.")
   data = load_data()
   for user_id, channels in data.items():
   for rss, last_video in channels.items():
   feed = feedparser.parse(rss)
   if not feed.entries:
   continue
   latest = feed.entries[0]
   if latest['id'] != last_video:
   data[user_id][rss] = latest['id']
   text = f"📢 Новое видео на канале *{feed.feed.title}*:\n[{latest.title}]({latest.link})"
   try:
   await application.bot.send_message(chat_id=int(user_id), text=text, parse_mode='Markdown')
   except Exception as e:
   logging.warning(f"Ошибка отправки {user_id}: {e}")
   elif not silent:
   logging.info(f"У {rss} нет новых видео (user {user_id})")
   save_data(data)
   async def force_check(update: Update, context: ContextTypes.DEFAULT_TYPE):
   await update.message.reply_text("🔄 Выполняю немедленную проверку каналов...")
   await check_new_videos(context.application, silent=True)
   await update.message.reply_text("✅ Проверка завершена.")
   async def set_interval(update: Update, context: ContextTypes.DEFAULT_TYPE):
   if not context.args:
   await update.message.reply_text("⚠️ Использование: /interval <минут>")
   return
   try:
   minutes = int(context.args[0])
   if minutes < 1:
   raise ValueError
   save_interval(minutes)
   await update.message.reply_text(f"✅ Интервал проверки установлен: каждые {minutes} минут.")
   schedule_job()
   except ValueError:
   await update.message.reply_text("⚠️ Введите положительное число минут.")
   async def main():
   global app, scheduler, loop
   app = ApplicationBuilder().token(TOKEN).build()
   app.add_handler(CommandHandler("start", start))
   app.add_handler(CommandHandler("add", add_channel))
   app.add_handler(CommandHandler("list", list_channels))
   app.add_handler(CommandHandler("remove", remove_channel))
   app.add_handler(CommandHandler("force", force_check))
   app.add_handler(CommandHandler("interval", set_interval))
   scheduler = BackgroundScheduler()
   loop = asyncio.get_event_loop()
   def schedule_job():
   scheduler.remove_all_jobs()
   def scheduled_job():
   asyncio.run_coroutine_threadsafe(
   check_new_videos(app, silent=False),
   loop
   )
   scheduler.add_job(scheduled_job, 'interval', minutes=load_interval())
   print(f"📅 Планировщик настроен: каждые {load_interval()} минут.")
   globals()["schedule_job"] = schedule_job
   schedule_job()
   scheduler.start()
   await app.run_polling()
   if __name__ == "__main__":
   try:
   asyncio.run(main())
   except RuntimeError:
   import nest_asyncio
   nest_asyncio.apply()
   asyncio.run(main())
5. Меняем “ТОКЕН СЮДА” на свой токен. Кавычки оставляем!
6. Сохраняем.
7. Запускаем командой:
   python bot.py
8. Тестируем.

Демонстрация работы

Подводя итоги

Как видите, колокольчик можно легко воспроизвести внутри Телеги, и ни потребность в РКН, ни странная работа самого YouTube больше не будут вам докучать. Подписывайтесь на наш Telegram-канал и первыми получайте инструкции по созданию ботов для арбитража трафика.