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
159 lines
5.8 KiB
Python
159 lines
5.8 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Скрипт для управления администраторами через CLI
|
||
Используется для быстрого доступа к функциям управления админами
|
||
"""
|
||
|
||
import asyncio
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
# Добавляем путь к проекту
|
||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||
|
||
from src.core.database import async_session_maker
|
||
from src.core.services import UserService
|
||
from src.core.config import ADMIN_IDS
|
||
|
||
|
||
async def list_admins():
|
||
"""Показать список всех администраторов"""
|
||
async with async_session_maker() as session:
|
||
from sqlalchemy import select
|
||
from src.core.models import User
|
||
|
||
# Получаем всех администраторов из БД
|
||
result = await session.execute(
|
||
select(User).where(User.is_admin == True).order_by(User.created_at.desc())
|
||
)
|
||
db_admins = result.scalars().all()
|
||
|
||
print("\n" + "="*60)
|
||
print("👑 СПИСОК АДМИНИСТРАТОРОВ")
|
||
print("="*60)
|
||
|
||
print("\n🔴 Главные администраторы (.env):")
|
||
if ADMIN_IDS:
|
||
for admin_id in ADMIN_IDS:
|
||
print(f" • ID: {admin_id}")
|
||
else:
|
||
print(" Нет главных администраторов")
|
||
|
||
print("\n🟠 Назначенные администраторы:")
|
||
if db_admins:
|
||
for admin in db_admins:
|
||
name = admin.first_name or admin.username or f"@ID_{admin.telegram_id}"
|
||
print(f" • {name} (Telegram ID: {admin.telegram_id})")
|
||
else:
|
||
print(" Нет назначенных администраторов")
|
||
|
||
print("\n" + "="*60 + "\n")
|
||
|
||
|
||
async def add_admin(telegram_id: int):
|
||
"""Добавить администратора"""
|
||
async with async_session_maker() as session:
|
||
# Проверяем, существует ли пользователь
|
||
user = await UserService.get_user_by_telegram_id(session, telegram_id)
|
||
|
||
if not user:
|
||
print(f"❌ Пользователь с ID {telegram_id} не найден")
|
||
return
|
||
|
||
if telegram_id in ADMIN_IDS:
|
||
print(f"❌ ID {telegram_id} - это главный администратор (.env)")
|
||
return
|
||
|
||
if user.is_admin:
|
||
print(f"❌ Пользователь {user.first_name or user.username} уже администратор")
|
||
return
|
||
|
||
# Назначаем админа
|
||
success = await UserService.set_admin(session, telegram_id, is_admin=True)
|
||
|
||
if success:
|
||
name = user.first_name or user.username or f"@ID_{telegram_id}"
|
||
print(f"✅ {name} назначен администратором")
|
||
else:
|
||
print(f"❌ Ошибка при назначении администратора")
|
||
|
||
|
||
async def remove_admin(telegram_id: int):
|
||
"""Удалить администратора"""
|
||
async with async_session_maker() as session:
|
||
if telegram_id in ADMIN_IDS:
|
||
print(f"❌ Нельзя удалить главного администратора (.env)")
|
||
print(f" Для изменения отредактируйте .env")
|
||
return
|
||
|
||
# Проверяем, существует ли пользователь
|
||
user = await UserService.get_user_by_telegram_id(session, telegram_id)
|
||
|
||
if not user:
|
||
print(f"❌ Пользователь с ID {telegram_id} не найден")
|
||
return
|
||
|
||
if not user.is_admin:
|
||
print(f"❌ Пользователь {user.first_name or user.username} не является администратором")
|
||
return
|
||
|
||
# Удаляем админа
|
||
success = await UserService.set_admin(session, telegram_id, is_admin=False)
|
||
|
||
if success:
|
||
name = user.first_name or user.username or f"@ID_{telegram_id}"
|
||
print(f"✅ Права администратора удалены у {name}")
|
||
else:
|
||
print(f"❌ Ошибка при удалении прав администратора")
|
||
|
||
|
||
async def main():
|
||
"""Главная функция"""
|
||
if len(sys.argv) < 2:
|
||
print("""
|
||
Использование: python scripts/manage_admins.py <команда> [аргументы]
|
||
|
||
Команды:
|
||
list - Показать список всех администраторов
|
||
add <id> - Добавить администратора (по Telegram ID)
|
||
remove <id> - Удалить администратора (по Telegram ID)
|
||
|
||
Примеры:
|
||
python scripts/manage_admins.py list
|
||
python scripts/manage_admins.py add 123456789
|
||
python scripts/manage_admins.py remove 123456789
|
||
""")
|
||
return
|
||
|
||
command = sys.argv[1].lower()
|
||
|
||
if command == "list":
|
||
await list_admins()
|
||
|
||
elif command == "add":
|
||
if len(sys.argv) < 3:
|
||
print("❌ Требуется указать Telegram ID")
|
||
return
|
||
try:
|
||
telegram_id = int(sys.argv[2])
|
||
await add_admin(telegram_id)
|
||
except ValueError:
|
||
print("❌ Telegram ID должен быть числом")
|
||
|
||
elif command == "remove":
|
||
if len(sys.argv) < 3:
|
||
print("❌ Требуется указать Telegram ID")
|
||
return
|
||
try:
|
||
telegram_id = int(sys.argv[2])
|
||
await remove_admin(telegram_id)
|
||
except ValueError:
|
||
print("❌ Telegram ID должен быть числом")
|
||
|
||
else:
|
||
print(f"❌ Неизвестная команда: {command}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|