81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""Проверка доступа пользователя к функциональности бота."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import time
|
||
|
||
from aiogram import Bot
|
||
from aiogram.enums import ChatMemberStatus
|
||
from aiogram.types import Message
|
||
|
||
from bot.services import users as users_service
|
||
from common.config import get_settings
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_ALLOWED_STATUSES = {
|
||
ChatMemberStatus.MEMBER,
|
||
ChatMemberStatus.CREATOR,
|
||
ChatMemberStatus.ADMINISTRATOR,
|
||
ChatMemberStatus.RESTRICTED,
|
||
}
|
||
|
||
#: TTL-кеш членства, чтобы не дёргать get_chat_member на каждое сообщение.
|
||
_MEMBERSHIP_TTL = 300.0
|
||
_membership_cache: dict[int, tuple[float, bool]] = {}
|
||
|
||
|
||
async def _is_member(bot: Bot, user_id: int) -> bool:
|
||
now = time.monotonic()
|
||
cached = _membership_cache.get(user_id)
|
||
if cached is not None and now - cached[0] < _MEMBERSHIP_TTL:
|
||
return cached[1]
|
||
|
||
member = await bot.get_chat_member(get_settings().chat_id, user_id)
|
||
allowed = member.status in _ALLOWED_STATUSES
|
||
_membership_cache[user_id] = (now, allowed)
|
||
return allowed
|
||
|
||
|
||
async def check_user(message: Message) -> bool:
|
||
"""Пропустить пользователя: проверить членство и обновить его запись.
|
||
|
||
В тестовом чате доступ разрешён всегда.
|
||
"""
|
||
settings = get_settings()
|
||
|
||
if message.chat.id == settings.test_chat_id:
|
||
return True
|
||
|
||
user = message.from_user
|
||
if user is None:
|
||
return False
|
||
|
||
if user.id in settings.admin_ids or await _is_member(message.bot, user.id):
|
||
await users_service.register_from_message(message)
|
||
return True
|
||
|
||
await users_service.delete_user(user.id)
|
||
logger.warning("Пользователь %s стучится в чат, доступ запрещён", user.id)
|
||
await _notify_denied(message)
|
||
return False
|
||
|
||
|
||
async def _notify_denied(message: Message) -> None:
|
||
settings = get_settings()
|
||
username = message.from_user.username if message.from_user else None
|
||
try:
|
||
await message.bot.send_message(
|
||
chat_id=settings.test_chat_id,
|
||
text=f"[check_user]\nСтучится в чат {username}!",
|
||
)
|
||
if settings.alert_chat_id:
|
||
await message.bot.forward_message(
|
||
chat_id=settings.alert_chat_id,
|
||
from_chat_id=message.chat.id,
|
||
message_id=message.message_id,
|
||
)
|
||
except Exception: # уведомление не должно ломать обработку
|
||
logger.exception("Не удалось отправить уведомление о запрете доступа")
|