# -*- 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 "Показывает список документов в базе знаний (личные/комнатные + глобальные)."