102 lines
4.2 KiB
Python
102 lines
4.2 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):
|
||
# Отвечаем об ошибке прав только в личке
|
||
if message.chat.type == 'private':
|
||
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
|