markov/fckbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Редактировать reindex_worker.py

Browse Source
This commit is contained in:
Markov Andrey
2026-06-30 20:57:13 +00:00
parent 6f09bce70a
commit b01c66cdd0
1 changed files with 5 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
bots/workers/reindex_worker.py
View File

@@ -88,13 +88,6 @@ class ReindexWorker:
Выбирает до batch_size документов с indexed=False и повторяет индексацию. Выбирает до batch_size документов с indexed=False и повторяет индексацию.
Для каждого документа проверяет, не превышено ли число попыток. Для каждого документа проверяет, не превышено ли число попыток.
""" """
# Получаем документы, у которых indexed=False и retry_count < max_retries
# (добавим колонку retry_count, если её нет, или используем существующую metadata)
# В текущей схеме нет retry_count, поэтому просто будем пытаться индексировать заново.
# Чтобы избежать бесконечных попыток, можно ограничить по времени последней ошибки.
# Для простоты будем считать, что документ помечен как неудачный, и мы пытаемся его
# переиндексировать, пока не удастся, но с ограничением по времени (например, не старше 7 дней).
# Реализуем простой вариант: берём документы с indexed=False, которые были созданы не более 7 дней назад.
import datetime import datetime
cutoff = datetime.datetime.now() - datetime.timedelta(days=7) cutoff = datetime.datetime.now() - datetime.timedelta(days=7)
async with self.db.pool.acquire() as conn: async with self.db.pool.acquire() as conn:
@@ -117,19 +110,13 @@ class ReindexWorker:
for row in rows: for row in rows:
doc_id = row['id'] doc_id = row['id']
try: try:
# Восстанавливаем данные для индексации. Текст документа должен храниться # Восстанавливаем текст документа. Предполагаем, что в metadata есть поле 'file_text'
# либо в метаданных (как file_text), либо в отдельном поле, либо мы можем # или мы добавили колонку file_text в таблицу. Для демонстрации используем metadata.
# попытаться извлечь из Qdrant? Лучше хранить исходный текст в БД.
# Предположим, что у нас есть колонка file_text в documents (или мы её добавим).
# Поскольку в текущей схеме её нет, мы не можем переиндексировать без текста.
# Здесь нужно решение: либо добавить колонку, либо хранить файлы на диске.
# Для демонстрации реализуем заглушку: если в metadata есть 'file_text', используем его.
metadata = row.get('metadata') or {} metadata = row.get('metadata') or {}
file_text = metadata.get('file_text') file_text = metadata.get('file_text')
if not file_text: if not file_text:
# Если текст не сохранён, пропускаем (логируем ошибку)
logger.error(f"Документ {doc_id} не содержит текста для переиндексации, пропускаем") logger.error(f"Документ {doc_id} не содержит текста для переиндексации, пропускаем")
# Можно обновить indexed=False и добавить метку "невозможно переиндексировать" # Можно обновить метаданные, чтобы отметить как невосстановимый
continue continue
# Повторно индексируем документ # Повторно индексируем документ
@@ -142,23 +129,12 @@ class ReindexWorker:
metadata=metadata, metadata=metadata,
room_jid=row['room_jid'], room_jid=row['room_jid'],
file_hash=row['file_hash'], file_hash=row['file_hash'],
update_if_exists=True, # перезаписать существующий документ update_if_exists=True,
) )
# Если индексация успешна, новая запись будет иметь indexed=True,
# а старая будет удалена (в add_document). Поэтому нам нужно убедиться,
# что старая запись удалена. Однако add_document создаёт новую запись,
# а затем удаляет старую, поэтому здесь мы можем просто отметить старую как удалённую.
# Но проще: если add_document вернул новый doc_id и chunk_count >0, считаем успех.
if chunk_count > 0: if chunk_count > 0:
logger.info(f"Документ {doc_id} успешно переиндексирован (новый doc_id={doc_id_new})") logger.info(f"Документ {doc_id} успешно переиндексирован (новый doc_id={doc_id_new})")
# Удаляем старую запись (она уже должна быть удалена в add_document, но на всякий случай)
# await self.db.delete_documents_by_ids([doc_id])
else: else:
logger.error(f"Не удалось переиндексировать документ {doc_id}, chunk_count=0") logger.error(f"Не удалось переиндексировать документ {doc_id}, chunk_count=0")
except Exception as e: except Exception as e:
logger.exception(f"Ошибка переиндексации документа {doc_id}: {e}") logger.exception(f"Ошибка переиндексации документа {doc_id}: {e}")
# Здесь можно увеличить счётчик ошибок, если он есть
# В противном случае просто продолжим
# После обработки пачки принудительно закроем соединение? Нет, пул сам управляет.