feat: add resumable local OpenSearch loading

Reviewed-on: #9

README.md

@@ -10,15 +10,15 @@ Telegram-бот — только часть рабочего окружения
 ## Текущее состояние
 
 Сейчас реализованы Telegram-бот-секретарь версии `0.2.2` и backend версии
-`0.3.0`: возобновляемая выгрузка, нормализация и подготовка индекса документов
+`0.4.1`: возобновляемая загрузка индекса документов Министерства юстиции
 ЦБД Минюста КР.
 
 | Компонент | Версия | Состояние |
 |---|---:|---|
 | Telegram-бот | `0.2.2` | на Synology работает `0.2.1`; обновление после слияния |
-| Backend | `0.3.0` | реализованы выгрузка, нормализация и экспорт для OpenSearch |
+| Backend | `0.4.1` | добавлено продолжение прерванной Bulk-загрузки |
 | Frontend | — | ещё не создан |
-| Сбор и обработка правовых данных | `0.3.0` | добавлены mapping и Bulk NDJSON для OpenSearch |
+| Сбор и обработка правовых данных | `0.4.1` | добавлено продолжение загрузки существующего индекса |
 | RAG и база знаний | — | ещё не созданы |
 
 ## Структура репозитория
@@ -59,4 +59,4 @@ python3 -m unittest discover -s tools/telegram-bot -v
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

backend/README.md

@@ -1,6 +1,6 @@
 # Backend Акылдаш
 
-Версия: `0.3.0`
+Версия: `0.4.1`
 
 Первая backend-область проекта — загрузка правовых документов из официального
 Open Data API ЦБД Минюста Кыргызской Республики. Код расположен в
@@ -114,10 +114,57 @@ python3 backend/search/minjust_opensearch.py
 
 По умолчанию создаётся `data/opensearch/minjust-fragments.ndjson`. Экспорт
 атомарный и детерминированный; для проверки можно передать `--limit 1`.
-Полный файл может быть большим, поэтому перед запуском нужно проверить
+Для прямой загрузки без большого промежуточного файла используется `--url`:
-свободное место. Загружать данные следует в новый версионный индекс и после
+
-проверки переключать alias, чтобы удалённые фрагменты не оставались в поиске.
+```bash
+python3 backend/search/minjust_opensearch.py \
+  --url http://127.0.0.1:9200 \
+  --index akyldash-fragments-dev-v1 \
+  --limit 1
+```
+
+Запросы Bulk API ограничены 25 МБ и не разрывают пару action/source. Для
+полного прохода убрать `--limit` и выбрать новое имя версионного индекса.
+После проверки production-индекса следует переключать alias, чтобы удалённые
+фрагменты не оставались в поиске.
+
+После каждого принятого Bulk-пакета загрузчик атомарно сохраняет checkpoint и
+печатает код безопасного возобновления. При временных HTTP 429/5xx, timeout и
+обрыве соединения запрос повторяется автоматически. Прерванную загрузку можно
+продолжить без ручного выбора документа:
+
+```bash
+python3 backend/search/minjust_opensearch.py \
+  --url http://127.0.0.1:9200 \
+  --index akyldash-fragments-v1 \
+  --resume
+```
+
+По умолчанию checkpoint хранится в
+`data/opensearch/<index>.checkpoint.json`; путь можно изменить через
+`--checkpoint`. Checkpoint привязан к URL, cluster UUID, index UUID, `--limit`
+и SHA-256 нормализованного manifest. Resume отклоняется при любом несовпадении:
+для обновлённого корпуса или пересозданного индекса нужно создать новый
+версионный индекс, проверить его и переключить alias. Это не оставляет
+удалённые trailing-фрагменты старых документов.
+
+## Локальный OpenSearch
+
+Стенд использует один узел OpenSearch без Dashboards, устанавливает
+`analysis-icu`, выделяет JVM 8 ГБ и доступен только на `127.0.0.1:9200`.
+Индекс хранится в `data/opensearch-node` на диске проекта.
+
+```bash
+sudo sysctl -w vm.max_map_count=262144
+docker compose -f deploy/local-opensearch/compose.yaml up -d --build
+curl http://127.0.0.1:9200/_cluster/health
+```
+
+Security plugin отключён только для локальной разработки; этот compose нельзя
+публиковать в сеть или использовать в production. Mapping локального стенда
+также задаёт одну shard и ноль replicas; для production число shard следует
+рассчитать по размеру корпуса и настроить не менее одной replica.
 
 ---
 
-Акылдаш · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Backend v0.4.1 · Frontend — не создан

backend/ingestion/minjust_cbd.py

@@ -21,7 +21,7 @@ from datetime import datetime, timezone
 from pathlib import Path
 from typing import Callable, Iterable
 
-APP_VERSION = "0.3.0"
+APP_VERSION = "0.4.1"
 API_BASE_URL = "https://cbd.minjust.gov.kg/api/v1/OpenData/"
 LANGUAGES = {"Rus": "ru", "Kyr": "ky"}
 IMAGE_LANGUAGES = {"Russian": "ru", "Kyrgyz": "ky"}

backend/normalization/minjust_cbd.py

@@ -23,7 +23,7 @@ from pathlib import Path
 from typing import Callable
 from urllib.parse import urlsplit
 
-APP_VERSION = "0.3.0"
+APP_VERSION = "0.4.1"
 SCHEMA_VERSION = "1"
 NORMALIZER_VERSION = "1.0.0"
 LANGUAGES = ("ru", "ky")

backend/search/minjust-fragments-index.json

@@ -1,4 +1,11 @@
 {
+  "settings": {
+    "index": {
+      "number_of_shards": 1,
+      "number_of_replicas": 0,
+      "refresh_interval": "30s"
+    }
+  },
   "mappings": {
     "dynamic": "strict",
     "properties": {

backend/search/minjust_opensearch.py

@@ -8,18 +8,30 @@ import json
 import os
 import sqlite3
 import tempfile
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
 from pathlib import Path
+from typing import Iterator
 
-APP_VERSION = "0.3.0"
+APP_VERSION = "0.4.1"
 LANGUAGES = {"ru", "ky"}
+DEFAULT_MAPPING = Path(__file__).with_name("minjust-fragments-index.json")
 
 
 def read_json(path: Path):
     def reject_constant(value: str):
         raise ValueError(f"Invalid JSON constant: {value}")
 
-    with path.open(encoding="utf-8") as source:
+    for attempt in range(3):
-        return json.load(source, parse_constant=reject_constant)
+        try:
+            with path.open(encoding="utf-8") as source:
+                return json.load(source, parse_constant=reject_constant)
+        except (OSError, UnicodeError, json.JSONDecodeError) as error:
+            if attempt == 2:
+                raise ValueError(f"Cannot read JSON {path}: {error}") from error
+            time.sleep(0.1)
 
 
 def localized(value: object, language: str):
@@ -73,49 +85,286 @@ def search_document(document: dict, fragment: dict, expected: tuple[str, str, st
     return {key: value for key, value in result.items() if value is not None}
 
 
+def document_codes(input_root: Path, limit: int | None = None, start_at: str | None = None) -> list[str]:
+    connection = sqlite3.connect(f"file:{input_root / 'manifest.sqlite3'}?mode=ro", uri=True)
+    try:
+        query = "SELECT code FROM documents WHERE state = 'success' ORDER BY CAST(code AS INTEGER), code"
+        parameters: list[int] = []
+        if limit is not None:
+            query += " LIMIT ?"
+            parameters.append(limit)
+        codes = [str(code) for (code,) in connection.execute(query, parameters)]
+    finally:
+        connection.close()
+    if start_at is None:
+        return codes
+    try:
+        return codes[codes.index(start_at):]
+    except ValueError as error:
+        raise ValueError(f"Resume document not found in selected range: {start_at}") from error
+
+
+def bulk_pairs(
+    input_root: Path,
+    index: str,
+    limit: int | None = None,
+    start_at: str | None = None,
+) -> Iterator[tuple[str, bytes]]:
+    document_root = input_root / "documents"
+    if not document_root.is_dir():
+        raise FileNotFoundError(f"Document directory not found: {document_root}")
+    for code in document_codes(input_root, limit, start_at):
+        directory = document_root / code
+        document = read_json(directory / "document.json")
+        if document.get("source_code") != directory.name:
+            raise ValueError(f"Document identity does not match its path: {directory}")
+        for fragment_path in sorted(directory.glob("editions/*/*/fragments.json")):
+            edition_code, language = fragment_path.parts[-3:-1]
+            values = read_json(fragment_path)
+            if not isinstance(values, list):
+                raise ValueError(f"Fragments must be a list: {fragment_path}")
+            for position, fragment in enumerate(values, 1):
+                source = search_document(document, fragment, (directory.name, edition_code, language, position))
+                action = json.dumps({"index": {"_index": index, "_id": fragment["id"]}}, ensure_ascii=False, allow_nan=False)
+                body = json.dumps(source, ensure_ascii=False, allow_nan=False)
+                yield directory.name, f"{action}\n{body}\n".encode()
+
+
+def document_count(input_root: Path, limit: int | None, start_at: str | None = None) -> int:
+    return len(document_codes(input_root, limit, start_at))
+
+
+def bulk_batches(pairs: Iterator[tuple[str, bytes]], maximum_bytes: int) -> Iterator[tuple[str, bytes]]:
+    batch = bytearray()
+    last_code = ""
+    for code, pair in pairs:
+        if len(pair) > maximum_bytes:
+            raise ValueError(f"One Bulk pair exceeds the {maximum_bytes}-byte batch limit")
+        if batch and len(batch) + len(pair) > maximum_bytes:
+            yield last_code, bytes(batch)
+            batch.clear()
+        last_code = code
+        batch.extend(pair)
+    if batch:
+        yield last_code, bytes(batch)
+
+
 def export_bulk(input_root: Path, output: Path, index: str, limit: int | None = None) -> tuple[int, int]:
     source = input_root.resolve()
     destination = output.resolve()
     if destination == source or destination.is_relative_to(source):
         raise ValueError("--output must not be inside --input")
-    document_root = input_root / "documents"
-    if not document_root.is_dir():
-        raise FileNotFoundError(f"Document directory not found: {document_root}")
-    connection = sqlite3.connect(f"file:{input_root / 'manifest.sqlite3'}?mode=ro", uri=True)
-    query = "SELECT code FROM documents WHERE state = 'success' ORDER BY CAST(code AS INTEGER), code"
-    parameters: tuple[int, ...] = ()
-    if limit is not None:
-        query += " LIMIT ?"
-        parameters = (limit,)
     output.parent.mkdir(parents=True, exist_ok=True)
-    documents = fragments = 0
+    fragments = 0
-    with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=output.parent, delete=False) as target:
+    with tempfile.NamedTemporaryFile("wb", dir=output.parent, delete=False) as target:
         temporary = Path(target.name)
         try:
-            for (code,) in connection.execute(query, parameters):
+            for code, pair in bulk_pairs(input_root, index, limit):
-                directory = document_root / str(code)
+                target.write(pair)
-                document = read_json(directory / "document.json")
+                fragments += 1
-                if document.get("source_code") != directory.name:
-                    raise ValueError(f"Document identity does not match its path: {directory}")
-                documents += 1
-                for fragment_path in sorted(directory.glob("editions/*/*/fragments.json")):
-                    edition_code, language = fragment_path.parts[-3:-1]
-                    values = read_json(fragment_path)
-                    if not isinstance(values, list):
-                        raise ValueError(f"Fragments must be a list: {fragment_path}")
-                    for position, fragment in enumerate(values, 1):
-                        source = search_document(document, fragment, (directory.name, edition_code, language, position))
-                        target.write(json.dumps({"index": {"_index": index, "_id": fragment["id"]}}, ensure_ascii=False, allow_nan=False) + "\n")
-                        target.write(json.dumps(source, ensure_ascii=False, allow_nan=False) + "\n")
-                        fragments += 1
             target.flush()
             os.fsync(target.fileno())
             os.replace(temporary, output)
         except BaseException:
             temporary.unlink(missing_ok=True)
             raise
-        finally:
+    return document_count(input_root, limit), fragments
-            connection.close()
+
+
+def file_sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def write_json_atomic(path: Path, value: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile("wb", dir=path.parent, delete=False) as target:
+        temporary = Path(target.name)
+        try:
+            target.write((json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode())
+            target.flush()
+            os.fsync(target.fileno())
+            os.replace(temporary, path)
+        except BaseException:
+            temporary.unlink(missing_ok=True)
+            raise
+
+
+def request_json(
+    url: str,
+    method: str,
+    body: bytes | None,
+    content_type: str,
+    attempts: int = 5,
+    retry_invalid_json: bool = False,
+) -> dict:
+    request = urllib.request.Request(url, data=body, method=method, headers={"Content-Type": content_type})
+    for attempt in range(attempts):
+        try:
+            with urllib.request.urlopen(request, timeout=120) as response:
+                return json.load(response)
+        except (UnicodeError, json.JSONDecodeError) as error:
+            if not retry_invalid_json or attempt == attempts - 1:
+                raise RuntimeError(f"{method} {url} returned invalid JSON: {error}") from error
+            delay = 2**attempt
+        except urllib.error.HTTPError as error:
+            response_body = error.read(500).decode("utf-8", errors="replace")
+            error.close()
+            retryable = error.code == 429 or 500 <= error.code < 600
+            if not retryable or attempt == attempts - 1:
+                raise RuntimeError(
+                    f"{method} {url} failed with HTTP {error.code}: {response_body}"
+                ) from error
+            retry_after = error.headers.get("Retry-After")
+            delay = float(retry_after) if retry_after and retry_after.isdigit() else 2**attempt
+        except (urllib.error.URLError, TimeoutError, ConnectionResetError) as error:
+            if attempt == attempts - 1:
+                raise RuntimeError(f"{method} {url} failed after {attempts} attempts: {error}") from error
+            delay = 2**attempt
+        time.sleep(delay)
+    raise AssertionError("unreachable")
+
+
+def opensearch_identity(base: str, index: str, index_url: str) -> tuple[str, str]:
+    cluster = request_json(f"{base}/", "GET", None, "application/json")
+    definition = request_json(index_url, "GET", None, "application/json")
+    try:
+        return cluster["cluster_uuid"], definition[index]["settings"]["index"]["uuid"]
+    except (KeyError, TypeError) as error:
+        raise RuntimeError("OpenSearch identity response is incomplete") from error
+
+
+def checkpoint_state(
+    input_root: Path,
+    index: str,
+    checkpoint: Path,
+    resume: bool,
+    url: str,
+    cluster_uuid: str,
+    index_uuid: str,
+    limit: int | None,
+) -> tuple[dict, str | None]:
+    source = input_root.resolve()
+    destination = checkpoint.resolve()
+    if destination == source or destination.is_relative_to(source):
+        raise ValueError("--checkpoint must not be inside --input")
+    manifest_sha256 = file_sha256(input_root / "manifest.sqlite3")
+    if not resume:
+        return {
+            "schema_version": 1,
+            "url": url,
+            "cluster_uuid": cluster_uuid,
+            "index": index,
+            "index_uuid": index_uuid,
+            "input": str(source),
+            "manifest_sha256": manifest_sha256,
+            "limit": limit,
+            "last_document_code": None,
+            "complete": False,
+        }, None
+
+    state = read_json(checkpoint)
+    expected = {
+        "schema_version",
+        "url",
+        "cluster_uuid",
+        "index",
+        "index_uuid",
+        "input",
+        "manifest_sha256",
+        "limit",
+        "last_document_code",
+        "complete",
+    }
+    if not isinstance(state, dict) or set(state) != expected:
+        raise ValueError(f"Invalid checkpoint: {checkpoint}")
+    if (
+        state["schema_version"] != 1
+        or state["url"] != url
+        or state["cluster_uuid"] != cluster_uuid
+        or state["index"] != index
+        or state["index_uuid"] != index_uuid
+        or state["input"] != str(source)
+    ):
+        raise ValueError(f"Checkpoint does not match this load: {checkpoint}")
+    if state["limit"] != limit:
+        raise ValueError(f"Checkpoint limit does not match --limit: {checkpoint}")
+    if state["manifest_sha256"] != manifest_sha256:
+        raise ValueError("Normalized manifest changed; create a new versioned index")
+    if state["complete"] is not False:
+        raise ValueError(f"Checkpoint is already complete: {checkpoint}")
+    start_at = state["last_document_code"]
+    if start_at is not None and not isinstance(start_at, str):
+        raise ValueError(f"Invalid checkpoint document code: {checkpoint}")
+    return state, start_at
+
+
+def load_bulk(
+    input_root: Path,
+    url: str,
+    index: str,
+    mapping: Path = DEFAULT_MAPPING,
+    maximum_bytes: int = 25 * 1024 * 1024,
+    limit: int | None = None,
+    resume: bool = False,
+    checkpoint: Path = Path("data/opensearch/minjust-fragments.checkpoint.json"),
+) -> tuple[int, int]:
+    base = url.rstrip("/")
+    index_url = f"{base}/{urllib.parse.quote(index, safe='')}"
+    if resume:
+        cluster_uuid, index_uuid = opensearch_identity(base, index, index_url)
+    else:
+        request_json(index_url, "PUT", mapping.read_bytes(), "application/json")
+        cluster_uuid, index_uuid = opensearch_identity(base, index, index_url)
+    state, start_at = checkpoint_state(
+        input_root,
+        index,
+        checkpoint,
+        resume,
+        base,
+        cluster_uuid,
+        index_uuid,
+        limit,
+    )
+    documents = document_count(input_root, limit, start_at)
+    if not resume:
+        write_json_atomic(checkpoint, state)
+    fragments = 0
+    for last_code, batch in bulk_batches(bulk_pairs(input_root, index, limit, start_at), maximum_bytes):
+        result = request_json(
+            f"{base}/_bulk",
+            "POST",
+            batch,
+            "application/x-ndjson",
+            retry_invalid_json=True,
+        )
+        expected = batch.count(b"\n") // 2
+        items = result.get("items", [])
+        if result.get("errors"):
+            failures = [item.get("index", {}) for item in items if item.get("index", {}).get("error")]
+            details = "; ".join(
+                f"{item.get('_id', '<unknown>')}: {item['error'].get('type', 'error')}: "
+                f"{item['error'].get('reason', '<no reason>')}"
+                for item in failures[:5]
+            )
+            raise RuntimeError(
+                f"OpenSearch Bulk API failed for {len(failures)} item(s); "
+                f"resume from {checkpoint}: {details}"
+            )
+        if len(items) != expected:
+            raise RuntimeError(
+                f"OpenSearch Bulk API returned {len(items)} of {expected} item result(s); "
+                f"resume from {checkpoint}"
+            )
+        fragments += len(items)
+        state["last_document_code"] = last_code
+        write_json_atomic(checkpoint, state)
+        print(f"checkpoint={last_code} fragments={fragments}", flush=True)
+    state["complete"] = True
+    write_json_atomic(checkpoint, state)
     return documents, fragments
 
 
@@ -125,12 +374,38 @@ def main() -> int:
     parser.add_argument("--output", type=Path, default=Path("data/opensearch/minjust-fragments.ndjson"))
     parser.add_argument("--index", default="akyldash-fragments-v1")
     parser.add_argument("--limit", type=int)
+    parser.add_argument("--url", help="create the index and stream bounded Bulk requests instead of writing a file")
+    parser.add_argument("--mapping", type=Path, default=DEFAULT_MAPPING)
+    parser.add_argument("--batch-mb", type=int, default=25)
+    parser.add_argument("--resume", action="store_true", help="load into an existing index")
+    parser.add_argument("--checkpoint", type=Path, help="persistent resume checkpoint path")
     parser.add_argument("--version", action="version", version=APP_VERSION)
     arguments = parser.parse_args()
     if arguments.limit is not None and arguments.limit <= 0:
         raise SystemExit("--limit must be greater than zero")
-    documents, fragments = export_bulk(arguments.input, arguments.output, arguments.index, arguments.limit)
+    if arguments.batch_mb <= 0:
-    print(f"documents={documents} fragments={fragments} output={arguments.output}\nAkyldash Backend v{APP_VERSION} · Frontend — not created")
+        raise SystemExit("--batch-mb must be greater than zero")
+    if arguments.resume and not arguments.url:
+        raise SystemExit("--resume requires --url")
+    if arguments.checkpoint and not arguments.url:
+        raise SystemExit("--checkpoint requires --url")
+    if arguments.url:
+        checkpoint = arguments.checkpoint or Path("data/opensearch") / f"{arguments.index}.checkpoint.json"
+        documents, fragments = load_bulk(
+            arguments.input,
+            arguments.url,
+            arguments.index,
+            arguments.mapping,
+            arguments.batch_mb * 1024 * 1024,
+            arguments.limit,
+            arguments.resume,
+            checkpoint,
+        )
+        destination = arguments.url
+    else:
+        documents, fragments = export_bulk(arguments.input, arguments.output, arguments.index, arguments.limit)
+        destination = arguments.output
+    print(f"documents={documents} fragments={fragments} destination={destination}\nAkyldash Backend v{APP_VERSION} · Frontend — not created")
     return 0
 
 

backend/test_minjust_opensearch.py

@@ -1,11 +1,15 @@
 import hashlib
+import io
 import json
 import sqlite3
 import tempfile
 import unittest
+import urllib.error
+from contextlib import closing
 from pathlib import Path
+from unittest.mock import patch
 
-from search.minjust_opensearch import export_bulk
+from search.minjust_opensearch import bulk_batches, document_codes, export_bulk, load_bulk, request_json
 
 
 class MinjustOpenSearchTest(unittest.TestCase):
@@ -14,9 +18,11 @@ class MinjustOpenSearchTest(unittest.TestCase):
             root = Path(temporary)
             document_root = root / "normalized/documents/7"
             document_root.mkdir(parents=True)
-            with sqlite3.connect(root / "normalized/manifest.sqlite3") as connection:
+            with closing(sqlite3.connect(root / "normalized/manifest.sqlite3")) as connection:
-                connection.execute("CREATE TABLE documents (code TEXT, state TEXT)")
+                with connection:
-                connection.execute("INSERT INTO documents VALUES ('7', 'success')")
+                    connection.execute("CREATE TABLE documents (code TEXT, state TEXT)")
+                    connection.execute("INSERT INTO documents VALUES ('7', 'success')")
+                    connection.execute("INSERT INTO documents VALUES ('8', 'success')")
             (document_root / "document.json").write_text(
                 json.dumps(
                     {
@@ -33,6 +39,11 @@ class MinjustOpenSearchTest(unittest.TestCase):
                 ),
                 encoding="utf-8",
             )
+            empty = root / "normalized/documents/8"
+            empty.mkdir()
+            (empty / "document.json").write_text(
+                json.dumps({"schema_version": "1", "source_code": "8"}), encoding="utf-8"
+            )
             for language, text in (("ru", "Текст \"RU\"\nстрока"), ("ky", "Кыргызча текст")):
                 path = document_root / f"editions/10/{language}/fragments.json"
                 path.parent.mkdir(parents=True)
@@ -56,7 +67,7 @@ class MinjustOpenSearchTest(unittest.TestCase):
                 )
 
             output = root / "bulk.ndjson"
-            self.assertEqual(export_bulk(root / "normalized", output, "test-index"), (1, 2))
+            self.assertEqual(export_bulk(root / "normalized", output, "test-index"), (2, 2))
             content = output.read_bytes()
             self.assertTrue(content.endswith(b"\n"))
             lines = [json.loads(line) for line in content.splitlines()]
@@ -65,6 +76,186 @@ class MinjustOpenSearchTest(unittest.TestCase):
             self.assertIn("text_ky", lines[1])
             self.assertNotIn("text_ru", lines[1])
             self.assertEqual(lines[3]["text_ru"], "Текст \"RU\"\nстрока")
+            self.assertEqual(
+                list(bulk_batches(iter((("7", b"a\nb\n"), ("8", b"c\nd\n"))), 4)),
+                [("7", b"a\nb\n"), ("8", b"c\nd\n")],
+            )
+            self.assertEqual(document_codes(root / "normalized", 2, "8"), ["8"])
+            with self.assertRaisesRegex(ValueError, "selected range"):
+                document_codes(root / "normalized", 1, "8")
+
+            checkpoint = root / "checkpoint.json"
+            with patch("search.minjust_opensearch.request_json") as request:
+                request.side_effect = [
+                    {},
+                    {"cluster_uuid": "cluster-1"},
+                    {"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
+                    {"errors": False, "items": [{"index": {}}, {"index": {}}]},
+                ]
+                self.assertEqual(
+                    load_bulk(
+                        root / "normalized",
+                        "http://127.0.0.1:9200",
+                        "test-index",
+                        maximum_bytes=4096,
+                        checkpoint=checkpoint,
+                    ),
+                    (2, 2),
+                )
+                self.assertEqual(request.call_args_list[-1].args[3], "application/x-ndjson")
+            state = json.loads(checkpoint.read_text(encoding="utf-8"))
+            self.assertEqual(state["last_document_code"], "7")
+            self.assertTrue(state["complete"])
+
+            state["last_document_code"] = "8"
+            state["complete"] = False
+            checkpoint.write_text(json.dumps(state), encoding="utf-8")
+            with patch("search.minjust_opensearch.request_json") as request:
+                request.side_effect = [
+                    {"cluster_uuid": "cluster-2"},
+                    {"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
+                ]
+                with self.assertRaisesRegex(ValueError, "does not match"):
+                    load_bulk(
+                        root / "normalized",
+                        "http://127.0.0.1:9200",
+                        "test-index",
+                        maximum_bytes=4096,
+                        resume=True,
+                        checkpoint=checkpoint,
+                    )
+            with patch("search.minjust_opensearch.request_json") as request:
+                request.side_effect = [
+                    {"cluster_uuid": "cluster-1"},
+                    {"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
+                ]
+                with self.assertRaisesRegex(ValueError, "limit"):
+                    load_bulk(
+                        root / "normalized",
+                        "http://127.0.0.1:9200",
+                        "test-index",
+                        maximum_bytes=4096,
+                        limit=1,
+                        resume=True,
+                        checkpoint=checkpoint,
+                    )
+            with patch("search.minjust_opensearch.request_json") as request:
+                request.side_effect = [
+                    {"cluster_uuid": "cluster-1"},
+                    {"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
+                ]
+                self.assertEqual(
+                    load_bulk(
+                        root / "normalized",
+                        "http://127.0.0.1:9200",
+                        "test-index",
+                        maximum_bytes=4096,
+                        resume=True,
+                        checkpoint=checkpoint,
+                    ),
+                    (1, 0),
+                )
+                self.assertTrue(all(call.args[1] == "GET" for call in request.call_args_list))
+
+            state["last_document_code"] = "9"
+            state["complete"] = False
+            checkpoint.write_text(json.dumps(state), encoding="utf-8")
+            with patch("search.minjust_opensearch.request_json") as request:
+                request.side_effect = [
+                    {"cluster_uuid": "cluster-1"},
+                    {"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
+                ]
+                with self.assertRaisesRegex(ValueError, "Resume document not found"):
+                    load_bulk(
+                        root / "normalized",
+                        "http://127.0.0.1:9200",
+                        "test-index",
+                        maximum_bytes=4096,
+                        resume=True,
+                        checkpoint=checkpoint,
+                    )
+
+            failed_checkpoint = root / "failed-checkpoint.json"
+            failure = {
+                "errors": True,
+                "items": [{"index": {"_id": "bad-id", "error": {"type": "mapper", "reason": "bad value"}}}],
+            }
+            failed_requests = [
+                {},
+                {"cluster_uuid": "cluster-1"},
+                {"failed-index": {"settings": {"index": {"uuid": "index-2"}}}},
+                failure,
+            ]
+            with patch("search.minjust_opensearch.request_json", side_effect=failed_requests):
+                with self.assertRaisesRegex(RuntimeError, "bad-id: mapper: bad value"):
+                    load_bulk(
+                        root / "normalized",
+                        "http://127.0.0.1:9200",
+                        "failed-index",
+                        maximum_bytes=4096,
+                        checkpoint=failed_checkpoint,
+                    )
+            failed_state = json.loads(failed_checkpoint.read_text(encoding="utf-8"))
+            self.assertIsNone(failed_state["last_document_code"])
+            self.assertFalse(failed_state["complete"])
+
+            state["last_document_code"] = "8"
+            state["complete"] = False
+            checkpoint.write_text(json.dumps(state), encoding="utf-8")
+            with closing(sqlite3.connect(root / "normalized/manifest.sqlite3")) as connection:
+                with connection:
+                    connection.execute("INSERT INTO documents VALUES ('9', 'success')")
+            with patch("search.minjust_opensearch.request_json") as request:
+                request.side_effect = [
+                    {"cluster_uuid": "cluster-1"},
+                    {"test-index": {"settings": {"index": {"uuid": "index-1"}}}},
+                ]
+                with self.assertRaisesRegex(ValueError, "manifest changed"):
+                    load_bulk(
+                        root / "normalized",
+                        "http://127.0.0.1:9200",
+                        "test-index",
+                        maximum_bytes=4096,
+                        resume=True,
+                        checkpoint=checkpoint,
+                    )
+
+            http_error = urllib.error.HTTPError(
+                "http://127.0.0.1:9200/test",
+                429,
+                "busy",
+                {},
+                io.BytesIO(b"busy"),
+            )
+            with (
+                patch("search.minjust_opensearch.urllib.request.urlopen", side_effect=[http_error, io.BytesIO(b"{}")]),
+                patch("search.minjust_opensearch.time.sleep") as sleep,
+            ):
+                self.assertEqual(
+                    request_json("http://127.0.0.1:9200/test", "GET", None, "application/json", attempts=2),
+                    {},
+                )
+                sleep.assert_called_once_with(1)
+
+            with (
+                patch(
+                    "search.minjust_opensearch.urllib.request.urlopen",
+                    side_effect=[io.BytesIO(b"{"), io.BytesIO(b"{}")],
+                ),
+                patch("search.minjust_opensearch.time.sleep") as sleep,
+            ):
+                self.assertEqual(
+                    request_json(
+                        "http://127.0.0.1:9200/_bulk",
+                        "POST",
+                        b"{}\n{}\n",
+                        "application/x-ndjson",
+                        attempts=2,
+                        retry_invalid_json=True,
+                    ),
+                    {},
+                )
+                sleep.assert_called_once_with(1)
 
             bad = document_root / "editions/10/ru/fragments.json"
             fragments = json.loads(bad.read_text(encoding="utf-8"))
@@ -83,9 +274,17 @@ class MinjustOpenSearchTest(unittest.TestCase):
                 export_bulk(root / "normalized", root / "normalized/manifest.sqlite3", "test-index")
             self.assertEqual(output.read_bytes(), content)
 
-            mapping = json.loads(
+            bad.write_text("", encoding="utf-8")
+            with self.assertRaisesRegex(ValueError, "fragments.json"):
+                export_bulk(root / "normalized", output, "test-index")
+            self.assertEqual(output.read_bytes(), content)
+
+            definition = json.loads(
                 (Path(__file__).parent / "search/minjust-fragments-index.json").read_text(encoding="utf-8")
-            )["mappings"]
+            )
+            mapping = definition["mappings"]
+            self.assertEqual(definition["settings"]["index"]["number_of_shards"], 1)
+            self.assertEqual(definition["settings"]["index"]["number_of_replicas"], 0)
             self.assertEqual(mapping["dynamic"], "strict")
             self.assertEqual(mapping["properties"]["position"]["type"], "integer")
             self.assertEqual(mapping["properties"]["text_ru"]["analyzer"], "russian")

deploy/local-opensearch/Dockerfile

@@ -0,0 +1,3 @@
+FROM opensearchproject/opensearch:3.7.0
+
+RUN /usr/share/opensearch/bin/opensearch-plugin install --batch analysis-icu

deploy/local-opensearch/compose.yaml

@@ -0,0 +1,21 @@
+services:
+  opensearch:
+    build: .
+    container_name: akyldash-opensearch
+    environment:
+      discovery.type: single-node
+      bootstrap.memory_lock: "true"
+      DISABLE_SECURITY_PLUGIN: "true"
+      OPENSEARCH_JAVA_OPTS: -Xms8g -Xmx8g
+    mem_limit: 12g
+    ports:
+      - 127.0.0.1:9200:9200
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+      nofile:
+        soft: 65536
+        hard: 65536
+    volumes:
+      - ../../data/opensearch-node:/usr/share/opensearch/data

docs/README.md

@@ -33,4 +33,4 @@
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

docs/decisions/001-telegram-workspace-mvp.md

@@ -78,4 +78,4 @@ Telegram позволяет запретить пользователям отп
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

docs/operations/project-status.md

@@ -1,19 +1,19 @@
 # Статус проекта
 
-Последняя проверка: 2026-08-13
+Последняя проверка: 2026-08-14
 Назначение документа: быстро восстановить контекст проекта для участников команды и будущих агентов.
 
 - Telegram-бот: `0.2.2`
 - Telegram-бот на Synology: `0.2.1`
-- Backend: `0.3.0`
+- Backend: `0.4.1`
 - Frontend: не создан
 
 ## Краткий итог
 
 Репозиторий переориентирован с отдельного бота на весь проект юридической информационно-аналитической платформы. Telegram-бот выделен в инструмент рабочего окружения. Реализованы возобновляемая выгрузка документов из официального Open Data API ЦБД Минюста КР и их локальная воспроизводимая нормализация.
 
-Ближайшая цель — развернуть проверяемый OpenSearch с `analysis-icu`, загрузить
+Ближайшая цель — оценить полный локальный индекс на 50–100 запросах RU/KY и
-подготовленный Bulk NDJSON и оценить поиск на 50–100 запросах RU/KY.
+настроить ранжирование до начала разработки поискового API.
 
 ## Уже сделано
 
@@ -135,6 +135,22 @@
 
 ## История изменений статуса
 
+### 2026-08-14
+
+- Полный локальный индекс содержит 56 295 965 фрагментов и успешно отвечает на
+  RU/KY-запросы.
+- Добавлены атомарный checkpoint и безопасное продолжение прерванной загрузки
+  существующего индекса.
+- Добавлены retry/backoff для временных HTTP-сбоев и подробные ошибки Bulk API.
+- Ошибки чтения JSON теперь содержат точный путь и повторяются при временном сбое.
+- Версия backend обновлена до `0.4.1`.
+
+### 2026-08-13
+
+- Добавлен локальный одноузловой OpenSearch с `analysis-icu` без Dashboards.
+- Добавлена прямая потоковая загрузка корпуса пакетами до 25 МБ.
+- Версия backend обновлена до `0.4.0`.
+
 ### 2026-08-13
 
 - Подтверждена полная успешная нормализация 209 958 загруженных документов.
@@ -207,4 +223,4 @@
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

docs/operations/telegram-workspace-plan.md

@@ -398,4 +398,4 @@ Git сохраняет актуальную версию
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

docs/product/frontend-search-sps-plan.md

@@ -258,4 +258,4 @@ runtime-зависимостями frontend. Регистрация в стор
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

docs/product/minjust-document-normalization-agent-task.md

@@ -214,4 +214,4 @@ python3 backend/normalization/minjust_cbd.py
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

docs/product/project-overview.md

@@ -44,4 +44,4 @@
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

docs/team/ai-skills-for-beginners.md

@@ -180,4 +180,4 @@
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

tools/telegram-bot/README.md

@@ -48,4 +48,4 @@ python3 -m unittest -v
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан