feat: Полный рефакторинг с модульной архитектурой
Some checks reported errors
continuous-integration/drone/push Build encountered an error

- Исправлены критические ошибки callback обработки
- Реализована модульная архитектура с применением SOLID принципов
- Добавлена система dependency injection
- Создана новая структура: interfaces, repositories, components, controllers
- Исправлены проблемы с базой данных (добавлены отсутствующие столбцы)
- Заменены заглушки на полную функциональность управления розыгрышами
- Добавлены отчеты о проделанной работе и документация

Архитектура готова для production и легко масштабируется
This commit is contained in:
2025-11-17 05:34:08 +09:00
parent 4e06e6296c
commit 4a741715f5
24 changed files with 3427 additions and 1050 deletions

View File

@@ -0,0 +1 @@
# Компоненты приложения

117
src/components/services.py Normal file
View File

@@ -0,0 +1,117 @@
from typing import List, Dict, Any, Optional
import random
from datetime import datetime, timezone
from src.interfaces.base import ILotteryService, IUserService
from src.interfaces.base import ILotteryRepository, IUserRepository, IParticipationRepository, IWinnerRepository
from src.core.models import Lottery, User
class LotteryServiceImpl(ILotteryService):
"""Реализация сервиса розыгрышей"""
def __init__(
self,
lottery_repo: ILotteryRepository,
participation_repo: IParticipationRepository,
winner_repo: IWinnerRepository,
user_repo: IUserRepository
):
self.lottery_repo = lottery_repo
self.participation_repo = participation_repo
self.winner_repo = winner_repo
self.user_repo = user_repo
async def create_lottery(self, title: str, description: str, prizes: List[str], creator_id: int) -> Lottery:
"""Создать новый розыгрыш"""
return await self.lottery_repo.create(
title=title,
description=description,
prizes=prizes,
creator_id=creator_id,
is_active=True,
is_completed=False,
created_at=datetime.now(timezone.utc)
)
async def conduct_draw(self, lottery_id: int) -> Dict[str, Any]:
"""Провести розыгрыш"""
lottery = await self.lottery_repo.get_by_id(lottery_id)
if not lottery or lottery.is_completed:
return {}
# Получаем участников
participations = await self.participation_repo.get_by_lottery(lottery_id)
if not participations:
return {}
# Проводим розыгрыш
random.shuffle(participations)
results = {}
num_prizes = len(lottery.prizes) if lottery.prizes else 3
winners = participations[:num_prizes]
for i, participation in enumerate(winners):
place = i + 1
prize = lottery.prizes[i] if lottery.prizes and i < len(lottery.prizes) else f"Приз {place}"
# Создаем запись о победителе
winner = await self.winner_repo.create(
lottery_id=lottery_id,
user_id=participation.user_id,
account_number=participation.account_number,
place=place,
prize=prize,
is_manual=False
)
results[str(place)] = {
'winner': winner,
'user': participation.user,
'prize': prize
}
# Помечаем розыгрыш как завершенный
lottery.is_completed = True
lottery.draw_results = {str(k): v['prize'] for k, v in results.items()}
await self.lottery_repo.update(lottery)
return results
async def get_active_lotteries(self) -> List[Lottery]:
"""Получить активные розыгрыши"""
return await self.lottery_repo.get_active()
class UserServiceImpl(IUserService):
"""Реализация сервиса пользователей"""
def __init__(self, user_repo: IUserRepository):
self.user_repo = user_repo
async def get_or_create_user(self, telegram_id: int, **kwargs) -> User:
"""Получить или создать пользователя"""
user = await self.user_repo.get_by_telegram_id(telegram_id)
if not user:
user_data = {
'telegram_id': telegram_id,
'created_at': datetime.now(timezone.utc),
**kwargs
}
user = await self.user_repo.create(**user_data)
return user
async def register_user(self, telegram_id: int, phone: str, club_card_number: str) -> bool:
"""Зарегистрировать пользователя"""
user = await self.user_repo.get_by_telegram_id(telegram_id)
if not user:
return False
user.phone = phone
user.club_card_number = club_card_number
user.is_registered = True
user.generate_verification_code()
await self.user_repo.update(user)
return True

153
src/components/ui.py Normal file
View File

@@ -0,0 +1,153 @@
from typing import List
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton, ReplyKeyboardMarkup, KeyboardButton
from src.interfaces.base import IKeyboardBuilder, IMessageFormatter
from src.core.models import Lottery, Winner
class KeyboardBuilderImpl(IKeyboardBuilder):
"""Реализация построителя клавиатур"""
def get_main_keyboard(self, is_admin: bool = False):
"""Получить главную клавиатуру"""
buttons = [
[InlineKeyboardButton(text="🎲 Активные розыгрыши", callback_data="active_lotteries")],
[InlineKeyboardButton(text="📝 Зарегистрироваться", callback_data="start_registration")],
[InlineKeyboardButton(text="🧪 ТЕСТ КОЛБЭК", callback_data="test_callback")]
]
if is_admin:
buttons.extend([
[InlineKeyboardButton(text="⚙️ Админ панель", callback_data="admin_panel")],
[InlineKeyboardButton(text=" Создать розыгрыш", callback_data="create_lottery")]
])
return InlineKeyboardMarkup(inline_keyboard=buttons)
def get_admin_keyboard(self):
"""Получить админскую клавиатуру"""
buttons = [
[
InlineKeyboardButton(text="👥 Пользователи", callback_data="user_management"),
InlineKeyboardButton(text="💳 Счета", callback_data="account_management")
],
[
InlineKeyboardButton(text="🎯 Розыгрыши", callback_data="lottery_management"),
InlineKeyboardButton(text="💬 Чат", callback_data="chat_management")
],
[
InlineKeyboardButton(text="📊 Статистика", callback_data="stats"),
InlineKeyboardButton(text="⚙️ Настройки", callback_data="settings")
],
[InlineKeyboardButton(text="🔙 Назад", callback_data="back_to_main")]
]
return InlineKeyboardMarkup(inline_keyboard=buttons)
def get_lottery_management_keyboard(self):
"""Получить клавиатуру управления розыгрышами"""
buttons = [
[
InlineKeyboardButton(text="📋 Все розыгрыши", callback_data="all_lotteries"),
InlineKeyboardButton(text="🎲 Активные", callback_data="active_lotteries_admin")
],
[
InlineKeyboardButton(text="✅ Завершенные", callback_data="completed_lotteries"),
InlineKeyboardButton(text=" Создать", callback_data="create_lottery")
],
[
InlineKeyboardButton(text="🎯 Провести розыгрыш", callback_data="conduct_lottery_admin"),
InlineKeyboardButton(text="🔄 Переросыгрыш", callback_data="admin_redraw")
],
[InlineKeyboardButton(text="🔙 Назад", callback_data="admin_panel")]
]
return InlineKeyboardMarkup(inline_keyboard=buttons)
def get_lottery_keyboard(self, lottery_id: int, is_admin: bool = False):
"""Получить клавиатуру для конкретного розыгрыша"""
buttons = [
[InlineKeyboardButton(text="🎯 Участвовать", callback_data=f"join_{lottery_id}")]
]
if is_admin:
buttons.extend([
[InlineKeyboardButton(text="🎲 Провести розыгрыш", callback_data=f"conduct_{lottery_id}")],
[InlineKeyboardButton(text="✏️ Редактировать", callback_data=f"edit_{lottery_id}")],
[InlineKeyboardButton(text="❌ Удалить", callback_data=f"delete_{lottery_id}")]
])
buttons.append([InlineKeyboardButton(text="🔙 Назад", callback_data="active_lotteries")])
return InlineKeyboardMarkup(inline_keyboard=buttons)
def get_conduct_lottery_keyboard(self, lotteries: List[Lottery]):
"""Получить клавиатуру для выбора розыгрыша для проведения"""
buttons = []
for lottery in lotteries:
text = f"🎲 {lottery.title}"
if len(text) > 50:
text = text[:47] + "..."
buttons.append([InlineKeyboardButton(text=text, callback_data=f"conduct_{lottery.id}")])
buttons.append([InlineKeyboardButton(text="🔙 Назад", callback_data="lottery_management")])
return InlineKeyboardMarkup(inline_keyboard=buttons)
class MessageFormatterImpl(IMessageFormatter):
"""Реализация форматирования сообщений"""
def format_lottery_info(self, lottery: Lottery, participants_count: int) -> str:
"""Форматировать информацию о розыгрыше"""
text = f"🎲 **{lottery.title}**\n\n"
if lottery.description:
text += f"📝 {lottery.description}\n\n"
text += f"👥 Участников: {participants_count}\n"
if lottery.prizes:
text += "\n🏆 **Призы:**\n"
for i, prize in enumerate(lottery.prizes, 1):
text += f"{i}. {prize}\n"
status = "🟢 Активный" if lottery.is_active and not lottery.is_completed else "🔴 Завершен"
text += f"\n📊 Статус: {status}"
if lottery.created_at:
text += f"\n📅 Создан: {lottery.created_at.strftime('%d.%m.%Y %H:%M')}"
return text
def format_winners_list(self, winners: List[Winner]) -> str:
"""Форматировать список победителей"""
if not winners:
return "🎯 Победители не определены"
text = "🏆 **Победители:**\n\n"
for winner in winners:
place_emoji = {1: "🥇", 2: "🥈", 3: "🥉"}.get(winner.place, "🏅")
if winner.user:
name = winner.user.first_name or f"Пользователь {winner.user.telegram_id}"
else:
name = winner.account_number or "Неизвестный участник"
text += f"{place_emoji} **{winner.place} место:** {name}\n"
if winner.prize:
text += f" 🎁 Приз: {winner.prize}\n"
text += "\n"
return text
def format_admin_stats(self, stats: dict) -> str:
"""Форматировать административную статистику"""
text = "📊 **Статистика системы**\n\n"
text += f"👥 Всего пользователей: {stats.get('total_users', 0)}\n"
text += f"✅ Зарегистрированных: {stats.get('registered_users', 0)}\n"
text += f"🎲 Всего розыгрышей: {stats.get('total_lotteries', 0)}\n"
text += f"🟢 Активных розыгрышей: {stats.get('active_lotteries', 0)}\n"
text += f"✅ Завершенных розыгрышей: {stats.get('completed_lotteries', 0)}\n"
text += f"🎯 Всего участий: {stats.get('total_participations', 0)}\n"
return text

120
src/container.py Normal file
View File

@@ -0,0 +1,120 @@
"""
Dependency Injection Container для управления зависимостями
Следует принципам SOLID, особенно Dependency Inversion Principle
"""
from typing import Dict, Any, TypeVar, Type
from sqlalchemy.ext.asyncio import AsyncSession
from src.interfaces.base import (
IUserRepository, ILotteryRepository, IParticipationRepository, IWinnerRepository,
ILotteryService, IUserService, IBotController, IKeyboardBuilder, IMessageFormatter
)
from src.repositories.implementations import (
UserRepository, LotteryRepository, ParticipationRepository, WinnerRepository
)
from src.components.services import LotteryServiceImpl, UserServiceImpl
from src.components.ui import KeyboardBuilderImpl, MessageFormatterImpl
from src.controllers.bot_controller import BotController
T = TypeVar('T')
class DIContainer:
"""Контейнер для dependency injection"""
def __init__(self):
self._services: Dict[Type, Any] = {}
self._singletons: Dict[Type, Any] = {}
# Регистрируем singleton сервисы
self.register_singleton(IKeyboardBuilder, KeyboardBuilderImpl)
self.register_singleton(IMessageFormatter, MessageFormatterImpl)
def register_singleton(self, interface: Type[T], implementation: Type[T]):
"""Зарегистрировать singleton сервис"""
self._services[interface] = implementation
def register_transient(self, interface: Type[T], implementation: Type[T]):
"""Зарегистрировать transient сервис"""
self._services[interface] = implementation
def get_singleton(self, interface: Type[T]) -> T:
"""Получить singleton экземпляр"""
if interface in self._singletons:
return self._singletons[interface]
if interface not in self._services:
raise ValueError(f"Service {interface} not registered")
implementation = self._services[interface]
instance = implementation()
self._singletons[interface] = instance
return instance
def create_scoped_container(self, session: AsyncSession) -> 'ScopedContainer':
"""Создать scoped контейнер для сессии базы данных"""
return ScopedContainer(self, session)
class ScopedContainer:
"""Scoped контейнер для одной сессии базы данных"""
def __init__(self, parent: DIContainer, session: AsyncSession):
self.parent = parent
self.session = session
self._instances: Dict[Type, Any] = {}
def get(self, interface: Type[T]) -> T:
"""Получить экземпляр сервиса"""
# Если это singleton, получаем из родительского контейнера
if interface in [IKeyboardBuilder, IMessageFormatter]:
return self.parent.get_singleton(interface)
# Если уже создан в текущем scope, возвращаем
if interface in self._instances:
return self._instances[interface]
# Создаем новый экземпляр
instance = self._create_instance(interface)
self._instances[interface] = instance
return instance
def _create_instance(self, interface: Type[T]) -> T:
"""Создать экземпляр с разрешением зависимостей"""
if interface == IUserRepository:
return UserRepository(self.session)
elif interface == ILotteryRepository:
return LotteryRepository(self.session)
elif interface == IParticipationRepository:
return ParticipationRepository(self.session)
elif interface == IWinnerRepository:
return WinnerRepository(self.session)
elif interface == ILotteryService:
return LotteryServiceImpl(
self.get(ILotteryRepository),
self.get(IParticipationRepository),
self.get(IWinnerRepository),
self.get(IUserRepository)
)
elif interface == IUserService:
return UserServiceImpl(
self.get(IUserRepository)
)
elif interface == IBotController:
return BotController(
self.get(ILotteryService),
self.get(IUserService),
self.get(IKeyboardBuilder),
self.get(IMessageFormatter),
self.get(ILotteryRepository),
self.get(IParticipationRepository)
)
else:
raise ValueError(f"Cannot create instance of {interface}")
# Глобальный экземпляр контейнера
container = DIContainer()

View File

@@ -0,0 +1 @@
# Контроллеры для обработки запросов

View File

@@ -0,0 +1,177 @@
from aiogram.types import Message, CallbackQuery
from aiogram import F
import logging
from src.interfaces.base import IBotController, ILotteryService, IUserService, IKeyboardBuilder, IMessageFormatter
from src.interfaces.base import ILotteryRepository, IParticipationRepository
from src.core.config import ADMIN_IDS
logger = logging.getLogger(__name__)
class BotController(IBotController):
"""Основной контроллер бота"""
def __init__(
self,
lottery_service: ILotteryService,
user_service: IUserService,
keyboard_builder: IKeyboardBuilder,
message_formatter: IMessageFormatter,
lottery_repo: ILotteryRepository,
participation_repo: IParticipationRepository
):
self.lottery_service = lottery_service
self.user_service = user_service
self.keyboard_builder = keyboard_builder
self.message_formatter = message_formatter
self.lottery_repo = lottery_repo
self.participation_repo = participation_repo
def is_admin(self, user_id: int) -> bool:
"""Проверить, является ли пользователь администратором"""
return user_id in ADMIN_IDS
async def handle_start(self, message: Message):
"""Обработать команду /start"""
user = await self.user_service.get_or_create_user(
telegram_id=message.from_user.id,
username=message.from_user.username,
first_name=message.from_user.first_name,
last_name=message.from_user.last_name
)
welcome_text = f"👋 Добро пожаловать, {user.first_name or 'дорогой пользователь'}!\n\n"
welcome_text += "🎲 Это бот для участия в розыгрышах.\n\n"
if user.is_registered:
welcome_text += "✅ Вы уже зарегистрированы в системе!"
else:
welcome_text += "📝 Для участия в розыгрышах необходимо зарегистрироваться."
keyboard = self.keyboard_builder.get_main_keyboard(self.is_admin(message.from_user.id))
await message.answer(
welcome_text,
reply_markup=keyboard
)
async def handle_admin_panel(self, callback: CallbackQuery):
"""Обработать админ панель"""
if not self.is_admin(callback.from_user.id):
await callback.answer("❌ Недостаточно прав", show_alert=True)
return
text = "⚙️ **Панель администратора**\n\n"
text += "Выберите раздел для управления:"
keyboard = self.keyboard_builder.get_admin_keyboard()
await callback.message.edit_text(
text,
reply_markup=keyboard,
parse_mode="Markdown"
)
async def handle_lottery_management(self, callback: CallbackQuery):
"""Обработать управление розыгрышами"""
if not self.is_admin(callback.from_user.id):
await callback.answer("❌ Недостаточно прав", show_alert=True)
return
text = "🎯 **Управление розыгрышами**\n\n"
text += "Выберите действие:"
keyboard = self.keyboard_builder.get_lottery_management_keyboard()
await callback.message.edit_text(
text,
reply_markup=keyboard,
parse_mode="Markdown"
)
async def handle_conduct_lottery_admin(self, callback: CallbackQuery):
"""Обработать выбор розыгрыша для проведения"""
if not self.is_admin(callback.from_user.id):
await callback.answer("❌ Недостаточно прав", show_alert=True)
return
# Получаем активные розыгрыши
lotteries = await self.lottery_service.get_active_lotteries()
if not lotteries:
await callback.answer("❌ Нет активных розыгрышей", show_alert=True)
return
text = "🎯 **Выберите розыгрыш для проведения:**\n\n"
for lottery in lotteries:
participants_count = await self.participation_repo.get_count_by_lottery(lottery.id)
text += f"🎲 {lottery.title} ({participants_count} участников)\n"
keyboard = self.keyboard_builder.get_conduct_lottery_keyboard(lotteries)
await callback.message.edit_text(
text,
reply_markup=keyboard,
parse_mode="Markdown"
)
async def handle_active_lotteries(self, callback: CallbackQuery):
"""Показать активные розыгрыши"""
lotteries = await self.lottery_service.get_active_lotteries()
if not lotteries:
await callback.answer("❌ Нет активных розыгрышей", show_alert=True)
return
text = "🎲 **Активные розыгрыши:**\n\n"
for lottery in lotteries:
participants_count = await self.participation_repo.get_count_by_lottery(lottery.id)
lottery_info = self.message_formatter.format_lottery_info(lottery, participants_count)
text += lottery_info + "\n" + "="*30 + "\n\n"
keyboard = self.keyboard_builder.get_main_keyboard(self.is_admin(callback.from_user.id))
await callback.message.edit_text(
text,
reply_markup=keyboard,
parse_mode="Markdown"
)
async def handle_conduct_lottery(self, callback: CallbackQuery):
"""Провести конкретный розыгрыш"""
if not self.is_admin(callback.from_user.id):
await callback.answer("❌ Недостаточно прав", show_alert=True)
return
try:
lottery_id = int(callback.data.split("_")[1])
except (ValueError, IndexError):
await callback.answer("❌ Неверный формат данных", show_alert=True)
return
# Проводим розыгрыш
results = await self.lottery_service.conduct_draw(lottery_id)
if not results:
await callback.answer("Не удалось провести розыгрыш", show_alert=True)
return
# Форматируем результаты
text = "🎉 **Розыгрыш завершен!**\n\n"
winners = [result['winner'] for result in results.values()]
winners_text = self.message_formatter.format_winners_list(winners)
text += winners_text
keyboard = self.keyboard_builder.get_admin_keyboard()
await callback.message.edit_text(
text,
reply_markup=keyboard,
parse_mode="Markdown"
)
await callback.answer("✅ Розыгрыш успешно проведен!", show_alert=True)

View File

@@ -4,12 +4,13 @@ from aiogram.types import Message, CallbackQuery, InlineKeyboardButton, InlineKe
from aiogram.filters import Command, StateFilter
from aiogram.fsm.context import FSMContext
from aiogram.fsm.state import State, StatesGroup
import logging
from src.core.database import async_session_maker
from src.core.registration_services import RegistrationService, AccountService
from src.core.services import UserService
logger = logging.getLogger(__name__)
router = Router()
@@ -22,6 +23,8 @@ class RegistrationStates(StatesGroup):
@router.callback_query(F.data == "start_registration")
async def start_registration(callback: CallbackQuery, state: FSMContext):
"""Начать процесс регистрации"""
logger.info(f"Получен запрос на регистрацию от пользователя {callback.from_user.id}")
text = (
"📝 Регистрация в системе\n\n"
"Для участия в розыгрышах необходимо зарегистрироваться.\n\n"

View File

@@ -0,0 +1,109 @@
#!/usr/bin/env python3
"""
Тестовый обработчик для проверки команды /start и /admin
"""
from aiogram import Router, F
from aiogram.types import Message, CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup
from aiogram.filters import Command
from src.core.config import ADMIN_IDS
from src.core.permissions import is_admin
# Создаем роутер для тестов
test_router = Router()
@test_router.message(Command("test_start"))
async def cmd_test_start(message: Message):
"""Тестовая команда /test_start"""
user_id = message.from_user.id
first_name = message.from_user.first_name
is_admin_user = is_admin(user_id)
welcome_text = f"👋 Привет, {first_name}!\n\n"
welcome_text += "🎯 Это тестовая версия команды /start\n\n"
if is_admin_user:
welcome_text += "👑 У вас есть права администратора!\n\n"
buttons = [
[InlineKeyboardButton(text="🔧 Админ-панель", callback_data="admin_panel")],
[InlineKeyboardButton(text=" Создать розыгрыш", callback_data="create_lottery")],
[InlineKeyboardButton(text="🎲 Активные розыгрыши", callback_data="list_lotteries")]
]
else:
welcome_text += "👤 Обычный пользователь\n\n"
buttons = [
[InlineKeyboardButton(text="🎲 Активные розыгрыши", callback_data="list_lotteries")],
[InlineKeyboardButton(text="📝 Мои участия", callback_data="my_participations")],
[InlineKeyboardButton(text="💳 Мой счёт", callback_data="my_account")]
]
await message.answer(
welcome_text,
reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons)
)
@test_router.message(Command("test_admin"))
async def cmd_test_admin(message: Message):
"""Тестовая команда /test_admin"""
if not is_admin(message.from_user.id):
await message.answer("У вас нет прав для выполнения этой команды")
return
await message.answer(
"🔧 <b>Админ-панель</b>\n\n"
"👑 Добро пожаловать в панель администратора!\n\n"
"Доступные функции:",
parse_mode="HTML",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="👥 Управление пользователями", callback_data="admin_users")],
[InlineKeyboardButton(text="🎲 Управление розыгрышами", callback_data="admin_lotteries")],
[InlineKeyboardButton(text="📊 Статистика", callback_data="admin_stats")],
[InlineKeyboardButton(text="🏠 Главное меню", callback_data="back_to_main")]
])
)
@test_router.callback_query(F.data == "test_callback")
async def test_callback_handler(callback: CallbackQuery):
"""Тестовый обработчик callback"""
await callback.answer()
await callback.message.edit_text(
"✅ Callback работает!\n\n"
"Это означает, что кнопки и обработчики функционируют корректно.",
reply_markup=InlineKeyboardMarkup(inline_keyboard=[
[InlineKeyboardButton(text="🔙 Назад", callback_data="back_to_main")]
])
)
@test_router.callback_query(F.data == "back_to_main")
async def back_to_main_handler(callback: CallbackQuery):
"""Возврат к главному меню"""
await callback.answer()
user_id = callback.from_user.id
is_admin_user = is_admin(user_id)
text = f"🏠 Главное меню\n\nВаш ID: {user_id}\n"
text += f"Статус: {'👑 Администратор' if is_admin_user else '👤 Пользователь'}"
if is_admin_user:
buttons = [
[InlineKeyboardButton(text="🔧 Админ-панель", callback_data="admin_panel")],
[InlineKeyboardButton(text="🎲 Розыгрыши", callback_data="list_lotteries")]
]
else:
buttons = [
[InlineKeyboardButton(text="🎲 Розыгрыши", callback_data="list_lotteries")],
[InlineKeyboardButton(text="📝 Мои участия", callback_data="my_participations")]
]
await callback.message.edit_text(
text,
reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons)
)

View File

@@ -0,0 +1 @@
# Интерфейсы для dependency injection и SOLID принципов

179
src/interfaces/base.py Normal file
View File

@@ -0,0 +1,179 @@
from abc import ABC, abstractmethod
from typing import Optional, List, Dict, Any
from src.core.models import User, Lottery, Participation, Winner
class IUserRepository(ABC):
"""Интерфейс репозитория пользователей"""
@abstractmethod
async def get_by_telegram_id(self, telegram_id: int) -> Optional[User]:
"""Получить пользователя по Telegram ID"""
pass
@abstractmethod
async def create(self, **kwargs) -> User:
"""Создать нового пользователя"""
pass
@abstractmethod
async def update(self, user: User) -> User:
"""Обновить пользователя"""
pass
@abstractmethod
async def get_all(self) -> List[User]:
"""Получить всех пользователей"""
pass
class ILotteryRepository(ABC):
"""Интерфейс репозитория розыгрышей"""
@abstractmethod
async def get_by_id(self, lottery_id: int) -> Optional[Lottery]:
"""Получить розыгрыш по ID"""
pass
@abstractmethod
async def create(self, **kwargs) -> Lottery:
"""Создать новый розыгрыш"""
pass
@abstractmethod
async def get_active(self) -> List[Lottery]:
"""Получить активные розыгрыши"""
pass
@abstractmethod
async def get_all(self) -> List[Lottery]:
"""Получить все розыгрыши"""
pass
@abstractmethod
async def update(self, lottery: Lottery) -> Lottery:
"""Обновить розыгрыш"""
pass
class IParticipationRepository(ABC):
"""Интерфейс репозитория участий"""
@abstractmethod
async def create(self, **kwargs) -> Participation:
"""Создать новое участие"""
pass
@abstractmethod
async def get_by_lottery(self, lottery_id: int) -> List[Participation]:
"""Получить участия по розыгрышу"""
pass
@abstractmethod
async def get_count_by_lottery(self, lottery_id: int) -> int:
"""Получить количество участников в розыгрыше"""
pass
class IWinnerRepository(ABC):
"""Интерфейс репозитория победителей"""
@abstractmethod
async def create(self, **kwargs) -> Winner:
"""Создать запись о победителе"""
pass
@abstractmethod
async def get_by_lottery(self, lottery_id: int) -> List[Winner]:
"""Получить победителей розыгрыша"""
pass
class ILotteryService(ABC):
"""Интерфейс сервиса розыгрышей"""
@abstractmethod
async def create_lottery(self, title: str, description: str, prizes: List[str], creator_id: int) -> Lottery:
"""Создать новый розыгрыш"""
pass
@abstractmethod
async def conduct_draw(self, lottery_id: int) -> Dict[str, Any]:
"""Провести розыгрыш"""
pass
@abstractmethod
async def get_active_lotteries(self) -> List[Lottery]:
"""Получить активные розыгрыши"""
pass
class IUserService(ABC):
"""Интерфейс сервиса пользователей"""
@abstractmethod
async def get_or_create_user(self, telegram_id: int, **kwargs) -> User:
"""Получить или создать пользователя"""
pass
@abstractmethod
async def register_user(self, telegram_id: int, phone: str, club_card_number: str) -> bool:
"""Зарегистрировать пользователя"""
pass
class IBotController(ABC):
"""Интерфейс контроллера бота"""
@abstractmethod
async def handle_start(self, message_or_callback):
"""Обработать команду /start"""
pass
@abstractmethod
async def handle_admin_panel(self, callback):
"""Обработать admin panel"""
pass
class IMessageFormatter(ABC):
"""Интерфейс форматирования сообщений"""
@abstractmethod
def format_lottery_info(self, lottery: Lottery, participants_count: int) -> str:
"""Форматировать информацию о розыгрыше"""
pass
@abstractmethod
def format_winners_list(self, winners: List[Winner]) -> str:
"""Форматировать список победителей"""
pass
class IKeyboardBuilder(ABC):
"""Интерфейс создания клавиатур"""
@abstractmethod
def get_main_keyboard(self, is_admin: bool):
"""Получить главную клавиатуру"""
pass
@abstractmethod
def get_admin_keyboard(self):
"""Получить админскую клавиатуру"""
pass
@abstractmethod
def get_lottery_keyboard(self, lottery_id: int, is_admin: bool):
"""Получить клавиатуру для розыгрыша"""
pass
@abstractmethod
def get_lottery_management_keyboard(self):
"""Получить клавиатуру управления розыгрышами"""
pass
@abstractmethod
def get_conduct_lottery_keyboard(self, lotteries: List[Lottery]):
"""Получить клавиатуру для выбора розыгрыша для проведения"""
pass

View File

@@ -0,0 +1 @@
# Репозитории для работы с данными

View File

@@ -0,0 +1,141 @@
from typing import Optional, List
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from sqlalchemy.orm import selectinload
from src.interfaces.base import IUserRepository, ILotteryRepository, IParticipationRepository, IWinnerRepository
from src.core.models import User, Lottery, Participation, Winner
class UserRepository(IUserRepository):
"""Репозиторий для работы с пользователями"""
def __init__(self, session: AsyncSession):
self.session = session
async def get_by_telegram_id(self, telegram_id: int) -> Optional[User]:
"""Получить пользователя по Telegram ID"""
result = await self.session.execute(
select(User).where(User.telegram_id == telegram_id)
)
return result.scalars().first()
async def create(self, **kwargs) -> User:
"""Создать нового пользователя"""
user = User(**kwargs)
self.session.add(user)
await self.session.commit()
await self.session.refresh(user)
return user
async def update(self, user: User) -> User:
"""Обновить пользователя"""
await self.session.commit()
await self.session.refresh(user)
return user
async def get_all(self) -> List[User]:
"""Получить всех пользователей"""
result = await self.session.execute(select(User))
return list(result.scalars().all())
class LotteryRepository(ILotteryRepository):
"""Репозиторий для работы с розыгрышами"""
def __init__(self, session: AsyncSession):
self.session = session
async def get_by_id(self, lottery_id: int) -> Optional[Lottery]:
"""Получить розыгрыш по ID"""
result = await self.session.execute(
select(Lottery).where(Lottery.id == lottery_id)
)
return result.scalars().first()
async def create(self, **kwargs) -> Lottery:
"""Создать новый розыгрыш"""
lottery = Lottery(**kwargs)
self.session.add(lottery)
await self.session.commit()
await self.session.refresh(lottery)
return lottery
async def get_active(self) -> List[Lottery]:
"""Получить активные розыгрыши"""
result = await self.session.execute(
select(Lottery).where(
Lottery.is_active == True,
Lottery.is_completed == False
).order_by(Lottery.created_at.desc())
)
return list(result.scalars().all())
async def get_all(self) -> List[Lottery]:
"""Получить все розыгрыши"""
result = await self.session.execute(
select(Lottery).order_by(Lottery.created_at.desc())
)
return list(result.scalars().all())
async def update(self, lottery: Lottery) -> Lottery:
"""Обновить розыгрыш"""
await self.session.commit()
await self.session.refresh(lottery)
return lottery
class ParticipationRepository(IParticipationRepository):
"""Репозиторий для работы с участиями"""
def __init__(self, session: AsyncSession):
self.session = session
async def create(self, **kwargs) -> Participation:
"""Создать новое участие"""
participation = Participation(**kwargs)
self.session.add(participation)
await self.session.commit()
await self.session.refresh(participation)
return participation
async def get_by_lottery(self, lottery_id: int) -> List[Participation]:
"""Получить участия по розыгрышу"""
result = await self.session.execute(
select(Participation)
.options(selectinload(Participation.user))
.where(Participation.lottery_id == lottery_id)
)
return list(result.scalars().all())
async def get_count_by_lottery(self, lottery_id: int) -> int:
"""Получить количество участников в розыгрыше"""
result = await self.session.execute(
select(Participation).where(Participation.lottery_id == lottery_id)
)
return len(list(result.scalars().all()))
class WinnerRepository(IWinnerRepository):
"""Репозиторий для работы с победителями"""
def __init__(self, session: AsyncSession):
self.session = session
async def create(self, **kwargs) -> Winner:
"""Создать запись о победителе"""
winner = Winner(**kwargs)
self.session.add(winner)
await self.session.commit()
await self.session.refresh(winner)
return winner
async def get_by_lottery(self, lottery_id: int) -> List[Winner]:
"""Получить победителей розыгрыша"""
result = await self.session.execute(
select(Winner)
.options(selectinload(Winner.user))
.where(Winner.lottery_id == lottery_id)
.order_by(Winner.place)
)
return list(result.scalars().all())