Moderation

Фильтрация нежелательного контента с поддержкой русского и казахского языков.

Возможности

• Словари для RU и KZ языков
• Levenshtein distance для обхода замен букв
• Кастомные словари
• Whitelist для исключений
• Детальный отчёт о нарушениях

Использование

Базовая проверка

from kit.core.moderation import ContentFilter

filter = ContentFilter(languages=["ru", "kz"])

result = filter.check("Какой-то текст для проверки")

if result.is_clean:
    print("Контент безопасен")
else:
    print(f"Найдены нарушения: {result.violations}")
    print(f"Уровень серьёзности: {result.severity}")

Результат проверки

@dataclass
class ModerationResult:
    is_clean: bool           # True если контент безопасен
    violations: List[str]    # Список найденных нарушений
    severity: str            # low, medium, high
    masked_text: str         # Текст с замаскированными словами
    details: Dict            # Детальная информация

Маскирование контента

result = filter.check("Текст с плохими словами")

if not result.is_clean:
    # Вернуть замаскированную версию
    safe_text = result.masked_text  # "Текст с ****** словами"

Кастомные словари

filter = ContentFilter(
    languages=["ru"],
    custom_words=["спам", "реклама", "казино"],
    whitelist=["анализ", "документация"]
)

# Добавление слов в runtime
filter.add_word("новое_слово")
filter.add_to_whitelist("исключение")

Настройка чувствительности

filter = ContentFilter(
    languages=["ru", "kz"],
    levenshtein_threshold=2,  # Допустимое расстояние (для обхода замен)
    min_word_length=3,        # Минимальная длина слова для проверки
    case_sensitive=False      # Регистронезависимая проверка
)

Алгоритм работы

1. Нормализация текста — приведение к нижнему регистру, удаление лишних символов
2. Токенизация — разбиение на слова
3. Прямое сравнение — проверка по словарю
4. Levenshtein — поиск похожих слов (для обхода "м0дерац1я" → "модерация")
5. Оценка severity — определение серьёзности нарушений

# Пример обхода через замену букв
filter.check("пр1в3т")  # Найдёт "привет" если в словаре

API Reference

ContentFilter

class ContentFilter:
    def __init__(
        self,
        languages: List[str] = ["ru"],
        custom_words: List[str] = None,
        whitelist: List[str] = None,
        levenshtein_threshold: int = 2
    )

    def check(self, text: str) -> ModerationResult
    def add_word(self, word: str) -> None
    def add_to_whitelist(self, word: str) -> None
    def remove_word(self, word: str) -> None

ModerationResult

@dataclass
class ModerationResult:
    is_clean: bool
    violations: List[str]
    severity: str  # "low", "medium", "high"
    masked_text: str
    details: Dict[str, Any]

Интеграция с FastAPI

from fastapi import FastAPI, HTTPException
from kit.core.moderation import ContentFilter

app = FastAPI()
filter = ContentFilter(languages=["ru", "kz"])

@app.post("/messages")
async def create_message(content: str):
    result = filter.check(content)

    if not result.is_clean:
        if result.severity == "high":
            raise HTTPException(400, "Недопустимый контент")
        # Для medium/low - возвращаем замаскированную версию
        content = result.masked_text

    return {"content": content}

Примеры из production

DedMoroz.ai — модерация поздравлений

class GreetingModerator:
    def __init__(self):
        self.filter = ContentFilter(
            languages=["ru", "kz"],
            custom_words=["политика", "религия"],
            whitelist=["дед мороз", "снегурочка"]
        )

    async def moderate(self, greeting: str) -> str:
        result = self.filter.check(greeting)

        if result.severity == "high":
            return "К сожалению, не могу создать такое поздравление."

        if not result.is_clean:
            # Логируем для анализа
            logger.warning(f"Moderated: {result.violations}")
            return result.masked_text

        return greeting

Telegram бот с модерацией

from aiogram import Router
from kit.core.moderation import ContentFilter

router = Router()
filter = ContentFilter()

@router.message()
async def handle_message(message: Message):
    result = filter.check(message.text)

    if not result.is_clean:
        await message.delete()
        await message.answer(
            "Сообщение удалено из-за нарушения правил.",
            reply_to_message_id=None
        )
        return

    # Обработка сообщения
    ...