Merge pull request 'Подготовить индекс фрагментов OpenSearch' (#9) from feature/opensearch-index-foundation into main

Reviewed-on: #9
2026-08-13 05:46:46 +00:00
parent 30065925e4 ac5ed95f3a
commit dddb07f393
16 changed files with 318 additions and 25 deletions
--- a/README.md
+++ b/README.md
@@ -10,14 +10,15 @@ Telegram-бот — только часть рабочего окружения
 ## Текущее состояние

 Сейчас реализованы Telegram-бот-секретарь версии `0.2.2` и backend версии
-`0.2.2`: возобновляемая выгрузка и нормализация документов ЦБД Минюста КР.
+`0.3.0`: возобновляемая выгрузка, нормализация и подготовка индекса документов
+ЦБД Минюста КР.

 | Компонент | Версия | Состояние |
 |---|---:|---|
 | Telegram-бот | `0.2.2` | на Synology работает `0.2.1`; обновление после слияния |
-| Backend | `0.2.2` | реализованы выгрузка и нормализация документов ЦБД Минюста КР |
+| Backend | `0.3.0` | реализованы выгрузка, нормализация и экспорт для OpenSearch |
 | Frontend | — | ещё не создан |
-| Сбор и обработка правовых данных | `0.2.2` | реализованы архиватор и нормализатор ЦБД Минюста КР |
+| Сбор и обработка правовых данных | `0.3.0` | добавлены mapping и Bulk NDJSON для OpenSearch |
 | RAG и база знаний | — | ещё не созданы |

 ## Структура репозитория
@@ -58,4 +59,4 @@ python3 -m unittest discover -s tools/telegram-bot -v

 ---

-Акылдаш · Telegram-бот v0.2.2 · Backend v0.2.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
--- a/backend/README.md
+++ b/backend/README.md
@@ -1,6 +1,6 @@
 # Backend Акылдаш

-Версия: `0.2.2`
+Версия: `0.3.0`

 Первая backend-область проекта — загрузка правовых документов из официального
 Open Data API ЦБД Минюста Кыргызской Республики. Код расположен в
@@ -96,8 +96,28 @@ data/minjust-normalized/
 ```bash
 PYTHONPATH=backend python3 -m unittest backend/test_minjust_cbd.py -v
 PYTHONPATH=backend python3 -m unittest backend/test_minjust_normalization.py -v
+PYTHONPATH=backend python3 -m unittest backend/test_minjust_opensearch.py -v
 ```

+## Подготовка индекса OpenSearch
+
+Mapping поискового индекса находится в
+`search/minjust-fragments-index.json`. Для кыргызского текста он использует
+`icu_analyzer`, поэтому в OpenSearch должен быть установлен плагин
+`analysis-icu`.
+
+Потоковый экспорт в формат Bulk API без внешних Python-зависимостей:
+
+```bash
+python3 backend/search/minjust_opensearch.py
+```
+
+По умолчанию создаётся `data/opensearch/minjust-fragments.ndjson`. Экспорт
+атомарный и детерминированный; для проверки можно передать `--limit 1`.
+Полный файл может быть большим, поэтому перед запуском нужно проверить
+свободное место. Загружать данные следует в новый версионный индекс и после
+проверки переключать alias, чтобы удалённые фрагменты не оставались в поиске.
+
 ---

-Акылдаш · Backend v0.2.2 · Frontend — не создан
+Акылдаш · Backend v0.3.0 · Frontend — не создан
--- a/backend/ingestion/minjust_cbd.py
+++ b/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.2.2"
+APP_VERSION = "0.3.0"
 API_BASE_URL = "https://cbd.minjust.gov.kg/api/v1/OpenData/"
 LANGUAGES = {"Rus": "ru", "Kyr": "ky"}
 IMAGE_LANGUAGES = {"Russian": "ru", "Kyrgyz": "ky"}
--- a/backend/normalization/minjust_cbd.py
+++ b/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.2.2"
+APP_VERSION = "0.3.0"
 SCHEMA_VERSION = "1"
 NORMALIZER_VERSION = "1.0.0"
 LANGUAGES = ("ru", "ky")
--- a/backend/search/minjust-fragments-index.json
+++ b/backend/search/minjust-fragments-index.json
@@ -0,0 +1,28 @@
+{
+  "mappings": {
+    "dynamic": "strict",
+    "properties": {
+      "schema_version": { "type": "keyword" },
+      "document_code": { "type": "keyword" },
+      "edition_code": { "type": "keyword" },
+      "language": { "type": "keyword" },
+      "position": { "type": "integer" },
+      "fragment_type": { "type": "keyword" },
+      "text_ru": { "type": "text", "analyzer": "russian" },
+      "text_ky": { "type": "text", "analyzer": "icu_analyzer" },
+      "document_name_ru": { "type": "text", "analyzer": "russian", "fields": { "keyword": { "type": "keyword", "ignore_above": 1024 } } },
+      "document_name_ky": { "type": "text", "analyzer": "icu_analyzer", "fields": { "keyword": { "type": "keyword", "ignore_above": 1024 } } },
+      "document_type_ru": { "type": "keyword" },
+      "document_type_ky": { "type": "keyword" },
+      "status_ru": { "type": "keyword" },
+      "status_ky": { "type": "keyword" },
+      "number": { "type": "keyword" },
+      "date_adopted": { "type": "date", "format": "strict_date" },
+      "authority_paths_ru": { "type": "keyword", "ignore_above": 2048 },
+      "authority_paths_ky": { "type": "keyword", "ignore_above": 2048 },
+      "source_path": { "type": "keyword", "index": false },
+      "source_sha256": { "type": "keyword", "index": false },
+      "text_sha256": { "type": "keyword", "index": false }
+    }
+  }
+}
--- a/backend/search/minjust_opensearch.py
+++ b/backend/search/minjust_opensearch.py
@@ -0,0 +1,138 @@
+"""Export normalized Ministry fragments for the OpenSearch Bulk API."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import sqlite3
+import tempfile
+from pathlib import Path
+
+APP_VERSION = "0.3.0"
+LANGUAGES = {"ru", "ky"}
+
+
+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:
+        return json.load(source, parse_constant=reject_constant)
+
+
+def localized(value: object, language: str):
+    return value.get(language) if isinstance(value, dict) else None
+
+
+def paths(document: dict, field: str, language: str) -> list[str]:
+    return [" > ".join(item[language]) for item in document.get(field, []) if item.get(language)]
+
+
+def search_document(document: dict, fragment: dict, expected: tuple[str, str, str, int]) -> dict:
+    document_code, edition_code, language, position = expected
+    fragment_id = f"document:{document_code}:edition:{edition_code}:lang:{language}:fragment:{position}"
+    required = ("id", "document_code", "edition_code", "language", "position", "type", "text", "text_sha256", "source_path", "source_sha256")
+    if not isinstance(fragment, dict) or any(key not in fragment for key in required):
+        raise ValueError(f"Incomplete fragment {fragment_id}")
+    if (fragment["document_code"], fragment["edition_code"], fragment["language"], fragment["position"], fragment["id"]) != (*expected, fragment_id):
+        raise ValueError(f"Fragment identity does not match its path: {fragment_id}")
+    if language not in LANGUAGES or not isinstance(fragment["text"], str) or not fragment["text"]:
+        raise ValueError(f"Invalid fragment content: {fragment_id}")
+    for key in ("text_sha256", "source_sha256"):
+        value = fragment[key]
+        if not isinstance(value, str) or len(value) != 64 or any(char not in "0123456789abcdef" for char in value):
+            raise ValueError(f"Invalid {key}: {fragment_id}")
+    if hashlib.sha256(fragment["text"].encode()).hexdigest() != fragment["text_sha256"]:
+        raise ValueError(f"Text checksum mismatch: {fragment_id}")
+
+    dates = document.get("dates") or {}
+    result = {
+        "schema_version": document["schema_version"],
+        "document_code": document_code,
+        "edition_code": edition_code,
+        "language": language,
+        "position": position,
+        "fragment_type": fragment["type"],
+        f"text_{language}": fragment["text"],
+        "document_name_ru": localized(document.get("name"), "ru"),
+        "document_name_ky": localized(document.get("name"), "ky"),
+        "document_type_ru": localized(document.get("type"), "ru"),
+        "document_type_ky": localized(document.get("type"), "ky"),
+        "status_ru": localized(document.get("status"), "ru"),
+        "status_ky": localized(document.get("status"), "ky"),
+        "number": document.get("number"),
+        "date_adopted": dates.get("DateAdopted"),
+        "authority_paths_ru": paths(document, "authority_paths", "ru"),
+        "authority_paths_ky": paths(document, "authority_paths", "ky"),
+        "source_path": fragment["source_path"],
+        "source_sha256": fragment["source_sha256"],
+        "text_sha256": fragment["text_sha256"],
+    }
+    return {key: value for key, value in result.items() if value is not None}
+
+
+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
+    with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=output.parent, delete=False) as target:
+        temporary = Path(target.name)
+        try:
+            for (code,) in connection.execute(query, parameters):
+                directory = document_root / str(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}")
+                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:
+            connection.close()
+    return documents, fragments
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--input", type=Path, default=Path("data/minjust-normalized"))
+    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("--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)
+    print(f"documents={documents} fragments={fragments} output={arguments.output}\nAkyldash Backend v{APP_VERSION} · Frontend — not created")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
--- a/backend/test_minjust_opensearch.py
+++ b/backend/test_minjust_opensearch.py
@@ -0,0 +1,96 @@
+import hashlib
+import json
+import sqlite3
+import tempfile
+import unittest
+from pathlib import Path
+
+from search.minjust_opensearch import export_bulk
+
+
+class MinjustOpenSearchTest(unittest.TestCase):
+    def test_exports_atomic_bulk_and_rejects_mismatched_fragment(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            root = Path(temporary)
+            document_root = root / "normalized/documents/7"
+            document_root.mkdir(parents=True)
+            with sqlite3.connect(root / "normalized/manifest.sqlite3") as connection:
+                connection.execute("CREATE TABLE documents (code TEXT, state TEXT)")
+                connection.execute("INSERT INTO documents VALUES ('7', 'success')")
+            (document_root / "document.json").write_text(
+                json.dumps(
+                    {
+                        "schema_version": "1",
+                        "source_code": "7",
+                        "name": {"ru": "Закон", "ky": "Мыйзам"},
+                        "type": {"ru": "Закон", "ky": "Мыйзам"},
+                        "status": {"ru": "Действует", "ky": "Күчүндө"},
+                        "number": "1",
+                        "dates": {"DateAdopted": "2026-01-01"},
+                        "authority_paths": [{"ru": ["Кабинет"], "ky": ["Кабинет"]}],
+                    },
+                    ensure_ascii=False,
+                ),
+                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)
+                path.write_text(
+                    json.dumps(
+                        [{
+                            "id": f"document:7:edition:10:lang:{language}:fragment:1",
+                            "document_code": "7",
+                            "edition_code": "10",
+                            "language": language,
+                            "position": 1,
+                            "type": "paragraph",
+                            "text": text,
+                            "text_sha256": hashlib.sha256(text.encode()).hexdigest(),
+                            "source_path": f"documents/7/editions/10/{language}.html",
+                            "source_sha256": "b" * 64,
+                        }],
+                        ensure_ascii=False,
+                    ),
+                    encoding="utf-8",
+                )
+
+            output = root / "bulk.ndjson"
+            self.assertEqual(export_bulk(root / "normalized", output, "test-index"), (1, 2))
+            content = output.read_bytes()
+            self.assertTrue(content.endswith(b"\n"))
+            lines = [json.loads(line) for line in content.splitlines()]
+            self.assertEqual(len(lines), 4)
+            self.assertEqual(lines[0]["index"]["_id"], "document:7:edition:10:lang:ky:fragment:1")
+            self.assertIn("text_ky", lines[1])
+            self.assertNotIn("text_ru", lines[1])
+            self.assertEqual(lines[3]["text_ru"], "Текст \"RU\"\nстрока")
+
+            bad = document_root / "editions/10/ru/fragments.json"
+            fragments = json.loads(bad.read_text(encoding="utf-8"))
+            fragments[0]["document_code"] = "8"
+            bad.write_text(json.dumps(fragments, ensure_ascii=False), encoding="utf-8")
+            with self.assertRaisesRegex(ValueError, "identity"):
+                export_bulk(root / "normalized", output, "test-index")
+            self.assertEqual(output.read_bytes(), content)
+
+            fragments[0]["document_code"] = "7"
+            fragments[0]["text"] = "Повреждено"
+            bad.write_text(json.dumps(fragments, ensure_ascii=False), encoding="utf-8")
+            with self.assertRaisesRegex(ValueError, "checksum"):
+                export_bulk(root / "normalized", output, "test-index")
+            with self.assertRaisesRegex(ValueError, "inside --input"):
+                export_bulk(root / "normalized", root / "normalized/manifest.sqlite3", "test-index")
+            self.assertEqual(output.read_bytes(), content)
+
+            mapping = json.loads(
+                (Path(__file__).parent / "search/minjust-fragments-index.json").read_text(encoding="utf-8")
+            )["mappings"]
+            self.assertEqual(mapping["dynamic"], "strict")
+            self.assertEqual(mapping["properties"]["position"]["type"], "integer")
+            self.assertEqual(mapping["properties"]["text_ru"]["analyzer"], "russian")
+            self.assertEqual(mapping["properties"]["text_ky"]["analyzer"], "icu_analyzer")
+
+
+if __name__ == "__main__":
+    unittest.main()
--- a/docs/README.md
+++ b/docs/README.md
@@ -33,4 +33,4 @@

 ---

-Акылдаш · Telegram-бот v0.2.2 · Backend v0.2.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
--- a/docs/decisions/001-telegram-workspace-mvp.md
+++ b/docs/decisions/001-telegram-workspace-mvp.md
@@ -78,4 +78,4 @@ Telegram позволяет запретить пользователям отп

 ---

-Акылдаш · Telegram-бот v0.2.2 · Backend v0.2.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
--- a/docs/operations/project-status.md
+++ b/docs/operations/project-status.md
@@ -1,19 +1,19 @@
 # Статус проекта

-Последняя проверка: 2026-08-10
+Последняя проверка: 2026-08-13
 Назначение документа: быстро восстановить контекст проекта для участников команды и будущих агентов.

 - Telegram-бот: `0.2.2`
 - Telegram-бот на Synology: `0.2.1`
- Backend: `0.2.2`
+- Backend: `0.3.0`
 - Frontend: не создан

 ## Краткий итог

 Репозиторий переориентирован с отдельного бота на весь проект юридической информационно-аналитической платформы. Telegram-бот выделен в инструмент рабочего окружения. Реализованы возобновляемая выгрузка документов из официального Open Data API ЦБД Минюста КР и их локальная воспроизводимая нормализация.

-Ближайшая цель — выполнить полный проход нормализатора, проверить отчёт ошибок
-и качество контрольной выборки RU/KY, затем подготовить индекс OpenSearch.
+Ближайшая цель — развернуть проверяемый OpenSearch с `analysis-icu`, загрузить
+подготовленный Bulk NDJSON и оценить поиск на 50–100 запросах RU/KY.

 ## Уже сделано

@@ -74,6 +74,9 @@
 - Реализован backend-нормализатор версии `0.2.2` без внешних зависимостей.
 - Нормализатор создаёт канонические метаданные, безопасный HTML, чистый текст и адресуемые фрагменты RU/KY.
 - SQLite-манифест обеспечивает возобновление, повтор ошибок и пропуск неизменившихся документов.
+- Полный проход завершён: 209 958 документов нормализованы без ошибок.
+- Контрольная выборка RU/KY прошла проверки текста, фрагментов, ID и SHA-256.
+- Добавлены строгий mapping и атомарный Bulk NDJSON-экспорт для OpenSearch.

 ### Развёртывание

@@ -132,6 +135,13 @@

 ## История изменений статуса

+### 2026-08-13
+
+- Подтверждена полная успешная нормализация 209 958 загруженных документов.
+- Единственный отсутствующий документ `6` повторно не отдан API Минюста.
+- Добавлены mapping фрагментов и потоковый экспорт для OpenSearch Bulk API.
+- Версия backend обновлена до `0.3.0`.
+
 ### 2026-08-12

 - Нормализатор запрещает пересекающиеся каталоги источника и результата,
@@ -197,4 +207,4 @@

 ---

-Акылдаш · Telegram-бот v0.2.2 · Backend v0.2.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
--- a/docs/operations/telegram-workspace-plan.md
+++ b/docs/operations/telegram-workspace-plan.md
@@ -398,4 +398,4 @@ Git сохраняет актуальную версию

 ---

-Акылдаш · Telegram-бот v0.2.2 · Backend v0.2.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
--- a/docs/product/frontend-search-sps-plan.md
+++ b/docs/product/frontend-search-sps-plan.md
@@ -98,10 +98,10 @@ MVP не входят.
 Интерфейс не должен предполагать наличие обоих языков. На момент полного
 скачивания архива распределение следующее:

- только русский язык — 29 432 документа;
- только кыргызский язык — 98 797 документов;
- оба языка — 80 901 документ;
- нет HTML-текста — 681 документ.
+- только русский язык — 29 433 документа;
+- только кыргызский язык — 98 905 документов;
+- оба языка — 80 930 документов;
+- нет HTML-текста — 690 документов.

 ## Рекомендуемая основа frontend

@@ -258,4 +258,4 @@ runtime-зависимостями frontend. Регистрация в стор

 ---

-Акылдаш · Telegram-бот v0.2.2 · Backend v0.2.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
--- a/docs/product/minjust-document-normalization-agent-task.md
+++ b/docs/product/minjust-document-normalization-agent-task.md
@@ -214,4 +214,4 @@ python3 backend/normalization/minjust_cbd.py

 ---

-Акылдаш · Telegram-бот v0.2.2 · Backend v0.2.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
--- a/docs/product/project-overview.md
+++ b/docs/product/project-overview.md
@@ -44,4 +44,4 @@

 ---

-Акылдаш · Telegram-бот v0.2.2 · Backend v0.2.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
--- a/docs/team/ai-skills-for-beginners.md
+++ b/docs/team/ai-skills-for-beginners.md
@@ -180,4 +180,4 @@

 ---

-Акылдаш · Telegram-бот v0.2.2 · Backend v0.2.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан
--- a/tools/telegram-bot/README.md
+++ b/tools/telegram-bot/README.md
@@ -48,4 +48,4 @@ python3 -m unittest -v

 ---

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