feat: Add admin management system with super admin controls
Some checks failed
continuous-integration/drone/pr Build is failing
Some checks failed
continuous-integration/drone/pr Build is failing
- Implemented two-level admin hierarchy (super admin from .env and assigned admins) - Only super admins (from ADMIN_IDS in .env) can manage admin assignments - Added admin management menu to settings (visible only for super admins) - Admins can add/remove other admins through the bot interface - Protected super admins from deletion - Added CLI tool for admin management (scripts/manage_admins.py) - Added database check script (scripts/check_db.py) - Added deployment scripts for server setup - Added comprehensive documentation on admin management system - Added backup and server deployment guides
This commit is contained in:
@@ -96,6 +96,13 @@ class AdminStates(StatesGroup):
|
||||
# Управление пользователями
|
||||
user_management_search = State() # Поиск пользователей
|
||||
user_management_view = State() # Просмотр пользователя
|
||||
|
||||
# Управление админами
|
||||
admin_management_action = State() # Выбор действия (добавить/удалить)
|
||||
admin_add_search = State() # Поиск пользователя для назначения админом
|
||||
admin_add_confirm = State() # Подтверждение назначения
|
||||
admin_remove_select = State() # Выбор админа для удаления
|
||||
admin_remove_confirm = State() # Подтверждение удаления
|
||||
|
||||
|
||||
admin_router = Router()
|
||||
@@ -106,6 +113,11 @@ def is_admin(user_id: int) -> bool:
|
||||
return user_id in ADMIN_IDS
|
||||
|
||||
|
||||
def is_super_admin(user_id: int) -> bool:
|
||||
"""Проверка, является ли пользователь главным администратором (из ADMIN_IDS)"""
|
||||
return user_id in ADMIN_IDS
|
||||
|
||||
|
||||
def get_admin_main_keyboard() -> InlineKeyboardMarkup:
|
||||
"""Главная админ-панель"""
|
||||
buttons = [
|
||||
@@ -3079,14 +3091,20 @@ async def show_admin_settings(callback: CallbackQuery):
|
||||
|
||||
text += "Доступные действия:"
|
||||
|
||||
buttons = [
|
||||
buttons = []
|
||||
|
||||
# Кнопка управления админами - только для главных админов
|
||||
if is_super_admin(callback.from_user.id):
|
||||
buttons.append([InlineKeyboardButton(text="👑 Управление админами", callback_data="admin_manage_admins")])
|
||||
|
||||
buttons.extend([
|
||||
[InlineKeyboardButton(text="💿 Экспорт пользователей", callback_data="admin_export_users")],
|
||||
[InlineKeyboardButton(text="⬆️ Импорт пользователей", callback_data="admin_import_users")],
|
||||
[InlineKeyboardButton(text="💿 Экспорт данных", callback_data="admin_export_data")],
|
||||
[InlineKeyboardButton(text="🧹 Очистка старых данных", callback_data="admin_cleanup")],
|
||||
[InlineKeyboardButton(text="📜 Системная информация", callback_data="admin_system_info")],
|
||||
[InlineKeyboardButton(text="◀️ Назад", callback_data="admin_panel")]
|
||||
]
|
||||
])
|
||||
|
||||
await callback.message.edit_text(text, reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons))
|
||||
|
||||
@@ -5216,5 +5234,259 @@ async def admin_user_unban(callback: CallbackQuery, state: FSMContext):
|
||||
await callback.answer("❌ Ошибка разблокировки", show_alert=True)
|
||||
|
||||
|
||||
# =========================
|
||||
# УПРАВЛЕНИЕ АДМИНИСТРАТОРАМИ
|
||||
# =========================
|
||||
|
||||
@admin_router.callback_query(F.data == "admin_manage_admins")
|
||||
async def manage_admins_menu(callback: CallbackQuery):
|
||||
"""Главное меню управления администраторами"""
|
||||
if not is_super_admin(callback.from_user.id):
|
||||
await callback.answer("❌ Только главные администраторы могут управлять правами", show_alert=True)
|
||||
return
|
||||
|
||||
await callback.answer()
|
||||
|
||||
text = "👑 <b>Управление администраторами</b>\n\n"
|
||||
text += f"Главные администраторы (.env): <code>{len(ADMIN_IDS)}</code>\n\n"
|
||||
text += "Выберите действие:"
|
||||
|
||||
buttons = [
|
||||
[InlineKeyboardButton(text="➕ Назначить админа", callback_data="admin_add_admin")],
|
||||
[InlineKeyboardButton(text="➖ Удалить админа", callback_data="admin_remove_admin")],
|
||||
[InlineKeyboardButton(text="📋 Список админов", callback_data="admin_list_admins_view")],
|
||||
[InlineKeyboardButton(text="◀️ Назад", callback_data="admin_settings")]
|
||||
]
|
||||
|
||||
await callback.message.edit_text(
|
||||
text,
|
||||
reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons),
|
||||
parse_mode="HTML"
|
||||
)
|
||||
|
||||
|
||||
@admin_router.callback_query(F.data == "admin_list_admins_view")
|
||||
async def list_admins_view(callback: CallbackQuery):
|
||||
"""Показать список всех администраторов"""
|
||||
if not is_super_admin(callback.from_user.id):
|
||||
await callback.answer("❌ Доступ запрещен", show_alert=True)
|
||||
return
|
||||
|
||||
await callback.answer()
|
||||
|
||||
async with async_session_maker() as session:
|
||||
from sqlalchemy import select
|
||||
|
||||
# Получаем всех администраторов (назначенных через БД)
|
||||
result = await session.execute(
|
||||
select(User).where(User.is_admin == True).order_by(User.created_at.desc())
|
||||
)
|
||||
db_admins = result.scalars().all()
|
||||
|
||||
text = "👑 <b>Список администраторов</b>\n\n"
|
||||
|
||||
# Главные администраторы из .env
|
||||
text += "<b>Главные администраторы (из .env):</b>\n"
|
||||
for admin_id in ADMIN_IDS:
|
||||
text += f"🔴 ID: <code>{admin_id}</code>\n"
|
||||
|
||||
text += "\n"
|
||||
|
||||
# Назначенные администраторы
|
||||
if db_admins:
|
||||
text += "<b>Назначенные администраторы:</b>\n"
|
||||
for admin in db_admins:
|
||||
icon = "🟠" # Назначенный админ
|
||||
name = admin.first_name or admin.username or f"@ID_{admin.telegram_id}"
|
||||
text += f"{icon} {name} (ID: <code>{admin.telegram_id}</code>)\n"
|
||||
else:
|
||||
text += "<b>Назначенные администраторы:</b> нет\n"
|
||||
|
||||
buttons = [
|
||||
[InlineKeyboardButton(text="◀️ Назад", callback_data="admin_manage_admins")]
|
||||
]
|
||||
|
||||
await callback.message.edit_text(
|
||||
text,
|
||||
reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons),
|
||||
parse_mode="HTML"
|
||||
)
|
||||
|
||||
|
||||
@admin_router.callback_query(F.data == "admin_add_admin")
|
||||
async def add_admin_start(callback: CallbackQuery, state: FSMContext):
|
||||
"""Начать добавление нового администратора"""
|
||||
if not is_super_admin(callback.from_user.id):
|
||||
await callback.answer("❌ Доступ запрещен", show_alert=True)
|
||||
return
|
||||
|
||||
await callback.answer()
|
||||
|
||||
text = "👤 <b>Назначение администратора</b>\n\n"
|
||||
text += "Введите Telegram ID пользователя или его имя для поиска:"
|
||||
|
||||
await callback.message.edit_text(
|
||||
text,
|
||||
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
|
||||
[InlineKeyboardButton(text="❌ Отмена", callback_data="admin_manage_admins")]
|
||||
]),
|
||||
parse_mode="HTML"
|
||||
)
|
||||
|
||||
await state.set_state(AdminStates.admin_add_search)
|
||||
|
||||
|
||||
@admin_router.message(StateFilter(AdminStates.admin_add_search))
|
||||
async def search_user_for_admin(message: Message, state: FSMContext):
|
||||
"""Поиск пользователя для назначения админом"""
|
||||
if not is_super_admin(message.from_user.id):
|
||||
await message.answer("❌ Доступ запрещен")
|
||||
return
|
||||
|
||||
search_query = message.text.strip()
|
||||
|
||||
async with async_session_maker() as session:
|
||||
user = None
|
||||
|
||||
# Пробуем найти по ID
|
||||
try:
|
||||
telegram_id = int(search_query)
|
||||
user = await UserService.get_user_by_telegram_id(session, telegram_id)
|
||||
except ValueError:
|
||||
# Если не число, ищем по имени или username
|
||||
users = await UserService.search_users(session, search_query, limit=5)
|
||||
if users:
|
||||
user = users[0]
|
||||
|
||||
if not user:
|
||||
await message.answer("❌ Пользователь не найден")
|
||||
await state.set_state(AdminStates.admin_add_search)
|
||||
return
|
||||
|
||||
# Проверяем, не главный ли админ из .env
|
||||
if user.telegram_id in ADMIN_IDS:
|
||||
await message.answer("❌ Это главный администратор (.env). Уже имеет максимальные права")
|
||||
await state.set_state(AdminStates.admin_add_search)
|
||||
return
|
||||
|
||||
# Проверяем, не админ ли уже
|
||||
if user.is_admin:
|
||||
await message.answer("❌ Этот пользователь уже администратор")
|
||||
await state.set_state(AdminStates.admin_add_search)
|
||||
return
|
||||
|
||||
# Сохраняем в state и просим подтверждение
|
||||
await state.update_data(admin_user_id=user.id, admin_telegram_id=user.telegram_id)
|
||||
|
||||
text = "👤 <b>Подтверждение назначения администратора</b>\n\n"
|
||||
text += f"Имя: {user.first_name or 'не указано'}\n"
|
||||
text += f"Username: {user.username or 'нет'}\n"
|
||||
text += f"Telegram ID: <code>{user.telegram_id}</code>\n"
|
||||
text += f"Зарегистрирован: {user.created_at.strftime('%d.%m.%Y %H:%M') if user.created_at else 'нет'}\n\n"
|
||||
text += "Вы уверены, что хотите дать этому пользователю права администратора?"
|
||||
|
||||
await message.answer(
|
||||
text,
|
||||
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
|
||||
[InlineKeyboardButton(text="✅ Да, назначить", callback_data="admin_add_confirm_yes"),
|
||||
InlineKeyboardButton(text="❌ Отмена", callback_data="admin_manage_admins")],
|
||||
]),
|
||||
parse_mode="HTML"
|
||||
)
|
||||
|
||||
await state.set_state(AdminStates.admin_add_confirm)
|
||||
|
||||
|
||||
@admin_router.callback_query(F.data == "admin_add_confirm_yes")
|
||||
async def confirm_add_admin(callback: CallbackQuery, state: FSMContext):
|
||||
"""Подтвердить назначение админа"""
|
||||
if not is_super_admin(callback.from_user.id):
|
||||
await callback.answer("❌ Доступ запрещен", show_alert=True)
|
||||
return
|
||||
|
||||
data = await state.get_data()
|
||||
admin_telegram_id = data.get('admin_telegram_id')
|
||||
|
||||
async with async_session_maker() as session:
|
||||
success = await UserService.set_admin(session, admin_telegram_id, is_admin=True)
|
||||
|
||||
if success:
|
||||
await callback.answer("✅ Администратор успешно назначен", show_alert=True)
|
||||
await state.clear()
|
||||
await manage_admins_menu(callback)
|
||||
else:
|
||||
await callback.answer("❌ Ошибка при назначении администратора", show_alert=True)
|
||||
|
||||
|
||||
@admin_router.callback_query(F.data == "admin_remove_admin")
|
||||
async def remove_admin_start(callback: CallbackQuery, state: FSMContext):
|
||||
"""Начать удаление администратора"""
|
||||
if not is_super_admin(callback.from_user.id):
|
||||
await callback.answer("❌ Доступ запрещен", show_alert=True)
|
||||
return
|
||||
|
||||
await callback.answer()
|
||||
|
||||
async with async_session_maker() as session:
|
||||
from sqlalchemy import select
|
||||
|
||||
# Получаем всех назначенных администраторов
|
||||
result = await session.execute(
|
||||
select(User).where(User.is_admin == True).order_by(User.created_at.desc())
|
||||
)
|
||||
admins = result.scalars().all()
|
||||
|
||||
if not admins:
|
||||
await callback.answer("❌ Нет назначенных администраторов", show_alert=True)
|
||||
return
|
||||
|
||||
text = "🗑️ <b>Выберите администратора для удаления</b>\n\n"
|
||||
|
||||
buttons = []
|
||||
for admin in admins[:20]: # Максимум 20 администраторов на странице
|
||||
name = admin.first_name or admin.username or f"@ID_{admin.telegram_id}"
|
||||
buttons.append([InlineKeyboardButton(
|
||||
text=f"🟠 {name}",
|
||||
callback_data=f"admin_remove_select:{admin.telegram_id}"
|
||||
)])
|
||||
|
||||
buttons.append([InlineKeyboardButton(text="❌ Отмена", callback_data="admin_manage_admins")])
|
||||
|
||||
await callback.message.edit_text(
|
||||
text,
|
||||
reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons),
|
||||
parse_mode="HTML"
|
||||
)
|
||||
|
||||
await state.set_state(AdminStates.admin_remove_select)
|
||||
|
||||
|
||||
@admin_router.callback_query(F.data.startswith("admin_remove_select:"))
|
||||
async def confirm_remove_admin(callback: CallbackQuery, state: FSMContext):
|
||||
"""Подтвердить удаление администратора"""
|
||||
if not is_super_admin(callback.from_user.id):
|
||||
await callback.answer("❌ Доступ запрещен", show_alert=True)
|
||||
return
|
||||
|
||||
admin_telegram_id = int(callback.data.split(":")[1])
|
||||
|
||||
async with async_session_maker() as session:
|
||||
user = await UserService.get_user_by_telegram_id(session, admin_telegram_id)
|
||||
|
||||
if not user:
|
||||
await callback.answer("❌ Пользователь не найден", show_alert=True)
|
||||
return
|
||||
|
||||
# Снять права администратора
|
||||
success = await UserService.set_admin(session, admin_telegram_id, is_admin=False)
|
||||
|
||||
if success:
|
||||
await callback.answer("✅ Права администратора удалены", show_alert=True)
|
||||
await state.clear()
|
||||
await manage_admins_menu(callback)
|
||||
else:
|
||||
await callback.answer("❌ Ошибка при удалении прав", show_alert=True)
|
||||
|
||||
|
||||
# Экспорт роутера
|
||||
__all__ = ['admin_router']
|
||||
Reference in New Issue
Block a user