347 lines
14 KiB
Python
347 lines
14 KiB
Python
"""
|
||
Обработчики команд режима тишины
|
||
"""
|
||
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 middleware.loggers import logger
|
||
from bot.utils.decorators import log_action
|
||
|
||
__all__ = ("router",)
|
||
|
||
router: Router = Router(name="silence_mode_router")
|
||
|
||
|
||
# ================= ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ =================
|
||
|
||
def parse_silence_args(text: str) -> tuple[bool, str | int]:
|
||
"""
|
||
Парсит аргументы команды для режима тишины.
|
||
|
||
Args:
|
||
text: Полный текст сообщения
|
||
|
||
Returns:
|
||
(success, result): result это либо минуты (int), либо текст ошибки (str)
|
||
"""
|
||
parts = text.split(maxsplit=1)
|
||
|
||
if len(parts) < 2:
|
||
return False, "❌ Использование: <code>/silence <минуты></code>"
|
||
|
||
return True, parts[1]
|
||
|
||
|
||
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("silence", ["silence"]), prefix=settings.PREFIX, ignore_case=True), IsAdmin())
|
||
@log_action(action_name="START_SILENCE_MODE", log_args=True)
|
||
async def start_silence_mode_cmd(message: Message) -> None:
|
||
"""
|
||
Активирует режим тишины на указанное время.
|
||
|
||
В этом режиме удаляются ВСЕ сообщения от обычных пользователей.
|
||
Администраторы могут продолжать писать.
|
||
|
||
Использование: /silence <минуты>
|
||
Примеры:
|
||
/silence 30 — на 30 минут
|
||
/silence 120 — на 2 часа
|
||
/silence 1440 — на сутки
|
||
"""
|
||
success, result = parse_silence_args(message.text)
|
||
|
||
if not success:
|
||
await message.answer(result, parse_mode="HTML")
|
||
return
|
||
|
||
# Валидация минут
|
||
try:
|
||
minutes = int(result)
|
||
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:
|
||
# Проверяем, уже активен ли режим
|
||
is_already_active = await manager.is_silence_active()
|
||
|
||
# Активируем режим (перезаписывает предыдущий, если был)
|
||
expires_at = await manager.set_silence_mode(minutes)
|
||
|
||
time_str = format_time_str(minutes)
|
||
expires_str = format_datetime(expires_at)
|
||
|
||
if is_already_active:
|
||
action_text = "🔄 <b>РЕЖИМ ТИШИНЫ ОБНОВЛЁН</b>"
|
||
else:
|
||
action_text = "🔇 <b>РЕЖИМ ТИШИНЫ АКТИВИРОВАН</b>"
|
||
|
||
text = (
|
||
f"{action_text}\n\n"
|
||
f"⏱ Длительность: {time_str}\n"
|
||
f"🕐 Окончание: {expires_str}\n\n"
|
||
f"⚠️ <b>Что происходит:</b>\n"
|
||
f"├─ Все сообщения от пользователей удаляются\n"
|
||
f"├─ Администраторы могут писать\n"
|
||
f"└─ Банворды временно отключены\n\n"
|
||
f"💡 <i>Используйте для успокоения спора или флуда</i>\n"
|
||
f"Отключить досрочно: /unsilence"
|
||
)
|
||
|
||
await message.answer(text, parse_mode="HTML")
|
||
|
||
logger.info(
|
||
f"Режим тишины {'обновлён' if is_already_active else 'активирован'} на {minutes} мин "
|
||
f"пользователем {message.from_user.id}",
|
||
log_type="SILENCE"
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка активации режима тишины: {e}", log_type="SILENCE")
|
||
await message.answer("❌ <b>Ошибка активации режима</b>\n\nПопробуйте позже", parse_mode="HTML")
|
||
|
||
|
||
@router.message(Command(*COMMANDS.get("unsilence", ["unsilence"]), prefix=settings.PREFIX, ignore_case=True), IsAdmin())
|
||
@log_action(action_name="STOP_SILENCE_MODE")
|
||
async def stop_silence_mode_cmd(message: Message) -> None:
|
||
"""
|
||
Отключает режим тишины.
|
||
|
||
Использование: /unsilence
|
||
"""
|
||
manager = get_manager()
|
||
|
||
try:
|
||
# Проверяем, активен ли режим
|
||
is_active = await manager.is_silence_active()
|
||
|
||
if not is_active:
|
||
await message.answer(
|
||
"⚠️ <b>Режим тишины не активен</b>\n\n"
|
||
"Активируйте командой: /silence <минуты>",
|
||
parse_mode="HTML"
|
||
)
|
||
return
|
||
|
||
# Отключаем режим
|
||
await manager.disable_silence_mode()
|
||
|
||
text = (
|
||
f"✅ <b>Режим тишины отключен</b>\n\n"
|
||
f"🔊 Пользователи снова могут отправлять сообщения\n"
|
||
f"🔄 Банворды снова активны"
|
||
)
|
||
|
||
await message.answer(text, parse_mode="HTML")
|
||
|
||
logger.info(
|
||
f"Режим тишины отключён пользователем {message.from_user.id}",
|
||
log_type="SILENCE"
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка отключения режима тишины: {e}", log_type="SILENCE")
|
||
await message.answer("❌ <b>Ошибка отключения режима</b>\n\nПопробуйте позже", parse_mode="HTML")
|
||
|
||
|
||
@router.message(Command(*COMMANDS.get("silencestatus", ["silencestatus"]), prefix=settings.PREFIX, ignore_case=True),
|
||
IsAdmin())
|
||
@log_action(action_name="SILENCE_STATUS")
|
||
async def silence_status_cmd(message: Message) -> None:
|
||
"""
|
||
Показывает статус режима тишины.
|
||
|
||
Использование: /silencestatus
|
||
"""
|
||
manager = get_manager()
|
||
|
||
try:
|
||
# Проверяем активность режима
|
||
is_active = await manager.is_silence_active()
|
||
|
||
if is_active:
|
||
# Режим активен - показываем детали
|
||
silence_until_str = await manager.repo.get_setting("silence_until")
|
||
silence_until = float(silence_until_str)
|
||
expires_at = datetime.fromtimestamp(silence_until)
|
||
|
||
now = datetime.now()
|
||
time_left_seconds = (expires_at - now).total_seconds()
|
||
time_left_minutes = int(time_left_seconds / 60)
|
||
|
||
# Расчёт процента прошедшего времени (для визуализации)
|
||
# Примерно определяем начальное время
|
||
started_minutes_ago = 0 # Можно было бы сохранять в БД
|
||
|
||
text = (
|
||
f"🔇 <b>РЕЖИМ ТИШИНЫ АКТИВЕН</b>\n\n"
|
||
f"⏱ Осталось: {format_time_str(time_left_minutes)}\n"
|
||
f"🕐 Окончание: {format_datetime(expires_at)}\n\n"
|
||
f"⚠️ <b>Что происходит:</b>\n"
|
||
f"├─ Все сообщения от пользователей удаляются\n"
|
||
f"├─ Администраторы могут писать\n"
|
||
f"└─ Банворды временно отключены\n\n"
|
||
f"💡 <i>Для успокоения конфликта или флуда</i>\n"
|
||
f"Отключить: /unsilence"
|
||
)
|
||
|
||
# Добавляем визуальную шкалу прогресса
|
||
if time_left_minutes <= 60:
|
||
progress_bar = create_progress_bar(time_left_minutes, 60)
|
||
text += f"\n\n{progress_bar}"
|
||
|
||
else:
|
||
# Режим не активен
|
||
text = (
|
||
f"💤 <b>Режим тишины НЕ активен</b>\n\n"
|
||
f"🔊 Пользователи могут отправлять сообщения\n"
|
||
f"🔄 Банворды работают в обычном режиме\n\n"
|
||
f"Активировать: /silence <минуты>"
|
||
)
|
||
|
||
await message.answer(text, parse_mode="HTML")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка получения статуса режима тишины: {e}", log_type="SILENCE")
|
||
await message.answer("❌ <b>Ошибка получения статуса</b>", parse_mode="HTML")
|
||
|
||
|
||
def create_progress_bar(minutes_left: int, total_minutes: int, length: int = 10) -> str:
|
||
"""
|
||
Создает визуальную шкалу прогресса.
|
||
|
||
Args:
|
||
minutes_left: Сколько минут осталось
|
||
total_minutes: Всего минут
|
||
length: Длина шкалы
|
||
|
||
Returns:
|
||
Строка с визуальной шкалой
|
||
"""
|
||
if total_minutes <= 0:
|
||
filled = 0
|
||
else:
|
||
filled = int((total_minutes - minutes_left) / total_minutes * length)
|
||
|
||
filled = max(0, min(filled, length))
|
||
empty = length - filled
|
||
|
||
bar = "█" * filled + "░" * empty
|
||
percentage = int((total_minutes - minutes_left) / total_minutes * 100) if total_minutes > 0 else 0
|
||
|
||
return f"[{bar}] {percentage}%"
|
||
|
||
|
||
@router.message(Command(*COMMANDS.get("extend_silence", ["extend_silence"]), prefix=settings.PREFIX, ignore_case=True),
|
||
IsAdmin())
|
||
@log_action(action_name="EXTEND_SILENCE_MODE", log_args=True)
|
||
async def extend_silence_mode_cmd(message: Message) -> None:
|
||
"""
|
||
Продлевает режим тишины на указанное время.
|
||
|
||
Использование: /extend_silence <минуты>
|
||
Пример: /extend_silence 30
|
||
"""
|
||
success, result = parse_silence_args(message.text)
|
||
|
||
if not success:
|
||
# Меняем текст ошибки для extend команды
|
||
await message.answer(
|
||
"❌ Использование: <code>/extend_silence <минуты></code>",
|
||
parse_mode="HTML"
|
||
)
|
||
return
|
||
|
||
# Проверяем, активен ли режим
|
||
manager = get_manager()
|
||
is_active = await manager.is_silence_active()
|
||
|
||
if not is_active:
|
||
await message.answer(
|
||
"⚠️ <b>Режим тишины не активен</b>\n\n"
|
||
"Сначала активируйте: /silence <минуты>",
|
||
parse_mode="HTML"
|
||
)
|
||
return
|
||
|
||
try:
|
||
add_minutes = int(result)
|
||
if add_minutes < 1 or add_minutes > 1440:
|
||
await message.answer(
|
||
"❌ Время продления должно быть от 1 до 1440 минут (24 часа)",
|
||
parse_mode="HTML"
|
||
)
|
||
return
|
||
except ValueError:
|
||
await message.answer("❌ Неверный формат времени. Укажите число минут", parse_mode="HTML")
|
||
return
|
||
|
||
try:
|
||
# Получаем текущее время окончания
|
||
silence_until_str = await manager.repo.get_setting("silence_until")
|
||
current_until = float(silence_until_str)
|
||
current_expires = datetime.fromtimestamp(current_until)
|
||
|
||
# Вычисляем сколько минут осталось + добавляем новые
|
||
now = datetime.now()
|
||
current_minutes_left = int((current_expires - now).total_seconds() / 60)
|
||
new_total_minutes = current_minutes_left + add_minutes
|
||
|
||
# Устанавливаем новое время
|
||
new_expires_at = await manager.set_silence_mode(new_total_minutes)
|
||
|
||
time_str = format_time_str(add_minutes)
|
||
new_expires_str = format_datetime(new_expires_at)
|
||
|
||
text = (
|
||
f"⏱ <b>РЕЖИМ ТИШИНЫ ПРОДЛЁН</b>\n\n"
|
||
f"➕ Добавлено: {time_str}\n"
|
||
f"🕐 Новое окончание: {new_expires_str}\n"
|
||
f"⏳ Всего осталось: {format_time_str(new_total_minutes)}\n\n"
|
||
f"Отключить: /unsilence"
|
||
)
|
||
|
||
await message.answer(text, parse_mode="HTML")
|
||
|
||
logger.info(
|
||
f"Режим тишины продлён на {add_minutes} мин (всего: {new_total_minutes} мин)",
|
||
log_type="SILENCE"
|
||
)
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка продления режима тишины: {e}", log_type="SILENCE")
|
||
await message.answer("❌ <b>Ошибка продления режима</b>\n\nПопробуйте позже", parse_mode="HTML")
|