diff --git a/bots/workers/indexing_worker.py b/bots/workers/indexing_worker.py index fb41bf8..efde168 100644 --- a/bots/workers/indexing_worker.py +++ b/bots/workers/indexing_worker.py @@ -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)