User
72 lines
2.6 KiB
Python
72 lines
2.6 KiB
Python
from loguru import logger
|
|
from typing import Dict, List
|
|
from datetime import datetime, timedelta
|
|
|
|
|
|
class MiddlewareChain:
|
|
"""Простая цепочка мидлваров (логирование, антиспам, аналитика)"""
|
|
def __init__(self):
|
|
self.middlewares: List[callable] = []
|
|
|
|
def add(self, middleware: callable):
|
|
self.middlewares.append(middleware)
|
|
|
|
def process(self, user_id: int, text: str) -> bool:
|
|
for mw in self.middlewares:
|
|
if not mw(user_id, text):
|
|
return False
|
|
return True
|
|
|
|
|
|
# ==================== Антиспам-фильтр ====================
|
|
|
|
class SpamFilter:
|
|
"""Простой антиспам-фильтр: блокирует пользователя, если он отправляет
|
|
более `max_messages` сообщений за `window_seconds` секунд."""
|
|
|
|
def __init__(self, max_messages: int = 5, window_seconds: int = 60):
|
|
self.max_messages = max_messages
|
|
self.window_seconds = window_seconds
|
|
# user_id -> [timestamp1, timestamp2, ...]
|
|
self._messages: Dict[int, List[float]] = {}
|
|
|
|
def is_spam(self, user_id: int) -> bool:
|
|
now = datetime.now().timestamp()
|
|
window_start = now - self.window_seconds
|
|
|
|
# Инициализируем список, если нет
|
|
if user_id not in self._messages:
|
|
self._messages[user_id] = []
|
|
|
|
# Удаляем старые записи за пределами окна
|
|
self._messages[user_id] = [
|
|
ts for ts in self._messages[user_id] if ts > window_start
|
|
]
|
|
|
|
# Проверяем, не превышен ли лимит
|
|
if len(self._messages[user_id]) >= self.max_messages:
|
|
return True
|
|
|
|
# Записываем текущее сообщение
|
|
self._messages[user_id].append(now)
|
|
return False
|
|
|
|
def reset(self, user_id: int):
|
|
"""Сброс счётчика для пользователя (например, при /start)"""
|
|
self._messages.pop(user_id, None)
|
|
|
|
|
|
# ==================== Логирование ====================
|
|
|
|
def logging_middleware(user_id: int, text: str) -> bool:
|
|
logger.info(f"Processed message from {user_id}: {text}")
|
|
return True
|
|
|
|
|
|
# ==================== Пример: мидлвар для аналитики ====================
|
|
|
|
def analytics_middleware(user_id: int, text: str) -> bool:
|
|
"""Заглушка для аналитики — можно расширить позже."""
|
|
logger.debug(f"Analytics: user {user_id} sent '{text}'")
|
|
return True
|