class ActivityMiddleware(BaseMiddleware):
async def __call__(
self,
handler: Callable[[Message, dict[str, Any]], Awaitable[Any]],
event: Message,
data: dict[str, Any],
) -> Any:
if not event.chat:
return await handler(event, data)
try:
raw_chat_type = event.chat.type
chat_type = str(getattr(raw_chat_type, "value", raw_chat_type))
except Exception:
chat_type = "unknown"
upsert_chat(chat_id=event.chat.id, title=event.chat.title or "", chat_type=chat_type)
if event.from_user and not event.from_user.is_bot:
upsert_user(
user_id=event.from_user.id,
username=event.from_user.username,
first_name=event.from_user.full_name,
)
ensure_chat_user_state(chat_id=event.chat.id, user_id=event.from_user.id)
text = event.text or ""
is_command = bool(text) and text[0] in {"/", "!"}
if text and not is_command and not is_system_message(event):
increment_message_stat(chat_id=event.chat.id, user_id=event.from_user.id)
if event.chat.type in {ChatType.GROUP, ChatType.SUPERGROUP} and text and not is_command:
bot: Bot | None = data.get("bot")
if bot is not None:
settings = get_chat_settings(event.chat.id)
level = await get_access_level(bot, event.chat.id, event.from_user.id)
if int(settings["silence"]) == 1 and level < 80:
with suppress(TelegramBadRequest):
await bot.delete_message(chat_id=event.chat.id, message_id=event.message_id)
return None
if level < 40:
filter_words = get_filter_words(event.chat.id)
low_text = text.lower()
if any(word in low_text for word in filter_words):
with suppress(TelegramBadRequest):
await bot.delete_message(chat_id=event.chat.id, message_id=event.message_id)
return None
return await handler(event, data)
