Update file indexing_worker.py
This commit is contained in:
Markov Andrey
@@ -8,8 +8,8 @@
|
||||
Все операции с БД и Qdrant выполняются на сервере. Воркер только отправляет данные.
|
||||
|
||||
ДОБАВЛЕНО:
|
||||
- Обработка изображений через RAG-сервер (OCR)
|
||||
- Обработка аудио через RAG-сервер (транскрибация)
|
||||
- Уведомления пользователя об ошибках индексации.
|
||||
- Обработка ошибок с отправкой сообщений в чат.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
@@ -48,6 +48,15 @@ class IndexingWorker:
|
||||
except Exception as e:
|
||||
logger.exception(f"Ошибка в воркере при обработке задачи: {e}")
|
||||
self.bot.metrics_handler.increment("errors_total")
|
||||
# Отправляем уведомление пользователю о критической ошибке
|
||||
user_jid = task.get("user_jid")
|
||||
room_jid = task.get("room_jid")
|
||||
if user_jid:
|
||||
await self._send_message(
|
||||
user_jid,
|
||||
f"❌ Критическая ошибка при обработке задачи: {e}",
|
||||
room_jid
|
||||
)
|
||||
finally:
|
||||
self.bot.indexing_queue.task_done()
|
||||
|
||||
@@ -83,10 +92,10 @@ class IndexingWorker:
|
||||
)
|
||||
return
|
||||
total = 0
|
||||
errors = []
|
||||
for page_url, page_title, page_text in pages:
|
||||
if page_text:
|
||||
doc_title = custom_title if custom_title else (page_title if page_title else page_url)
|
||||
# Отправляем на сервер для индексации
|
||||
result = await self.bot.rag_client.index_document(
|
||||
file_name=page_url,
|
||||
file_text=page_text,
|
||||
@@ -99,12 +108,22 @@ class IndexingWorker:
|
||||
if result.get('doc_id'):
|
||||
total += 1
|
||||
else:
|
||||
logger.error(f"Ошибка индексации {page_url}: {result.get('error')}")
|
||||
await self._send_message(
|
||||
user_jid,
|
||||
f"✅ Проиндексировано {total} страниц (рекурсивно).",
|
||||
room_jid
|
||||
)
|
||||
error_msg = result.get('error', 'неизвестная ошибка')
|
||||
errors.append(f"{page_url}: {error_msg}")
|
||||
logger.error(f"Ошибка индексации {page_url}: {error_msg}")
|
||||
# Отправляем итоговое сообщение
|
||||
if errors:
|
||||
await self._send_message(
|
||||
user_jid,
|
||||
f"⚠️ Проиндексировано {total} страниц, ошибок: {len(errors)}. Первая ошибка: {errors[0]}",
|
||||
room_jid
|
||||
)
|
||||
else:
|
||||
await self._send_message(
|
||||
user_jid,
|
||||
f"✅ Проиндексировано {total} страниц (рекурсивно).",
|
||||
room_jid
|
||||
)
|
||||
else:
|
||||
title, text = await fetch_any_url(url)
|
||||
if text:
|
||||
@@ -126,9 +145,10 @@ class IndexingWorker:
|
||||
room_jid
|
||||
)
|
||||
else:
|
||||
error_msg = result.get('error', 'неизвестная ошибка')
|
||||
await self._send_message(
|
||||
user_jid,
|
||||
f"❌ Ошибка индексации {url}: {result.get('error')}",
|
||||
f"❌ Ошибка индексации {url}: {error_msg}",
|
||||
room_jid
|
||||
)
|
||||
else:
|
||||
@@ -195,8 +215,6 @@ class IndexingWorker:
|
||||
self.bot.last_file_texts[user_jid] = full_text
|
||||
|
||||
# --- 2. Сохранение постоянной копии в data/ (локально) ---
|
||||
# Это нужно для команд !summary и хирургической замены, которые работают с файлом локально.
|
||||
# В будущем можно перенести и их на сервер, но пока оставляем.
|
||||
with open(file_path, 'rb') as f:
|
||||
file_hash = hashlib.sha256(f.read()).hexdigest()
|
||||
short_hash = file_hash[:12]
|
||||
@@ -251,9 +269,9 @@ class IndexingWorker:
|
||||
return
|
||||
|
||||
# --- 5. Индексация документов через RAG-сервер ---
|
||||
# Отправляем на сервер каждый извлечённый файл (включая вложенные из архивов)
|
||||
if mode in ('personal', 'global') or (mode == 'room' and room_jid):
|
||||
saved = 0
|
||||
errors = []
|
||||
for sub_name, sub_text in files_list:
|
||||
if sub_text:
|
||||
result = await self.bot.rag_client.index_document(
|
||||
@@ -263,29 +281,41 @@ class IndexingWorker:
|
||||
room_jid=room_jid,
|
||||
is_global=is_global,
|
||||
title=title,
|
||||
# file_hash=file_hash, # можно передать, если нужно
|
||||
update_if_exists=True
|
||||
)
|
||||
if result.get('doc_id'):
|
||||
saved += 1
|
||||
else:
|
||||
logger.error(f"Ошибка индексации {sub_name}: {result.get('error')}")
|
||||
if saved:
|
||||
error_msg = result.get('error', 'неизвестная ошибка')
|
||||
errors.append(f"{sub_name}: {error_msg}")
|
||||
logger.error(f"Ошибка индексации {sub_name}: {error_msg}")
|
||||
|
||||
if saved > 0:
|
||||
if room_jid:
|
||||
scope = f"комнату {room_jid}"
|
||||
elif is_global:
|
||||
scope = "глобальную"
|
||||
else:
|
||||
scope = "личную"
|
||||
await self._send_message(
|
||||
user_jid,
|
||||
f"🎓 {saved} файлов добавлено в {scope} БЗ.",
|
||||
room_jid
|
||||
)
|
||||
|
||||
if errors:
|
||||
await self._send_message(
|
||||
user_jid,
|
||||
f"🎓 {saved} файлов добавлено в {scope} БЗ, но возникли ошибки: {errors[0]}",
|
||||
room_jid
|
||||
)
|
||||
else:
|
||||
await self._send_message(
|
||||
user_jid,
|
||||
f"🎓 {saved} файлов добавлено в {scope} БЗ.",
|
||||
room_jid
|
||||
)
|
||||
else:
|
||||
# Если ни один файл не проиндексирован
|
||||
error_summary = "; ".join(errors[:3]) # покажем первые 3 ошибки
|
||||
await self._send_message(
|
||||
user_jid,
|
||||
f"⚠️ Не удалось проиндексировать {original_fname}. Проверьте логи.",
|
||||
f"⚠️ Не удалось проиндексировать {original_fname}. Ошибки: {error_summary}",
|
||||
room_jid
|
||||
)
|
||||
else:
|
||||
@@ -301,11 +331,11 @@ class IndexingWorker:
|
||||
self.bot.metrics_handler.increment("errors_total")
|
||||
await self._send_message(
|
||||
user_jid,
|
||||
f"❌ Ошибка при обработке файла '{original_fname}'. Проверьте логи.",
|
||||
f"❌ Ошибка при обработке файла '{original_fname}': {e}",
|
||||
room_jid
|
||||
)
|
||||
finally:
|
||||
# Удаляем временный файл (постоянная копия уже сохранена)
|
||||
# Удаляем временный файл
|
||||
try:
|
||||
if os.path.exists(file_path):
|
||||
os.unlink(file_path)
|
||||
|
||||
Reference in New Issue
Block a user