Редактировать kb.py
This commit is contained in:
Markov Andrey
@@ -1,9 +1,7 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Команды управления базой знаний: !clear, !global_clear, !kb.
|
||||
Поддержка комнат:
|
||||
- !clear в комнате очищает комнатную БЗ (требует прав администратора бота).
|
||||
- !kb в комнате показывает документы комнаты + глобальные.
|
||||
Команда !kb – список документов в базе знаний (личные/комнатные + глобальные).
|
||||
Использует RAGClient.
|
||||
"""
|
||||
|
||||
import logging
|
||||
@@ -12,120 +10,35 @@ 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("Не удалось отправить сообщение о пустой комнатной БЗ")
|
||||
try:
|
||||
result = await self.bot.rag_client.list_documents(user_jid, room_jid)
|
||||
sources = result.get('documents', [])
|
||||
if not sources:
|
||||
reply = msg.reply("📭 База знаний пуста.")
|
||||
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))
|
||||
global_docs = [s['source_name'] for s in sources if s.get('is_global')]
|
||||
other_docs = [s['source_name'] for s in sources if not s.get('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))
|
||||
except Exception as e:
|
||||
logger.exception(f"Ошибка в !kb: {e}")
|
||||
reply = msg.reply(f"❌ Ошибка: {e}")
|
||||
if reply:
|
||||
reply.send() # исправлено
|
||||
else:
|
||||
logger.error("Не удалось отправить список документов БЗ")
|
||||
reply.send()
|
||||
|
||||
def get_help(self) -> str:
|
||||
return "Показывает список документов в базе знаний (личные/комнатные + глобальные)."
|
||||
Reference in New Issue
Block a user