95 lines
3.8 KiB
Python
95 lines
3.8 KiB
Python
"""
|
||
Management command для запуска Telegram бота.
|
||
"""
|
||
from django.core.management.base import BaseCommand
|
||
from apps.notifications.telegram_bot import TelegramBot
|
||
import asyncio
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class Command(BaseCommand):
|
||
help = 'Запуск Telegram бота для уведомлений'
|
||
|
||
def add_arguments(self, parser):
|
||
"""Добавление аргументов команды."""
|
||
parser.add_argument(
|
||
'--no-polling',
|
||
action='store_true',
|
||
help='Не запускать polling (только инициализация)'
|
||
)
|
||
|
||
def handle(self, *args, **options):
|
||
"""Запуск бота."""
|
||
self.stdout.write(self.style.SUCCESS('🤖 Запуск Telegram бота...'))
|
||
|
||
bot = TelegramBot()
|
||
|
||
try:
|
||
if options['no_polling']:
|
||
self.stdout.write(self.style.WARNING('Режим без polling'))
|
||
asyncio.run(self._init_only(bot))
|
||
else:
|
||
# Получаем информацию о боте перед запуском
|
||
asyncio.run(self._show_bot_info(bot))
|
||
asyncio.run(self._run_bot(bot))
|
||
|
||
except KeyboardInterrupt:
|
||
self.stdout.write(self.style.WARNING('\n⏹️ Остановка бота...'))
|
||
asyncio.run(bot.stop())
|
||
self.stdout.write(self.style.SUCCESS('✅ Бот остановлен'))
|
||
|
||
except Exception as e:
|
||
self.stdout.write(self.style.ERROR(f'❌ Ошибка: {str(e)}'))
|
||
logger.error(f'Error running Telegram bot: {e}', exc_info=True)
|
||
|
||
async def _init_only(self, bot):
|
||
"""Только инициализация без polling."""
|
||
await bot.application.initialize()
|
||
self.stdout.write(self.style.SUCCESS('✅ Бот инициализирован'))
|
||
|
||
async def _run_bot(self, bot):
|
||
"""Запустить бота и держать его активным."""
|
||
import signal
|
||
|
||
# Запускаем бота
|
||
await bot.start()
|
||
|
||
# Держим бота активным
|
||
stop_event = asyncio.Event()
|
||
|
||
def signal_handler(signum, frame):
|
||
stop_event.set()
|
||
|
||
signal.signal(signal.SIGINT, signal_handler)
|
||
signal.signal(signal.SIGTERM, signal_handler)
|
||
|
||
self.stdout.write(self.style.SUCCESS('✅ Бот работает. Нажмите Ctrl+C для остановки.'))
|
||
|
||
# Ждем сигнала остановки
|
||
await stop_event.wait()
|
||
|
||
async def _show_bot_info(self, bot):
|
||
"""Показать информацию о боте."""
|
||
try:
|
||
from telegram import Bot
|
||
telegram_bot = Bot(token=bot.token)
|
||
me = await telegram_bot.get_me()
|
||
|
||
self.stdout.write(self.style.SUCCESS(''))
|
||
self.stdout.write(self.style.SUCCESS('=' * 60))
|
||
self.stdout.write(self.style.SUCCESS('✅ Бот успешно авторизован!'))
|
||
self.stdout.write(self.style.SUCCESS('=' * 60))
|
||
self.stdout.write(self.style.SUCCESS(f'📛 Имя: {me.first_name}'))
|
||
self.stdout.write(self.style.SUCCESS(f'👤 Username: @{me.username}'))
|
||
self.stdout.write(self.style.SUCCESS(f'🆔 ID: {me.id}'))
|
||
self.stdout.write(self.style.SUCCESS(f'🔗 Ссылка: https://t.me/{me.username}'))
|
||
self.stdout.write(self.style.SUCCESS('=' * 60))
|
||
self.stdout.write(self.style.SUCCESS(''))
|
||
|
||
await telegram_bot.close()
|
||
except Exception as e:
|
||
self.stdout.write(self.style.WARNING(f'⚠️ Не удалось получить информацию о боте: {str(e)}'))
|
||
|