refactor(ops): массовая заливка — в общий конвейер вместо отдельных скриптов

Заливку Википедии и PMC я сделал отдельными скриптами мимо существующей
инфраструктуры: запуск руками через ssh, состояние в файле в /tmp, никакой
видимости. Результат предсказуем — за неделю обе умерли молча (обрыв базы на
20 008 статьях из 2 млн и таймаут сети на 85 тыс. из 100 тыс.), прогресс
потерялся, а узнали мы об этом через неделю. При том что рядом лежит готовый
механизм: parse_sources, прогоны со шкалой, журнал, кнопки, ретраи Celery.

Теперь это обычные типы источника — wikipedia_ru и pmc_bulk:

- заводятся и запускаются из админки, как OpenAlex или КиберЛенинка;
- показывают ту же шкалу, журнал и кнопку остановки;
- падение воркера больше не теряет прогресс: позиция продолжения хранится в
  parse_sources.resume_token (номер статьи в дампе / токен страницы бакета),
  повторный запуск берёт следующую порцию;
- укладываются в бюджет времени таска — заливка идёт порциями, а не одним
  многосуточным процессом.

Чего не хватало конвейеру для миллионов и что добавлено:
- парсеры отдают генератор, а не список: 2 млн статей в память не влезают;
- app/bulk_writer.py — запись пачками через COPY (21 тыс. строк/с против
  6.7 тыс. построчно) с переподключением к базе при обрыве;
- эмбеддинги при массовой заливке не диспатчатся: они на порядок медленнее и
  стали бы узким местом, вектора досчитываются отдельно (reembed_missing.py).

scripts/ops/bulk_ingest_*.py удалены — их работу делает конвейер.

Проверено на проде: оба парсера отдают документы, прогон через run_parser
завершается штатно, позиция продолжения сдвигается (300 → 600), повторный
запуск продолжает с неё.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

services/api/alembic/versions/007_parse_source_resume.py

@@ -0,0 +1,29 @@
+"""Позиция продолжения массовой заливки в источнике.
+
+Revision ID: 007
+Revises: 006
+Create Date: 2026-09-05
+
+Массовые источники (дамп Википедии, бакет PMC) заливаются частями: за один
+прогон берётся столько статей, сколько влезает в бюджет времени таска. Чтобы
+следующий прогон продолжал с того же места, а не перечитывал залитое заново,
+позиция сохраняется прямо в источнике: для Википедии это номер статьи в дампе,
+для PMC — токен страницы бакета.
+"""
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision = "007"
+down_revision = "006"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("parse_sources", sa.Column("resume_token", sa.Text(), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("parse_sources", "resume_token")

services/api/app/api/admin.py

@@ -50,8 +50,10 @@ logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/admin", tags=["admin"], dependencies=[Depends(get_admin_user)])
 
-# Типы источников, для которых в worker-indexer есть парсер (index.run_parser)
-SOURCE_TYPES = ("openalex", "cyberleninka", "arxiv", "pmc")
+# Типы источников, для которых в worker-indexer есть парсер (index.run_parser).
+# wikipedia_ru и pmc_bulk — массовые: читают дамп/бакет потоком и пишут пачками,
+# продолжая с сохранённой позиции (parse_sources.resume_token)
+SOURCE_TYPES = ("openalex", "cyberleninka", "arxiv", "pmc", "wikipedia_ru", "pmc_bulk")
 # Прогон в этих статусах ещё может двигаться — повторно запускать источник нельзя
 RUN_ACTIVE_STATUSES = ("queued", "running")
 

services/api/app/models/admin.py

@@ -35,6 +35,9 @@ class ParseSource(Base):
     docs_added: Mapped[int] = mapped_column(default=0)
     # Последний (пока идёт — текущий) прогон, см. ParseRun
     last_run_id: Mapped[int | None] = mapped_column(nullable=True)
+    # Позиция продолжения для массовых источников: номер статьи в дампе
+    # Википедии или токен страницы бакета PMC
+    resume_token: Mapped[str | None] = mapped_column(Text, nullable=True)
     created_at: Mapped[datetime] = mapped_column(server_default=func.now())
 
     def __repr__(self) -> str:

services/api/app/schemas/admin.py

@@ -120,7 +120,9 @@ class ParseRunDetail(ParseRunResponse):
 
 # ─── Источники парсинга ───────────────────────────────────────────────────────
 class ParseSourceCreate(BaseModel):
-    source_type: str = Field(description="openalex / cyberleninka / arxiv / pmc")
+    source_type: str = Field(
+        description="openalex / cyberleninka / arxiv / pmc / wikipedia_ru / pmc_bulk"
+    )
     name: str = Field(min_length=1, max_length=255)
     query: str | None = None
     lang: str | None = None
@@ -164,7 +166,9 @@ class ParseSourceResponse(BaseModel):
 class ParseSourceBulkCreate(BaseModel):
     """Пакетное добавление: один тип/лимит, много запросов (по строке на тему)."""
 
-    source_type: str = Field(description="openalex / cyberleninka / arxiv / pmc")
+    source_type: str = Field(
+        description="openalex / cyberleninka / arxiv / pmc / wikipedia_ru / pmc_bulk"
+    )
     queries: list[str] = Field(min_length=1, max_length=500)
     name_prefix: str = ""
     lang: str | None = None

services/frontend/src/pages/admin/Sources.tsx

@@ -32,7 +32,11 @@ const SOURCE_TYPES = [
   { value: 'openalex', label: 'OpenAlex' },
   { value: 'cyberleninka', label: 'КиберЛенинка' },
   { value: 'arxiv', label: 'arXiv' },
-  { value: 'pmc', label: 'PubMed Central' },
+  { value: 'pmc', label: 'PubMed Central (по теме)' },
+  // Массовые: читают дамп/бакет потоком и продолжают с места остановки,
+  // поэтому запускаются повторно до исчерпания источника
+  { value: 'wikipedia_ru', label: 'Википедия ru (дамп, массово)' },
+  { value: 'pmc_bulk', label: 'PubMed Central (бакет, массово)' },
 ];
 
 const STATUS_COLORS: Record<string, string> = {

services/worker-indexer/app/bulk_writer.py

@@ -0,0 +1,132 @@
+"""Пакетная запись документов корпуса — путь для массовых источников.
+
+Обычный `index.add_document` пишет по одному документу через ORM: удобно для
+парсеров, отдающих сотни статей, но на миллионах это тупик. Замер на проде:
+построчная вставка отпечатков — 6.7 тыс. строк/с, COPY — 21 тыс. строк/с, а
+миллион статей это ~2 млрд отпечатков. Разница между сутками и неделями.
+
+Поэтому массовые источники (дампы Википедии, бакет PMC) идут сюда: документы
+вставляются одной командой с ON CONFLICT, отпечатки — через COPY.
+
+Elasticsearch и эмбеддинги здесь намеренно не трогаются: L1 и L2 начинают
+работать сразу, а векторы для L3 досчитываются отдельно
+(`scripts/ops/reembed_missing.py`) — иначе заливка упирается в скорость
+сервера эмбеддингов и тормозит на порядок.
+"""
+
+import contextlib
+import io
+import logging
+import time
+from typing import Any
+
+import psycopg2
+
+from app.algorithms.winnowing import sample_evenly, winnow_ordered
+from app.config import settings
+
+logger = logging.getLogger(__name__)
+
+# Сколько попыток пережить обрыв соединения с базой. Массовая заливка идёт
+# часами, и разрыв сети за это время — норма, а не исключение.
+DB_RETRIES = 5
+
+
+def connect():
+    """Отдельное подключение psycopg2: COPY недоступен через ORM-сессию."""
+    return psycopg2.connect(
+        host=settings.POSTGRES_HOST,
+        port=settings.POSTGRES_PORT,
+        dbname=settings.POSTGRES_DB,
+        user=settings.POSTGRES_USER,
+        password=settings.POSTGRES_PASSWORD,
+    )
+
+
+def write_batch(conn, docs: list[dict[str, Any]], fp_limit: int) -> tuple[Any, dict[str, int]]:
+    """Записать пачку документов с отпечатками, пережив обрыв соединения.
+
+    Args:
+        conn: активное подключение psycopg2 (может быть заменено при обрыве)
+        docs: документы в унифицированном формате парсеров; нужны ext_id,
+            source, title и text — остальное необязательно
+        fp_limit: максимум отпечатков на документ
+
+    Returns:
+        (соединение, счётчики) — соединение может оказаться новым, если
+        пришлось переподключаться; счётчики: added / duplicates / fingerprints
+    """
+    if not docs:
+        return conn, {"added": 0, "duplicates": 0, "fingerprints": 0}
+
+    for attempt in range(1, DB_RETRIES + 1):
+        try:
+            return conn, _write_once(conn, docs, fp_limit)
+        except psycopg2.OperationalError as e:
+            wait = min(60, 5 * attempt)
+            logger.warning(
+                "bulk_writer: база недоступна (%s), повтор через %sс [%s/%s]",
+                str(e).strip()[:80], wait, attempt, DB_RETRIES,
+            )
+            time.sleep(wait)
+            with contextlib.suppress(Exception):
+                conn.close()
+            try:
+                conn = connect()
+            except Exception:
+                continue
+
+    raise RuntimeError(f"не удалось записать пачку после {DB_RETRIES} попыток")
+
+
+def _write_once(conn, docs: list[dict[str, Any]], fp_limit: int) -> dict[str, int]:
+    """Одна попытка записи; счётчики возвращаются только при успехе."""
+    added = duplicates = fingerprints = 0
+
+    with conn.cursor() as cur:
+        cur.executemany(
+            """INSERT INTO documents (ext_id, source, title, doi, year, lang, journal,
+                                      abstract, url, authors, indexed_at)
+               VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,'[]'::json, now())
+               ON CONFLICT (ext_id) DO NOTHING""",
+            [
+                (
+                    d["ext_id"], d["source"], (d.get("title") or "")[:1000], d.get("doi"),
+                    d.get("year"), d.get("lang"), (d.get("journal") or None),
+                    (d.get("text") or "")[:2000], d.get("url"),
+                )
+                for d in docs
+            ],
+        )
+
+        cur.execute(
+            "SELECT id, ext_id FROM documents WHERE ext_id = ANY(%s)",
+            ([d["ext_id"] for d in docs],),
+        )
+        id_by_ext = {ext: doc_id for doc_id, ext in cur.fetchall()}
+
+        buf = io.StringIO()
+        for d in docs:
+            doc_id = id_by_ext.get(d["ext_id"])
+            if doc_id is None:
+                continue
+            # Уже с отпечатками — значит документ залит прошлым прогоном
+            cur.execute("SELECT 1 FROM fingerprints WHERE doc_id = %s LIMIT 1", (doc_id,))
+            if cur.fetchone():
+                duplicates += 1
+                continue
+
+            hashes = sample_evenly(winnow_ordered(d.get("text") or ""), fp_limit)
+            if not hashes:
+                continue
+            for pos, h in enumerate(hashes):
+                buf.write(f"{doc_id}\t{h}\t{pos}\n")
+            fingerprints += len(hashes)
+            added += 1
+
+        buf.seek(0)
+        if added:
+            cur.copy_from(buf, "fingerprints", columns=("doc_id", "hash_value", "position"))
+
+    conn.commit()
+    return {"added": added, "duplicates": duplicates, "fingerprints": fingerprints}

services/worker-indexer/app/config.py

@@ -59,6 +59,12 @@ class Settings(BaseSettings):
     FULL_TEXT_MIN_CHARS: int = 500  # Минимум символов, иначе считаем извлечение неудачным
     EMBED_BATCH_SIZE: int = 64  # Размер пачки документов для диспатча эмбеддингов
 
+    # Сколько документов массового источника пишется одной транзакцией.
+    # Больше — меньше накладных расходов, но длиннее транзакция: 500 статей по
+    # 2000 отпечатков это миллион строк за раз, и на таком объёме соединение
+    # обрывалось. 150 — компромисс, проверенный на заливке Википедии.
+    BULK_WRITE_BATCH: int = 150
+
     # Бюджет времени одного прогона index.run_parser. RabbitMQ рвёт канал
     # consumer'а, не сделавшего ack за consumer_timeout (по умолчанию 1800с):
     # воркер падает, сообщение передоставляется, таск начинается заново —

services/worker-indexer/app/models/__init__.py

@@ -78,6 +78,7 @@ class ParseSource(Base):
     last_run_at: Mapped[datetime | None] = mapped_column(nullable=True)
     docs_added: Mapped[int] = mapped_column(default=0)
     last_run_id: Mapped[int | None] = mapped_column(nullable=True)
+    resume_token: Mapped[str | None] = mapped_column(Text, nullable=True)
     created_at: Mapped[datetime] = mapped_column(server_default=func.now())
 
 

services/worker-indexer/app/tasks/index.py

@@ -1,5 +1,6 @@
 """Celery задачи индексации документов и проверки плагиата (уровни 1-2)."""
 
+import contextlib
 import io
 from datetime import UTC, datetime
 from pathlib import Path
@@ -560,12 +561,180 @@ def _parser_for(source_type: str, cfg: dict[str, Any]) -> tuple[Any, dict[str, A
             "year_from": cfg.get("year_from"),
             "year_to": cfg.get("year_to"),
         }
+    elif source_type == "wikipedia_ru":
+        # Массовый источник: дамп читается с позиции прошлого прогона
+        from wikipedia_ru import WikipediaRuParser as P
+        kwargs = {
+            "limit": limit,
+            "dump_path": query or None,     # query = путь к дампу, если задан
+            "skip": int(cfg.get("resume_token") or 0),
+        }
+    elif source_type == "pmc_bulk":
+        from pmc_bulk import PMCBulkParser as P
+        kwargs = {"limit": limit, "resume_token": cfg.get("resume_token")}
     else:
         raise ValueError(f"неизвестный тип источника: {source_type}")
 
     return P(), kwargs
 
 
+def _ingest_one_by_one(parser: Any, raw_docs: Any, prog: Any, run_id: int) -> tuple[bool, bool]:
+    """Обычный путь: документ за документом через add_document.
+
+    Подходит источникам, отдающим сотни статей: работает ORM-логика с
+    дедупликацией, индексацией в Elasticsearch и обогащением полным текстом.
+
+    Returns:
+        (отменён, исчерпан бюджет времени)
+    """
+    cancelled = exhausted = False
+
+    prog.stage = "index"
+    prog.count_fetched(len(raw_docs))
+    prog.log("info", f"получено {len(raw_docs)} документов, индексация")
+    _write_run(run_id, **prog.snapshot())
+
+    # Эмбеддинги диспатчим пачками, а не по одному документу: так 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:
+        if not cancelled and prog.over_budget:
+            exhausted = True
+            prog.log("warning", f"бюджет времени {prog.budget_s:.0f}с исчерпан на индексации")
+        if cancelled or exhausted:
+            break
+        try:
+            doc = parser.transform(raw)
+            if not (doc and doc.get("title") and doc.get("ext_id")):
+                prog.count_result("skipped")
+                continue
+            result = add_document(doc, dispatch_embed=False)
+            prog.count_result(result.get("status", "error"))
+            if result.get("status") == "indexed":
+                embed_batch.append(result["doc_id"])
+                if len(embed_batch) >= batch_size:
+                    _flush_embed()
+        except Exception as e:
+            prog.count_result("error")
+            prog.log("warning", f"документ пропущен: {e}")
+            logger.warning(f"run_parser: ошибка документа: {e}")
+
+        if prog.should_flush():
+            cancelled = _write_run(run_id, **prog.snapshot())
+            prog.mark_flushed()
+            if cancelled:
+                prog.log("warning", "отмена по запросу из админки")
+
+    _flush_embed()
+    return cancelled, exhausted
+
+
+def _ingest_bulk(
+    parser: Any, raw_docs: Any, prog: Any, run_id: int, source_id: int
+) -> tuple[bool, bool]:
+    """Массовый путь: статьи приходят потоком и пишутся пачками через COPY.
+
+    Отличия от обычного пути и почему они нужны:
+    - парсер отдаёт генератор, поэтому счётчик «получено» растёт по ходу дела,
+      а не известен заранее;
+    - запись идёт пакетно (bulk_writer), иначе миллионы отпечатков занимают
+      недели вместо суток;
+    - позиция продолжения сохраняется в источнике, чтобы следующий прогон
+      начинал с места остановки, а не перечитывал дамп с начала.
+
+    Эмбеддинги здесь не диспатчатся: они считаются заметно медленнее заливки и
+    делали бы её узким местом. Векторы досчитываются отдельно —
+    `scripts/ops/reembed_missing.py`.
+
+    Returns:
+        (отменён, исчерпан бюджет времени)
+    """
+    from app.bulk_writer import connect, write_batch
+    from app.models import ParseSource
+
+    cancelled = exhausted = False
+    prog.stage = "index"
+    prog.log("info", "массовый источник: запись пачками")
+    _write_run(run_id, **prog.snapshot())
+
+    conn = connect()
+    batch: list[dict[str, Any]] = []
+    resume_token: str | None = None
+    batch_size = settings.BULK_WRITE_BATCH
+
+    def save_resume() -> None:
+        """Запомнить позицию в источнике — с неё продолжит следующий прогон."""
+        if resume_token is None:
+            return
+        with db_session() as session:
+            src = session.get(ParseSource, source_id)
+            if src:
+                src.resume_token = str(resume_token)
+                session.commit()
+
+    def flush() -> bool:
+        """Записать накопленное; True — попросили остановиться."""
+        nonlocal conn, batch
+        if not batch:
+            return False
+        conn, counts = write_batch(conn, batch, settings.MAX_FINGERPRINTS_PER_DOC)
+        for _ in range(counts["added"]):
+            prog.count_result("indexed")
+        for _ in range(counts["duplicates"]):
+            prog.count_result("duplicate")
+        batch = []
+        save_resume()
+        stop = _write_run(run_id, **prog.snapshot())
+        prog.mark_flushed()
+        return stop
+
+    try:
+        for raw in raw_docs:
+            if prog.over_budget:
+                exhausted = True
+                prog.log("warning", f"бюджет времени {prog.budget_s:.0f}с исчерпан")
+                break
+            try:
+                doc = parser.transform(raw)
+            except Exception as e:
+                prog.count_result("error")
+                logger.warning(f"run_parser bulk: ошибка документа: {e}")
+                continue
+
+            if not (doc and doc.get("ext_id") and doc.get("text")):
+                prog.count_result("skipped")
+                continue
+
+            # Позиция продолжения: у Википедии — номер статьи в дампе,
+            # у PMC — токен страницы бакета
+            resume_token = doc.get("dump_position") or doc.get("resume_token") or resume_token
+            batch.append(doc)
+            prog.count_fetched(prog.fetched + 1)
+
+            if len(batch) >= batch_size and flush():
+                cancelled = True
+                prog.log("warning", "отмена по запросу из админки")
+                break
+
+        if not cancelled and flush():
+            cancelled = True
+    finally:
+        with contextlib.suppress(Exception):
+            conn.close()
+
+    return cancelled, exhausted
+
+
 def _write_run(run_id: int, **fields: Any) -> bool:
     """Записать прогресс прогона и вернуть True, если запрошена отмена.
 
@@ -622,6 +791,8 @@ def run_parser(self, source_id: int, run_id: int | None = None) -> dict[str, Any
             "year_from": src.year_from,
             "year_to": src.year_to,
             "limit": src.limit,
+            # Массовые источники продолжают с сохранённой позиции
+            "resume_token": src.resume_token,
         }
         src.last_status = "running"
         src.last_error = None
@@ -676,53 +847,13 @@ def run_parser(self, source_id: int, run_id: int | None = None) -> dict[str, Any
         # fetch+transform без записи в JSONL — работаем in-memory
         raw_docs = parser.fetch(**fetch_kwargs, progress_cb=on_fetch_progress)
 
-        prog.stage = "index"
-        prog.count_fetched(len(raw_docs))
-        prog.log("info", f"получено {len(raw_docs)} документов, индексация")
-        _write_run(run_id, **prog.snapshot())
-
-        # Эмбеддинги диспатчим пачками, а не по одному документу: так 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:
-            if not cancelled and prog.over_budget:
-                exhausted = True
-                prog.log("warning", f"бюджет времени {prog.budget_s:.0f}с исчерпан на индексации")
-            if cancelled or exhausted:
-                break
-            try:
-                doc = parser.transform(raw)
-                if not (doc and doc.get("title") and doc.get("ext_id")):
-                    prog.count_result("skipped")
-                    continue
-                result = add_document(doc, dispatch_embed=False)
-                prog.count_result(result.get("status", "error"))
-                if result.get("status") == "indexed":
-                    embed_batch.append(result["doc_id"])
-                    if len(embed_batch) >= batch_size:
-                        _flush_embed()
-            except Exception as e:
-                prog.count_result("error")
-                prog.log("warning", f"документ пропущен: {e}")
-                logger.warning(f"run_parser: ошибка документа: {e}")
-
-            if prog.should_flush():
-                cancelled = _write_run(run_id, **prog.snapshot())
-                prog.mark_flushed()
-                if cancelled:
-                    prog.log("warning", "отмена по запросу из админки")
-
-        _flush_embed()
+        if getattr(parser, "bulk", False):
+            # Массовые источники (дамп Википедии, бакет PMC) отдают генератор:
+            # миллионы статей нельзя ни держать в памяти, ни писать по одной
+            # через ORM — для них отдельный путь с пакетной записью
+            cancelled, exhausted = _ingest_bulk(parser, raw_docs, prog, run_id, source_id)
+        else:
+            cancelled, exhausted = _ingest_one_by_one(parser, raw_docs, prog, run_id)
     except Exception as e:
         error_msg = str(e)[:500]
         prog.log("error", f"прогон упал: {error_msg}")