fix(ops): читать дамп Википедии с диска — сетевой поток рвётся на 5.6 ГБ

Wikimedia закрывает долгие потоковые соединения: обрыв пришёлся на 32 МБ из
5.9 ГБ. Скачать файл с докачкой (curl -C -) и читать локально надёжнее, поэтому
у скрипта появился --dump-file; чтение из сети осталось запасным путём.

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

scripts/ops/bulk_ingest_wikipedia_ru.py

@@ -63,47 +63,75 @@ def clean_wikitext(raw: str) -> str:
     return text.strip()
 
 
-def iter_pages(min_chars: int):
+def _pages_from_chunks(chunks, min_chars: int):
-    """Потоково читать дамп и отдавать статьи основного пространства имён."""
+    """Разобрать поток распакованного XML на статьи основного пространства имён."""
+    buf = ""
+    for raw in chunks:
+        buf += raw.decode("utf-8", errors="replace")
+
+        while True:
+            m = PAGE_RE.search(buf)
+            if not m:
+                break
+            page, buf = m.group(1), buf[m.end():]
+
+            if REDIRECT_RE.search(page):
+                continue
+            ns = NS_RE.search(page)
+            if not ns or ns.group(1) != "0":          # только статьи
+                continue
+            tm, im, xm = TITLE_RE.search(page), ID_RE.search(page), TEXT_RE.search(page)
+            if not (tm and im and xm):
+                continue
+            text = clean_wikitext(xm.group(1))
+            if len(text) < min_chars:
+                continue
+            yield im.group(1), tm.group(1), text
+
+        if len(buf) > 20 * 1024 * 1024:               # страховка от разбухания
+            buf = buf[-1024 * 1024:]
+
+
+def iter_pages(min_chars: int, dump_file: str | None):
+    """Статьи из локального дампа либо, если файла нет, прямо из сети.
+
+    Локальный файл предпочтителен: Wikimedia рвёт долгие потоковые соединения
+    (проверено — обрыв на 32 МБ из 5.9 ГБ), а скачать файл можно с докачкой
+    (`curl -C -`) и потом читать сколько угодно.
+    """
+    decomp = bz2.BZ2Decompressor()
+
+    if dump_file:
+        def chunks():
+            with open(dump_file, "rb") as fh:
+                while True:
+                    part = fh.read(4 * 1024 * 1024)
+                    if not part:
+                        return
+                    try:
+                        yield decomp.decompress(part)
+                    except EOFError:
+                        return
+        yield from _pages_from_chunks(chunks(), min_chars)
+        return
+
     import httpx
 
-    decomp = bz2.BZ2Decompressor()
-    buf = ""
     # Wikimedia отдаёт 403 без осмысленного User-Agent — по их правилам он должен
     # называть приложение и давать контакт
     headers = {"User-Agent": "AcademicHelper/1.0 (https://academic.jze9.ru; noreply@jze9.ru)"}
-    with httpx.stream("GET", DUMP_URL, timeout=120, follow_redirects=True, headers=headers) as resp:
-        resp.raise_for_status()
-        for chunk in resp.iter_bytes(4 * 1024 * 1024):
-            try:
-                raw = decomp.decompress(chunk)
-            except EOFError:
-                break
-            if not raw:
-                continue
-            buf += raw.decode("utf-8", errors="replace")
 
-            while True:
+    def net_chunks():
-                m = PAGE_RE.search(buf)
+        with httpx.stream("GET", DUMP_URL, timeout=120, follow_redirects=True,
-                if not m:
+                          headers=headers) as resp:
-                    break
+            resp.raise_for_status()
-                page, buf = m.group(1), buf[m.end():]
+            for chunk in resp.iter_bytes(4 * 1024 * 1024):
+                try:
+                    yield decomp.decompress(chunk)
+                except EOFError:
+                    return
 
-                if REDIRECT_RE.search(page):
+    yield from _pages_from_chunks(net_chunks(), min_chars)
-                    continue
-                ns = NS_RE.search(page)
-                if not ns or ns.group(1) != "0":      # только статьи
-                    continue
-                tm, im, xm = TITLE_RE.search(page), ID_RE.search(page), TEXT_RE.search(page)
-                if not (tm and im and xm):
-                    continue
-                text = clean_wikitext(xm.group(1))
-                if len(text) < min_chars:
-                    continue
-                yield im.group(1), tm.group(1), text
-
-            if len(buf) > 20 * 1024 * 1024:            # страховка от разбухания
-                buf = buf[-1024 * 1024:]
 
 
 def main() -> None:
@@ -113,6 +141,8 @@ def main() -> None:
     ap.add_argument("--batch", type=int, default=500, help="статей в транзакции")
     ap.add_argument("--min-chars", type=int, default=2000, help="минимальная длина текста")
     ap.add_argument("--fp-per-doc", type=int, default=2000, help="максимум отпечатков на статью")
+    ap.add_argument("--dump-file", default=None,
+                    help="путь к заранее скачанному дампу (надёжнее, чем читать из сети)")
     args = ap.parse_args()
 
     import psycopg2
@@ -165,7 +195,7 @@ def main() -> None:
             added += fresh
         conn.commit()
 
-    for pid, title, text in iter_pages(args.min_chars):
+    for pid, title, text in iter_pages(args.min_chars, args.dump_file):
         batch.append((pid, title, text))
         if len(batch) >= args.batch:
             flush(batch)