class Logger:
def __init__(self, dp: Dispatcher):
self.dp = dp
self.register_handlers()
def register_handlers(self):
@self.dp.message(Command("logger_show"))
@superuser_permission
def register_handlers(self):
@self.dp.message(Command("logger_show"))
@superuser_permission
async def command_logger_show(message: Message) -> None:
try:
# 1. Получаем ID оператора, который отправил команду
operator_id = int(message.from_user.id)
# 2. Извлекаем ID пользователя из текста команды
# Проверяем, что текст сообщения содержит ID
try:
_, user_id_str = message.text.split() # Ожидается формат: "/logger_show <user_id>"
user_id = int(user_id_str)
except ValueError:
await message.answer("Пожалуйста, укажите корректный ID пользователя: /logger_show <user_id>")
return
# 3. Получаем информацию о пользователе и операторе из базы данных
user_operator = get_user_by_id_from_database(operator_id)
user = get_user_by_id_from_database(user_id)
if not user_operator or not user:
await message.answer("Не удалось найти информацию об операторе или пользователе.")
return
# 4. Проверяем статус суперпользователя оператора и пользователя
operator_is_superuser = user_operator[USER_IS_SUPERUSER_INDEX]
user_is_superuser = user[USER_IS_SUPERUSER_INDEX]
# 5. Проверка прав доступа
if user_is_superuser and not operator_is_superuser:
await message.answer("У вас нет прав для выполнения этой команды.")
return
# 6. Извлечение данных из Excel-файла
try:
# Путь к Excel-файлу
excel_file_path = "logs.xlsx"
# Чтение данных
df = pd.read_excel(excel_file_path)
# Проверяем, что файл содержит данные
if df.empty:
await message.answer("Нет данных для выгрузки.")
return
# 7. Формируем ответ
log_messages = "\n".join(
[f"{row['pk']}: {row['name']} - {row['action']}" for _, row in df.iterrows()]
)
await message.answer(f"Логи:\n{log_messages}")
except FileNotFoundError:
await message.answer("Файл с логами не найден.")
except Exception as e:
await message.answer(f"Произошла ошибка при чтении Excel-файла: {e}")
except Exception as e:
await message.answer(f"Произошла ошибка: {e}")
@self.dp.message(Command("logger_delete"))
@superuser_permission
def register_handlers(self):
@self.dp.message(Command("logger_delete"))
@superuser_permission
async def command_logger_delete(message: Message) -> None:
try:
# 1. Получаем ID оператора, который отправил команду
operator_id = int(message.from_user.id)
# 2. Проверяем текст команды на наличие ключа "all" (удаление всего файла)
try:
_, log_identifier = message.text.split() # Ожидается формат: "/logger_delete all"
except ValueError:
await message.answer("Пожалуйста, укажите 'all' для удаления файла логов: /logger_delete all")
return
# 3. Проверяем права оператора
user_operator = get_user_by_id_from_database(operator_id)
if not user_operator:
await message.answer("Не удалось найти информацию об операторе.")
return
operator_is_superuser = user_operator[USER_IS_SUPERUSER_INDEX]
if not operator_is_superuser:
await message.answer("У вас нет прав для выполнения этой команды.")
return
# 4. Удаление файла логов
if log_identifier.lower() == "all":
# Путь к Excel-файлу
excel_file_path = "logs.xlsx"
# Проверяем существование файла
if os.path.exists(excel_file_path):
os.remove(excel_file_path) # Удаляем файл
await message.answer("Файл логов успешно удалён.")
else:
await message.answer("Файл логов не найден.")
else:
await message.answer("Некорректная команда. Используйте: /logger_delete all")
except Exception as e:
await message.answer(f"Произошла ошибка: {e}")
@self.dp.message(Command("logger_add"))
@superuser_permission
async def command_loger_add(message: Message) -> None:
...