added utils.py

1. Добавил message_thread_id для всех команд, убрав костыль, который отвечал за корректную отправку
   сообщений в топик.
2. Функции (определение администратора, удаление сообщений) вынес в utils.py, от куда они будут вызываться в командах. Модули стали более читаемы из-за уменьшения количества строк кода в них.
3. Дописал manual_unban и добавил error в config.py
4. Оптимизация
This commit is contained in:
2025-07-28 21:03:28 +03:00
parent d73c0079f0
commit 853bfcdbb4
8 changed files with 692 additions and 492 deletions

177
src/utils.py Normal file
View File

@@ -0,0 +1,177 @@
from telebot.async_telebot import AsyncTeleBot
from telebot.types import Message
import asyncio
import logging
# Получаем логгер для текущего модуля
logger = logging.getLogger(__name__)
# Удаляет определённое количество сообщения
async def delete_messages(bot: AsyncTeleBot, message: Message, time_sleep: int, number_message: int):
await asyncio.sleep(time_sleep)
for i in range(number_message):
await bot.delete_message(message.chat.id, message.message_id+i)
# Проверяет, является ли отправитель администратором
async def check_admin_status(bot: AsyncTeleBot, message: Message):
if message.reply_to_message and message.is_topic_message is None:
message.message_thread_id = None
try:
# Получаем статус отправителя
admin_status = await bot.get_chat_member(message.chat.id, message.from_user.id)
# Проверка наличия прав администратора/создателя
if admin_status.status not in ('administrator', 'creator'):
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['no_admin_rights'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
return 1
# Проверка права на ограничение участников
if admin_status.status == 'administrator' and not admin_status.can_restrict_members:
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['no_restrict_rights'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
return 1
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)
return 1
# Проверяет статус целевого пользователя
async def check_target_status(bot: AsyncTeleBot, message: Message, target_user):
if message.reply_to_message and message.is_topic_message is None:
message.message_thread_id = None
try:
# Получаем статус пользователя
target_status = await bot.get_chat_member(
chat_id=message.chat.id,
user_id=target_user.id,
)
# Проверяем, является ли цель администратором или создателем
if target_status.status in ('administrator', 'creator'):
# Отправляем предупреждение
await bot.send_message(
chat_id=message.chat.id,
text=COMMAND_MESSAGES['cant_mute_admin'],
message_thread_id=message.message_thread_id,
)
# Удаляем сообщения через 5 секунд
await delete_messages(bot, message, time_sleep=5, number_message=2)
return
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)
return
# Возвращает количество секунд
def parse_mute_time(time_str: str) -> int | None:
# Нормализация входной строки
normalized = time_str.strip().lower()
# Словарь для конвертации единиц времени в секунды
time_units = {
'm': 60, 'м': 60, # минуты
'h': 3600, 'ч': 3600, # часы
'd': 86400, 'д': 86400 # дни
}
# Проверям, существует ли строка
if not normalized:
return None
# Проверяем последний символ (единица измерения)
unit = normalized[-1]
if unit not in time_units:
return None
# Парсим числовую часть
try:
value = int(normalized[:-1])
if value <= 0:
return None
return value * time_units[unit]
except (ValueError, TypeError):
return None
# Форматирует время в нормальный вид
def format_mute_time(seconds: int) -> str:
# Обработка минут
minutes = seconds // 60
if minutes < 60:
if minutes % 10 == 1 and minutes % 100 != 11:
return f"{minutes} минуту"
elif 2 <= minutes % 10 <= 4 and (minutes % 100 < 10 or minutes % 100 >= 20):
return f"{minutes} минуты"
else:
return f"{minutes} минут"
# Обработка часов
hours = minutes // 60
if hours < 24:
if hours % 10 == 1 and hours % 100 != 11:
return f"{hours} час"
elif 2 <= hours % 10 <= 4 and (hours % 100 < 10 or hours % 100 >= 20):
return f"{hours} часа"
else:
return f"{hours} часов"
# Обработка дней
days = hours // 24
if days % 10 == 1 and days % 100 != 11:
return f"{days} день"
elif 2 <= days % 10 <= 4 and (days % 100 < 10 or days % 100 >= 20):
return f"{days} дня"
else:
return f"{days} дней"