124 lines
4.9 KiB
Python
124 lines
4.9 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Утилиты для управления ботом
|
||
"""
|
||
import asyncio
|
||
import sys
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
from database import async_session_maker, init_db
|
||
from services import UserService
|
||
from config import ADMIN_IDS
|
||
|
||
|
||
async def setup_admin_users():
|
||
"""Установить права администратора для пользователей из ADMIN_IDS"""
|
||
if not ADMIN_IDS:
|
||
print("❌ Список ADMIN_IDS пуст")
|
||
return
|
||
|
||
async with async_session_maker() as session:
|
||
for admin_id in ADMIN_IDS:
|
||
success = await UserService.set_admin(session, admin_id, True)
|
||
if success:
|
||
print(f"✅ Права администратора установлены для ID: {admin_id}")
|
||
else:
|
||
print(f"⚠️ Пользователь с ID {admin_id} не найден в базе")
|
||
|
||
|
||
async def create_sample_lottery():
|
||
"""Создать пример розыгрыша для тестирования"""
|
||
from services import LotteryService
|
||
|
||
async with async_session_maker() as session:
|
||
# Берем первого администратора как создателя
|
||
if not ADMIN_IDS:
|
||
print("❌ Нет администраторов для создания розыгрыша")
|
||
return
|
||
|
||
admin_user = await UserService.get_user_by_telegram_id(session, ADMIN_IDS[0])
|
||
if not admin_user:
|
||
print("❌ Пользователь-администратор не найден в базе")
|
||
return
|
||
|
||
lottery = await LotteryService.create_lottery(
|
||
session,
|
||
title="🎉 Тестовый розыгрыш",
|
||
description="Это тестовый розыгрыш для демонстрации работы бота",
|
||
prizes=[
|
||
"🥇 Главный приз - 10,000 рублей",
|
||
"🥈 Второй приз - iPhone 15",
|
||
"🥉 Третий приз - AirPods Pro"
|
||
],
|
||
creator_id=admin_user.id
|
||
)
|
||
|
||
print(f"✅ Создан тестовый розыгрыш с ID: {lottery.id}")
|
||
print(f"📝 Название: {lottery.title}")
|
||
|
||
|
||
async def init_database():
|
||
"""Инициализация базы данных"""
|
||
print("🔄 Инициализация базы данных...")
|
||
await init_db()
|
||
print("✅ База данных инициализирована")
|
||
|
||
|
||
async def show_stats():
|
||
"""Показать статистику бота"""
|
||
from services import LotteryService, ParticipationService
|
||
from models import User, Lottery, Participation
|
||
from sqlalchemy import select, func
|
||
|
||
async with async_session_maker() as session:
|
||
# Количество пользователей
|
||
result = await session.execute(select(func.count(User.id)))
|
||
users_count = result.scalar()
|
||
|
||
# Количество розыгрышей
|
||
result = await session.execute(select(func.count(Lottery.id)))
|
||
lotteries_count = result.scalar()
|
||
|
||
# Количество активных розыгрышей
|
||
result = await session.execute(
|
||
select(func.count(Lottery.id))
|
||
.where(Lottery.is_active == True, Lottery.is_completed == False)
|
||
)
|
||
active_lotteries = result.scalar()
|
||
|
||
# Количество участий
|
||
result = await session.execute(select(func.count(Participation.id)))
|
||
participations_count = result.scalar()
|
||
|
||
print("\n📊 Статистика бота:")
|
||
print(f"👥 Всего пользователей: {users_count}")
|
||
print(f"🎲 Всего розыгрышей: {lotteries_count}")
|
||
print(f"🟢 Активных розыгрышей: {active_lotteries}")
|
||
print(f"🎫 Всего участий: {participations_count}")
|
||
|
||
|
||
def main():
|
||
"""Главная функция утилиты"""
|
||
if len(sys.argv) < 2:
|
||
print("Использование:")
|
||
print(" python utils.py init - Инициализация базы данных")
|
||
print(" python utils.py setup-admins - Установка прав администратора")
|
||
print(" python utils.py sample - Создание тестового розыгрыша")
|
||
print(" python utils.py stats - Показать статистику")
|
||
return
|
||
|
||
command = sys.argv[1]
|
||
|
||
if command == "init":
|
||
asyncio.run(init_database())
|
||
elif command == "setup-admins":
|
||
asyncio.run(setup_admin_users())
|
||
elif command == "sample":
|
||
asyncio.run(create_sample_lottery())
|
||
elif command == "stats":
|
||
asyncio.run(show_stats())
|
||
else:
|
||
print(f"❌ Неизвестная команда: {command}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |