668 lines
32 KiB
Python
668 lines
32 KiB
Python
"""
|
||
Обработчики для администраторов
|
||
"""
|
||
from datetime import datetime
|
||
|
||
from aiogram import F, Router
|
||
from aiogram.filters import Command
|
||
from aiogram.types import CallbackQuery, Message
|
||
|
||
from config import config
|
||
from config.constants import DEFAULT_PAGE_SIZE, MAX_PAGE_SIZE, MIN_PAGE_NUMBER, PERCENTAGE_DECIMAL_PLACES, TIME_DECIMAL_PLACES
|
||
from keyboards.inline import (
|
||
get_admin_keyboard, get_ban_confirm_keyboard, get_ban_duration_keyboard,
|
||
get_ban_user_keyboard, get_rate_limit_keyboard, get_stats_keyboard,
|
||
get_superuser_assignment_keyboard, get_superuser_confirm_keyboard, get_unban_keyboard
|
||
)
|
||
from services.auth.auth_new import AuthService
|
||
from services.infrastructure.database import DatabaseService
|
||
from services.infrastructure.logger import get_logger
|
||
from services.business.message_service import MessageService
|
||
from services.business.pagination_service import PaginationService
|
||
from services.permissions.decorators import require_admin_or_superuser, require_permission
|
||
from services.rate_limiting.rate_limit_service import RateLimitService
|
||
from services.business.user_service import UserService
|
||
from services.utils import format_stats
|
||
from dependencies import inject_admin_services
|
||
|
||
logger = get_logger(__name__)
|
||
router = Router()
|
||
|
||
|
||
async def _format_users_list(
|
||
users: list,
|
||
page: int,
|
||
per_page: int,
|
||
pagination_service: PaginationService
|
||
) -> str:
|
||
"""Форматирование списка пользователей для отображения"""
|
||
try:
|
||
# Рассчитываем пагинацию
|
||
page_users, total_users, current_page, total_pages = pagination_service.calculate_pagination(
|
||
users, page, per_page
|
||
)
|
||
|
||
# Формируем информацию о пагинации
|
||
start_idx = current_page * per_page
|
||
end_idx = min(start_idx + per_page, total_users)
|
||
pagination_info = pagination_service.format_pagination_info(
|
||
current_page, total_pages, start_idx, end_idx, total_users
|
||
)
|
||
|
||
# Формируем текст сообщения
|
||
users_text = f"🔍 <b>Управление суперпользователями</b>\n\n"
|
||
users_text += pagination_info
|
||
|
||
# Добавляем информацию о пользователях
|
||
for i, user in enumerate(page_users, start_idx + 1):
|
||
status_emoji = "🔍" if user.is_superuser else "👤"
|
||
users_text += f"{i}. {status_emoji} {user.display_name}\n"
|
||
if user.username:
|
||
users_text += f" @{user.username}\n"
|
||
created_at_str = user.created_at.strftime('%d.%m.%Y') if user.created_at else 'Неизвестно'
|
||
users_text += f" 📅 {created_at_str}\n\n"
|
||
|
||
users_text += "💡 Нажмите на пользователя для изменения его статуса."
|
||
|
||
return users_text
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при форматировании списка пользователей: {e}")
|
||
return "❌ Ошибка при форматировании списка пользователей."
|
||
|
||
|
||
# Функция is_admin теперь импортируется из services.auth
|
||
|
||
|
||
@router.message(Command("stats"))
|
||
@require_permission("view_stats", "❌ У вас нет прав для выполнения этой команды.")
|
||
async def cmd_stats(message: Message, database: DatabaseService = None):
|
||
"""Обработчик команды /stats"""
|
||
await show_stats(message, database)
|
||
|
||
|
||
@router.message(F.text == "👑 Админ панель")
|
||
@require_permission("admin_panel", "❌ У вас нет прав для доступа к админ панели.")
|
||
async def admin_panel_button(message: Message):
|
||
"""Обработчик кнопки 'Админ панель'"""
|
||
admin_text = "👑 <b>Админ панель</b>\n\n"
|
||
admin_text += "Выберите действие:"
|
||
|
||
await message.answer(
|
||
admin_text,
|
||
reply_markup=get_admin_keyboard(),
|
||
parse_mode="HTML"
|
||
)
|
||
|
||
|
||
@router.message(F.text == "📊 Статистика")
|
||
@require_permission("view_stats", "❌ У вас нет прав для просмотра статистики.")
|
||
async def stats_button(message: Message, database: DatabaseService = None):
|
||
"""Обработчик кнопки 'Статистика'"""
|
||
await show_stats(message, database)
|
||
|
||
|
||
async def show_stats(message: Message, database: DatabaseService = None):
|
||
"""Показать статистику"""
|
||
try:
|
||
if not database:
|
||
from dependencies import get_database
|
||
database = get_database()
|
||
|
||
# Получаем статистику
|
||
users_stats = await database.get_users_stats()
|
||
questions_stats = await database.get_questions_stats()
|
||
|
||
# Объединяем статистику
|
||
all_stats = {**users_stats, **questions_stats}
|
||
|
||
# Форматируем и отправляем
|
||
stats_text = format_stats(all_stats)
|
||
|
||
await message.answer(
|
||
stats_text,
|
||
reply_markup=get_stats_keyboard(),
|
||
parse_mode="HTML"
|
||
)
|
||
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении статистики: {e}")
|
||
await message.answer(
|
||
"❌ Произошла ошибка при получении статистики. Попробуйте позже."
|
||
)
|
||
|
||
|
||
|
||
|
||
|
||
|
||
@router.message(F.text == "📢 Рассылка")
|
||
@require_permission("broadcast", "❌ У вас нет прав для рассылки.")
|
||
async def broadcast_button(message: Message):
|
||
"""Обработчик кнопки 'Рассылка'"""
|
||
await message.answer(
|
||
"📢 <b>Рассылка</b>\n\n"
|
||
"Функция рассылки будет добавлена в следующих версиях.\n\n"
|
||
"💡 Планируемые возможности:\n"
|
||
"• Рассылка сообщений всем пользователям\n"
|
||
"• Рассылка по группам пользователей\n"
|
||
"• Планирование рассылок\n"
|
||
"• Статистика доставки",
|
||
reply_markup=get_admin_keyboard(),
|
||
parse_mode="HTML"
|
||
)
|
||
|
||
|
||
@router.message(F.text == "⚙️ Настройки")
|
||
@require_permission("admin_panel", "❌ У вас нет прав для изменения настроек.")
|
||
async def settings_button(message: Message):
|
||
"""Обработчик кнопки 'Настройки'"""
|
||
settings_text = "⚙️ <b>Настройки бота</b>\n\n"
|
||
settings_text += f"🔧 <b>Текущие настройки:</b>\n"
|
||
settings_text += f"• Режим отладки: {'Включен' if config.DEBUG else 'Выключен'}\n"
|
||
settings_text += f"• Макс. длина вопроса: {config.MAX_QUESTION_LENGTH} символов\n"
|
||
settings_text += f"• Макс. длина ответа: {config.MAX_ANSWER_LENGTH} символов\n"
|
||
settings_text += f"• Путь к БД: {config.DATABASE_PATH}\n\n"
|
||
settings_text += f"👑 <b>Администраторы:</b>\n"
|
||
for admin_id in config.ADMINS:
|
||
settings_text += f"• {admin_id}\n"
|
||
|
||
settings_text += "\n💡 Настройки изменяются в файле .env"
|
||
|
||
await message.answer(
|
||
settings_text,
|
||
reply_markup=get_admin_keyboard(),
|
||
parse_mode="HTML"
|
||
)
|
||
|
||
|
||
@router.callback_query(F.data == "admin_stats")
|
||
@require_permission("view_stats", "❌ У вас нет прав")
|
||
async def admin_stats_callback(callback: CallbackQuery, database: DatabaseService = None):
|
||
"""Обработчик кнопки 'Статистика' в админ панели"""
|
||
await show_stats(callback.message, database)
|
||
await callback.answer()
|
||
|
||
|
||
|
||
|
||
@router.callback_query(F.data == "admin_broadcast")
|
||
@require_permission("broadcast", "❌ У вас нет прав")
|
||
async def admin_broadcast_callback(callback: CallbackQuery):
|
||
"""Обработчик кнопки 'Рассылка' в админ панели"""
|
||
await callback.message.edit_text(
|
||
"📢 <b>Рассылка</b>\n\n"
|
||
"Функция рассылки будет добавлена в следующих версиях.\n\n"
|
||
"💡 Планируемые возможности:\n"
|
||
"• Рассылка сообщений всем пользователям\n"
|
||
"• Рассылка по группам пользователей\n"
|
||
"• Планирование рассылок\n"
|
||
"• Статистика доставки",
|
||
reply_markup=get_admin_keyboard(),
|
||
parse_mode="HTML"
|
||
)
|
||
await callback.answer()
|
||
|
||
|
||
@router.callback_query(F.data == "stats_general")
|
||
@require_permission("view_stats", "❌ У вас нет прав")
|
||
async def stats_general_callback(callback: CallbackQuery, database: DatabaseService = None):
|
||
"""Обработчик кнопки 'Общая статистика'"""
|
||
await show_stats(callback.message, database)
|
||
await callback.answer()
|
||
|
||
|
||
|
||
|
||
@router.callback_query(F.data == "back_to_admin")
|
||
@require_permission("admin_panel", "❌ У вас нет прав")
|
||
async def back_to_admin_callback(callback: CallbackQuery):
|
||
"""Обработчик кнопки 'Назад' в админ панель"""
|
||
admin_text = "👑 <b>Админ панель</b>\n\n"
|
||
admin_text += "Выберите действие:"
|
||
|
||
await callback.message.edit_text(
|
||
admin_text,
|
||
reply_markup=get_admin_keyboard(),
|
||
parse_mode="HTML"
|
||
)
|
||
await callback.answer()
|
||
|
||
|
||
@router.callback_query(F.data == "admin_assign_superuser")
|
||
@require_permission("manage_users", "❌ У вас нет прав для управления пользователями.")
|
||
async def admin_assign_superuser_callback(
|
||
callback: CallbackQuery,
|
||
user_service: UserService,
|
||
message_service: MessageService
|
||
):
|
||
"""Обработчик кнопки 'Назначить суперпользователя'"""
|
||
|
||
try:
|
||
# Получаем пользователей с пагинацией (оптимизированный запрос)
|
||
users = await user_service.database.get_all_users(limit=MAX_PAGE_SIZE, offset=MIN_PAGE_NUMBER)
|
||
|
||
if not users:
|
||
await message_service.edit_message(
|
||
callback,
|
||
"👥 <b>Управление суперпользователями</b>\n\n"
|
||
"❌ Пользователи не найдены.",
|
||
get_admin_keyboard()
|
||
)
|
||
await message_service.send_callback_answer(callback)
|
||
return
|
||
|
||
# Показываем первую страницу
|
||
await show_superuser_assignment_page(
|
||
callback,
|
||
users,
|
||
page=MIN_PAGE_NUMBER,
|
||
message_service=message_service
|
||
)
|
||
await message_service.send_callback_answer(callback)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении списка пользователей: {e}")
|
||
await message_service.edit_message(
|
||
callback,
|
||
"❌ Произошла ошибка при загрузке списка пользователей.",
|
||
get_admin_keyboard()
|
||
)
|
||
await message_service.send_callback_answer(callback)
|
||
|
||
|
||
@router.callback_query(F.data.startswith("superuser_page_"))
|
||
@require_permission("manage_users", "❌ У вас нет прав для управления пользователями.")
|
||
async def superuser_page_callback(callback: CallbackQuery, database: DatabaseService = None):
|
||
"""Обработчик пагинации списка пользователей для назначения суперпользователей"""
|
||
|
||
try:
|
||
# Извлекаем номер страницы
|
||
page = int(callback.data.split("_")[-1])
|
||
|
||
if not database:
|
||
from dependencies import get_database
|
||
database = get_database()
|
||
|
||
# Получаем всех пользователей
|
||
users = await database.get_all_users(limit=MAX_PAGE_SIZE, offset=MIN_PAGE_NUMBER)
|
||
|
||
if not users:
|
||
await callback.answer("❌ Пользователи не найдены", show_alert=True)
|
||
return
|
||
|
||
# Показываем нужную страницу
|
||
await show_superuser_assignment_page(callback, users, page)
|
||
await callback.answer()
|
||
|
||
except ValueError:
|
||
await callback.answer("❌ Неверный номер страницы", show_alert=True)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при пагинации пользователей: {e}")
|
||
await callback.answer("❌ Ошибка при загрузке страницы", show_alert=True)
|
||
|
||
|
||
@router.callback_query(F.data.startswith("assign_superuser_"))
|
||
@require_permission("manage_users", "❌ У вас нет прав для управления пользователями.")
|
||
async def assign_superuser_callback(callback: CallbackQuery, database: DatabaseService = None, validator = None):
|
||
"""Обработчик выбора пользователя для назначения суперпользователем"""
|
||
|
||
try:
|
||
# Извлекаем ID пользователя
|
||
user_id_str = callback.data.split("_")[-1]
|
||
|
||
# Валидируем callback data
|
||
if validator:
|
||
validation_result = validator.validate_callback_data(callback.data)
|
||
if not validation_result:
|
||
logger.warning(f"⚠️ Невалидный callback data от пользователя {callback.from_user.id}: {callback.data}")
|
||
await callback.answer("❌ Неверные данные", show_alert=True)
|
||
return
|
||
|
||
# Валидируем user_id
|
||
try:
|
||
user_id = int(user_id_str)
|
||
if validator:
|
||
user_id_validation = validator.validate_telegram_id(user_id)
|
||
if not user_id_validation:
|
||
logger.warning(f"⚠️ Невалидный user_id в callback: {user_id}")
|
||
await callback.answer("❌ Неверный ID пользователя", show_alert=True)
|
||
return
|
||
except ValueError:
|
||
logger.warning(f"⚠️ Неверный формат user_id в callback: {user_id_str}")
|
||
await callback.answer("❌ Неверный формат ID", show_alert=True)
|
||
return
|
||
|
||
if not database:
|
||
from dependencies import get_database
|
||
database = get_database()
|
||
|
||
# Получаем информацию о пользователе
|
||
user = await database.get_user(user_id)
|
||
if not user:
|
||
await callback.answer("❌ Пользователь не найден", show_alert=True)
|
||
return
|
||
|
||
# Формируем текст с информацией о пользователе
|
||
user_text = f"👤 <b>Информация о пользователе</b>\n\n"
|
||
user_text += f"🆔 ID: {user.telegram_id}\n"
|
||
user_text += f"👤 Имя: {user.display_name}\n"
|
||
user_text += f"📝 Полное имя: {user.full_name}\n"
|
||
user_text += f"🔗 Ссылка: {user.profile_link}\n"
|
||
created_at_str = user.created_at.strftime('%d.%m.%Y %H:%M') if user.created_at else 'Неизвестно'
|
||
user_text += f"📅 Регистрация: {created_at_str}\n"
|
||
user_text += f"✅ Активен: {'Да' if user.is_active else 'Нет'}\n"
|
||
user_text += f"🔍 Суперпользователь: {'Да' if user.is_superuser else 'Нет'}\n\n"
|
||
|
||
if user.is_superuser:
|
||
user_text += "❓ <b>Хотите снять права суперпользователя?</b>"
|
||
else:
|
||
user_text += "❓ <b>Хотите назначить суперпользователем?</b>"
|
||
|
||
# Показываем информацию и кнопки подтверждения
|
||
await callback.message.edit_text(
|
||
user_text,
|
||
reply_markup=get_superuser_confirm_keyboard(user_id, user.is_superuser),
|
||
parse_mode="HTML"
|
||
)
|
||
await callback.answer()
|
||
|
||
except ValueError:
|
||
await callback.answer("❌ Неверный ID пользователя", show_alert=True)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении информации о пользователе: {e}")
|
||
await callback.answer("❌ Ошибка при загрузке информации", show_alert=True)
|
||
|
||
|
||
@router.callback_query(F.data.startswith("confirm_superuser_"))
|
||
@require_permission("manage_users", "❌ У вас нет прав для управления пользователями.")
|
||
async def confirm_superuser_callback(callback: CallbackQuery, database: DatabaseService = None, validator = None):
|
||
"""Обработчик подтверждения назначения суперпользователем"""
|
||
|
||
try:
|
||
# Извлекаем ID пользователя
|
||
user_id_str = callback.data.split("_")[-1]
|
||
|
||
# Валидируем callback data
|
||
if validator:
|
||
validation_result = validator.validate_callback_data(callback.data)
|
||
if not validation_result:
|
||
logger.warning(f"⚠️ Невалидный callback data от пользователя {callback.from_user.id}: {callback.data}")
|
||
await callback.answer("❌ Неверные данные", show_alert=True)
|
||
return
|
||
|
||
# Валидируем user_id
|
||
try:
|
||
user_id = int(user_id_str)
|
||
if validator:
|
||
user_id_validation = validator.validate_telegram_id(user_id)
|
||
if not user_id_validation:
|
||
logger.warning(f"⚠️ Невалидный user_id в callback: {user_id}")
|
||
await callback.answer("❌ Неверный ID пользователя", show_alert=True)
|
||
return
|
||
except ValueError:
|
||
logger.warning(f"⚠️ Неверный формат user_id в callback: {user_id_str}")
|
||
await callback.answer("❌ Неверный формат ID", show_alert=True)
|
||
return
|
||
|
||
if not database:
|
||
from dependencies import get_database
|
||
database = get_database()
|
||
|
||
# Получаем пользователя
|
||
user = await database.get_user(user_id)
|
||
if not user:
|
||
await callback.answer("❌ Пользователь не найден", show_alert=True)
|
||
return
|
||
|
||
# Назначаем суперпользователем
|
||
user.is_superuser = True
|
||
await database.update_user(user)
|
||
|
||
await callback.message.edit_text(
|
||
f"✅ <b>Права назначены!</b>\n\n"
|
||
f"👤 Пользователь {user.display_name} теперь является суперпользователем.\n\n"
|
||
f"🔍 Суперпользователи могут видеть информацию об авторах вопросов.",
|
||
reply_markup=get_superuser_confirm_keyboard(user_id, True),
|
||
parse_mode="HTML"
|
||
)
|
||
await callback.answer("✅ Права суперпользователя назначены!")
|
||
|
||
except ValueError:
|
||
await callback.answer("❌ Неверный ID пользователя", show_alert=True)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при назначении суперпользователя: {e}")
|
||
await callback.answer("❌ Ошибка при назначении прав", show_alert=True)
|
||
|
||
|
||
@router.callback_query(F.data.startswith("remove_superuser_"))
|
||
@require_permission("manage_users", "❌ У вас нет прав для управления пользователями.")
|
||
async def remove_superuser_callback(callback: CallbackQuery, database: DatabaseService = None, validator = None):
|
||
"""Обработчик снятия прав суперпользователя"""
|
||
|
||
logger.info(f"🔧 Обработка снятия прав суперпользователя: {callback.data}")
|
||
|
||
try:
|
||
# Извлекаем ID пользователя
|
||
user_id_str = callback.data.split("_")[-1]
|
||
logger.info(f"🔧 Извлечен user_id: {user_id_str}")
|
||
|
||
# Валидируем callback data
|
||
if validator:
|
||
validation_result = validator.validate_callback_data(callback.data)
|
||
if not validation_result:
|
||
logger.warning(f"⚠️ Невалидный callback data от пользователя {callback.from_user.id}: {callback.data}")
|
||
await callback.answer("❌ Неверные данные", show_alert=True)
|
||
return
|
||
|
||
# Валидируем user_id
|
||
try:
|
||
user_id = int(user_id_str)
|
||
if validator:
|
||
user_id_validation = validator.validate_telegram_id(user_id)
|
||
if not user_id_validation:
|
||
logger.warning(f"⚠️ Невалидный user_id в callback: {user_id}")
|
||
await callback.answer("❌ Неверный ID пользователя", show_alert=True)
|
||
return
|
||
except ValueError:
|
||
logger.warning(f"⚠️ Неверный формат user_id в callback: {user_id_str}")
|
||
await callback.answer("❌ Неверный формат ID", show_alert=True)
|
||
return
|
||
|
||
if not database:
|
||
from dependencies import get_database
|
||
database = get_database()
|
||
|
||
# Получаем пользователя
|
||
user = await database.get_user(user_id)
|
||
if not user:
|
||
logger.warning(f"⚠️ Пользователь с ID {user_id} не найден")
|
||
await callback.answer("❌ Пользователь не найден", show_alert=True)
|
||
return
|
||
|
||
logger.info(f"🔧 Найден пользователь: {user.display_name}, текущий статус суперпользователя: {user.is_superuser}")
|
||
|
||
# Снимаем права суперпользователя
|
||
user.is_superuser = False
|
||
await database.update_user(user)
|
||
logger.info(f"✅ Права суперпользователя сняты для пользователя {user.display_name}")
|
||
|
||
await callback.message.edit_text(
|
||
f"❌ <b>Права сняты!</b>\n\n"
|
||
f"👤 Пользователь {user.display_name} больше не является суперпользователем.\n\n"
|
||
f"👤 Теперь он видит анонимные вопросы без информации об авторах.",
|
||
reply_markup=get_superuser_confirm_keyboard(user_id, False),
|
||
parse_mode="HTML"
|
||
)
|
||
await callback.answer("❌ Права суперпользователя сняты!")
|
||
|
||
except ValueError:
|
||
await callback.answer("❌ Неверный ID пользователя", show_alert=True)
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при снятии прав суперпользователя: {e}")
|
||
await callback.answer("❌ Ошибка при снятии прав", show_alert=True)
|
||
|
||
|
||
async def show_superuser_assignment_page(
|
||
callback: CallbackQuery,
|
||
users: list,
|
||
page: int = MIN_PAGE_NUMBER,
|
||
per_page: int = DEFAULT_PAGE_SIZE,
|
||
pagination_service: PaginationService = None,
|
||
message_service: MessageService = None
|
||
):
|
||
"""Показать страницу с пользователями для назначения суперпользователей"""
|
||
try:
|
||
# Используем сервисы если они переданы, иначе создаем временные
|
||
if not pagination_service:
|
||
from dependencies import get_pagination_service
|
||
pagination_service = get_pagination_service()
|
||
|
||
if not message_service:
|
||
from dependencies import get_message_service
|
||
message_service = get_message_service()
|
||
|
||
# Формируем текст сообщения
|
||
users_text = await _format_users_list(users, page, per_page, pagination_service)
|
||
|
||
# Создаем клавиатуру
|
||
keyboard = get_superuser_assignment_keyboard(users, page, per_page)
|
||
|
||
# Отправляем или редактируем сообщение
|
||
await message_service.edit_message(callback, users_text, keyboard)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при отображении страницы пользователей: {e}")
|
||
await message_service.edit_message(
|
||
callback,
|
||
"❌ Произошла ошибка при отображении списка пользователей.",
|
||
get_admin_keyboard()
|
||
)
|
||
|
||
|
||
# ===========================================
|
||
# Rate Limiting callback обработчики
|
||
# ===========================================
|
||
|
||
@router.callback_query(F.data == "admin_rate_limit")
|
||
async def admin_rate_limit_menu(
|
||
callback: CallbackQuery,
|
||
message_service: MessageService,
|
||
auth: AuthService
|
||
):
|
||
"""Показать меню rate limiting"""
|
||
try:
|
||
if not auth.is_admin(callback.from_user.id):
|
||
await callback.answer("❌ Доступ запрещен. Только для администраторов.", show_alert=True)
|
||
return
|
||
|
||
text = "🚦 <b>Управление Rate Limiting</b>\n\n"
|
||
text += "Выберите действие для управления системой ограничения скорости отправки сообщений:"
|
||
|
||
await message_service.edit_message(callback, text, get_rate_limit_keyboard())
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при отображении меню rate limiting: {e}")
|
||
await callback.answer("❌ Произошла ошибка при отображении меню.", show_alert=True)
|
||
|
||
|
||
@router.callback_query(F.data == "rate_limit_stats")
|
||
@inject_admin_services
|
||
async def rate_limit_stats_callback(
|
||
callback: CallbackQuery,
|
||
rate_limit_service: RateLimitService,
|
||
message_service: MessageService,
|
||
auth: AuthService,
|
||
**kwargs
|
||
):
|
||
"""Показать статистику rate limiting"""
|
||
try:
|
||
if not auth.is_admin(callback.from_user.id):
|
||
await callback.answer("❌ Доступ запрещен. Только для администраторов.", show_alert=True)
|
||
return
|
||
|
||
stats = rate_limit_service.get_stats()
|
||
|
||
stats_text = "📊 <b>Статистика Rate Limiting</b>\n\n"
|
||
stats_text += "🔢 <b>Общая статистика:</b>\n"
|
||
stats_text += f"• Всего запросов: {stats.get('total_requests', 0)}\n"
|
||
stats_text += f"• Успешных запросов: {stats.get('successful_requests', 0)}\n"
|
||
stats_text += f"• Неудачных запросов: {stats.get('failed_requests', 0)}\n"
|
||
stats_text += f"• Процент успеха: {stats.get('success_rate', 0.0):.{PERCENTAGE_DECIMAL_PLACES}%}\n"
|
||
stats_text += f"• Процент ошибок: {stats.get('error_rate', 0.0):.{PERCENTAGE_DECIMAL_PLACES}%}\n"
|
||
stats_text += f"• Среднее время ожидания: {stats.get('average_wait_time', 0.0):.{TIME_DECIMAL_PLACES}f}с\n\n"
|
||
|
||
stats_text += "🔍 <b>Детальная статистика:</b>\n"
|
||
stats_text += f"• RetryAfter ошибок: {stats.get('retry_after_errors', 0)}\n"
|
||
stats_text += f"• Других ошибок: {stats.get('other_errors', 0)}\n"
|
||
stats_text += f"• Процент RetryAfter: {stats.get('retry_after_rate', 0.0):.{PERCENTAGE_DECIMAL_PLACES}%}\n"
|
||
stats_text += f"• Процент других ошибок: {stats.get('other_error_rate', 0.0):.{PERCENTAGE_DECIMAL_PLACES}%}\n"
|
||
|
||
await message_service.edit_message(callback, stats_text, get_rate_limit_keyboard())
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении статистики rate limiting: {e}")
|
||
await callback.answer("❌ Произошла ошибка при получении статистики.", show_alert=True)
|
||
|
||
|
||
@router.callback_query(F.data == "rate_limit_reset_stats")
|
||
@inject_admin_services
|
||
async def reset_rate_limit_stats_callback(
|
||
callback: CallbackQuery,
|
||
rate_limit_service: RateLimitService,
|
||
message_service: MessageService,
|
||
auth: AuthService,
|
||
**kwargs
|
||
):
|
||
"""Сбросить статистику rate limiting"""
|
||
try:
|
||
if not auth.is_admin(callback.from_user.id):
|
||
await callback.answer("❌ Доступ запрещен. Только для администраторов.", show_alert=True)
|
||
return
|
||
|
||
rate_limit_service.reset_stats()
|
||
await callback.answer("✅ Статистика rate limiting сброшена.", show_alert=True)
|
||
|
||
# Обновляем сообщение
|
||
text = "🚦 <b>Управление Rate Limiting</b>\n\n"
|
||
text += "Выберите действие для управления системой ограничения скорости отправки сообщений:"
|
||
await message_service.edit_message(callback, text, get_rate_limit_keyboard())
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при сбросе статистики rate limiting: {e}")
|
||
await callback.answer("❌ Произошла ошибка при сбросе статистики.", show_alert=True)
|
||
|
||
|
||
@router.callback_query(F.data == "rate_limit_adapt")
|
||
@inject_admin_services
|
||
async def adapt_rate_limit_config_callback(
|
||
callback: CallbackQuery,
|
||
rate_limit_service: RateLimitService,
|
||
message_service: MessageService,
|
||
auth: AuthService,
|
||
**kwargs
|
||
):
|
||
"""Адаптировать конфигурацию rate limiting"""
|
||
try:
|
||
if not auth.is_admin(callback.from_user.id):
|
||
await callback.answer("❌ Доступ запрещен. Только для администраторов.", show_alert=True)
|
||
return
|
||
|
||
if rate_limit_service.should_adapt_config():
|
||
await rate_limit_service.adapt_config_if_needed()
|
||
await callback.answer("✅ Конфигурация rate limiting адаптирована на основе текущей производительности.", show_alert=True)
|
||
else:
|
||
await callback.answer("ℹ️ Адаптация конфигурации не требуется. Недостаточно данных или производительность в норме.", show_alert=True)
|
||
|
||
# Обновляем сообщение
|
||
text = "🚦 <b>Управление Rate Limiting</b>\n\n"
|
||
text += "Выберите действие для управления системой ограничения скорости отправки сообщений:"
|
||
await message_service.edit_message(callback, text, get_rate_limit_keyboard())
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при адаптации конфигурации rate limiting: {e}")
|
||
await callback.answer("❌ Произошла ошибка при адаптации конфигурации.", show_alert=True)
|
||
|
||
|