forked from Muzifs/LGBot
336 lines
13 KiB
Python
336 lines
13 KiB
Python
from telebot.async_telebot import AsyncTeleBot
|
||
from telebot.types import Message, User, ChatPermissions
|
||
import asyncio
|
||
import logging
|
||
import time
|
||
from database import db
|
||
|
||
from action_reporter import action_reporter
|
||
from utils import (
|
||
delete_messages,
|
||
check_admin_status,
|
||
check_target_status,
|
||
parse_mute_time,
|
||
format_mute_time,
|
||
)
|
||
|
||
from config import COMMAND_MESSAGES
|
||
|
||
# Получаем логгер для текущего модуля
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# Регистрирует все обработчики команд
|
||
def register_handlers(bot: AsyncTeleBot):
|
||
|
||
# Обработчик команды /mute
|
||
@bot.message_handler(commands=['mute'])
|
||
async def _mute_command_wrapper(message: Message):
|
||
await mute_command(bot, message)
|
||
|
||
# Основная функция команды /mute
|
||
async def mute_command(bot: AsyncTeleBot, message: Message, photo_path: str = None):
|
||
|
||
# Определяем целевого пользователя
|
||
target_user = None
|
||
|
||
# Определяем время
|
||
time_arg = None
|
||
|
||
# Определяем причину
|
||
reason = None
|
||
|
||
# Разбиваем текст сообщения на части
|
||
parts_msg = message.text.split()
|
||
|
||
# Команда /mute help
|
||
if len(parts_msg) == 2 and parts_msg[1].strip() in ('help', 'помощь'):
|
||
|
||
# Отправляем инструкцию
|
||
await bot.send_message(
|
||
chat_id=message.chat.id,
|
||
text=COMMAND_MESSAGES['manual_mute'],
|
||
message_thread_id=message.message_thread_id,
|
||
)
|
||
|
||
# Удаляем сообщения через 30 секунд
|
||
await delete_messages(bot, message, time_sleep=30, number_message=2)
|
||
return
|
||
|
||
try:
|
||
|
||
# Проверяем, является ли отправитель администратором
|
||
if await check_admin_status(bot, message) == 1:
|
||
return
|
||
|
||
# Если одно слово (/mute)
|
||
if len(parts_msg) == 1:
|
||
|
||
# Удаляем сообщение через 3 секунды
|
||
await delete_messages(bot, message, time_sleep=3, number_message=1)
|
||
return
|
||
|
||
# Команда через ответ на сообщение, если два слова (/mute 2m)
|
||
elif len(parts_msg) == 2:
|
||
|
||
# Если это топик
|
||
if message.is_topic_message:
|
||
|
||
# Если без ответа на сообщение
|
||
if not message.reply_to_message or message.message_thread_id == message.reply_to_message.message_id:
|
||
|
||
# Удаляем сообщение через 3 секунды
|
||
await delete_messages(bot, message, time_sleep=3, number_message=1)
|
||
return
|
||
|
||
# Если с ответом на сообщение
|
||
else:
|
||
|
||
# Собираем данные
|
||
target_user = message.reply_to_message.from_user
|
||
time_arg = parts_msg[1]
|
||
reason = 'отсутствует'
|
||
|
||
# Если это General (обычный чат)
|
||
elif message.reply_to_message and not message.is_topic_message:
|
||
|
||
# Собираем данные
|
||
target_user = message.reply_to_message.from_user
|
||
time_arg = parts_msg[1]
|
||
reason = 'отсутствует'
|
||
message.message_thread_id = None
|
||
|
||
# Если команда неправильная
|
||
else:
|
||
|
||
# Удаляем сообщение через 3 секунды
|
||
await delete_messages(bot, message, time_sleep=3, number_message=1)
|
||
return
|
||
|
||
# В сообщении больше двух слов
|
||
else:
|
||
|
||
# Если второе слово это тег или ID
|
||
if parts_msg[1].strip().isdigit() or parts_msg[1].startswith('@'):
|
||
|
||
# Собираем данные
|
||
identifier = parts_msg[1].strip()
|
||
time_arg = parts_msg[2]
|
||
reason = ' '.join(parts_msg[3:]) if parts_msg[3:] != [] else 'отсутствует'
|
||
|
||
# Делаем поиск по ID
|
||
if identifier.isdigit():
|
||
|
||
# Ищем пользователя в базе данных
|
||
user_info = db.get_user(int(identifier))
|
||
|
||
# Если нашли пользователя
|
||
if user_info:
|
||
|
||
# Создаем объект пользователя
|
||
target_user = User(
|
||
id=user_info[0],
|
||
first_name=user_info[1],
|
||
username=user_info[2],
|
||
is_bot=False
|
||
)
|
||
|
||
# Делаем поиск по тегу
|
||
elif identifier.startswith('@'):
|
||
|
||
# Ищем пользователя в базе данных (убрали @)
|
||
user_info = db.get_user_by_username(identifier[1:])
|
||
|
||
# Если нашли пользователя
|
||
if user_info:
|
||
|
||
# Создаем объект пользователя
|
||
target_user = User(
|
||
id=user_info[0],
|
||
first_name=user_info[1],
|
||
username=user_info[2],
|
||
is_bot=False
|
||
)
|
||
|
||
# Команда через ответ на сообщение с указанием причины
|
||
else:
|
||
|
||
# Если это топик
|
||
if message.is_topic_message:
|
||
|
||
# Если без ответа на сообщение
|
||
if not message.reply_to_message or message.message_thread_id == message.reply_to_message.message_id:
|
||
|
||
# Удаляем сообщение через 3 секунды
|
||
await delete_messages(bot, message, time_sleep=3, number_message=1)
|
||
return
|
||
|
||
# Если с ответом на сообщение
|
||
else:
|
||
|
||
# Собираем данные
|
||
target_user = message.reply_to_message.from_user
|
||
time_arg = parts_msg[1]
|
||
reason = ' '.join(parts_msg[2:])
|
||
|
||
# Если это General (обычный чат)
|
||
elif message.reply_to_message and not message.is_topic_message:
|
||
|
||
# Собираем данные
|
||
target_user = message.reply_to_message.from_user
|
||
time_arg = parts_msg[1]
|
||
reason = ' '.join(parts_msg[2:])
|
||
message.message_thread_id = None
|
||
|
||
# Если команда неправильная
|
||
else:
|
||
|
||
# Удаляем сообщение через 3 секунды
|
||
await delete_messages(bot, message, time_sleep=3, number_message=1)
|
||
return
|
||
|
||
# Если пользователь не найден в базе данных
|
||
if not target_user:
|
||
|
||
# Отправляем предупреждение
|
||
await bot.send_message(
|
||
chat_id=message.chat.id,
|
||
text=COMMAND_MESSAGES['user_not_found'],
|
||
message_thread_id=message.message_thread_id,
|
||
)
|
||
|
||
# Удаляем сообщения через 5 секунд
|
||
await delete_messages(bot, message, time_sleep=5, number_message=2)
|
||
return
|
||
|
||
# Парсинг времени мута
|
||
mute_seconds = parse_mute_time(time_arg)
|
||
|
||
# Если не указали время
|
||
if mute_seconds is None:
|
||
|
||
# Отправляем предупреждение
|
||
await bot.send_message(
|
||
chat_id=message.chat.id,
|
||
text=COMMAND_MESSAGES['incorrect_time_format'],
|
||
message_thread_id=message.message_thread_id,
|
||
)
|
||
|
||
# Удаляем сообщения через 5 секунд
|
||
await delete_messages(bot, message, time_sleep=5, number_message=2)
|
||
return
|
||
|
||
# Минимальное время мута - 1 минута (60 секунд)
|
||
if mute_seconds < 60:
|
||
|
||
# Отправляем предупреждение
|
||
await bot.send_message(
|
||
chat_id=message.chat.id,
|
||
text=COMMAND_MESSAGES['min_mute'],
|
||
message_thread_id=message.message_thread_id,
|
||
)
|
||
|
||
# Удаляем сообщения через 5 секунд
|
||
await delete_messages(bot, message, time_sleep=5, number_message=2)
|
||
return
|
||
|
||
# Импортируем максимальное время мута
|
||
from config import MAX_MUTE_TIME
|
||
|
||
# Максимальное время мута - 30 дней
|
||
if mute_seconds > MAX_MUTE_TIME:
|
||
|
||
# Отправляем предупреждение
|
||
await bot.send_message(
|
||
chat_id=message.chat.id,
|
||
text=COMMAND_MESSAGES['max_mute'],
|
||
message_thread_id=message.message_thread_id,
|
||
)
|
||
|
||
# Удаляем сообщения через 5 секунд
|
||
await delete_messages(bot, message, time_sleep=5, number_message=2)
|
||
return
|
||
|
||
# Проверяем статус целевого пользователя
|
||
if await check_target_status(bot, message, target_user) == 1:
|
||
return
|
||
|
||
try:
|
||
|
||
# Вычисляем время окончания мута
|
||
until_date = int(time.time()) + mute_seconds
|
||
|
||
# Устанавливаем ограничения (только чтение)
|
||
permissions = ChatPermissions(
|
||
can_send_messages=False,
|
||
can_send_media_messages=False,
|
||
can_send_polls=False,
|
||
can_send_other_messages=False,
|
||
can_add_web_page_previews=False,
|
||
can_change_info=False,
|
||
can_invite_users=False,
|
||
can_pin_messages=False,
|
||
)
|
||
|
||
# Выполняем мут
|
||
await bot.restrict_chat_member(
|
||
chat_id=message.chat.id,
|
||
user_id=target_user.id,
|
||
permissions=permissions,
|
||
until_date=until_date
|
||
)
|
||
|
||
# Форматируем время в удобный формат
|
||
time_display = format_mute_time(mute_seconds)
|
||
|
||
# Отправляем сообщение-лог в админ-чат
|
||
await action_reporter.log_action(
|
||
action="МУТ",
|
||
user_id=target_user.id,
|
||
admin_id=message.from_user.id,
|
||
reason=reason,
|
||
duration=time_display,
|
||
photo_path=photo_path,
|
||
)
|
||
|
||
# Отправляем сообщение, что пользователь получил мут
|
||
await bot.send_message(
|
||
chat_id=message.chat.id,
|
||
text=COMMAND_MESSAGES['muted'].format(time_display=time_display),
|
||
message_thread_id=message.message_thread_id,
|
||
)
|
||
|
||
# Записываем действие в логи
|
||
logger.info(f"Пользователь {target_user.id} получил мут на {time_display} от администратора {message.from_user.id}.")
|
||
|
||
# Удаляем сообщения через 5 секунд
|
||
await delete_messages(bot, message, time_sleep=5, number_message=2)
|
||
|
||
except Exception as e:
|
||
|
||
# Отправляем ошибку
|
||
await bot.send_message(
|
||
chat_id=message.chat.id,
|
||
text=COMMAND_MESSAGES['error'].format(e=str(e)),
|
||
message_thread_id=message.message_thread_id,
|
||
)
|
||
|
||
# Записываем ошибку в логи
|
||
logger.error(f"Ошибка мута: {str(e)}")
|
||
|
||
# Удаляем сообщения через 5 секунд
|
||
await delete_messages(bot, message, time_sleep=5, number_message=2)
|
||
|
||
except Exception as e:
|
||
|
||
# Отправляем ошибку
|
||
await bot.send_message(
|
||
chat_id=message.chat.id,
|
||
text=COMMAND_MESSAGES['general_error'],
|
||
message_thread_id=message.message_thread_id,
|
||
)
|
||
|
||
# Записываем ошибку в логи
|
||
logger.error(f"Общая ошибка в mute_command: {str(e)}")
|
||
|
||
# Удаляем сообщения через 5 секунд
|
||
await delete_messages(bot, message, time_sleep=5, number_message=2) |