130 lines
5.6 KiB
Python
130 lines
5.6 KiB
Python
"""Admin-bot для получения уведомлений о скачанных файлах."""
|
||
import logging
|
||
from telegram import Update
|
||
from telegram.ext import (
|
||
Application,
|
||
CommandHandler,
|
||
MessageHandler,
|
||
filters,
|
||
ContextTypes,
|
||
)
|
||
from telegram.request import HTTPXRequest
|
||
|
||
from app.config import Config
|
||
from app.admin_manager import AdminManager
|
||
from app.statistics import Statistics
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def get_user_language(update: Update) -> str:
|
||
"""Определить язык пользователя для локализации."""
|
||
user = update.effective_user
|
||
lang_code = user.language_code or 'en'
|
||
# Если язык русский или начинается с ru, возвращаем 'ru', иначе 'en'
|
||
return 'ru' if lang_code.startswith('ru') else 'en'
|
||
|
||
|
||
def get_start_message(language: str, first_name: str) -> str:
|
||
"""Получить приветственное сообщение для admin-bot в зависимости от языка."""
|
||
if language == 'ru':
|
||
return (
|
||
f"Привет, {first_name}! 👋\n\n"
|
||
"Это административный бот для мониторинга сервиса YouTube → MP3.\n\n"
|
||
"Ты зарегистрирован как администратор. "
|
||
"Ты будешь получать уведомления о всех скачанных файлах с метаданными:\n"
|
||
"• Название файла\n"
|
||
"• Пользователь, который запросил\n"
|
||
"• Исходная ссылка на видео\n"
|
||
"• Сам MP3 файл\n\n"
|
||
"Используй /stat для просмотра статистики."
|
||
)
|
||
else:
|
||
return (
|
||
f"Hello, {first_name}! 👋\n\n"
|
||
"This is an admin bot for monitoring the YouTube → MP3 service.\n\n"
|
||
"You are registered as an administrator. "
|
||
"You will receive notifications about all downloaded files with metadata:\n"
|
||
"• File name\n"
|
||
"• User who requested\n"
|
||
"• Original video link\n"
|
||
"• The MP3 file itself\n\n"
|
||
"Use /stat to view statistics."
|
||
)
|
||
|
||
|
||
def get_stat_message(language: str, user_count: int, processed_urls: int) -> str:
|
||
"""Получить сообщение со статистикой в зависимости от языка."""
|
||
if language == 'ru':
|
||
return (
|
||
"📊 Статистика сервиса:\n\n"
|
||
f"👥 Уникальных пользователей: {user_count}\n"
|
||
f"🔗 Обработано ссылок: {processed_urls}"
|
||
)
|
||
else:
|
||
return (
|
||
"📊 Service Statistics:\n\n"
|
||
f"👥 Unique users: {user_count}\n"
|
||
f"🔗 Processed links: {processed_urls}"
|
||
)
|
||
|
||
|
||
async def start_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""Обработчик команды /start для регистрации администратора."""
|
||
user = update.effective_user
|
||
admin_manager: AdminManager = context.bot_data.get('admin_manager')
|
||
language = get_user_language(update)
|
||
|
||
if admin_manager:
|
||
admin_manager.add_admin(user.id)
|
||
message = get_start_message(language, user.first_name or "Admin")
|
||
await update.message.reply_text(message)
|
||
logger.info(f"Admin registered: {user.id} (@{user.username})")
|
||
else:
|
||
error_msg = "Ошибка: менеджер администраторов не инициализирован." if language == 'ru' else "Error: admin manager not initialized."
|
||
await update.message.reply_text(error_msg)
|
||
|
||
|
||
async def stat_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
|
||
"""Обработчик команды /stat для вывода статистики."""
|
||
statistics: Statistics = context.bot_data.get('statistics')
|
||
language = get_user_language(update)
|
||
|
||
if not statistics:
|
||
error_msg = "Ошибка: статистика не инициализирована." if language == 'ru' else "Error: statistics not initialized."
|
||
await update.message.reply_text(error_msg)
|
||
return
|
||
|
||
user_count = statistics.get_user_count()
|
||
processed_urls = statistics.get_processed_urls_count()
|
||
|
||
message = get_stat_message(language, user_count, processed_urls)
|
||
await update.message.reply_text(message)
|
||
logger.info(f"Statistics requested by admin {update.effective_user.id}")
|
||
|
||
|
||
def create_admin_bot_application(config: Config, admin_manager: AdminManager, statistics: Statistics) -> Application:
|
||
"""Создать и настроить приложение admin-bot."""
|
||
request = HTTPXRequest(
|
||
connect_timeout=config.tg_connect_timeout,
|
||
read_timeout=config.tg_read_timeout,
|
||
write_timeout=config.tg_write_timeout,
|
||
pool_timeout=config.tg_pool_timeout,
|
||
)
|
||
app = Application.builder().token(config.admin_bot_token).request(request).build()
|
||
|
||
# Сохраняем в bot_data
|
||
app.bot_data['config'] = config
|
||
app.bot_data['admin_manager'] = admin_manager
|
||
app.bot_data['statistics'] = statistics
|
||
|
||
# Обработчик команды /start
|
||
app.add_handler(CommandHandler("start", start_command))
|
||
|
||
# Обработчик команды /stat
|
||
app.add_handler(CommandHandler("stat", stat_command))
|
||
|
||
# Обработчик всех остальных сообщений (просто игнорируем)
|
||
app.add_handler(MessageHandler(filters.ALL, lambda u, c: None))
|
||
|
||
return app
|