
Напишем и запустим телеграм бота который будет присылать новости с сайта https://newxboxone.ru/
📋 Содержание
- Подготовительные шаги
- Установка необходимых пакетов
- Создание структуры проекта
- Создание файлов бота
- Настройка виртуального окружения
- Создание systemd службы
- Запуск и управление ботом
- Мониторинг и логи
- Устранение неполадок
- Обновление бота
1. Подготовительные шаги
1.1. Подключение к серверу
ssh root@ваш_ip_сервера # или, если используете пользователя с sudo правами: ssh username@ваш_ip_сервера
1.2. Получение токена бота Telegram
- Откройте Telegram
- Найдите @BotFather
- Отправьте команду
/newbot - Следуйте инструкциям для создания бота
- Сохраните полученный токен (он выглядит примерно так:
1234567890:ABCdefGHIjklMNOpqrsTUVwxyz)
2. Установка необходимых пакетов
2.1. Обновление системы
apt update && apt upgrade -y
2.2. Установка Python и инструментов
apt install -y python3 python3-pip python3-venv python3-dev
2.3. Установка дополнительных зависимостей
apt install -y build-essential libssl-dev libffi-dev curl wget git
2.4. Проверка установки Python
python3 --version # Должно показать Python 3.8 или выше
3. Создание структуры проекта
3.1. Создание директории для бота
mkdir -p /opt/telegram-bot cd /opt/telegram-bot
3.2. Создание поддиректорий
mkdir -p {logs,data,backup}
3.3. Проверка структуры
ls -la /opt/telegram-bot/ # Должно показать: # backup/ data/ logs/
4. Создание файлов бота
4.1. Создание файла зависимостей (requirements.txt)
nano /opt/telegram-bot/requirements.txt
Добавьте следующий код:
python-telegram-bot==20.3 beautifulsoup4==4.12.2 python-dotenv==1.0.0 requests==2.31.0 aiohttp==3.9.1
Сохраните файл: Ctrl+X, затем Y, затем Enter.
4.2. Создание файла с токеном (.env)
nano /opt/telegram-bot/.env
Добавьте ваш токен:
TELEGRAM_BOT_TOKEN=ВАШ_ТОКЕН_ЗДЕСЬ
Пример:
TELEGRAM_BOT_TOKEN=1234567890:ABCdefGHIjklMNOpqrsTUVwxyz
Сохраните файл: Ctrl+X, затем Y, затем Enter.
4.3. Создание основного файла бота
nano /opt/telegram-bot/newxboxone.py
Скопируйте и вставьте полный код бота:
import os
import asyncio
import aiohttp
from bs4 import BeautifulSoup
from dotenv import load_dotenv
from telegram import Update
from telegram.ext import Application, CommandHandler, ContextTypes
import logging
import pickle
import hashlib
import json
import time
# Настройка логирования
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.INFO
)
logger = logging.getLogger(__name__)
# Загрузка переменных из .env
load_dotenv()
# Получение токена из .env
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
# ID чатов/каналов для рассылки
SUBSCRIBED_CHATS_FILE = "/opt/telegram-bot/data/subscribed_chats.pkl"
LAST_NEWS_FILE = "/opt/telegram-bot/data/last_news.json"
class NewsBot:
def __init__(self, max_stored_news=50):
self.url = "https://newxboxone.ru/"
self.subscribed_chats = self.load_subscribed_chats()
self.last_news_hashes = self.load_last_news_hashes()
self.max_stored_news = max_stored_news
self.application = None
def load_subscribed_chats(self):
try:
with open(SUBSCRIBED_CHATS_FILE, 'rb') as f:
return pickle.load(f)
except FileNotFoundError:
return set()
def save_subscribed_chats(self):
os.makedirs(os.path.dirname(SUBSCRIBED_CHATS_FILE), exist_ok=True)
with open(SUBSCRIBED_CHATS_FILE, 'wb') as f:
pickle.dump(self.subscribed_chats, f)
def load_last_news_hashes(self):
try:
with open(LAST_NEWS_FILE, 'r', encoding='utf-8') as f:
data = json.load(f)
return set(data.get('hashes', []))
except (FileNotFoundError, json.JSONDecodeError):
return set()
def save_last_news_hashes(self):
try:
os.makedirs(os.path.dirname(LAST_NEWS_FILE), exist_ok=True)
data = {
'hashes': list(self.last_news_hashes),
'timestamp': time.time()
}
with open(LAST_NEWS_FILE, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
except Exception as e:
logger.error(f"Ошибка при сохранении хешей: {e}")
def add_news_hash(self, news_hash):
self.last_news_hashes.add(news_hash)
if len(self.last_news_hashes) > self.max_stored_news:
oldest = next(iter(self.last_news_hashes))
self.last_news_hashes.remove(oldest)
self.save_last_news_hashes()
async def get_xbox_news_headlines(self):
try:
async with aiohttp.ClientSession() as session:
async with session.get(self.url, timeout=10) as response:
if response.status != 200:
logger.error(f"Ошибка при запросе к сайту: {response.status}")
return []
html = await response.text()
soup = BeautifulSoup(html, 'html.parser')
news_items = []
posts = soup.find_all('li', class_='post-item')
if not posts:
posts = soup.select('.posts-items .post-item')
for post in posts[:10]:
try:
title_elem = post.find('h3', class_='post-title')
if not title_elem:
continue
title_link = title_elem.find('a')
if not title_link:
continue
title = title_link.text.strip()
link = title_link.get('href')
if not link or not title:
continue
date_elem = post.find('span', class_='date')
date = date_elem.text.strip() if date_elem else "Сегодня"
img_elem = post.find('img')
image_url = None
if img_elem:
if img_elem.get('data-src'):
image_url = img_elem.get('data-src')
elif img_elem.get('src'):
src = img_elem.get('src')
if src and not src.startswith('data:image/svg'):
image_url = src
news_hash = hashlib.md5(f"{title}{link}".encode()).hexdigest()
news_items.append({
'title': title,
'link': link,
'date': date,
'image_url': image_url,
'hash': news_hash
})
except Exception as e:
logger.error(f"Ошибка при парсинге новости: {e}")
continue
return news_items
except Exception as e:
logger.error(f"Ошибка при запросе к сайту: {e}")
return []
async def check_new_news(self):
try:
news_items = await self.get_xbox_news_headlines()
if not news_items:
logger.warning("Не удалось получить новости")
return []
latest_news = news_items[0]
latest_hash = latest_news['hash']
if latest_hash not in self.last_news_hashes:
logger.info(f"Найдена новая новость: {latest_news['title']}")
self.add_news_hash(latest_hash)
return [latest_news]
logger.info("Новых новостей нет")
return []
except Exception as e:
logger.error(f"Ошибка при проверке новостей: {e}")
return []
def format_news_message(self, news_item):
message = f"🎮 <b>Новая новость Xbox!</b>\n\n"
message += f"📰 <b>{news_item['title']}</b>\n"
message += f"📅 {news_item['date']}\n\n"
message += f"🔗 <a href='{news_item['link']}'>Читать полностью</a>"
return message
async def send_news_to_subscribers(self, news_item):
if not self.subscribed_chats:
logger.info("Нет подписчиков для отправки новостей")
return
if not self.application:
logger.error("Application не установлена")
return
message = self.format_news_message(news_item)
successful_sends = 0
for chat_id in self.subscribed_chats:
try:
if news_item['image_url'] and not news_item['image_url'].startswith('data:'):
try:
await self.application.bot.send_photo(
chat_id=chat_id,
photo=news_item['image_url'],
caption=message,
parse_mode='HTML'
)
except Exception as e:
logger.error(f"Не удалось отправить с фото в чат {chat_id}: {e}")
await self.application.bot.send_message(
chat_id=chat_id,
text=message,
parse_mode='HTML',
disable_web_page_preview=False
)
else:
await self.application.bot.send_message(
chat_id=chat_id,
text=message,
parse_mode='HTML',
disable_web_page_preview=False
)
successful_sends += 1
logger.info(f"Новость отправлена в чат {chat_id}")
await asyncio.sleep(0.1)
except Exception as e:
logger.error(f"Не удалось отправить новость в чат {chat_id}: {e}")
logger.info(f"Новость отправлена {successful_sends} из {len(self.subscribed_chats)} подписчиков")
async def subscribe_chat(self, chat_id):
if chat_id not in self.subscribed_chats:
self.subscribed_chats.add(chat_id)
self.save_subscribed_chats()
logger.info(f"Чат {chat_id} подписался на новости")
return True
return False
async def unsubscribe_chat(self, chat_id):
if chat_id in self.subscribed_chats:
self.subscribed_chats.remove(chat_id)
self.save_subscribed_chats()
logger.info(f"Чат {chat_id} отписался от новостей")
return True
return False
async def initialize(self):
try:
logger.info("Инициализация бота...")
news_items = await self.get_xbox_news_headlines()
if news_items:
for item in news_items[:5]:
self.add_news_hash(item['hash'])
logger.info(f"Добавлено {min(5, len(news_items))} начальных новостей в кэш")
logger.info(f"Загружено {len(self.last_news_hashes)} хешей новостей")
logger.info(f"Загружено {len(self.subscribed_chats)} подписчиков")
except Exception as e:
logger.error(f"Ошибка при инициализации: {e}")
# Создаем экземпляр бота
news_bot = NewsBot()
# Фоновая задача для периодической проверки новостей
async def check_news_periodically():
while True:
try:
logger.info("Проверка новых новостей...")
new_news = await news_bot.check_new_news()
for news_item in new_news:
await news_bot.send_news_to_subscribers(news_item)
logger.info(f"Отправлена новая новость: {news_item['title']}")
if new_news:
logger.info(f"Найдена и отправлена 1 новая новость")
else:
logger.info("Новых новостей не найдено")
except Exception as e:
logger.error(f"Ошибка в фоновой задаче: {e}")
await asyncio.sleep(300)
# Команды бота
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"Привет! Я бот для получения новостей о Xbox.\n\n"
"Команды:\n"
"/news - получить последние новости\n"
"/subscribe - подписаться на автоматическую рассылку\n"
"/unsubscribe - отписаться от рассылки\n"
"/help - помощь\n"
"/status - статус подписки"
)
async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"📚 <b>Помощь по командам:</b>\n\n"
"/start - Начало работы\n"
"/news - Получить последние новости\n"
"/subscribe - Подписаться на автоматическую рассылку\n"
"/unsubscribe - Отписаться от рассылки\n"
"/status - Статус подписки\n"
"/help - Эта справка\n\n"
"Бот автоматически проверяет сайт каждые 5 минут и присылает новые новости!",
parse_mode='HTML'
)
async def status_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
chat_id = update.effective_chat.id
if chat_id in news_bot.subscribed_chats:
await update.message.reply_text(
"✅ Вы подписаны на автоматическую рассылку новостей.\n"
f"Всего подписчиков: {len(news_bot.subscribed_chats)}\n"
f"Отслеживаемых новостей: {len(news_bot.last_news_hashes)}"
)
else:
await update.message.reply_text(
"❌ Вы не подписаны на рассылку.\n"
"Используйте /subscribe чтобы подписаться."
)
async def subscribe(update: Update, context: ContextTypes.DEFAULT_TYPE):
chat_id = update.effective_chat.id
if await news_bot.subscribe_chat(chat_id):
await update.message.reply_text(
"✅ Вы успешно подписались на автоматическую рассылку новостей!\n"
"Теперь вы будете получать новые новости сразу после их публикации на сайте.\n\n"
"Бот проверяет новости каждые 5 минут."
)
else:
await update.message.reply_text("ℹ️ Вы уже подписаны на рассылку.")
async def unsubscribe(update: Update, context: ContextTypes.DEFAULT_TYPE):
chat_id = update.effective_chat.id
if await news_bot.unsubscribe_chat(chat_id):
await update.message.reply_text(
"❌ Вы отписались от автоматической рассылки новостей.\n"
"Чтобы снова подписаться, используйте команду /subscribe."
)
else:
await update.message.reply_text("ℹ️ Вы не были подписаны на рассылку.")
async def news(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text("🔄 Получаю последние новости...")
news_items = await news_bot.get_xbox_news_headlines()
if not news_items:
await update.message.reply_text("Не удалось получить новости. Попробуйте позже.")
return
for i, item in enumerate(news_items[:5], 1):
message = news_bot.format_news_message(item)
try:
await update.message.reply_text(message, parse_mode='HTML', disable_web_page_preview=False)
except Exception as e:
logger.error(f"Ошибка при отправке новости: {e}")
await update.message.reply_text(f"{i}. {item['title']}\n{item['link']}")
if i < len(news_items[:5]):
await asyncio.sleep(0.5)
async def post_init(application):
news_bot.application = application
await news_bot.initialize()
asyncio.create_task(check_news_periodically())
def main():
application = Application.builder() \
.token(TELEGRAM_BOT_TOKEN) \
.post_init(post_init) \
.build()
application.add_handler(CommandHandler("start", start))
application.add_handler(CommandHandler("help", help_command))
application.add_handler(CommandHandler("news", news))
application.add_handler(CommandHandler("subscribe", subscribe))
application.add_handler(CommandHandler("unsubscribe", unsubscribe))
application.add_handler(CommandHandler("status", status_command))
logger.info("Бот запущен...")
application.run_polling()
if __name__ == "__main__":
main()
Сохраните файл: Ctrl+X, затем Y, затем Enter.
4.4. Установка прав на файлы
chmod +x /opt/telegram-bot/newxboxone.py chown -R root:root /opt/telegram-bot
5. Настройка виртуального окружения
5.1. Создание виртуального окружения
cd /opt/telegram-bot python3 -m venv venv
5.2. Активация виртуального окружения
source venv/bin/activate
Примечание: После активации в командной строке появится (venv).
5.3. Обновление pip
pip install --upgrade pip
5.4. Установка зависимостей
pip install -r requirements.txt
5.5. Проверка установки
pip list | grep telegram # Должно показать python-telegram-bot 20.3
5.6. Тестовый запуск бота
python newxboxone.py
Что должно произойти:
- Вы должны увидеть «Бот запущен…»
- Если в .env правильный токен, бот подключится к Telegram
- Нажмите Ctrl+C для остановки
Если видите ошибку:
- Проверьте токен в .env
- Проверьте интернет-соединение
- Проверьте, что все зависимости установлены
6. Создание systemd службы
6.1. Создание файла службы
nano /etc/systemd/system/newxboxone.service
6.2. Добавление конфигурации службы
[Unit] Description=Telegram Bot for Xbox News After=network.target Wants=network.target [Service] Type=simple User=root Group=root WorkingDirectory=/opt/telegram-bot Environment="PATH=/opt/telegram-bot/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" Environment="PYTHONPATH=/opt/telegram-bot" ExecStart=/opt/telegram-bot/venv/bin/python /opt/telegram-bot/newxboxone.py Restart=always RestartSec=10 StandardOutput=append:/opt/telegram-bot/logs/bot.log StandardError=append:/opt/telegram-bot/logs/bot_error.log # Защита службы ProtectSystem=strict ReadWritePaths=/opt/telegram-bot/data /opt/telegram-bot/logs NoNewPrivileges=true PrivateTmp=true [Install] WantedBy=multi-user.target
Сохраните файл: Ctrl+X, затем Y, затем Enter.
6.3. Создание файла ротации логов (опционально, но рекомендуется)
nano /etc/logrotate.d/newxboxone
Добавьте:
/opt/telegram-bot/logs/*.log {
daily
rotate 7
compress
delaycompress
missingok
notifempty
create 644 root root
}
Сохраните файл: Ctrl+X, затем Y, затем Enter.
7. Запуск и управление ботом
7.1. Перезагрузка systemd
systemctl daemon-reload
7.2. Включение автозапуска
systemctl enable newxboxone.service
7.3. Запуск бота
systemctl start newxboxone.service
7.4. Проверка статуса
systemctl status newxboxone.service
Успешный статус должен показывать:
Active: active (running)- Без ошибок в логах
7.5. Основные команды управления
# Запуск бота systemctl start newxboxone # Остановка бота systemctl stop newxboxone # Перезапуск бота systemctl restart newxboxone # Проверка статуса systemctl status newxboxone # Включение автозапуска при загрузке системы systemctl enable newxboxone # Отключение автозапуска systemctl disable newxboxone
8. Мониторинг и логи
8.1. Просмотр логов в реальном времени
# Используя systemd journalctl -u newxboxone -f # Или через файлы логов tail -f /opt/telegram-bot/logs/bot.log
8.2. Просмотр ошибок
tail -f /opt/telegram-bot/logs/bot_error.log
8.3. Просмотр последних 50 записей
journalctl -u newxboxone -n 50
8.4. Просмотр логов за определенную дату
journalctl -u newxboxone --since "2024-01-24" --until "2024-01-25"
8.5. Скрипт для проверки состояния бота
nano /opt/telegram-bot/check_bot.sh
Добавьте:
#!/bin/bash
SERVICE="newxboxone"
LOG_FILE="/opt/telegram-bot/logs/health_check.log"
echo "$(date): Проверка состояния бота $SERVICE" >> $LOG_FILE
if systemctl is-active --quiet $SERVICE; then
echo "$(date): ✓ $SERVICE работает" >> $LOG_FILE
else
echo "$(date): ✗ $SERVICE не работает. Перезапускаю..." >> $LOG_FILE
systemctl restart $SERVICE
fi
Сделайте скрипт исполняемым:
chmod +x /opt/telegram-bot/check_bot.sh
Добавьте в crontab для автоматической проверки каждые 5 минут:
crontab -e
Добавьте строку:
*/5 * * * * /opt/telegram-bot/check_bot.sh
Сохраните: Ctrl+X, затем Y, затем Enter.
9. Устранение неполадок
9.1. Бот не запускается
# Проверьте токен cat /opt/telegram-bot/.env # Проверьте ошибки journalctl -u newxboxone -xe # Проверьте права ls -la /opt/telegram-bot/
9.2. Проблемы с зависимостями
cd /opt/telegram-bot source venv/bin/activate # Переустановите зависимости pip install -r requirements.txt --upgrade --force-reinstall
9.3. Бот падает
# Проверьте свободное место df -h # Проверьте использование памяти free -h # Проверьте сетевую доступность ping api.telegram.org curl -I https://newxboxone.ru/
9.4. Сброс службы
systemctl daemon-reload systemctl reset-failed newxboxone systemctl restart newxboxone
9.5. Полная переустановка
# Остановите бота systemctl stop newxboxone # Удалите старые файлы rm -rf /opt/telegram-bot/venv rm -rf /opt/telegram-bot/data/* # Начните с шага 5.1
10. Обновление бота
10.1. Процедура обновления
# 1. Остановите бота systemctl stop newxboxone # 2. Сделайте резервную копию BACKUP_DIR="/opt/telegram-bot/backup/$(date +%Y%m%d-%H%M%S)" mkdir -p $BACKUP_DIR cp -r /opt/telegram-bot/data $BACKUP_DIR/ cp /opt/telegram-bot/.env $BACKUP_DIR/ # 3. Обновите код (если нужно) # nano /opt/telegram-bot/newxboxone.py # 4. Обновите зависимости cd /opt/telegram-bot source venv/bin/activate pip install -r requirements.txt --upgrade # 5. Запустите бота systemctl start newxboxone # 6. Проверьте статус systemctl status newxboxone
10.2. Скрипт для автоматического обновления
nano /opt/telegram-bot/update_bot.sh
Добавьте:
#!/bin/bash
set -e
SERVICE="newxboxone"
BACKUP_DIR="/opt/telegram-bot/backup"
DATE=$(date +%Y%m%d-%H%M%S)
echo "=== Начало обновления бота ==="
echo "Время: $(date)"
# Создание резервной копии
echo "1. Создание резервной копии..."
mkdir -p $BACKUP_DIR
tar -czf $BACKUP_DIR/bot-backup-$DATE.tar.gz \
/opt/telegram-bot/data \
/opt/telegram-bot/.env \
/opt/telegram-bot/requirements.txt
# Остановка службы
echo "2. Остановка службы..."
systemctl stop $SERVICE || true
# Обновление зависимостей
echo "3. Обновление зависимостей..."
cd /opt/telegram-bot
source venv/bin/activate
pip install -r requirements.txt --upgrade
# Запуск службы
echo "4. Запуск службы..."
systemctl start $SERVICE
# Проверка
echo "5. Проверка состояния..."
sleep 5
systemctl status $SERVICE --no-pager
echo "=== Обновление завершено ==="
Сделайте скрипт исполняемым:
chmod +x /opt/telegram-bot/update_bot.sh
📝 Проверка работоспособности
1. Проверьте, что бот работает:
systemctl status newxboxone
2. Проверьте логи:
tail -f /opt/telegram-bot/logs/bot.log
3. Проверьте бота в Telegram:
- Найдите вашего бота в Telegram
- Отправьте команду
/start - Должно прийти приветственное сообщение
- Отправьте
/subscribeдля подписки - Отправьте
/newsдля получения новостей
4. Проверьте автоматическую отправку:
- Подождите 5 минут
- Проверьте логи на наличие проверок новостей
- Когда на сайте появится новая новость, бот отправит её автоматически
🔧 Дополнительные возможности
Настройка интервала проверки
Чтобы изменить интервал проверки новостей (по умолчанию 5 минут), измените строку в коде:
В файле /opt/telegram-bot/newxboxone.py найдите:
await asyncio.sleep(300) # 300 секунд = 5 минут
Измените на нужное количество секунд, например:
await asyncio.sleep(600) # 10 минут
После изменения перезапустите бота:
systemctl restart newxboxone
Добавление администраторов
Вы можете добавить ID администраторов для получения уведомлений о состоянии бота.
🎯 Готово!
Ваш Telegram-бот для отслеживания новостей Xbox теперь:
- ✅ Установлен на VPS
- ✅ Настроен на автозапуск
- ✅ Автоматически проверяет новости каждые 5 минут
- ✅ Отправляет новые новости подписчикам
- ✅ Сохраняет логи
- ✅ Автоматически перезапускается при падении
Для получения помощи по командам бота отправьте /help в чат с ботом.