Fix admin management

This commit is contained in:
Capybara-z
2025-07-11 19:27:36 +03:00
parent 9427cb20d9
commit cafe576478
4 changed files with 31 additions and 10 deletions
+3 -3
View File
@@ -1,16 +1,16 @@
from aiogram.filters import BaseFilter
from aiogram.types import Message
from aiogram.types import Message, CallbackQuery
from sqlalchemy import select
from database.models import Admin
from database.db import async_session_maker
class IsAdminFilter(BaseFilter):
async def __call__(self, message: Message) -> bool:
async def __call__(self, event: Message | CallbackQuery) -> bool:
try:
async with async_session_maker() as session:
result = await session.execute(
select(Admin).where(Admin.tg_id == message.from_user.id)
select(Admin).where(Admin.tg_id == event.from_user.id)
)
admin = result.scalar_one_or_none()
return admin is not None
@@ -254,6 +254,13 @@ async def set_admin_role(callback: CallbackQuery, callback_data: AdminPanelCallb
await callback.message.edit_text("❌ Неверный формат.")
return
if tg_id == callback.from_user.id:
await callback.message.edit_text(
"🚫 <b>Нельзя изменить свою собственную роль!</b>",
reply_markup=build_single_admin_menu(tg_id)
)
return
result = await session.execute(select(Admin).where(Admin.tg_id == tg_id))
admin = result.scalar_one_or_none()
if not admin:
+1 -1
View File
@@ -25,10 +25,10 @@ def register_middleware(
"""Регистрирует middleware в диспетчере."""
if middlewares is None:
available_middlewares = {
"admin": AdminMiddleware(),
"session": (
SessionMiddleware(sessionmaker) if sessionmaker else SessionMiddleware()
),
"admin": AdminMiddleware(),
"maintenance": MaintenanceModeMiddleware(),
"logging": LoggingMiddleware(),
"throttling": ThrottlingMiddleware(),
+20 -6
View File
@@ -3,8 +3,10 @@ from typing import Any
from aiogram import BaseMiddleware
from aiogram.types import CallbackQuery, Message, TelegramObject
from sqlalchemy import select
from config import ADMIN_ID
from database.models import Admin
class AdminMiddleware(BaseMiddleware):
@@ -25,18 +27,30 @@ class AdminMiddleware(BaseMiddleware):
data: dict[str, Any],
) -> Any:
"""Обрабатывает событие и добавляет флаг администратора в data."""
data["admin"] = self._check_admin_access(event)
data["admin"] = await self._check_admin_access(event, data.get("session"))
return await handler(event, data)
def _check_admin_access(self, event: TelegramObject) -> bool:
async def _check_admin_access(self, event: TelegramObject, session) -> bool:
"""Проверяет, имеет ли пользователь права администратора."""
try:
user_id = None
if isinstance(event, Message):
return event.from_user and event.from_user.id in self._admin_ids
user_id = event.from_user.id if event.from_user else None
elif isinstance(event, CallbackQuery):
return event.from_user and event.from_user.id in self._admin_ids
user_id = event.from_user.id if event.from_user else None
else:
user_id = getattr(getattr(event, "from_user", None), "id", None)
user_id = getattr(getattr(event, "from_user", None), "id", None)
return user_id in self._admin_ids if user_id else False
if not user_id:
return False
if user_id in self._admin_ids:
return True
if session:
result = await session.execute(select(Admin).where(Admin.tg_id == user_id))
return result.scalar_one_or_none() is not None
return False
except Exception:
return False