Muzifs/LGBot
2
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'finalized the ban command' (#5) from updade/ban-command into master

Browse Source 
Reviewed-on: #5
This commit is contained in:
Maks Kozhevyatov
2025-07-12 09:12:11 +00:00
parent 0f26a05b54 8c8604d465
commit c81a37f9ac
1 changed files with 135 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

170
src/modules/ban.py
View File

@ -8,6 +8,22 @@ from config import COMMAND_MESSAGES
logger = logging.getLogger(__name__) # Получаем логгер для текущего модуля logger = logging.getLogger(__name__) # Получаем логгер для текущего модуля
# Возвращает причину бана
def extract_reason(words: str) -> str:
if words == []:
reason = 'отсутствует'
else:
reason = ' '.join(words)
return reason
# Удаляет два последних сообщения
async def delete_messages(bot: AsyncTeleBot, message: Message, time_sleep: int):
await asyncio.sleep(time_sleep)
await bot.delete_message(message.chat.id, message.message_id)
await bot.delete_message(message.chat.id, message.message_id+1)
def register_handlers(bot: AsyncTeleBot): # Регистрирует все обработчики команд def register_handlers(bot: AsyncTeleBot): # Регистрирует все обработчики команд
@bot.message_handler(commands=['ban']) # Обработчик команды /ban @bot.message_handler(commands=['ban']) # Обработчик команды /ban
@ -20,6 +36,9 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
# Определяем целевого пользователя # Определяем целевого пользователя
target_user = None target_user = None
# Определяем причину
reason = None
# Разбиваем текст сообщения на части # Разбиваем текст сообщения на части
parts_msg = message.text.split() parts_msg = message.text.split()
@ -31,83 +50,139 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
# Проверяем статус администратора (создателя) # Проверяем статус администратора (создателя)
if admin_status.status not in ['administrator', 'creator']: if admin_status.status not in ['administrator', 'creator']:
await send_message(chat_id, COMMAND_MESSAGES['no_admin_rights']) await send_message(chat_id, COMMAND_MESSAGES['no_admin_rights'])
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return return
# Проверяем право администратора на бан # Проверяем право администратора на бан
if admin_status.status == 'administrator' and not admin_status.can_restrict_members: if admin_status.status == 'administrator' and not admin_status.can_restrict_members:
await send_message(chat_id, COMMAND_MESSAGES['no_restrict_rights']) await send_message(chat_id, COMMAND_MESSAGES['no_restrict_rights'])
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return return
except Exception as e: except Exception as e:
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}") await send_message(chat_id, f"⚠️ Ошибка: {str(e)}")
logger.error(f"Ошибка при получении статуса администратора: {str(e)}") logger.error(f"Ошибка при получении статуса администратора: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return return
# Случай №1 - Команда используется в ответ на сообщение # Выводим помощь (/ban help)
if len(parts_msg) == 2 and parts_msg[1].strip() in ['help', 'помощь']:
await send_message(chat_id, COMMAND_MESSAGES['manual_ban'])
# Удаляем сообщения через 30 секунд
await delete_messages(bot, message, 30)
return
# Если одно слово, то ответом на сообщение
if len(parts_msg) == 1: if len(parts_msg) == 1:
# Если банят в теме # Если это тема
if message.is_topic_message: if message.is_topic_message:
# Если без ответа на сообщение # Если без ответа на сообщение, ошибка
if message.message_thread_id == message.reply_to_message.message_id: if message.message_thread_id == message.reply_to_message.message_id:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id)
return return
# Если команда правильная # Если с ответом на сообщение
else: else:
target_user = message.reply_to_message.from_user target_user = message.reply_to_message.from_user
reason = extract_reason(parts_msg[1:])
# Если банят в обычном чате # Если это обычный чат
elif message.reply_to_message and message.is_topic_message is None: elif message.reply_to_message and message.is_topic_message is None:
target_user = message.reply_to_message.from_user target_user = message.reply_to_message.from_user
reason = extract_reason(parts_msg[1:])
# Не выводим сообщение, что команда неправильная # Удаляем сообщение, если команда неправильная
else: else:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id)
return return
# Случай №2 - Команда с аргументом (/ban @username или /ban 12345) else:
elif len(parts_msg) == 2:
# Выводим помощь (/ban help) # Если второе слово это тег или ID
if parts_msg[1].strip() in ['help', 'помощь']: if parts_msg[1].strip().isdigit() or parts_msg[1].startswith('@'):
await send_message(chat_id, COMMAND_MESSAGES['manual_ban'])
return
identifier = parts_msg[1].strip() identifier = parts_msg[1].strip()
reason = extract_reason(parts_msg[2:])
# Поиск по ID # Поиск по ID
if identifier.isdigit(): if identifier.isdigit():
# Делаем в int и ищем # Делаем в int и ищем
user_info = db.get_user(int(identifier)) user_info = db.get_user(int(identifier))
if user_info: if user_info:
# Создаем объект пользователя из данных базы # Создаем объект пользователя из данных базы
target_user = User( target_user = User(
id=user_info[0], id=user_info[0],
first_name=user_info[1], first_name=user_info[1],
username=user_info[2], username=user_info[2],
is_bot=False is_bot=False
) )
# Поиск по тэгу # Поиск по тэгу
elif identifier.startswith('@'): elif identifier.startswith('@'):
# Убираем @ и ищем # Убираем @ и ищем
user_info = db.get_user_by_username(identifier[1:]) user_info = db.get_user_by_username(identifier[1:])
if user_info: if user_info:
# Создаем объект пользователя из данных базы # Создаем объект пользователя из данных базы
target_user = User( target_user = User(
id=user_info[0], id=user_info[0],
first_name=user_info[1], first_name=user_info[1],
username=user_info[2], username=user_info[2],
is_bot=False is_bot=False
) )
else:
# Если это тема
if message.is_topic_message:
# Если без ответа на сообщение, ошибка
if message.message_thread_id == message.reply_to_message.message_id:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id)
return
# Если с ответом на сообщение
else:
target_user = message.reply_to_message.from_user
reason = extract_reason(parts_msg[1:])
# Если это обычный чат
elif message.reply_to_message and message.is_topic_message is None:
target_user = message.reply_to_message.from_user
reason = extract_reason(parts_msg[1:])
# Удаляем сообщение, если команда неправильная
else:
await asyncio.sleep(3)
await bot.delete_message(message.chat.id, message.message_id)
return
# Если пользователь не найден # Если пользователь не найден
if not target_user: if not target_user:
await send_message(chat_id, COMMAND_MESSAGES['user_not_found']) await send_message(chat_id, COMMAND_MESSAGES['user_not_found'])
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return return
# Проверяем статус целевого пользователя # Проверяем статус целевого пользователя
@ -117,27 +192,44 @@ def register_handlers(bot: AsyncTeleBot): # Регистрирует все об
# Проверяем, является ли цель администратором или создателем # Проверяем, является ли цель администратором или создателем
if target_status.status in ['administrator', 'creator']: if target_status.status in ['administrator', 'creator']:
await send_message(chat_id, COMMAND_MESSAGES['cant_ban_admin']) await send_message(chat_id, COMMAND_MESSAGES['cant_ban_admin'])
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return return
except Exception as e: except Exception as e:
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}") await send_message(chat_id, f"⚠️ Ошибка: {str(e)}")
logger.error(f"Ошибка при получении статуса пользователя: {str(e)}") logger.error(f"Ошибка при получении статуса пользователя: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
return return
# Выполняем бан # Выполняем бан
try: try:
await bot.ban_chat_member(message.chat.id, target_user.id) await bot.ban_chat_member(message.chat.id, target_user.id)
await send_message(chat_id, COMMAND_MESSAGES['banned'])
# Отправляем сообщения, что пользователь получил бан
await send_message(chat_id, COMMAND_MESSAGES['banned'])
logger.info(f"Пользователь {target_user.id} забанен администратором {message.from_user.id}.")
# Удаляем сообщения через 5 секунд
await asyncio.sleep(5) await asyncio.sleep(5)
await bot.delete_message(message.chat.id, message.message_id) await bot.delete_message(message.chat.id, message.message_id)
await bot.delete_message(message.chat.id, message.message_id+2) await bot.delete_message(message.chat.id, message.message_id+2)
logger.info(f"Пользователь {target_user.id} забанен администратором {message.from_user.id}.")
except Exception as e: except Exception as e:
await send_message(chat_id, f"⚠️ Ошибка: {str(e)}") await send_message(chat_id, f"⚠️ Ошибка: {str(e)}")
logger.error(f"Ошибка бана: {str(e)}") logger.error(f"Ошибка бана: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)
except Exception as e: except Exception as e:
await send_message(chat_id, COMMAND_MESSAGES['general_error']) await send_message(chat_id, COMMAND_MESSAGES['general_error'])
logger.error(f"Общая ошибка в ban_command: {str(e)}") logger.error(f"Общая ошибка в ban_command: {str(e)}")
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, 5)