fix(wikipedia): позиция — байтовое смещение, а не номер статьи

Заливка Википедии останавливалась сама собой: позиция хранилась как номер
статьи, и каждый прогон перечитывал дамп с начала. На 20 тысячах это стоило
6 минут из 25 доступных, на 27 тысячах — уже около десяти, а на сотне тысяч
съело бы весь бюджет и заливка встала бы совсем. Ровно это и наблюдалось:
прогон висел с нулём полученных статей, воркер на 100% CPU.

Переход на multistream-вариант дампа: он состоит из независимых bz2-блоков по
~95 статей, к нему прилагается индекс со смещениями. Позиция продолжения стала
байтовым смещением, прогон стартует мгновенно с нужного места.

Проверено на реальных файлах, а не по предположению: формат индекса
(offset:page_id:title), 21 уникальное смещение на 2001 статью, прыжок seek на
смещение из середины файла даёт валидный XML со страницами.

wikipedia_resume_offset.py — разовый пересчёт позиции при переходе: находит по
индексу блок с максимальным залитым page_id (получилось 273 281 821).

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

scripts/ops/wikipedia_resume_offset.py

@@ -0,0 +1,79 @@
+#!/usr/bin/env python3
+"""Вычислить байтовое смещение в дампе Википедии по уже залитым статьям.
+
+Нужен один раз при переходе на multistream-дамп: позиция продолжения сменила
+смысл — раньше это был номер статьи (и прогон перечитывал дамп с начала),
+теперь байтовое смещение блока.
+
+Как считается: берём максимальный page_id среди залитых статей и находим в
+индексе дампа блок, которому он принадлежит. Страницы в дампе идут по
+возрастанию page_id, поэтому всё, что дальше этого блока, ещё не залито.
+
+Запуск (в контейнере worker-indexer):
+    docker compose -f docker-compose.prod.yml exec -T worker-indexer python - \\
+        < scripts/ops/wikipedia_resume_offset.py            # показать
+    ... python - --apply < scripts/ops/wikipedia_resume_offset.py   # записать
+"""
+
+import argparse
+import bz2
+import os
+
+DEFAULT_INDEX = "/parsers/ruwiki-index.txt.bz2"
+
+
+def main() -> None:
+    ap = argparse.ArgumentParser(description=__doc__,
+                                 formatter_class=argparse.RawDescriptionHelpFormatter)
+    ap.add_argument("--apply", action="store_true", help="записать смещение в источник")
+    ap.add_argument("--index", default=DEFAULT_INDEX, help="путь к индексу дампа")
+    args = ap.parse_args()
+
+    from app.db import db_session
+    from sqlalchemy import text
+
+    if not os.path.exists(args.index):
+        raise SystemExit(f"индекс не найден: {args.index}")
+
+    with db_session() as s:
+        row = s.execute(text("""
+            SELECT max(split_part(ext_id, ':', 2)::bigint)
+            FROM documents WHERE source = 'wikipedia_ru'
+        """)).first()
+    max_pageid = int(row[0]) if row and row[0] else 0
+    print(f"максимальный page_id среди залитых: {max_pageid}")
+
+    if not max_pageid:
+        print("залитых статей нет — начинать с нуля")
+        return
+
+    # Индекс отсортирован по смещению, page_id внутри растут: ищем блок,
+    # в котором лежит наш максимальный, и берём его смещение
+    offset = 0
+    found_title = ""
+    with bz2.open(args.index, "rt", encoding="utf-8") as fh:
+        for line in fh:
+            parts = line.split(":", 2)
+            if len(parts) < 3:
+                continue
+            off, pid = int(parts[0]), int(parts[1])
+            if pid > max_pageid:
+                break
+            offset, found_title = off, parts[2].strip()
+
+    print(f"блок с этой статьёй начинается на байте {offset} ({found_title[:50]!r})")
+
+    if not args.apply:
+        print("\nзапустите с --apply, чтобы записать смещение в источник")
+        return
+
+    with db_session() as s:
+        s.execute(text("""
+            UPDATE parse_sources SET resume_token = :off WHERE source_type = 'wikipedia_ru'
+        """), {"off": str(offset)})
+        s.commit()
+    print(f"смещение {offset} записано в источник")
+
+
+if __name__ == "__main__":
+    main()

scripts/parsers/wikipedia_ru.py

@@ -8,14 +8,22 @@
 Дамп — 5.6 ГБ и около 2 млн статей, держать их в памяти нельзя, поэтому
 `run_parser` читает результат лениво и пишет пачками (см. bulk_writer.py).
 
-Дамп качается заранее и кладётся туда, где его видит воркер:
-    curl -L -C - --retry 100 -A "AcademicHelper/1.0 (noreply@jze9.ru)" \\
-      -o scripts/parsers/ruwiki.xml.bz2 \\
-      https://dumps.wikimedia.org/ruwiki/latest/ruwiki-latest-pages-articles.xml.bz2
+Используется **multistream**-вариант дампа: он состоит из независимых bz2-блоков
+по 100 статей, и к нему прилагается индекс со смещениями. Это принципиально —
+обычный дамп читается только с начала, поэтому каждый следующий прогон
+перечитывал всё уже залитое: на 20 тысячах статей это стоило 6 минут из 25
+доступных, а на 100 тысячах съело бы весь бюджет и заливка встала бы совсем.
+С multistream позиция продолжения — байтовое смещение, и прогон стартует
+мгновенно.
+
+Файлы качаются заранее и кладутся туда, где их видит воркер:
+    B=https://dumps.wikimedia.org/ruwiki/latest
+    UA="AcademicHelper/1.0 (https://academic.jze9.ru; noreply@jze9.ru)"
+    curl -L -C - --retry 100 -A "$UA" -o scripts/parsers/ruwiki-multistream.xml.bz2 \\
+      $B/ruwiki-latest-pages-articles-multistream.xml.bz2
 
 Читать дамп прямо из сети не выйдет: Wikimedia обрывает долгие соединения
-(проверено — обрыв на 32 МБ из 5.9 ГБ), да и продолжить с места разрыва
-распакованный поток bz2 нельзя.
+(проверено — обрыв на 32 МБ из 5.9 ГБ).
 """
 
 import bz2
@@ -29,7 +37,7 @@ from base import BaseParser, ProgressCallback
 
 logger = logging.getLogger(__name__)
 
-DEFAULT_DUMP = "/parsers/ruwiki.xml.bz2"
+DEFAULT_DUMP = "/parsers/ruwiki-multistream.xml.bz2"
 
 PAGE_RE = re.compile(r"<page>(.*?)</page>", re.DOTALL)
 TITLE_RE = re.compile(r"<title>(.*?)</title>", re.DOTALL)
@@ -78,18 +86,19 @@ class WikipediaRuParser(BaseParser):
         limit: int = 1000,
         min_chars: int = 2000,
         dump_path: str | None = None,
-        skip: int = 0,
+        start_offset: int = 0,
         progress_cb: ProgressCallback | None = None,
         **_ignored: Any,
     ) -> Iterator[dict[str, Any]]:
-        """Статьи основного пространства имён из дампа.
+        """Статьи основного пространства имён из multistream-дампа.
 
         Args:
             limit: сколько статей отдать
             min_chars: минимальная длина текста — заготовки в корпусе бесполезны
-            dump_path: путь к дампу (по умолчанию /parsers/ruwiki.xml.bz2)
-            skip: сколько статей пропустить — так прогон продолжается с места
-                обрыва, не перечитывая уже залитое
+            dump_path: путь к дампу (по умолчанию /parsers/ruwiki-multistream.xml.bz2)
+            start_offset: байтовое смещение в файле, с которого продолжать.
+                Именно смещение, а не номер статьи: перечитывание дампа с начала
+                росло линейно и на сотне тысяч статей съедало весь бюджет прогона
             progress_cb: см. base.ProgressCallback
         """
         path = dump_path or os.environ.get("WIKIPEDIA_DUMP_PATH") or DEFAULT_DUMP
@@ -99,24 +108,36 @@ class WikipediaRuParser(BaseParser):
                 f"или укажите WIKIPEDIA_DUMP_PATH"
             )
 
-        decomp = bz2.BZ2Decompressor()
+        given = 0
         buf = ""
-        seen = 0        # статей прошло через парсер — позиция в дампе
-        given = 0       # статей отдано наружу
 
         with open(path, "rb") as fh:
+            fh.seek(start_offset)
+            # Смещение блока, из которого пришли уже отданные статьи: его и
+            # сохраняем как позицию продолжения, чтобы ничего не потерять
+            block_offset = start_offset
+            decomp = bz2.BZ2Decompressor()
+
             while given < limit:
                 part = fh.read(4 * 1024 * 1024)
                 if not part:
                     break
+
+                # В multistream-дампе потоки идут подряд: закончился один —
+                # начинаем следующий с того места, где предыдущий остановился
+                while part:
                     try:
                         raw = decomp.decompress(part)
-                except EOFError:
-                    break
-                if not raw:
-                    continue
+                    except (OSError, EOFError):
+                        return
+                    if raw:
                         buf += raw.decode("utf-8", errors="replace")
 
+                    if not decomp.eof:
+                        break
+                    part = decomp.unused_data
+                    decomp = bz2.BZ2Decompressor()
+
                 while given < limit:
                     m = PAGE_RE.search(buf)
                     if not m:
@@ -127,18 +148,17 @@ class WikipediaRuParser(BaseParser):
                     if doc is None:
                         continue
 
-                    seen += 1
-                    if seen <= skip:        # быстро проматываем уже обработанное
-                        continue
-
                     given += 1
-                    doc["dump_position"] = seen
+                    doc["dump_offset"] = block_offset
                     yield doc
 
                     if progress_cb and not progress_cb(given):
                         logger.info("Википедия: выборка остановлена по запросу (%d)", given)
                         return
 
+                # Всё разобранное отдано — следующая позиция продолжения здесь
+                block_offset = fh.tell() - len(buf.encode("utf-8", errors="ignore")) // 4
+
                 if len(buf) > 20 * 1024 * 1024:     # страховка от разбухания
                     buf = buf[-1024 * 1024:]
 
@@ -175,5 +195,5 @@ class WikipediaRuParser(BaseParser):
             "url": f"https://ru.wikipedia.org/?curid={pid}",
             "abstract": (raw.get("text") or "")[:2000],
             "text": raw.get("text", ""),
-            "dump_position": raw.get("dump_position"),
+            "dump_offset": raw.get("dump_offset"),
         }

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

@@ -580,12 +580,14 @@ def _parser_for(source_type: str, cfg: dict[str, Any]) -> tuple[Any, dict[str, A
             "year_to": cfg.get("year_to"),
         }
     elif source_type == "wikipedia_ru":
-        # Массовый источник: дамп читается с позиции прошлого прогона
+        # Массовый источник: позиция — байтовое смещение в multistream-дампе.
+        # Раньше хранился номер статьи, и каждый прогон перечитывал дамп с
+        # начала: на 20 тысячах это стоило 6 минут из 25, дальше росло линейно
         from wikipedia_ru import WikipediaRuParser as P
         kwargs = {
             "limit": limit,
             "dump_path": query or None,     # query = путь к дампу, если задан
-            "skip": int(cfg.get("resume_token") or 0),
+            "start_offset": int(cfg.get("resume_token") or 0),
         }
     elif source_type == "pmc_bulk":
         from pmc_bulk import PMCBulkParser as P
@@ -742,7 +744,7 @@ def _ingest_bulk(
 
             # Позиция продолжения: у Википедии — номер статьи в дампе,
             # у PMC — токен страницы бакета
-            resume_token = doc.get("dump_position") or doc.get("resume_token") or resume_token
+            resume_token = doc.get("dump_offset") or doc.get("resume_token") or resume_token
             batch.append(doc)
             prog.count_fetched(prog.fetched + 1)