116 lines
4.7 KiB
Python
116 lines
4.7 KiB
Python
"""
|
||
Django management команда для управления Telegram webhook.
|
||
"""
|
||
from django.core.management.base import BaseCommand
|
||
from django.conf import settings
|
||
from telegram import Bot
|
||
from telegram.error import TelegramError
|
||
import asyncio
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class Command(BaseCommand):
|
||
help = 'Управление Telegram webhook (setup/remove/info)'
|
||
|
||
def add_arguments(self, parser):
|
||
parser.add_argument(
|
||
'action',
|
||
type=str,
|
||
choices=['setup', 'remove', 'info'],
|
||
help='Действие: setup - установить webhook, remove - удалить, info - информация'
|
||
)
|
||
|
||
def handle(self, *args, **options):
|
||
action = options['action']
|
||
token = getattr(settings, 'TELEGRAM_BOT_TOKEN', None)
|
||
|
||
if not token:
|
||
self.stdout.write(self.style.ERROR('TELEGRAM_BOT_TOKEN not set'))
|
||
return
|
||
|
||
webhook_url = getattr(settings, 'TELEGRAM_WEBHOOK_URL', None)
|
||
webhook_secret_token = getattr(settings, 'TELEGRAM_WEBHOOK_SECRET_TOKEN', None)
|
||
|
||
if action == 'setup':
|
||
if not webhook_url:
|
||
self.stdout.write(self.style.ERROR('TELEGRAM_WEBHOOK_URL not set'))
|
||
return
|
||
|
||
asyncio.run(self.setup_webhook(token, webhook_url, webhook_secret_token))
|
||
|
||
elif action == 'remove':
|
||
asyncio.run(self.remove_webhook(token))
|
||
|
||
elif action == 'info':
|
||
asyncio.run(self.get_webhook_info(token))
|
||
|
||
async def setup_webhook(self, token, webhook_url, secret_token=None):
|
||
"""Установить webhook."""
|
||
try:
|
||
bot = Bot(token=token)
|
||
|
||
webhook_kwargs = {
|
||
'url': webhook_url,
|
||
'allowed_updates': ['message', 'callback_query', 'inline_query', 'chosen_inline_result'],
|
||
}
|
||
|
||
if secret_token:
|
||
webhook_kwargs['secret_token'] = secret_token
|
||
|
||
await bot.set_webhook(**webhook_kwargs)
|
||
await bot.close()
|
||
|
||
self.stdout.write(self.style.SUCCESS(f'Webhook установлен: {webhook_url}'))
|
||
|
||
# Получаем информацию
|
||
await self.get_webhook_info(token)
|
||
|
||
except TelegramError as e:
|
||
self.stdout.write(self.style.ERROR(f'Ошибка установки webhook: {e}'))
|
||
except Exception as e:
|
||
self.stdout.write(self.style.ERROR(f'Неожиданная ошибка: {e}'))
|
||
|
||
async def remove_webhook(self, token):
|
||
"""Удалить webhook."""
|
||
try:
|
||
bot = Bot(token=token)
|
||
await bot.delete_webhook(drop_pending_updates=True)
|
||
await bot.close()
|
||
|
||
self.stdout.write(self.style.SUCCESS('Webhook удален'))
|
||
|
||
except TelegramError as e:
|
||
self.stdout.write(self.style.ERROR(f'Ошибка удаления webhook: {e}'))
|
||
except Exception as e:
|
||
self.stdout.write(self.style.ERROR(f'Неожиданная ошибка: {e}'))
|
||
|
||
async def get_webhook_info(self, token):
|
||
"""Получить информацию о webhook."""
|
||
try:
|
||
bot = Bot(token=token)
|
||
info = await bot.get_webhook_info()
|
||
await bot.close()
|
||
|
||
self.stdout.write('\n' + '=' * 50)
|
||
self.stdout.write('Информация о webhook:')
|
||
self.stdout.write('=' * 50)
|
||
self.stdout.write(f'URL: {info.url or "Не установлен"}')
|
||
self.stdout.write(f'Has custom certificate: {info.has_custom_certificate}')
|
||
self.stdout.write(f'Pending updates: {info.pending_update_count}')
|
||
if info.last_error_date:
|
||
self.stdout.write(f'Last error date: {info.last_error_date}')
|
||
if info.last_error_message:
|
||
self.stdout.write(f'Last error: {info.last_error_message}')
|
||
self.stdout.write(f'Max connections: {info.max_connections}')
|
||
if info.allowed_updates:
|
||
self.stdout.write(f'Allowed updates: {", ".join(info.allowed_updates)}')
|
||
self.stdout.write('=' * 50 + '\n')
|
||
|
||
except TelegramError as e:
|
||
self.stdout.write(self.style.ERROR(f'Ошибка получения информации: {e}'))
|
||
except Exception as e:
|
||
self.stdout.write(self.style.ERROR(f'Неожиданная ошибка: {e}'))
|
||
|