Init commit

This commit is contained in:
Kirill Samoylenkov 2025-08-22 11:37:26 +07:00
commit d6e00c56e7
49 changed files with 2324 additions and 0 deletions

View file

@ -0,0 +1,9 @@
__all__ = [
"add_filters",
"ChatTypeFilter",
"UserRoleFilter",
]
from .utils import add_filters
from .chat_type import ChatTypeFilter
from .user_role import UserRoleFilter

View file

@ -0,0 +1,24 @@
from typing import Union
from aiogram.filters import BaseFilter
from aiogram.types import Message, CallbackQuery
class ChatTypeFilter(BaseFilter):
"""
Фильтрация обновлений в зависимости от чата.
"""
def __init__(self, chat_type: Union[str, list]) -> None:
self.chat_type = chat_type
async def __call__(self, update: Union[Message, CallbackQuery]) -> bool:
if isinstance(update, Message):
chat_type = update.chat.type
elif isinstance(update, CallbackQuery):
chat_type = update.message.chat.type
if isinstance(self.chat_type, str):
return chat_type == self.chat_type
else:
return chat_type in self.chat_type

View file

@ -0,0 +1,29 @@
from typing import Union
from aiogram.filters import BaseFilter
from aiogram.types import Message, CallbackQuery
from Mousey.Misc import UserRole
from Mousey.Database import check_user
class UserRoleFilter(BaseFilter):
'''
Фильтрация сообщений на основе их роли в БД.
'''
def __init__(self, user_role: Union[UserRole, list]) -> None:
if isinstance(user_role, list):
self.user_role = [ role.value for role in user_role ]
else:
self.user_role = [ user_role.value ]
async def __call__(self, update: Union[Message, CallbackQuery], session) -> bool:
if isinstance(update, Message):
telegram_id = update.from_user.id
elif isinstance(update, CallbackQuery):
telegram_id = update.from_user.id
user = await check_user(session, telegram_id)
return user.role in self.user_role

View file

@ -0,0 +1,17 @@
from aiogram import Router
from aiogram.filters import BaseFilter
def add_filters(router: Router, update_type: str, *filters: BaseFilter) -> None:
"""
Добавить фильтр на указанных тип обновлений к перечисленным роутерам.
update_type: str = "message", "callback" или "all".
"""
for filter in filters:
if update_type in ("message", "all"):
router.message.filter(filter)
if update_type in ("callback", "all"):
router.callback_query.filter(filter)