Add new file

core/commands/kb.py

@@ -0,0 +1,131 @@
+# -*- coding: utf-8 -*-
+"""
+Команды управления базой знаний: !clear, !global_clear, !kb.
+Поддержка комнат:
+- !clear в комнате очищает комнатную БЗ (требует прав администратора бота).
+- !kb в комнате показывает документы комнаты + глобальные.
+"""
+
+import logging
+from core.commands.base import Command
+
+logger = logging.getLogger(__name__)
+
+
+class ClearCommand(Command):
+    name = "!clear"
+    aliases = ["/clear"]
+
+    async def execute(self, msg, args: str, user_jid: str, room_jid: str = None):
+        """
+        Очищает базу знаний:
+        - В личном чате – личную БЗ пользователя.
+        - В групповом чате – БЗ комнаты (требует прав администратора бота).
+        """
+        if room_jid:
+            # Для очистки комнатной БЗ нужны права администратора бота
+            if user_jid not in self.bot.config.admin_jids:
+                reply = msg.reply("⛔ Только администраторы бота могут очищать комнатную БЗ.")
+                if reply:
+                    reply.send()   # исправлено
+                else:
+                    logger.error("Не удалось отправить сообщение о правах на очистку комнаты")
+                return
+            await self.bot.kb.clear_room_kb(room_jid)
+            reply = msg.reply(f"🗑 Комнатная БЗ {room_jid} очищена.")
+            if reply:
+                reply.send()   # исправлено
+            else:
+                logger.error("Не удалось отправить подтверждение очистки комнатной БЗ")
+        else:
+            await self.bot.kb.clear_user_kb(user_jid)
+            reply = msg.reply("🗑 Ваша ЛИЧНАЯ база знаний очищена.")
+            if reply:
+                reply.send()   # исправлено
+            else:
+                logger.error("Не удалось отправить подтверждение очистки личной БЗ")
+
+    def get_help(self) -> str:
+        return "Очищает вашу личную базу знаний. В групповом чате очищает БЗ комнаты (требует прав администратора бота)."
+
+
+class GlobalClearCommand(Command):
+    name = "!global_clear"
+    aliases = ["/global_clear"]
+    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()   # исправлено
+            else:
+                logger.error("Не удалось отправить сообщение о недостатке прав (kb)")
+            return
+        if room_jid:
+            reply = msg.reply("⚠️ Команда !global_clear работает только в личном чате.")
+            if reply:
+                reply.send()   # исправлено
+            else:
+                logger.error("Не удалось отправить сообщение об ограничении !global_clear")
+            return
+        await self.bot.kb.clear_global_kb()
+        reply = msg.reply("🌐 ГЛОБАЛЬНАЯ база знаний очищена.")
+        if reply:
+            reply.send()   # исправлено
+        else:
+            logger.error("Не удалось отправить подтверждение очистки глобальной БЗ")
+
+    def get_help(self) -> str:
+        return "Очищает глобальную базу знаний (админ, только в личном чате)."
+
+
+class KbCommand(Command):
+    name = "!kb"
+    aliases = ["/kb"]
+
+    async def execute(self, msg, args: str, user_jid: str, room_jid: str = None):
+        """
+        Показывает список документов в базе знаний:
+        - В личном чате: личные + глобальные.
+        - В групповом чате: документы комнаты + глобальные.
+        """
+        sources = await self.bot.kb.get_sources_with_type(user_jid, room_jid)
+        if not sources:
+            if room_jid:
+                reply = msg.reply("📭 База знаний этой комнаты пуста.")
+                if reply:
+                    reply.send()   # исправлено
+                else:
+                    logger.error("Не удалось отправить сообщение о пустой комнатной БЗ")
+            else:
+                reply = msg.reply("📭 Ваша личная база знаний пуста.")
+                if reply:
+                    reply.send()   # исправлено
+                else:
+                    logger.error("Не удалось отправить сообщение о пустой личной БЗ")
+            return
+        global_docs = [s['source_name'] for s in sources if s['is_global']]
+        other_docs = [s['source_name'] for s in sources if not s['is_global']]
+        lines = []
+        if global_docs:
+            lines.append("🌍 **Глобальные документы:**")
+            lines.extend(f"  • {name}" for name in global_docs)
+        if other_docs:
+            if room_jid:
+                lines.append(f"🏠 **Документы комнаты {room_jid}:**")
+            else:
+                lines.append("👤 **Ваши личные документы:**")
+            lines.extend(f"  • {name}" for name in other_docs)
+        reply = msg.reply("\n".join(lines))
+        if reply:
+            reply.send()   # исправлено
+        else:
+            logger.error("Не удалось отправить список документов БЗ")
+
+    def get_help(self) -> str:
+        return "Показывает список документов в базе знаний (личные/комнатные + глобальные)."