100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
Административные команды: !stats, !status, !clean.
|
||
Используют RAGClient.
|
||
"""
|
||
|
||
import os
|
||
import shutil
|
||
import logging
|
||
from datetime import datetime
|
||
from core.commands.base import Command
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class StatsCommand(Command):
|
||
name = "!stats"
|
||
aliases = ["/stats"]
|
||
permission = "admin"
|
||
|
||
async def execute(self, msg, args: str, user_jid: str, room_jid: str = None):
|
||
if not self.check_permission(user_jid):
|
||
reply = msg.reply("⛔ Нет прав.")
|
||
if reply: reply.send()
|
||
return
|
||
try:
|
||
stats = await self.bot.rag_client.get_stats()
|
||
uptime = datetime.now() - self.bot.start_time
|
||
response = (
|
||
f"📊 Статистика бота {self.bot.config.name}\n"
|
||
f"⏱ Uptime: {str(uptime).split('.')[0]}\n"
|
||
f"📚 Документов в БЗ: {stats.get('documents', 0)}\n"
|
||
f"⏱ Время работы сервера: {stats.get('uptime', 0):.0f} сек"
|
||
)
|
||
reply = msg.reply(response)
|
||
except Exception as e:
|
||
logger.exception(f"Ошибка в !stats: {e}")
|
||
reply = msg.reply(f"❌ Ошибка получения статистики: {e}")
|
||
if reply:
|
||
reply.send()
|
||
|
||
def get_help(self) -> str:
|
||
return "Показывает статистику (админ)."
|
||
|
||
|
||
class StatusCommand(Command):
|
||
name = "!status"
|
||
aliases = ["/status"]
|
||
permission = "admin"
|
||
|
||
async def execute(self, msg, args: str, user_jid: str, room_jid: str = None):
|
||
if not self.check_permission(user_jid):
|
||
reply = msg.reply("⛔ Нет прав.")
|
||
if reply: reply.send()
|
||
return
|
||
try:
|
||
health = await self.bot.rag_client.health_check()
|
||
if health:
|
||
status = "✅ Все сервисы работают"
|
||
else:
|
||
status = "⚠️ Некоторые сервисы недоступны"
|
||
reply = msg.reply(f"🛰 Статус бота: {status}")
|
||
except Exception as e:
|
||
logger.exception(f"Ошибка в !status: {e}")
|
||
reply = msg.reply(f"❌ Ошибка проверки статуса: {e}")
|
||
if reply:
|
||
reply.send()
|
||
|
||
def get_help(self) -> str:
|
||
return "Проверяет состояние сервисов (админ)."
|
||
|
||
|
||
class CleanCommand(Command):
|
||
name = "!clean"
|
||
aliases = ["/clean"]
|
||
permission = "admin"
|
||
|
||
async def execute(self, msg, args: str, user_jid: str, room_jid: str = None):
|
||
if not self.check_permission(user_jid):
|
||
reply = msg.reply("⛔ Нет прав.")
|
||
if reply: reply.send()
|
||
return
|
||
temp_dir = self.bot.config.temp_dir
|
||
count = 0
|
||
for f in os.listdir(temp_dir):
|
||
try:
|
||
p = os.path.join(temp_dir, f)
|
||
if os.path.isfile(p):
|
||
os.unlink(p)
|
||
elif os.path.isdir(p):
|
||
shutil.rmtree(p)
|
||
count += 1
|
||
except:
|
||
pass
|
||
reply = msg.reply(f"🧹 Удалено объектов из temp_dir: {count}")
|
||
if reply:
|
||
reply.send()
|
||
|
||
def get_help(self) -> str:
|
||
return "Очищает временную директорию (админ)." |