19c5fdd0ac
✅ Модель InitiativeGroup в БД ✅ Сервис initiative_group.py ✅ Хендлеры: /ig_add, /ig_remove, /ig_list, /ig_stats, /ig_check, /ig_broadcast ✅ Декораторы для ИГ ✅ Миграция v2.1 выполнена 📦 Бекап: backups/versions/v2.1_initiative_group_2026-02-27/ Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
100 lines
4 KiB
Python
100 lines
4 KiB
Python
"""
|
|
Декораторы для хендлеров
|
|
"""
|
|
from functools import wraps
|
|
from aiogram.types import Message, CallbackQuery
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select
|
|
from database.models import User
|
|
import config
|
|
|
|
|
|
async def get_user_from_db(session: AsyncSession, user_id: int) -> User | None:
|
|
"""Получить пользователя из БД"""
|
|
result = await session.execute(select(User).where(User.user_id == user_id))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
def admin_only(func):
|
|
"""Только для администраторов"""
|
|
@wraps(func)
|
|
async def wrapper(message: Message, *args, **kwargs):
|
|
if not config.is_admin(message.from_user.id):
|
|
await message.answer('❌ Эта команда доступна только администраторам')
|
|
return
|
|
return await func(message, *args, **kwargs)
|
|
return wrapper
|
|
|
|
|
|
def initiative_group_only(func):
|
|
"""Только для участников инициативной группы"""
|
|
@wraps(func)
|
|
async def wrapper(message: Message, *args, **kwargs):
|
|
from database.db import AsyncSessionLocal
|
|
from services.initiative_group import InitiativeGroupService
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
ig_service = InitiativeGroupService(session)
|
|
is_member = await ig_service.is_member(message.from_user.id)
|
|
|
|
# Также разрешаем админам
|
|
is_admin = config.is_admin(message.from_user.id)
|
|
|
|
if not (is_member or is_admin):
|
|
await message.answer(
|
|
'❌ Эта команда доступна только участникам инициативной группы'
|
|
)
|
|
return
|
|
return await func(message, *args, **kwargs)
|
|
return wrapper
|
|
|
|
|
|
def verified_only(func):
|
|
"""Только для верифицированных жильцов"""
|
|
@wraps(func)
|
|
async def wrapper(message: Message, *args, **kwargs):
|
|
# Проверяем в БД
|
|
from database.db import AsyncSessionLocal
|
|
from sqlalchemy import select
|
|
from database.models import User
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
user = await get_user_from_db(session, message.from_user.id)
|
|
if not user or not user.verified:
|
|
await message.answer(
|
|
'❗ Для использования этой команды пройдите верификацию\n'
|
|
'Отправьте номер квартиры фото квитанции'
|
|
)
|
|
return
|
|
return await func(message, *args, **kwargs)
|
|
return wrapper
|
|
|
|
|
|
def admin_callback_only(func):
|
|
"""Только для администраторов (callback queries)"""
|
|
@wraps(func)
|
|
async def wrapper(callback: CallbackQuery, *args, **kwargs):
|
|
if not config.is_admin(callback.from_user.id):
|
|
await callback.answer('❌ Только для администраторов', show_alert=True)
|
|
return
|
|
return await func(callback, *args, **kwargs)
|
|
return wrapper
|
|
|
|
|
|
def ig_callback_only(func):
|
|
"""Только для участников ИГ (callback queries)"""
|
|
@wraps(func)
|
|
async def wrapper(callback: CallbackQuery, *args, **kwargs):
|
|
from database.db import AsyncSessionLocal
|
|
from services.initiative_group import InitiativeGroupService
|
|
|
|
async with AsyncSessionLocal() as session:
|
|
ig_service = InitiativeGroupService(session)
|
|
is_member = await ig_service.is_member(callback.from_user.id)
|
|
is_admin = config.is_admin(callback.from_user.id)
|
|
|
|
if not (is_member or is_admin):
|
|
await callback.answer('❌ Только для участников ИГ', show_alert=True)
|
|
return
|
|
return await func(callback, *args, **kwargs)
|
|
return wrapper
|