43034eda33
add_document слал index.enrich_full_text для КАЖДОГО документа с url, даже когда full_text уже пришёл от парсера (у КиберЛенинки — OCR-фрагмент прямо в ответе поиска). Страница статьи там HTML, не PDF — enrich гарантированно ничего не находит, только тратит запрос впустую. При массовой заливке (десятки источников параллельно) это давало шквал запросов к чужому сайту и 503 в ответ — саму заливку не ломало, но съедало время и нагружало источник без всякой пользы.
626 lines
27 KiB
Python
626 lines
27 KiB
Python
"""Celery задачи индексации документов и проверки плагиата (уровни 1-2)."""
|
||
|
||
import io
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from celery.utils.log import get_task_logger
|
||
from sqlalchemy import func, select
|
||
from sqlalchemy.exc import IntegrityError
|
||
|
||
from app.algorithms.minhash import add_to_lsh, find_similar
|
||
from app.algorithms.winnowing import winnow
|
||
from app.celery_app import celery_app
|
||
from app.config import settings
|
||
from app.db import db_session, get_minio, refund_plagiarism_quota, update_task_status
|
||
from app.extractors.docx import extract_text_from_docx, extract_text_from_txt
|
||
from app.extractors.pdf import extract_text_from_pdf
|
||
from app.fragments import split_into_fragments
|
||
from app.staging import staged_work_to_doc_data
|
||
|
||
logger = get_task_logger(__name__)
|
||
|
||
|
||
@celery_app.task(
|
||
name="index.extract_and_check",
|
||
bind=True,
|
||
max_retries=3,
|
||
default_retry_delay=60,
|
||
)
|
||
def extract_and_check(
|
||
self,
|
||
task_id: str,
|
||
minio_key: str,
|
||
filename: str,
|
||
) -> dict[str, Any]:
|
||
"""
|
||
Извлечь текст из документа и проверить плагиат (уровни 1-2).
|
||
|
||
Алгоритм:
|
||
1. Скачать файл из MinIO
|
||
2. Извлечь текст (PDF/DOCX/TXT)
|
||
3. Разбить на фрагменты по 200 слов с перекрытием 50 слов
|
||
4. Уровень 1: Winnowing против базы fingerprints в PostgreSQL
|
||
5. Уровень 2: MinHash LSH нечёткий поиск
|
||
6. Диспатч gpu.check_plagiarism для уровней 3 и 4
|
||
|
||
Args:
|
||
task_id: ID задачи в PostgreSQL
|
||
minio_key: Ключ объекта в MinIO
|
||
filename: Оригинальное имя файла
|
||
"""
|
||
logger.info(f"Извлечение текста для задачи {task_id!r}, файл: {filename!r}")
|
||
|
||
update_task_status(task_id, "processing")
|
||
|
||
try:
|
||
# Скачать из MinIO
|
||
minio = get_minio()
|
||
response = minio.get_object(settings.MINIO_BUCKET_DOCS, minio_key)
|
||
file_data = response.read()
|
||
response.close()
|
||
response.release_conn()
|
||
|
||
logger.info(f"Файл скачан из MinIO: {minio_key} ({len(file_data)} байт)")
|
||
|
||
# Извлечь текст в зависимости от формата
|
||
ext = Path(filename).suffix.lower()
|
||
if ext == ".pdf":
|
||
text = extract_text_from_pdf(file_data)
|
||
elif ext == ".docx":
|
||
text = extract_text_from_docx(file_data)
|
||
else:
|
||
text = extract_text_from_txt(file_data)
|
||
|
||
if not text.strip():
|
||
raise ValueError("Не удалось извлечь текст из документа")
|
||
|
||
word_count = len(text.split())
|
||
logger.info(f"Текст извлечён: {len(text)} символов, {word_count} слов")
|
||
|
||
# ──── Автосбор в отстойник (буфер для пополнения базы) ────────────────
|
||
# Сохраняем извлечённый текст и метаданные для последующего ручного
|
||
# одобрения админом. Не дублируем в базу источников до решения.
|
||
_stage_work(task_id, minio_key, filename, text, word_count)
|
||
|
||
# Разбить на фрагменты
|
||
fragments = split_into_fragments(
|
||
text,
|
||
window=settings.FRAGMENT_WINDOW_WORDS,
|
||
overlap=settings.FRAGMENT_OVERLAP_WORDS,
|
||
)
|
||
logger.info(f"Фрагментов создано: {len(fragments)}")
|
||
|
||
# ──── Уровень 1: Winnowing fingerprints (пофрагментно, с локализацией) ───
|
||
# Winnow'им КАЖДЫЙ фрагмент отдельно и ищем, с каким источником и на
|
||
# какой позиции он совпадает. Так в отчёте видно не «документ похож на X»,
|
||
# а «фрагмент на символах A–B скопирован из источника X».
|
||
level1_matches: list[dict] = []
|
||
from app.models import Document, Fingerprint
|
||
|
||
with db_session() as session:
|
||
doc_cache: dict[int, Document] = {}
|
||
|
||
for frag in fragments:
|
||
frag_fp = winnow(frag["text"])
|
||
if not frag_fp:
|
||
continue
|
||
frag_hashes = list(frag_fp)
|
||
|
||
# Источник, разделяющий больше всего отпечатков с этим фрагментом
|
||
row = session.execute(
|
||
select(
|
||
Fingerprint.doc_id,
|
||
func.count(Fingerprint.id).label("cnt"),
|
||
)
|
||
.where(Fingerprint.hash_value.in_(frag_hashes))
|
||
.group_by(Fingerprint.doc_id)
|
||
.order_by(func.count(Fingerprint.id).desc())
|
||
.limit(1)
|
||
).first()
|
||
|
||
if row is None:
|
||
continue
|
||
|
||
doc_id, cnt = row
|
||
# Доля отпечатков фрагмента, найденных в источнике
|
||
similarity = cnt / len(frag_hashes) * 100
|
||
if similarity < settings.EXACT_FRAGMENT_THRESHOLD:
|
||
continue
|
||
|
||
doc = doc_cache.get(doc_id)
|
||
if doc is None:
|
||
doc = session.get(Document, doc_id)
|
||
if doc is None:
|
||
continue
|
||
doc_cache[doc_id] = doc
|
||
|
||
level1_matches.append({
|
||
"fragment": frag["text"][:300],
|
||
"position_start": frag["start"],
|
||
"position_end": frag["end"],
|
||
"similarity": round(min(similarity, 100.0), 1),
|
||
"method": "exact",
|
||
"source_title": doc.title,
|
||
"source_url": doc.url,
|
||
"source_db": doc.source,
|
||
})
|
||
|
||
logger.info(
|
||
f"Уровень 1 (Winnowing): {len(level1_matches)} совпадений-фрагментов"
|
||
)
|
||
|
||
# ──── Уровень 2: MinHash LSH ────────────────────────────────────────────
|
||
level2_matches: list[dict] = []
|
||
similar_keys = find_similar(text)
|
||
|
||
if similar_keys:
|
||
from app.models import Document
|
||
|
||
with db_session() as session:
|
||
for key in similar_keys[:10]:
|
||
# Ключ формата "doc:{id}"
|
||
try:
|
||
doc_id = int(key.split(":")[-1])
|
||
doc = session.get(Document, doc_id)
|
||
if not doc:
|
||
continue
|
||
|
||
level2_matches.append({
|
||
"fragment": text[:200],
|
||
"position_start": 0,
|
||
"position_end": len(text),
|
||
"similarity": 60.0, # MinHash даёт только факт похожести
|
||
"method": "fuzzy",
|
||
"source_title": doc.title,
|
||
"source_url": doc.url,
|
||
"source_db": doc.source,
|
||
})
|
||
except (ValueError, IndexError):
|
||
continue
|
||
|
||
logger.info(f"Уровень 2 (MinHash): {len(level2_matches)} совпадений")
|
||
|
||
# ──── Диспатч GPU задачи (уровни 3-4) ─────────────────────────────────
|
||
# Передаём только текстовые данные (JSON-сериализуемые)
|
||
celery_app.send_task(
|
||
"gpu.check_plagiarism",
|
||
args=[task_id, text, fragments],
|
||
kwargs={
|
||
"level1_matches": level1_matches,
|
||
"level2_matches": level2_matches,
|
||
},
|
||
queue="queue.gpu",
|
||
)
|
||
|
||
logger.info(f"GPU задача отправлена для задачи {task_id!r}")
|
||
return {
|
||
"task_id": task_id,
|
||
"fragments": len(fragments),
|
||
"level1": len(level1_matches),
|
||
"level2": len(level2_matches),
|
||
}
|
||
|
||
except ValueError as exc:
|
||
# Детерминированная ошибка (битый файл, не тот формат, пустой текст) —
|
||
# повтор не поможет: результат будет тем же на 2-й и 3-й попытке. Не
|
||
# ретраим (быстрее фидбек юзеру) и возвращаем месячную квоту — реальная
|
||
# проверка так и не началась, списывать не за что.
|
||
logger.warning(f"Задача {task_id!r}: не удалось обработать файл: {exc}")
|
||
update_task_status(task_id, "failed", str(exc))
|
||
refund_plagiarism_quota(task_id)
|
||
return {"task_id": task_id, "status": "failed", "error": str(exc)}
|
||
|
||
except Exception as exc:
|
||
logger.error(f"Ошибка при обработке задачи {task_id!r}: {exc}", exc_info=True)
|
||
update_task_status(task_id, "failed", str(exc))
|
||
raise self.retry(exc=exc, countdown=60) from exc
|
||
|
||
|
||
@celery_app.task(name="index.add_document")
|
||
def add_document(doc_data: dict[str, Any], dispatch_embed: bool = True) -> dict[str, Any]:
|
||
"""
|
||
Добавить документ из внешнего источника в систему.
|
||
|
||
Алгоритм:
|
||
1. Дедупликация по ext_id
|
||
2. Сохранить метаданные в PostgreSQL
|
||
3. Вычислить провизорные Winnowing fingerprints (из аннотации)
|
||
4. Добавить в MinHash LSH
|
||
5. Индексировать в Elasticsearch
|
||
6. Диспатч gpu.embed_documents для FAISS эмбеддингов (если dispatch_embed)
|
||
7. Если включён FETCH_FULL_TEXT и есть url — диспатч index.enrich_full_text,
|
||
который скачает полный текст и пересчитает fingerprints по нему.
|
||
|
||
Args:
|
||
doc_data: Словарь с метаданными документа
|
||
dispatch_embed: Диспатчить ли эмбеддинг по одному документу. При массовой
|
||
заливке run_parser выключает это и батчит эмбеддинги сам.
|
||
|
||
Returns:
|
||
dict со статусом операции и doc_id
|
||
"""
|
||
from app.models import Document, Fingerprint
|
||
|
||
ext_id = doc_data.get("ext_id")
|
||
if not ext_id:
|
||
return {"status": "error", "reason": "ext_id обязателен"}
|
||
|
||
with db_session() as session:
|
||
# Проверить дублирование
|
||
existing = session.execute(
|
||
select(Document).where(Document.ext_id == ext_id)
|
||
).scalar_one_or_none()
|
||
|
||
if existing:
|
||
return {"status": "duplicate", "doc_id": existing.id}
|
||
|
||
# Создать документ
|
||
allowed_fields = {c.key for c in Document.__table__.columns}
|
||
doc_kwargs = {k: v for k, v in doc_data.items() if k in allowed_fields}
|
||
|
||
doc = Document(**doc_kwargs)
|
||
session.add(doc)
|
||
try:
|
||
session.flush()
|
||
except IntegrityError:
|
||
# Гонка: другой воркер параллельно вставил этот ext_id между нашей
|
||
# проверкой existing и flush. Уникальный индекс отработал — это дубль.
|
||
session.rollback()
|
||
existing = session.execute(
|
||
select(Document).where(Document.ext_id == ext_id)
|
||
).scalar_one_or_none()
|
||
return {"status": "duplicate", "doc_id": existing.id if existing else None}
|
||
doc_id = doc.id
|
||
|
||
# Вычислить fingerprints
|
||
text = doc_data.get("full_text") or doc_data.get("abstract", "") or ""
|
||
if text:
|
||
fp = winnow(text)
|
||
fingerprints_to_add = list(fp)[: settings.MAX_FINGERPRINTS_PER_DOC]
|
||
for i, hash_val in enumerate(fingerprints_to_add):
|
||
session.add(Fingerprint(doc_id=doc_id, hash_value=hash_val, position=i))
|
||
|
||
# Добавить в MinHash LSH (in-memory)
|
||
add_to_lsh(f"doc:{doc_id}", text)
|
||
|
||
session.commit()
|
||
|
||
logger.info(f"Документ {doc_id} добавлен в PostgreSQL: {doc_data.get('title', '')[:50]!r}")
|
||
|
||
# Индексация в Elasticsearch
|
||
try:
|
||
from elasticsearch import Elasticsearch
|
||
|
||
from app.config import settings as cfg
|
||
|
||
es = Elasticsearch(cfg.ELASTICSEARCH_URL)
|
||
es_doc = {
|
||
"doc_id": doc_id,
|
||
"source": doc_data.get("source"),
|
||
"title": doc_data.get("title", ""),
|
||
"abstract": doc_data.get("abstract", ""),
|
||
"authors": " ".join(
|
||
f"{a.get('last_name', '')} {a.get('first_name', '')}"
|
||
for a in doc_data.get("authors", [])
|
||
),
|
||
"year": doc_data.get("year"),
|
||
"lang": doc_data.get("lang"),
|
||
"journal": doc_data.get("journal"),
|
||
"doi": doc_data.get("doi"),
|
||
"url": doc_data.get("url"),
|
||
}
|
||
es.index(index="documents", id=str(doc_id), document=es_doc)
|
||
logger.info(f"Документ {doc_id} проиндексирован в Elasticsearch")
|
||
except Exception as e:
|
||
logger.warning(f"Ошибка индексации в ES для документа {doc_id}: {e}")
|
||
|
||
# Диспатч FAISS эмбеддингов (по одному документу; при массовой заливке
|
||
# run_parser выключает это и батчит сам)
|
||
if dispatch_embed:
|
||
celery_app.send_task(
|
||
"gpu.embed_documents",
|
||
args=[[doc_id]],
|
||
queue="queue.gpu",
|
||
)
|
||
|
||
# Обогащение полным текстом: скачать PDF по url и пересчитать fingerprints.
|
||
# Если full_text уже есть (например, OCR-фрагмент из ответа поиска
|
||
# КиберЛенинки) — не дублируем запрос: страница статьи там HTML, а не PDF,
|
||
# enrich_full_text гарантированно ничего не найдёт, только зря нагрузит
|
||
# источник (и получит 503 при массовой заливке).
|
||
if settings.FETCH_FULL_TEXT and doc_data.get("url") and not doc_data.get("full_text"):
|
||
celery_app.send_task(
|
||
"index.enrich_full_text",
|
||
args=[doc_id, doc_data["url"]],
|
||
queue="queue.index",
|
||
)
|
||
|
||
return {"status": "indexed", "doc_id": doc_id}
|
||
|
||
|
||
@celery_app.task(
|
||
name="index.enrich_full_text",
|
||
bind=True,
|
||
max_retries=2,
|
||
default_retry_delay=120,
|
||
)
|
||
def enrich_full_text(self, doc_id: int, url: str) -> dict[str, Any]:
|
||
"""Скачать полный текст статьи и пересчитать по нему fingerprints.
|
||
|
||
Метаданные и провизорные fingerprints (из аннотации) уже сохранены в
|
||
add_document. Здесь мы:
|
||
1. Скачиваем PDF по url и извлекаем текст (best-effort).
|
||
2. Сохраняем полный текст в MinIO (bucket documents, префикс corpus/).
|
||
3. Заменяем fingerprints документа на посчитанные по полному тексту.
|
||
4. Обновляем MinHash LSH.
|
||
|
||
Недоступный/не-PDF источник — не ошибка: возвращаем no_fulltext.
|
||
"""
|
||
from app.fulltext import fetch_full_text
|
||
from app.models import Document, Fingerprint
|
||
|
||
text = fetch_full_text(url)
|
||
if not text:
|
||
return {"status": "no_fulltext", "doc_id": doc_id}
|
||
|
||
# Сохранить полный текст в MinIO
|
||
try:
|
||
minio = get_minio()
|
||
key = f"corpus/{doc_id}.txt"
|
||
data = text.encode("utf-8")
|
||
minio.put_object(
|
||
settings.MINIO_BUCKET_DOCS, key, io.BytesIO(data), length=len(data),
|
||
content_type="text/plain; charset=utf-8",
|
||
)
|
||
except Exception as exc:
|
||
logger.error(f"enrich_full_text: не удалось сохранить текст в MinIO для {doc_id}: {exc}")
|
||
raise self.retry(exc=exc, countdown=120) from exc
|
||
|
||
# Пересчитать fingerprints по полному тексту
|
||
doc_fp = winnow(text)
|
||
hashes = list(doc_fp)[: settings.MAX_FINGERPRINTS_PER_DOC]
|
||
|
||
from sqlalchemy import delete
|
||
|
||
with db_session() as session:
|
||
doc = session.get(Document, doc_id)
|
||
if doc is None:
|
||
return {"status": "doc_gone", "doc_id": doc_id}
|
||
doc.minio_key = key
|
||
# Удалить провизорные fingerprints и записать новые
|
||
session.execute(delete(Fingerprint).where(Fingerprint.doc_id == doc_id))
|
||
for i, hash_val in enumerate(hashes):
|
||
session.add(Fingerprint(doc_id=doc_id, hash_value=hash_val, position=i))
|
||
session.commit()
|
||
|
||
# Обновить MinHash LSH по полному тексту
|
||
add_to_lsh(f"doc:{doc_id}", text)
|
||
|
||
logger.info(
|
||
f"enrich_full_text: doc={doc_id} полный текст {len(text)} симв., "
|
||
f"fingerprints={len(hashes)}"
|
||
)
|
||
return {"status": "ok", "doc_id": doc_id, "chars": len(text), "fingerprints": len(hashes)}
|
||
|
||
|
||
@celery_app.task(
|
||
name="index.auto_approve_submission",
|
||
bind=True,
|
||
max_retries=2,
|
||
default_retry_delay=60,
|
||
)
|
||
def auto_approve_submission(self, task_id: str) -> dict[str, Any]:
|
||
"""Автоматически добавить проверенную работу студента в базу для сравнения.
|
||
|
||
Вызывается из gpu.check_plagiarism ПОСЛЕ сохранения результата проверки —
|
||
так работа не сматчится сама с собой. Идемпотентно: пропускает, если
|
||
StagedWork не 'pending' (уже одобрена/отклонена вручную или повторный вызов)
|
||
или если AUTO_APPROVE_SUBMISSIONS выключен настройкой — тогда работа так и
|
||
остаётся ждать ручного решения в админке.
|
||
"""
|
||
from app.models import StagedWork
|
||
|
||
if not settings.AUTO_APPROVE_SUBMISSIONS:
|
||
return {"status": "skipped", "reason": "auto-approve выключен настройкой"}
|
||
|
||
with db_session() as session:
|
||
sw = session.execute(
|
||
select(StagedWork).where(StagedWork.task_id == task_id)
|
||
).scalar_one_or_none()
|
||
if sw is None:
|
||
return {"status": "skipped", "reason": "StagedWork не найден"}
|
||
if sw.status != "pending":
|
||
return {"status": "skipped", "reason": f"уже {sw.status}"}
|
||
|
||
full_text = ""
|
||
if sw.text_key:
|
||
try:
|
||
minio = get_minio()
|
||
obj = minio.get_object(settings.MINIO_BUCKET_STAGING, sw.text_key)
|
||
full_text = obj.read().decode("utf-8", errors="replace")
|
||
except Exception as e:
|
||
logger.warning(f"auto_approve_submission: не прочитан текст {task_id!r}: {e}")
|
||
return {"status": "error", "reason": str(e)}
|
||
|
||
doc_data = staged_work_to_doc_data(sw, full_text)
|
||
sw.status = "approved"
|
||
sw.reviewed_by = None # None = одобрено автоматически, не человеком
|
||
sw.reviewed_at = datetime.now(UTC)
|
||
session.commit()
|
||
|
||
result = add_document(doc_data, dispatch_embed=True)
|
||
logger.info(f"auto_approve_submission: задача {task_id!r} → {result}")
|
||
return result
|
||
|
||
|
||
def _stage_work(
|
||
task_id: str,
|
||
minio_key: str,
|
||
filename: str,
|
||
text: str,
|
||
word_count: int,
|
||
) -> None:
|
||
"""Сохранить проверенную работу в отстойник (StagedWork) + текст в MinIO.
|
||
|
||
Идемпотентно по task_id: повторный вызов (например при requeue) не дублирует.
|
||
Ошибки логируются, но не валят основную проверку плагиата.
|
||
"""
|
||
try:
|
||
from app.models import StagedWork
|
||
|
||
# Сохранить извлечённый текст в bucket staging
|
||
text_key = f"text/{task_id}.txt"
|
||
minio = get_minio()
|
||
bucket = settings.MINIO_BUCKET_STAGING
|
||
if not minio.bucket_exists(bucket):
|
||
minio.make_bucket(bucket)
|
||
data = text.encode("utf-8")
|
||
minio.put_object(
|
||
bucket, text_key, io.BytesIO(data), length=len(data),
|
||
content_type="text/plain; charset=utf-8",
|
||
)
|
||
|
||
with db_session() as session:
|
||
existing = session.execute(
|
||
select(StagedWork).where(StagedWork.task_id == task_id)
|
||
).scalar_one_or_none()
|
||
if existing:
|
||
return
|
||
|
||
# Достать user_id из задачи
|
||
from app.models import Task
|
||
task = session.get(Task, task_id)
|
||
user_id = task.user_id if task else None
|
||
|
||
session.add(StagedWork(
|
||
user_id=user_id,
|
||
task_id=task_id,
|
||
filename=filename,
|
||
minio_key=minio_key,
|
||
text_key=text_key,
|
||
title=Path(filename).stem if filename else None,
|
||
word_count=word_count,
|
||
status="pending",
|
||
))
|
||
session.commit()
|
||
logger.info(f"Работа задачи {task_id!r} добавлена в отстойник")
|
||
except Exception as e:
|
||
logger.warning(f"Не удалось добавить работу {task_id!r} в отстойник: {e}")
|
||
|
||
|
||
@celery_app.task(name="index.run_parser")
|
||
def run_parser(source_id: int) -> dict[str, Any]:
|
||
"""Запустить парсинг источника и наполнить базу документов.
|
||
|
||
Переиспользует парсеры из scripts/parsers (BaseParser.run) и
|
||
задачу add_document для каждого полученного документа.
|
||
"""
|
||
import sys
|
||
from datetime import datetime
|
||
|
||
from app.models import ParseSource
|
||
|
||
# Парсеры лежат в /parsers (скопированы в образ)
|
||
if "/parsers" not in sys.path:
|
||
sys.path.insert(0, "/parsers")
|
||
|
||
with db_session() as session:
|
||
src = session.get(ParseSource, source_id)
|
||
if src is None:
|
||
return {"status": "error", "reason": "источник не найден"}
|
||
cfg = {
|
||
"source_type": src.source_type,
|
||
"query": src.query,
|
||
"lang": src.lang,
|
||
"year_from": src.year_from,
|
||
"year_to": src.year_to,
|
||
"limit": src.limit,
|
||
}
|
||
src.last_status = "running"
|
||
session.commit()
|
||
|
||
added = 0
|
||
error_msg = None
|
||
try:
|
||
# Выбрать парсер по типу и собрать совместимые с его fetch() аргументы
|
||
stype = cfg["source_type"]
|
||
if stype == "openalex":
|
||
from openalex import OpenAlexParser as P
|
||
fetch_kwargs = {
|
||
"query": cfg.get("query") or "",
|
||
"limit": cfg["limit"],
|
||
"lang": cfg.get("lang"),
|
||
"year_from": cfg.get("year_from"),
|
||
"year_to": cfg.get("year_to"),
|
||
"open_access_only": settings.INGEST_OPEN_ACCESS_ONLY,
|
||
}
|
||
elif stype == "cyberleninka":
|
||
from cyberleninka import CyberLeninkaParser as P
|
||
fetch_kwargs = {"query": cfg.get("query") or "", "limit": cfg["limit"]}
|
||
elif stype == "arxiv":
|
||
from arxiv import ArxivParser as P
|
||
fetch_kwargs = {
|
||
"query": cfg.get("query") or "",
|
||
"limit": cfg["limit"],
|
||
"year_from": cfg.get("year_from"),
|
||
}
|
||
elif stype == "pmc":
|
||
from pmc import PMCParser as P
|
||
fetch_kwargs = {
|
||
"query": cfg.get("query") or "",
|
||
"limit": cfg["limit"],
|
||
"year_from": cfg.get("year_from"),
|
||
"year_to": cfg.get("year_to"),
|
||
}
|
||
else:
|
||
raise ValueError(f"неизвестный тип источника: {stype}")
|
||
|
||
parser = P()
|
||
# fetch+transform без записи в JSONL — работаем in-memory
|
||
raw_docs = parser.fetch(**fetch_kwargs)
|
||
|
||
# Эмбеддинги диспатчим пачками, а не по одному документу: так worker-gpu
|
||
# кодирует батч разом и переписывает FAISS-индекс на диск раз в N добавлений,
|
||
# а не на каждый документ.
|
||
embed_batch: list[int] = []
|
||
batch_size = settings.EMBED_BATCH_SIZE
|
||
|
||
def _flush_embed() -> None:
|
||
if embed_batch:
|
||
celery_app.send_task(
|
||
"gpu.embed_documents", args=[list(embed_batch)], queue="queue.gpu"
|
||
)
|
||
embed_batch.clear()
|
||
|
||
for raw in raw_docs:
|
||
try:
|
||
doc = parser.transform(raw)
|
||
if not (doc and doc.get("title") and doc.get("ext_id")):
|
||
continue
|
||
result = add_document(doc, dispatch_embed=False)
|
||
if result.get("status") == "indexed":
|
||
added += 1
|
||
embed_batch.append(result["doc_id"])
|
||
if len(embed_batch) >= batch_size:
|
||
_flush_embed()
|
||
except Exception as e:
|
||
logger.warning(f"run_parser: ошибка документа: {e}")
|
||
|
||
_flush_embed()
|
||
except Exception as e:
|
||
error_msg = str(e)[:500]
|
||
logger.error(f"run_parser source={source_id} ошибка: {e}", exc_info=True)
|
||
|
||
with db_session() as session:
|
||
src = session.get(ParseSource, source_id)
|
||
if src:
|
||
src.last_status = "error" if error_msg else "done"
|
||
src.last_error = error_msg
|
||
src.docs_added = (src.docs_added or 0) + added
|
||
src.last_run_at = datetime.now(UTC)
|
||
session.commit()
|
||
|
||
return {"status": "error" if error_msg else "done", "added": added, "error": error_msg}
|