Модуль анти-конфликта, для блокировки слов конфликтных
This commit is contained in:
435
bot/handlers/commands/users/conflict.py
Normal file
435
bot/handlers/commands/users/conflict.py
Normal file
@@ -0,0 +1,435 @@
|
||||
"""
|
||||
Обработчики команд режима антиконфликта
|
||||
"""
|
||||
from datetime import datetime
|
||||
from aiogram import Router
|
||||
from aiogram.filters import Command
|
||||
from aiogram.types import Message
|
||||
|
||||
from bot.filters.admin import IsAdmin
|
||||
from configs import settings, COMMANDS
|
||||
from database import get_manager
|
||||
from database.models import BanWordType
|
||||
from middleware.loggers import logger
|
||||
from bot.utils.decorators import log_action
|
||||
|
||||
__all__ = ("router",)
|
||||
|
||||
router: Router = Router(name="conflict_mode_router")
|
||||
|
||||
|
||||
# ================= ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ =================
|
||||
|
||||
def parse_conflict_args(text: str, command: str, need_minutes: bool = False) -> tuple[bool, str | list]:
|
||||
"""
|
||||
Парсит аргументы команды для конфликтного режима.
|
||||
|
||||
Args:
|
||||
text: Полный текст сообщения
|
||||
command: Название команды
|
||||
need_minutes: Требуется ли параметр минут
|
||||
|
||||
Returns:
|
||||
(success, result): result это либо список аргументов, либо текст ошибки
|
||||
"""
|
||||
parts = text.split(maxsplit=2 if need_minutes else 1)
|
||||
|
||||
min_args = 1 if need_minutes else 1
|
||||
|
||||
if len(parts) < min_args + 1:
|
||||
if need_minutes:
|
||||
return False, f"❌ Использование: <code>/{command} [минуты]</code>"
|
||||
else:
|
||||
return False, f"❌ Использование: <code>/{command} [слово]</code>"
|
||||
|
||||
args = parts[1:]
|
||||
|
||||
# Валидация слова
|
||||
if not need_minutes:
|
||||
if len(args[0]) < 2:
|
||||
return False, "❌ Слово должно содержать минимум 2 символа"
|
||||
|
||||
if len(args[0]) > 100:
|
||||
return False, "❌ Слово слишком длинное (максимум 100 символов)"
|
||||
|
||||
return True, args
|
||||
|
||||
|
||||
def format_time_str(minutes: int) -> str:
|
||||
"""Форматирует время в читабельный формат"""
|
||||
if minutes < 60:
|
||||
return f"{minutes} мин"
|
||||
elif minutes < 1440:
|
||||
hours = minutes // 60
|
||||
mins = minutes % 60
|
||||
return f"{hours}ч {mins}м" if mins else f"{hours}ч"
|
||||
else:
|
||||
days = minutes // 1440
|
||||
hours = (minutes % 1440) // 60
|
||||
return f"{days}д {hours}ч" if hours else f"{days}д"
|
||||
|
||||
|
||||
def format_datetime(dt: datetime) -> str:
|
||||
"""Форматирует datetime в читабельный формат"""
|
||||
return dt.strftime("%d.%m.%Y %H:%M:%S")
|
||||
|
||||
|
||||
# ================= ДОБАВЛЕНИЕ КОНФЛИКТНЫХ СЛОВ =================
|
||||
|
||||
@router.message(
|
||||
Command(*COMMANDS.get("addconflictword", ["addconflictword"]), prefix=settings.PREFIX, ignore_case=True), IsAdmin())
|
||||
@log_action(action_name="ADD_CONFLICT_WORD", log_args=True)
|
||||
async def add_conflict_word_cmd(message: Message) -> None:
|
||||
"""
|
||||
Добавляет конфликтное слово-подстроку.
|
||||
|
||||
Конфликтные слова работают только в режиме /stopconflict.
|
||||
|
||||
Использование: /addconflictword <слово>
|
||||
"""
|
||||
success, result = parse_conflict_args(message.text, "addconflictword", need_minutes=False)
|
||||
|
||||
if not success:
|
||||
await message.answer(result, parse_mode="HTML")
|
||||
return
|
||||
|
||||
word = result[0].lower().strip()
|
||||
manager = get_manager()
|
||||
|
||||
try:
|
||||
added = await manager.add_banword(
|
||||
word=word,
|
||||
word_type=BanWordType.CONFLICT_SUBSTRING,
|
||||
added_by=message.from_user.id,
|
||||
reason="Конфликтное слово"
|
||||
)
|
||||
|
||||
if added:
|
||||
text = (
|
||||
f"✅ <b>Конфликтное слово добавлено</b>\n\n"
|
||||
f"📝 Слово: <code>{word}</code>\n"
|
||||
f"🔍 Тип: подстрока\n\n"
|
||||
f"⚔️ <i>Будет работать только в режиме антиконфликта</i>\n"
|
||||
f"Активируйте: <code>/stopconflict [минуты]</code>"
|
||||
)
|
||||
else:
|
||||
text = f"⚠️ Конфликтное слово <code>{word}</code> уже существует"
|
||||
|
||||
await message.answer(text, parse_mode="HTML")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка добавления конфликтного слова: {e}", log_type="CONFLICT")
|
||||
await message.answer("❌ <b>Ошибка добавления</b>\n\nПопробуйте позже", parse_mode="HTML")
|
||||
|
||||
|
||||
@router.message(
|
||||
Command(*COMMANDS.get("addconflictlemma", ["addconflictlemma"]), prefix=settings.PREFIX, ignore_case=True),
|
||||
IsAdmin())
|
||||
@log_action(action_name="ADD_CONFLICT_LEMMA", log_args=True)
|
||||
async def add_conflict_lemma_cmd(message: Message) -> None:
|
||||
"""
|
||||
Добавляет конфликтную лемму.
|
||||
|
||||
Конфликтные леммы работают только в режиме /stopconflict.
|
||||
|
||||
Использование: /addconflictlemma <слово>
|
||||
"""
|
||||
success, result = parse_conflict_args(message.text, "addconflictlemma", need_minutes=False)
|
||||
|
||||
if not success:
|
||||
await message.answer(result, parse_mode="HTML")
|
||||
return
|
||||
|
||||
word = result[0].lower().strip()
|
||||
manager = get_manager()
|
||||
|
||||
try:
|
||||
added = await manager.add_banword(
|
||||
word=word,
|
||||
word_type=BanWordType.CONFLICT_LEMMA,
|
||||
added_by=message.from_user.id,
|
||||
reason="Конфликтная лемма"
|
||||
)
|
||||
|
||||
if added:
|
||||
text = (
|
||||
f"✅ <b>Конфликтная лемма добавлена</b>\n\n"
|
||||
f"🔤 Слово: <code>{word}</code>\n"
|
||||
f"🔍 Тип: лемма (все формы слова)\n\n"
|
||||
f"⚔️ <i>Будет работать только в режиме антиконфликта</i>\n"
|
||||
f"Активируйте: <code>/stopconflict [минуты]</code>"
|
||||
)
|
||||
else:
|
||||
text = f"⚠️ Конфликтная лемма <code>{word}</code> уже существует"
|
||||
|
||||
await message.answer(text, parse_mode="HTML")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка добавления конфликтной леммы: {e}", log_type="CONFLICT")
|
||||
await message.answer("❌ <b>Ошибка добавления</b>\n\nПопробуйте позже", parse_mode="HTML")
|
||||
|
||||
|
||||
# ================= УДАЛЕНИЕ КОНФЛИКТНЫХ СЛОВ =================
|
||||
|
||||
@router.message(
|
||||
Command(*COMMANDS.get("remconflictword", ["remconflictword"]), prefix=settings.PREFIX, ignore_case=True), IsAdmin())
|
||||
@log_action(action_name="REMOVE_CONFLICT_WORD", log_args=True)
|
||||
async def remove_conflict_word_cmd(message: Message) -> None:
|
||||
"""
|
||||
Удаляет конфликтное слово-подстроку.
|
||||
|
||||
Использование: /remconflictword <слово>
|
||||
"""
|
||||
success, result = parse_conflict_args(message.text, "remconflictword", need_minutes=False)
|
||||
|
||||
if not success:
|
||||
await message.answer(result, parse_mode="HTML")
|
||||
return
|
||||
|
||||
word = result[0].lower().strip()
|
||||
manager = get_manager()
|
||||
|
||||
try:
|
||||
removed = await manager.remove_banword(
|
||||
word=word,
|
||||
word_type=BanWordType.CONFLICT_SUBSTRING
|
||||
)
|
||||
|
||||
if removed:
|
||||
text = f"🗑 <b>Конфликтное слово удалено</b>\n\n📝 Слово: <code>{word}</code>"
|
||||
else:
|
||||
text = f"⚠️ Конфликтное слово <code>{word}</code> не найдено"
|
||||
|
||||
await message.answer(text, parse_mode="HTML")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка удаления конфликтного слова: {e}", log_type="CONFLICT")
|
||||
await message.answer("❌ <b>Ошибка удаления</b>\n\nПопробуйте позже", parse_mode="HTML")
|
||||
|
||||
|
||||
@router.message(
|
||||
Command(*COMMANDS.get("remconflictlemma", ["remconflictlemma"]), prefix=settings.PREFIX, ignore_case=True),
|
||||
IsAdmin())
|
||||
@log_action(action_name="REMOVE_CONFLICT_LEMMA", log_args=True)
|
||||
async def remove_conflict_lemma_cmd(message: Message) -> None:
|
||||
"""
|
||||
Удаляет конфликтную лемму.
|
||||
|
||||
Использование: /remconflictlemma <слово>
|
||||
"""
|
||||
success, result = parse_conflict_args(message.text, "remconflictlemma", need_minutes=False)
|
||||
|
||||
if not success:
|
||||
await message.answer(result, parse_mode="HTML")
|
||||
return
|
||||
|
||||
word = result[0].lower().strip()
|
||||
manager = get_manager()
|
||||
|
||||
try:
|
||||
removed = await manager.remove_banword(
|
||||
word=word,
|
||||
word_type=BanWordType.CONFLICT_LEMMA
|
||||
)
|
||||
|
||||
if removed:
|
||||
text = f"🗑 <b>Конфликтная лемма удалена</b>\n\n🔤 Слово: <code>{word}</code>"
|
||||
else:
|
||||
text = f"⚠️ Конфликтная лемма <code>{word}</code> не найдена"
|
||||
|
||||
await message.answer(text, parse_mode="HTML")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка удаления конфликтной леммы: {e}", log_type="CONFLICT")
|
||||
await message.answer("❌ <b>Ошибка удаления</b>\n\nПопробуйте позже", parse_mode="HTML")
|
||||
|
||||
|
||||
# ================= УПРАВЛЕНИЕ РЕЖИМОМ АНТИКОНФЛИКТА =================
|
||||
|
||||
@router.message(Command(*COMMANDS.get("stopconflict", ["stopconflict"]), prefix=settings.PREFIX, ignore_case=True),
|
||||
IsAdmin())
|
||||
@log_action(action_name="START_CONFLICT_MODE", log_args=True)
|
||||
async def start_conflict_mode_cmd(message: Message) -> None:
|
||||
"""
|
||||
Активирует режим антиконфликта на указанное время.
|
||||
|
||||
В этом режиме работают только конфликтные слова/леммы.
|
||||
Обычные банворды временно отключаются.
|
||||
|
||||
Использование: /stopconflict <минуты>
|
||||
Пример: /stopconflict 30
|
||||
"""
|
||||
success, result = parse_conflict_args(message.text, "stopconflict", need_minutes=True)
|
||||
|
||||
if not success:
|
||||
await message.answer(result, parse_mode="HTML")
|
||||
return
|
||||
|
||||
# Валидация минут
|
||||
try:
|
||||
minutes = int(result[0])
|
||||
if minutes < 1 or minutes > 10080: # Максимум неделя
|
||||
await message.answer(
|
||||
"❌ Время должно быть от 1 минуты до 10080 минут (7 дней)",
|
||||
parse_mode="HTML"
|
||||
)
|
||||
return
|
||||
except ValueError:
|
||||
await message.answer("❌ Неверный формат времени. Укажите число минут", parse_mode="HTML")
|
||||
return
|
||||
|
||||
manager = get_manager()
|
||||
|
||||
try:
|
||||
# Получаем статистику конфликтных слов
|
||||
data = await manager.get_all_words_list()
|
||||
conflict_words_count = len(data.get('conflict_substring', set()))
|
||||
conflict_lemmas_count = len(data.get('conflict_lemma', set()))
|
||||
total_conflict = conflict_words_count + conflict_lemmas_count
|
||||
|
||||
if total_conflict == 0:
|
||||
await message.answer(
|
||||
"⚠️ <b>Нет конфликтных слов</b>\n\n"
|
||||
"Сначала добавьте конфликтные слова:\n"
|
||||
"• <code>/addconflictword [слово]</code>\n"
|
||||
"• <code>/addconflictlemma [слово]</code>",
|
||||
parse_mode="HTML"
|
||||
)
|
||||
return
|
||||
|
||||
# Активируем режим
|
||||
expires_at = await manager.set_conflict_mode(minutes)
|
||||
|
||||
time_str = format_time_str(minutes)
|
||||
expires_str = format_datetime(expires_at)
|
||||
|
||||
text = (
|
||||
f"⚔️ <b>РЕЖИМ АНТИКОНФЛИКТА АКТИВИРОВАН</b>\n\n"
|
||||
f"⏱ Длительность: {time_str}\n"
|
||||
f"🕐 Окончание: {expires_str}\n\n"
|
||||
f"📊 Активные правила:\n"
|
||||
f"├─ Конфликтные слова: <code>{conflict_words_count}</code>\n"
|
||||
f"└─ Конфликтные леммы: <code>{conflict_lemmas_count}</code>\n\n"
|
||||
f"⚠️ <i>Обычные банворды временно отключены</i>\n"
|
||||
f"Отключить режим: /unstopconflict"
|
||||
)
|
||||
|
||||
await message.answer(text, parse_mode="HTML")
|
||||
|
||||
logger.info(
|
||||
f"Режим антиконфликта активирован на {minutes} мин "
|
||||
f"(конфликтных правил: {total_conflict})",
|
||||
log_type="CONFLICT"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка активации режима антиконфликта: {e}", log_type="CONFLICT")
|
||||
await message.answer("❌ <b>Ошибка активации режима</b>\n\nПопробуйте позже", parse_mode="HTML")
|
||||
|
||||
|
||||
@router.message(Command(*COMMANDS.get("unstopconflict", ["unstopconflict"]), prefix=settings.PREFIX, ignore_case=True),
|
||||
IsAdmin())
|
||||
@log_action(action_name="STOP_CONFLICT_MODE")
|
||||
async def stop_conflict_mode_cmd(message: Message) -> None:
|
||||
"""
|
||||
Отключает режим антиконфликта.
|
||||
|
||||
Использование: /unstopconflict
|
||||
"""
|
||||
manager = get_manager()
|
||||
|
||||
try:
|
||||
# Проверяем, активен ли режим
|
||||
is_active = await manager.is_conflict_active()
|
||||
|
||||
if not is_active:
|
||||
await message.answer(
|
||||
"⚠️ <b>Режим антиконфликта не активен</b>\n\n"
|
||||
"Активируйте: <code>/stopconflict [минуты]</code>",
|
||||
parse_mode="HTML"
|
||||
)
|
||||
return
|
||||
|
||||
# Отключаем режим
|
||||
await manager.disable_conflict_mode()
|
||||
|
||||
text = (
|
||||
f"✅ <b>Режим антиконфликта отключен</b>\n\n"
|
||||
f"🔄 Обычные банворды снова активны\n"
|
||||
f"⚔️ Конфликтные слова деактивированы"
|
||||
)
|
||||
|
||||
await message.answer(text, parse_mode="HTML")
|
||||
|
||||
logger.info("Режим антиконфликта отключён", log_type="CONFLICT")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка отключения режима антиконфликта: {e}", log_type="CONFLICT")
|
||||
await message.answer("❌ <b>Ошибка отключения режима</b>\n\nПопробуйте позже", parse_mode="HTML")
|
||||
|
||||
|
||||
@router.message(Command(*COMMANDS.get("conflictstatus", ["conflictstatus"]), prefix=settings.PREFIX, ignore_case=True),
|
||||
IsAdmin())
|
||||
@log_action(action_name="CONFLICT_STATUS")
|
||||
async def conflict_status_cmd(message: Message) -> None:
|
||||
"""
|
||||
Показывает статус режима антиконфликта.
|
||||
|
||||
Использование: /conflictstatus
|
||||
"""
|
||||
manager = get_manager()
|
||||
|
||||
try:
|
||||
# Проверяем активность режима
|
||||
is_active = await manager.is_conflict_active()
|
||||
|
||||
# Получаем статистику
|
||||
data = await manager.get_all_words_list()
|
||||
conflict_words_count = len(data.get('conflict_substring', set()))
|
||||
conflict_lemmas_count = len(data.get('conflict_lemma', set()))
|
||||
total_conflict = conflict_words_count + conflict_lemmas_count
|
||||
|
||||
if is_active:
|
||||
# Режим активен - показываем детали
|
||||
conflict_until_str = await manager.repo.get_setting("conflict_until")
|
||||
conflict_until = float(conflict_until_str)
|
||||
expires_at = datetime.fromtimestamp(conflict_until)
|
||||
|
||||
now = datetime.now()
|
||||
time_left_seconds = (expires_at - now).total_seconds()
|
||||
time_left_minutes = int(time_left_seconds / 60)
|
||||
|
||||
text = (
|
||||
f"⚔️ <b>РЕЖИМ АНТИКОНФЛИКТА АКТИВЕН</b>\n\n"
|
||||
f"⏱ Осталось: {format_time_str(time_left_minutes)}\n"
|
||||
f"🕐 Окончание: {format_datetime(expires_at)}\n\n"
|
||||
f"📊 Активные правила:\n"
|
||||
f"├─ Конфликтные слова: <code>{conflict_words_count}</code>\n"
|
||||
f"└─ Конфликтные леммы: <code>{conflict_lemmas_count}</code>\n\n"
|
||||
f"⚠️ <i>Обычные банворды отключены</i>\n"
|
||||
f"Отключить: /unstopconflict"
|
||||
)
|
||||
else:
|
||||
# Режим не активен
|
||||
text = (
|
||||
f"💤 <b>Режим антиконфликта НЕ активен</b>\n\n"
|
||||
f"📊 Конфликтных правил в базе:\n"
|
||||
f"├─ Слова: <code>{conflict_words_count}</code>\n"
|
||||
f"└─ Леммы: <code>{conflict_lemmas_count}</code>\n\n"
|
||||
)
|
||||
|
||||
if total_conflict > 0:
|
||||
text += f"Активировать: <code>/stopconflict [минуты]</code>"
|
||||
else:
|
||||
text += (
|
||||
f"⚠️ <i>Нет конфликтных слов</i>\n"
|
||||
f"Добавьте:\n"
|
||||
f"• <code>/addconflictword [слово]</code>\n"
|
||||
f"• <code>/addconflictlemma [слово]</code>"
|
||||
)
|
||||
|
||||
await message.answer(text, parse_mode="HTML")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка получения статуса режима: {e}", log_type="CONFLICT")
|
||||
await message.answer("❌ <b>Ошибка получения статуса</b>", parse_mode="HTML")
|
||||
Reference in New Issue
Block a user