feat: add resumable local OpenSearch loading

Reviewed-on: #9

.gitignore

@@ -13,3 +13,6 @@ __pycache__/
 
 # Local application archives
 data/
+
+# Local runtime logs
+logs/

README.md

@@ -9,15 +9,16 @@ Telegram-бот — только часть рабочего окружения
 
 ## Текущее состояние
 
-Сейчас реализованы Telegram-бот-секретарь версии `0.2.2` и первая backend-функция
-версии `0.1.2`: возобновляемая выгрузка документов из ЦБД Минюста КР.
+Сейчас реализованы Telegram-бот-секретарь версии `0.2.2` и backend версии
+`0.4.1`: возобновляемая загрузка индекса документов Министерства юстиции
+ЦБД Минюста КР.
 
 | Компонент | Версия | Состояние |
 |---|---:|---|
 | Telegram-бот | `0.2.2` | на Synology работает `0.2.1`; обновление после слияния |
-| Backend | `0.1.2` | реализована выгрузка документов ЦБД Минюста КР |
+| Backend | `0.4.1` | добавлено продолжение прерванной Bulk-загрузки |
 | Frontend | — | ещё не создан |
-| Сбор и обработка правовых данных | `0.1.2` | реализован архиватор ЦБД Минюста КР |
+| Сбор и обработка правовых данных | `0.4.1` | добавлено продолжение загрузки существующего индекса |
 | RAG и база знаний | — | ещё не созданы |
 
 ## Структура репозитория
@@ -32,6 +33,7 @@ tools/
   telegram-bot/        бот рабочего Telegram-пространства
 backend/
   ingestion/           получение и обновление правовых источников
+  normalization/       воспроизводимая нормализация исходного архива
 ```
 
 Каталоги для загрузки и обработки источников, RAG, backend и frontend будут
@@ -57,4 +59,4 @@ python3 -m unittest discover -s tools/telegram-bot -v
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.1.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

backend/README.md

@@ -1,6 +1,6 @@
 # Backend Акылдаш
 
-Версия: `0.1.2`
+Версия: `0.4.1`
 
 Первая backend-область проекта — загрузка правовых документов из официального
 Open Data API ЦБД Минюста Кыргызской Республики. Код расположен в
@@ -21,8 +21,9 @@ python3 backend/ingestion/minjust_cbd.py --limit 10
 Полная загрузка выполняется без `--limit`. По умолчанию архив сохраняется в
 `data/minjust-cbd`, который исключён из Git. Повторный запуск пропускает уже
 загруженные документы; `--refresh` принудительно проверяет их заново.
-Если временный идентификатор списка API истечёт во время многодневной загрузки,
-скрипт пересоздаст список на текущей странице и продолжит автоматически.
+Если временный идентификатор списка API истечёт или запрос страницы исчерпает
+повторы во время многодневной загрузки, скрипт пересоздаст список на текущей
+странице и продолжит автоматически.
 
 В версии `0.1.0` обновление существующих документов выполняется полной проверкой
 через `--refresh`. Инкрементальную проверку по `lastmod` из sitemap следует
@@ -53,12 +54,117 @@ result = sync_archive(output_path)
 Планировщик, очередь задач и PostgreSQL пока не добавлены: модуль не зависит от
 выбора будущего backend-фреймворка.
 
+## Нормализация архива
+
+Нормализатор читает исходный архив без изменений и создаёт отдельный набор
+данных для будущих поиска, API и RAG:
+
+```bash
+python3 backend/normalization/minjust_cbd.py --limit 10
+python3 backend/normalization/minjust_cbd.py
+```
+
+По умолчанию источник читается из `data/minjust-cbd`, а результат записывается
+в `data/minjust-normalized`. Пути можно изменить параметрами `--input` и
+`--output`; `--refresh` принудительно обрабатывает неизменившиеся документы,
+`--log-level` задаёт уровень журнала.
+Каталоги `--input` и `--output` не должны совпадать, содержать друг друга или
+пересекаться через разрешённые абсолютные пути.
+
+```text
+data/minjust-normalized/
+  manifest.sqlite3
+  documents/<code>/document.json
+  documents/<code>/editions/<edition>/edition.json
+  documents/<code>/editions/<edition>/<lang>/content.html
+  documents/<code>/editions/<edition>/<lang>/content.txt
+  documents/<code>/editions/<edition>/<lang>/fragments.json
+```
+
+`content.html` содержит только разрешённую безопасную разметку, `content.txt` —
+текст для поиска, а `fragments.json` — адресуемые блоки с детерминированными ID
+и SHA-256. Манифест пропускает документы с неизменившимися исходниками и
+повторяет документы, обработка которых завершилась ошибкой.
+
+Первая версия не выполняет OCR, перевод, юридические выводы о редакциях,
+сопоставление фрагментов, загрузку в PostgreSQL/OpenSearch и построение RAG.
+Внешние и встроенные `data:`-изображения из HTML удаляются; сведения и пути к
+локальным изображениям исходного архива сохраняются в `edition.json`.
+
 ## Проверка
 
 ```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`.
+Для прямой загрузки без большого промежуточного файла используется `--url`:
+
+```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.1.2 · 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.1.2"
+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"}
@@ -121,8 +121,8 @@ class CbdClient:
                         ("PageNumber", page_number),
                     ),
                 )
-            except urllib.error.HTTPError as error:
-                if error.code != 404:
+            except (urllib.error.HTTPError, RuntimeError) as error:
+                if isinstance(error, urllib.error.HTTPError) and error.code != 404:
                     raise
                 first = page = query_page(page_number)
                 self.total_documents = min(

backend/normalization/minjust_cbd.py

@@ -0,0 +1,729 @@
+#!/usr/bin/env python3
+"""Normalize the local Ministry of Justice CBD archive."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import html
+import json
+import logging
+import os
+import re
+import shutil
+import sqlite3
+import sys
+import tempfile
+import time
+import unicodedata
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from html.parser import HTMLParser
+from pathlib import Path
+from typing import Callable
+from urllib.parse import urlsplit
+
+APP_VERSION = "0.4.1"
+SCHEMA_VERSION = "1"
+NORMALIZER_VERSION = "1.0.0"
+LANGUAGES = ("ru", "ky")
+LOGGER = logging.getLogger(__name__)
+
+ALLOWED_TAGS = {
+    "a", "b", "blockquote", "br", "div", "em", "h1", "h2", "h3", "h4",
+    "h5", "h6", "i", "img", "li", "ol", "p", "pre", "span", "strong",
+    "sub", "sup", "table", "tbody", "td", "tfoot", "th", "thead", "tr",
+    "u", "ul",
+}
+VOID_TAGS = {"br", "img"}
+DROP_CONTENT_TAGS = {"applet", "iframe", "noscript", "object", "script", "style", "svg"}
+DROP_ELEMENT_TAGS = {"link", "meta"}
+BLOCK_TAGS = {"blockquote", "h1", "h2", "h3", "h4", "h5", "h6", "li", "p", "pre", "td", "th"}
+AUTO_CLOSE = {
+    "li": {"li"},
+    "p": {"blockquote", "div", "h1", "h2", "h3", "h4", "h5", "h6", "li", "ol", "p", "pre", "table", "ul"},
+    "td": {"td", "th"},
+    "th": {"td", "th"},
+    "tr": {"tr"},
+}
+
+
+@dataclass(frozen=True)
+class NormalizeResult:
+    discovered: int = 0
+    normalized: int = 0
+    skipped: int = 0
+    failed: int = 0
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def atomic_write(path: Path, content: bytes) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as temporary:
+        temporary.write(content)
+        temporary_path = Path(temporary.name)
+    os.replace(temporary_path, path)
+
+
+def json_bytes(value: object) -> bytes:
+    return (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
+
+
+def sha256_bytes(content: bytes) -> str:
+    return hashlib.sha256(content).hexdigest()
+
+
+def source_inventory(document_directory: Path, input_root: Path) -> tuple[list[dict], str]:
+    files = []
+    combined = hashlib.sha256()
+    for path in sorted(item for item in document_directory.rglob("*") if item.is_file()):
+        relative = path.relative_to(input_root).as_posix()
+        digest = hashlib.sha256()
+        with path.open("rb") as source:
+            for chunk in iter(lambda: source.read(1024 * 1024), b""):
+                digest.update(chunk)
+        checksum = digest.hexdigest()
+        files.append({"path": relative, "sha256": checksum})
+        combined.update(relative.encode("utf-8"))
+        combined.update(b"\0")
+        combined.update(checksum.encode("ascii"))
+        combined.update(b"\0")
+    return files, combined.hexdigest()
+
+
+def clean_value(value):
+    if isinstance(value, str):
+        normalized = unicodedata.normalize("NFC", value)
+        return normalized if normalized.strip() else None
+    if isinstance(value, dict):
+        return {key: clean_value(item) for key, item in value.items()}
+    if isinstance(value, list):
+        return [clean_value(item) for item in value]
+    return value
+
+
+def bilingual(value) -> dict[str, object]:
+    value = value if isinstance(value, dict) else {}
+    return {"ru": clean_value(value.get("Rus")), "ky": clean_value(value.get("Kyr"))}
+
+
+def hierarchy_paths(items: object, child_key: str) -> list[dict]:
+    paths: list[dict] = []
+
+    def visit(nodes: object, ancestors: dict[str, list[str]]) -> None:
+        for node in nodes if isinstance(nodes, list) else []:
+            if not isinstance(node, dict):
+                continue
+            names = bilingual(node.get("Name"))
+            current = {language: list(ancestors[language]) for language in LANGUAGES}
+            for language in LANGUAGES:
+                name = names[language]
+                if name:
+                    current[language].append(str(name))
+            children = node.get(child_key)
+            if children:
+                visit(children, current)
+            else:
+                paths.append(current)
+
+    visit(items, {"ru": [], "ky": []})
+    return paths
+
+
+class SafeHtmlParser(HTMLParser):
+    def __init__(self, edition_directory: Path) -> None:
+        super().__init__(convert_charrefs=True)
+        self.edition_directory = edition_directory.resolve()
+        self.parts: list[str] = []
+        self.stack: list[str] = []
+        self.drop_depth = 0
+        self.removed_elements = 0
+        self.removed_attributes = 0
+        self.removed_images = 0
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        tag = tag.lower()
+        if self.drop_depth:
+            if tag in DROP_CONTENT_TAGS:
+                self.drop_depth += 1
+            return
+        if tag in DROP_CONTENT_TAGS:
+            self.drop_depth = 1
+            self.removed_elements += 1
+            return
+        if tag in DROP_ELEMENT_TAGS:
+            self.removed_elements += 1
+            return
+        if tag not in ALLOWED_TAGS:
+            self.removed_elements += 1
+            return
+        for open_tag, closing_tags in AUTO_CLOSE.items():
+            if tag in closing_tags and open_tag in self.stack:
+                self._close(open_tag)
+        safe_attrs = self._attributes(tag, attrs)
+        if tag == "img" and not any(name == "src" for name, _ in safe_attrs):
+            self.removed_images += 1
+            return
+        rendered = "".join(
+            f' {name}="{html.escape(value, quote=True)}"' for name, value in safe_attrs
+        )
+        self.parts.append(f"<{tag}{rendered}>")
+        if tag not in VOID_TAGS:
+            self.stack.append(tag)
+
+    def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        self.handle_starttag(tag, attrs)
+        if tag.lower() not in VOID_TAGS:
+            self.handle_endtag(tag)
+
+    def handle_endtag(self, tag: str) -> None:
+        tag = tag.lower()
+        if self.drop_depth:
+            if tag in DROP_CONTENT_TAGS:
+                self.drop_depth -= 1
+            return
+        if tag in self.stack:
+            self._close(tag)
+
+    def handle_data(self, data: str) -> None:
+        if not self.drop_depth:
+            self.parts.append(html.escape(unicodedata.normalize("NFC", data), quote=False))
+
+    def close(self) -> None:
+        super().close()
+        while self.stack:
+            self.parts.append(f"</{self.stack.pop()}>")
+
+    def _close(self, tag: str) -> None:
+        while self.stack:
+            current = self.stack.pop()
+            self.parts.append(f"</{current}>")
+            if current == tag:
+                break
+
+    def _attributes(self, tag: str, attrs: list[tuple[str, str | None]]) -> list[tuple[str, str]]:
+        allowed = {"title"}
+        if tag == "a":
+            allowed |= {"href"}
+        elif tag == "img":
+            allowed |= {"alt", "src"}
+        elif tag in {"td", "th"}:
+            allowed |= {"colspan", "rowspan"}
+        safe = []
+        for raw_name, raw_value in attrs:
+            name = raw_name.lower()
+            value = unicodedata.normalize("NFC", raw_value or "")
+            if name not in allowed:
+                self.removed_attributes += 1
+                continue
+            if name == "href" and not safe_link(value):
+                self.removed_attributes += 1
+                continue
+            if name == "src" and not self.safe_image(value):
+                self.removed_attributes += 1
+                continue
+            if name in {"colspan", "rowspan"} and not value.isdigit():
+                self.removed_attributes += 1
+                continue
+            safe.append((name, value))
+        if tag == "a" and any(name == "href" and urlsplit(value).scheme in {"http", "https"} for name, value in safe):
+            safe.append(("rel", "noopener noreferrer"))
+        return safe
+
+    def safe_image(self, value: str) -> bool:
+        try:
+            parsed = urlsplit(value)
+        except ValueError:
+            return False
+        if parsed.scheme or parsed.netloc or not parsed.path or parsed.path.startswith(("/", "\\")):
+            return False
+        candidate = (self.edition_directory / parsed.path.replace("\\", "/")).resolve()
+        try:
+            candidate.relative_to(self.edition_directory)
+        except ValueError:
+            return False
+        return candidate.is_file()
+
+
+def safe_link(value: str) -> bool:
+    value = value.strip()
+    if not value or value.startswith(("//", "\\\\")):
+        return False
+    try:
+        parsed = urlsplit(value)
+    except ValueError:
+        return False
+    return parsed.scheme.lower() in {"", "http", "https", "mailto"} and not (
+        not parsed.scheme and parsed.netloc
+    )
+
+
+def sanitize_html(source: str, edition_directory: Path) -> tuple[str, dict]:
+    parser = SafeHtmlParser(edition_directory)
+    parser.feed(source)
+    parser.close()
+    return unicodedata.normalize("NFC", "".join(parser.parts)), {
+        "removed_elements": parser.removed_elements,
+        "removed_attributes": parser.removed_attributes,
+        "removed_images": parser.removed_images,
+    }
+
+
+class TextBlockParser(HTMLParser):
+    def __init__(self) -> None:
+        super().__init__(convert_charrefs=True)
+        self.blocks: list[tuple[str, str]] = []
+        self.active_tag: str | None = None
+        self.active: list[str] = []
+        self.loose: list[str] = []
+
+    def handle_starttag(self, tag: str, attrs) -> None:
+        if tag in BLOCK_TAGS:
+            self._flush_active()
+            self._flush_loose()
+            self.active_tag = tag
+        elif tag == "br":
+            (self.active if self.active_tag else self.loose).append("\n")
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag == self.active_tag:
+            self._flush_active()
+
+    def handle_data(self, data: str) -> None:
+        (self.active if self.active_tag else self.loose).append(data)
+
+    def close(self) -> None:
+        super().close()
+        self._flush_active()
+        self._flush_loose()
+
+    def _flush_active(self) -> None:
+        if self.active_tag:
+            text = clean_text("".join(self.active))
+            if text:
+                self.blocks.append((self.active_tag, text))
+        self.active_tag = None
+        self.active = []
+
+    def _flush_loose(self) -> None:
+        text = clean_text("".join(self.loose))
+        if text:
+            self.blocks.append(("p", text))
+        self.loose = []
+
+
+def clean_text(value: str) -> str:
+    lines = []
+    for line in unicodedata.normalize("NFC", value).replace("\xa0", " ").splitlines():
+        line = re.sub(r"[ \t\f\v]+", " ", line).strip()
+        if line:
+            lines.append(line)
+    return "\n".join(lines)
+
+
+def fragment_type(tag: str, text: str, language: str) -> str:
+    lowered = text.casefold()
+    article_words = ("статья", "ст.") if language == "ru" else ("берене", "статья")
+    if any(re.match(rf"^{re.escape(word)}\s*\d", lowered) for word in article_words):
+        return "article"
+    if re.match(r"^\d+(?:\.\d+)*[.)]?\s+", text):
+        return "point"
+    if tag.startswith("h"):
+        return "heading"
+    return {"li": "list_item", "td": "table_cell", "th": "table_header"}.get(tag, "paragraph")
+
+
+def extract_text_and_fragments(
+    sanitized: str,
+    document_code: str,
+    edition_code: str,
+    language: str,
+    source_path: str,
+    source_sha256: str,
+) -> tuple[str, list[dict]]:
+    parser = TextBlockParser()
+    parser.feed(sanitized)
+    parser.close()
+    fragments = []
+    for position, (tag, text) in enumerate(parser.blocks, 1):
+        fragments.append(
+            {
+                "id": f"document:{document_code}:edition:{edition_code}:lang:{language}:fragment:{position}",
+                "document_code": document_code,
+                "edition_code": edition_code,
+                "language": language,
+                "position": position,
+                "type": fragment_type(tag, text, language),
+                "text": text,
+                "text_sha256": sha256_bytes(text.encode("utf-8")),
+                "source_path": source_path,
+                "source_sha256": source_sha256,
+            }
+        )
+    return "\n\n".join(fragment["text"] for fragment in fragments), fragments
+
+
+def load_json(path: Path) -> dict:
+    value = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(value, dict):
+        raise ValueError(f"Expected JSON object: {path}")
+    return value
+
+
+def edition_summary(edition_directory: Path, input_root: Path) -> dict:
+    metadata = load_json(edition_directory / "metadata.json")
+    languages = [language for language in LANGUAGES if (edition_directory / f"{language}.html").is_file()]
+    return {
+        "source_code": str(metadata.get("Code", edition_directory.name)),
+        "name": bilingual(metadata.get("Name")),
+        "source_type": clean_value(metadata.get("Type")),
+        "available_languages": languages,
+        "source_path": edition_directory.relative_to(input_root).as_posix(),
+    }
+
+
+def normalize_document(
+    document_directory: Path,
+    input_root: Path,
+    destination: Path,
+    files: list[dict] | None = None,
+    source_checksum: str | None = None,
+) -> None:
+    metadata_path = document_directory / "metadata.json"
+    metadata = load_json(metadata_path)
+    document_code = str(metadata.get("Code", document_directory.name))
+    if document_code != document_directory.name:
+        raise ValueError(f"Document code mismatch in {metadata_path}")
+    if files is None or source_checksum is None:
+        files, source_checksum = source_inventory(document_directory, input_root)
+    file_checksums = {item["path"]: item["sha256"] for item in files}
+    edition_root = document_directory / "editions"
+    edition_directories = sorted(
+        (path for path in edition_root.iterdir() if path.is_dir()),
+        key=lambda path: (not path.name.isdigit(), int(path.name) if path.name.isdigit() else path.name),
+    ) if edition_root.is_dir() else []
+    summaries = [edition_summary(path, input_root) for path in edition_directories]
+    available_languages = [language for language in LANGUAGES if any(language in item["available_languages"] for item in summaries)]
+    normalized_metadata = clean_value(metadata)
+    document = {
+        "schema_version": SCHEMA_VERSION,
+        "source_code": document_code,
+        "class": bilingual(metadata.get("Class")),
+        "type": bilingual(metadata.get("Type")),
+        "title": bilingual(metadata.get("Title")),
+        "name": bilingual(metadata.get("Name")),
+        "status": bilingual(metadata.get("Status")),
+        "number": clean_value(metadata.get("Number")),
+        "dates": {key: value for key, value in normalized_metadata.items() if key.startswith("Date")},
+        "registration_number": clean_value(metadata.get("NumberRegistration")),
+        "publication_number": clean_value(metadata.get("NumberPublication")),
+        "is_public_in_cdb": metadata.get("IsPublicInCdb"),
+        "is_public_in_register": metadata.get("IsPublicInRegister"),
+        "authorities": normalized_metadata.get("Authorities") or [],
+        "authority_paths": hierarchy_paths(metadata.get("Authorities"), "Authorities"),
+        "source_publications": normalized_metadata.get("SourcePublications") or [],
+        "source_publication_paths": hierarchy_paths(metadata.get("SourcePublications"), "SourcePublications"),
+        "keywords": normalized_metadata.get("Keywords") or [],
+        "keyword_paths": hierarchy_paths(metadata.get("Keywords"), "Keywords"),
+        "general_classifiers": normalized_metadata.get("GeneralClassifiers") or [],
+        "general_classifier_paths": hierarchy_paths(metadata.get("GeneralClassifiers"), "GeneralClassifiers"),
+        "references": normalized_metadata.get("References") or [],
+        "source_metadata": normalized_metadata,
+        "available_languages": available_languages,
+        "editions": summaries,
+        "source": {
+            "path": document_directory.relative_to(input_root).as_posix(),
+            "files": files,
+            "sha256": source_checksum,
+        },
+        "normalizer": {"version": NORMALIZER_VERSION, "processed_at": utc_now()},
+    }
+    atomic_write(destination / "document.json", json_bytes(document))
+
+    for edition_directory, summary in zip(edition_directories, summaries):
+        edition_code = summary["source_code"]
+        edition_metadata = load_json(edition_directory / "metadata.json")
+        edition_destination = destination / "editions" / edition_directory.name
+        image_records = []
+        for image in edition_metadata.get("Images") or []:
+            language = {"Russian": "ru", "Kyrgyz": "ky"}.get(image.get("Lang"), "unknown")
+            name = Path(str(image.get("Name") or "").replace("\\", "/")).name
+            source_path = edition_directory / "images" / language / name
+            relative = source_path.relative_to(input_root).as_posix()
+            image_records.append(
+                {
+                    "language": language,
+                    "name": clean_value(image.get("Name")),
+                    "source_path": relative if source_path.is_file() else None,
+                    "source_sha256": file_checksums.get(relative),
+                    "source_metadata": clean_value(image),
+                }
+            )
+        quality = {"has_html": bool(summary["available_languages"]), "languages": {}}
+        for language in summary["available_languages"]:
+            html_path = edition_directory / f"{language}.html"
+            relative = html_path.relative_to(input_root).as_posix()
+            raw = html_path.read_text(encoding="utf-8")
+            sanitized, sanitizer_quality = sanitize_html(raw, edition_directory)
+            text, fragments = extract_text_and_fragments(
+                sanitized, document_code, edition_code, language, relative, file_checksums[relative]
+            )
+            language_destination = edition_destination / language
+            atomic_write(language_destination / "content.html", sanitized.encode("utf-8"))
+            atomic_write(language_destination / "content.txt", (text + ("\n" if text else "")).encode("utf-8"))
+            atomic_write(language_destination / "fragments.json", json_bytes(fragments))
+            quality["languages"][language] = {
+                **sanitizer_quality,
+                "empty_text": not bool(text),
+                "fragment_count": len(fragments),
+            }
+        edition = {
+            "schema_version": SCHEMA_VERSION,
+            "source_code": edition_code,
+            "name": summary["name"],
+            "source_type": summary["source_type"],
+            "available_languages": summary["available_languages"],
+            "images": image_records,
+            "source_metadata": clean_value(edition_metadata),
+            "source": {
+                "path": summary["source_path"],
+                "files": [item for item in files if item["path"].startswith(summary["source_path"] + "/")],
+            },
+            "quality": quality,
+        }
+        atomic_write(edition_destination / "edition.json", json_bytes(edition))
+
+
+def connect_manifest(path: Path) -> sqlite3.Connection:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    connection = sqlite3.connect(path)
+    connection.execute(
+        """
+        CREATE TABLE IF NOT EXISTS documents (
+            code TEXT PRIMARY KEY,
+            source_sha256 TEXT,
+            schema_version TEXT NOT NULL,
+            normalizer_version TEXT NOT NULL,
+            processed_at TEXT,
+            state TEXT NOT NULL,
+            error TEXT,
+            failed_at TEXT
+        )
+        """
+    )
+    columns = {row[1] for row in connection.execute("PRAGMA table_info(documents)")}
+    if "failed_at" not in columns:
+        connection.execute("ALTER TABLE documents ADD COLUMN failed_at TEXT")
+    return connection
+
+
+def publish_directory(staged: Path, target: Path) -> None:
+    target.parent.mkdir(parents=True, exist_ok=True)
+    backup = target.parent / f".{target.name}.previous"
+    if backup.exists():
+        shutil.rmtree(backup)
+    if target.exists():
+        os.replace(target, backup)
+    try:
+        os.replace(staged, target)
+    except Exception:
+        if backup.exists():
+            os.replace(backup, target)
+        raise
+    if backup.exists():
+        shutil.rmtree(backup)
+
+
+def recover_directory(target: Path) -> None:
+    backup = target.parent / f".{target.name}.previous"
+    if not backup.exists():
+        return
+    if target.exists():
+        shutil.rmtree(backup)
+    else:
+        os.replace(backup, target)
+
+
+def validate_roots(input_root: Path, output: Path) -> None:
+    source = input_root.resolve()
+    destination = output.resolve()
+    if source == destination or source.is_relative_to(destination) or destination.is_relative_to(source):
+        raise ValueError("--input and --output must not overlap")
+
+
+def normalize_archive(
+    input_root: Path = Path("data/minjust-cbd"),
+    output: Path = Path("data/minjust-normalized"),
+    limit: int | None = None,
+    refresh: bool = False,
+    progress: Callable[[NormalizeResult, int], None] | None = None,
+) -> NormalizeResult:
+    validate_roots(input_root, output)
+    document_root = input_root / "documents"
+    if not document_root.is_dir():
+        raise FileNotFoundError(f"Document directory not found: {document_root}")
+    output.mkdir(parents=True, exist_ok=True)
+    connection = connect_manifest(output / "manifest.sqlite3")
+    known = {
+        row[0]: (row[1], row[2], row[3], row[4])
+        for row in connection.execute(
+            "SELECT code, source_sha256, schema_version, normalizer_version, state FROM documents"
+        )
+    }
+    directories = sorted(
+        (path for path in document_root.iterdir() if path.is_dir()),
+        key=lambda path: (not path.name.isdigit(), int(path.name) if path.name.isdigit() else path.name),
+    )
+    if limit is not None:
+        directories = directories[:limit]
+    total = len(directories)
+    discovered = normalized = skipped = failed = 0
+    staging_root = output / ".staging"
+    staging_root.mkdir(exist_ok=True)
+    try:
+        for source_directory in directories:
+            discovered += 1
+            code = source_directory.name
+            target = output / "documents" / code
+            checksum = None
+            try:
+                recover_directory(target)
+                files, checksum = source_inventory(source_directory, input_root)
+                if target.is_dir() and not refresh and known.get(code) == (
+                    checksum, SCHEMA_VERSION, NORMALIZER_VERSION, "success"
+                ):
+                    skipped += 1
+                else:
+                    with tempfile.TemporaryDirectory(dir=staging_root) as temporary:
+                        staged = Path(temporary) / code
+                        normalize_document(source_directory, input_root, staged, files, checksum)
+                        publish_directory(staged, target)
+                    with connection:
+                        connection.execute(
+                            """
+                            INSERT INTO documents (
+                                code, source_sha256, schema_version,
+                                normalizer_version, processed_at, state, error,
+                                failed_at
+                            ) VALUES (?, ?, ?, ?, ?, 'success', NULL, NULL)
+                            ON CONFLICT(code) DO UPDATE SET
+                                source_sha256=excluded.source_sha256,
+                                schema_version=excluded.schema_version,
+                                normalizer_version=excluded.normalizer_version,
+                                processed_at=excluded.processed_at,
+                                state='success', error=NULL, failed_at=NULL
+                            """,
+                            (code, checksum, SCHEMA_VERSION, NORMALIZER_VERSION, utc_now()),
+                        )
+                    normalized += 1
+            except Exception as error:  # Keep a corpus run alive after one malformed record.
+                LOGGER.exception("Failed to normalize document %s", code)
+                with connection:
+                    connection.execute(
+                        """
+                        INSERT INTO documents (
+                            code, source_sha256, schema_version,
+                            normalizer_version, processed_at, state, error,
+                            failed_at
+                        ) VALUES (?, ?, ?, ?, NULL, 'error', ?, ?)
+                        ON CONFLICT(code) DO UPDATE SET
+                            source_sha256=excluded.source_sha256,
+                            schema_version=excluded.schema_version,
+                            normalizer_version=excluded.normalizer_version,
+                            state='error', error=excluded.error,
+                            failed_at=excluded.failed_at
+                        """,
+                        (
+                            code,
+                            checksum,
+                            SCHEMA_VERSION,
+                            NORMALIZER_VERSION,
+                            str(error),
+                            utc_now(),
+                        ),
+                    )
+                failed += 1
+            result = NormalizeResult(discovered, normalized, skipped, failed)
+            if progress:
+                progress(result, total)
+            elif discovered % 100 == 0:
+                LOGGER.info("discovered=%s normalized=%s skipped=%s failed=%s", discovered, normalized, skipped, failed)
+    finally:
+        connection.close()
+        try:
+            staging_root.rmdir()
+        except OSError:
+            pass
+    return NormalizeResult(discovered, normalized, skipped, failed)
+
+
+def format_duration(seconds: float) -> str:
+    seconds = max(0, round(seconds))
+    hours, seconds = divmod(seconds, 3600)
+    minutes, seconds = divmod(seconds, 60)
+    return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
+
+
+def progress_line(result: NormalizeResult, total: int, elapsed: float, width: int = 24) -> str:
+    fraction = result.discovered / total if total else 0
+    filled = min(width, round(width * fraction))
+    rate = result.discovered / elapsed if elapsed > 0 else 0
+    eta = (total - result.discovered) / rate if rate else 0
+    return (
+        f"[{'#' * filled}{'-' * (width - filled)}] {fraction:6.2%} "
+        f"{result.discovered}/{total} normalized={result.normalized} "
+        f"skipped={result.skipped} failed={result.failed} "
+        f"rate={rate:.2f}/s ETA={format_duration(eta)}"
+    )
+
+
+def terminal_progress() -> Callable[[NormalizeResult, int], None]:
+    started = time.monotonic()
+
+    def update(result: NormalizeResult, total: int) -> None:
+        print(
+            f"\r{progress_line(result, total, time.monotonic() - started)}",
+            end="\n" if result.discovered >= total else "",
+            file=sys.stderr,
+            flush=True,
+        )
+
+    return update
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--input", type=Path, default=Path("data/minjust-cbd"))
+    parser.add_argument("--output", type=Path, default=Path("data/minjust-normalized"))
+    parser.add_argument("--limit", type=int, help="normalize only the first N documents")
+    parser.add_argument("--refresh", action="store_true", help="renormalize unchanged documents")
+    parser.add_argument("--log-level", choices=("DEBUG", "INFO", "WARNING", "ERROR"), default="INFO")
+    parser.add_argument("--version", action="version", version=APP_VERSION)
+    return parser.parse_args()
+
+
+def main() -> int:
+    arguments = parse_args()
+    if arguments.limit is not None and arguments.limit <= 0:
+        raise SystemExit("--limit must be greater than zero")
+    logging.basicConfig(level=arguments.log_level, format="%(asctime)s %(levelname)s %(message)s")
+    result = normalize_archive(
+        arguments.input,
+        arguments.output,
+        arguments.limit,
+        arguments.refresh,
+        terminal_progress() if sys.stderr.isatty() else None,
+    )
+    print(
+        f"discovered={result.discovered} normalized={result.normalized} "
+        f"skipped={result.skipped} failed={result.failed}\n"
+        f"Akyldash Backend v{APP_VERSION} · Frontend — not created"
+    )
+    return int(result.failed > 0)
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

backend/search/minjust-fragments-index.json

@@ -0,0 +1,35 @@
+{
+  "settings": {
+    "index": {
+      "number_of_shards": 1,
+      "number_of_replicas": 0,
+      "refresh_interval": "30s"
+    }
+  },
+  "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 }
+    }
+  }
+}

backend/search/minjust_opensearch.py

@@ -0,0 +1,413 @@
+"""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
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from pathlib import Path
+from typing import Iterator
+
+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}")
+
+    for attempt in range(3):
+        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):
+    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 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")
+    output.parent.mkdir(parents=True, exist_ok=True)
+    fragments = 0
+    with tempfile.NamedTemporaryFile("wb", dir=output.parent, delete=False) as target:
+        temporary = Path(target.name)
+        try:
+            for code, pair in bulk_pairs(input_root, index, limit):
+                target.write(pair)
+                fragments += 1
+            target.flush()
+            os.fsync(target.fileno())
+            os.replace(temporary, output)
+        except BaseException:
+            temporary.unlink(missing_ok=True)
+            raise
+    return document_count(input_root, limit), fragments
+
+
+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
+
+
+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("--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")
+    if arguments.batch_mb <= 0:
+        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
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

backend/test_minjust_cbd.py

@@ -99,6 +99,29 @@ class MinjustCbdTest(unittest.TestCase):
         self.assertEqual(recovery_call.args[0], "GetDocumentListByQuery")
         self.assertIn(("PageNumber", 2), recovery_call.args[1])
 
+    def test_recreates_list_after_page_retries_are_exhausted(self):
+        client = CbdClient(requests_per_second=1000)
+        client.request_json = Mock(
+            side_effect=[
+                {
+                    "Id": "failed-list",
+                    "TotalCount": 3,
+                    "Documents": [{"Code": 1}, {"Code": 2}],
+                },
+                RuntimeError("Ministry of Justice API request failed"),
+                {
+                    "Id": "new-list",
+                    "TotalCount": 3,
+                    "Documents": [{"Code": 3}],
+                },
+            ]
+        )
+
+        self.assertEqual(list(client.document_codes(page_size=2)), [1, 2, 3])
+        recovery_call = client.request_json.call_args_list[2]
+        self.assertEqual(recovery_call.args[0], "GetDocumentListByQuery")
+        self.assertIn(("PageNumber", 2), recovery_call.args[1])
+
     def test_formats_progress_with_rate_and_eta(self):
         line = progress_line(
             SyncResult(discovered=50, downloaded=48, skipped=1, failed=1),

backend/test_minjust_normalization.py

@@ -0,0 +1,152 @@
+import json
+import os
+import sqlite3
+import tempfile
+import unittest
+from pathlib import Path
+
+from normalization.minjust_cbd import normalize_archive
+
+
+class MinjustNormalizationTest(unittest.TestCase):
+    def write_json(self, path: Path, value: object) -> None:
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
+
+    def edition(self, root: Path, code: int, languages: dict[str, str]) -> None:
+        directory = root / "editions" / str(code)
+        self.write_json(
+            directory / "metadata.json",
+            {"Code": code, "Name": {"Rus": "Редакция", "Kyr": "Редакция"}, "Type": "edition", "Images": []},
+        )
+        for language, content in languages.items():
+            (directory / f"{language}.html").write_text(content, encoding="utf-8")
+
+    def document(self, root: Path, code: int = 1) -> Path:
+        directory = root / "documents" / str(code)
+        self.write_json(
+            directory / "metadata.json",
+            {
+                "Code": code,
+                "Class": {"Rus": "Акты", "Kyr": "Актылар"},
+                "Type": {"Rus": "Закон", "Kyr": "Мыйзам"},
+                "Title": {"Rus": " ", "Kyr": None},
+                "Name": {"Rus": "Документ", "Kyr": "Документ"},
+                "Status": {"Rus": "Действует", "Kyr": "Күчүндө"},
+                "Number": "1",
+                "DateAdopted": "2026-01-01",
+                "IsPublicInCdb": True,
+                "IsPublicInRegister": True,
+                "Authorities": [],
+                "SourcePublications": [],
+                "Keywords": [],
+                "GeneralClassifiers": [],
+                "References": [],
+            },
+        )
+        return directory
+
+    def test_languages_safety_empty_document_and_deterministic_fragments(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            base = Path(temporary)
+            source = base / "source"
+            output = base / "normalized"
+            document = self.document(source)
+            self.edition(
+                document,
+                10,
+                {
+                    "ru": '<meta charset=unicode><style>x</style><p style="color:red">Статья 1 Закон<script>bad()</script></p><a href="javascript:bad">ссылка</a><a href="http://[">сломанная ссылка</a><img src="http://[">',
+                },
+            )
+            self.edition(document, 20, {"ky": "<p>1. Кыргызча жобо</p>"})
+            self.edition(document, 30, {"ru": "<p>Русский</p>", "ky": "<p>Кыргызча</p>"})
+            self.edition(document, 40, {})
+
+            first = normalize_archive(source, output)
+            fragments_path = output / "documents/1/editions/10/ru/fragments.json"
+            fragments = fragments_path.read_bytes()
+            second = normalize_archive(source, output)
+
+            self.assertEqual((first.normalized, second.skipped), (1, 1))
+            self.assertEqual(fragments, fragments_path.read_bytes())
+            safe_html = (output / "documents/1/editions/10/ru/content.html").read_text(encoding="utf-8")
+            self.assertNotIn("script", safe_html)
+            self.assertNotIn("style=", safe_html)
+            self.assertNotIn("javascript:", safe_html)
+            self.assertNotIn("http://[", safe_html)
+            self.assertIn("Статья 1 Закон", safe_html)
+            parsed = json.loads(fragments)
+            self.assertEqual(parsed[0]["type"], "article")
+            self.assertEqual(parsed[0]["id"], "document:1:edition:10:lang:ru:fragment:1")
+            self.assertEqual(len(parsed[0]["text_sha256"]), 64)
+            canonical = json.loads((output / "documents/1/document.json").read_text(encoding="utf-8"))
+            self.assertEqual(canonical["available_languages"], ["ru", "ky"])
+            self.assertIsNone(canonical["title"]["ru"])
+            empty = json.loads((output / "documents/1/editions/40/edition.json").read_text(encoding="utf-8"))
+            self.assertFalse(empty["quality"]["has_html"])
+
+    def test_continues_after_bad_document_and_clears_repaired_error(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            base = Path(temporary)
+            source = base / "source"
+            output = base / "normalized"
+            bad = source / "documents/1"
+            bad.mkdir(parents=True)
+            (bad / "metadata.json").write_text("not json", encoding="utf-8")
+            good = self.document(source, 2)
+            self.edition(good, 10, {"ky": "<p>Берене 1 Текст</p>"})
+
+            with self.assertLogs("normalization.minjust_cbd", level="ERROR"):
+                failed = normalize_archive(source, output)
+            with sqlite3.connect(output / "manifest.sqlite3") as connection:
+                state, failed_at = connection.execute(
+                    "SELECT state, failed_at FROM documents WHERE code='1'"
+                ).fetchone()
+                self.assertEqual(state, "error")
+                self.assertIsNotNone(failed_at)
+            self.write_json(bad / "metadata.json", {"Code": 1, "Name": {"Rus": "Исправлен", "Kyr": None}})
+            repaired = normalize_archive(source, output)
+
+            self.assertEqual((failed.failed, failed.normalized), (1, 1))
+            self.assertEqual((repaired.normalized, repaired.skipped, repaired.failed), (1, 1, 0))
+            with sqlite3.connect(output / "manifest.sqlite3") as connection:
+                self.assertEqual(
+                    connection.execute(
+                        "SELECT state, error, failed_at FROM documents WHERE code='1'"
+                    ).fetchone(),
+                    ("success", None, None),
+                )
+
+    def test_rejects_overlapping_input_and_output(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            source = Path(temporary) / "source"
+            metadata = self.document(source) / "metadata.json"
+            original = metadata.read_bytes()
+
+            for output in (source, source / "normalized", source.parent):
+                with self.subTest(output=output):
+                    with self.assertRaisesRegex(ValueError, "must not overlap"):
+                        normalize_archive(source, output)
+                    self.assertEqual(metadata.read_bytes(), original)
+
+    def test_recovers_interrupted_directory_publication_before_skip(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            base = Path(temporary)
+            source = base / "source"
+            output = base / "normalized"
+            self.document(source)
+            first = normalize_archive(source, output)
+            target = output / "documents/1"
+            backup = output / "documents/.1.previous"
+            os.replace(target, backup)
+
+            second = normalize_archive(source, output)
+
+            self.assertEqual((first.normalized, second.skipped), (1, 1))
+            self.assertTrue((target / "document.json").is_file())
+            self.assertFalse(backup.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()

backend/test_minjust_opensearch.py

@@ -0,0 +1,295 @@
+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 bulk_batches, document_codes, export_bulk, load_bulk, request_json
+
+
+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 closing(sqlite3.connect(root / "normalized/manifest.sqlite3")) as connection:
+                with connection:
+                    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(
+                    {
+                        "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",
+            )
+            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)
+                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"), (2, 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строка")
+            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"))
+            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)
+
+            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")
+            )
+            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")
+            self.assertEqual(mapping["properties"]["text_ky"]["analyzer"], "icu_analyzer")
+
+
+if __name__ == "__main__":
+    unittest.main()

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

@@ -4,6 +4,10 @@
 
 - [Обзор проекта](product/project-overview.md) — назначение, основные области и
   правила работы с данными.
+- [План frontend поисковой СПС](product/frontend-search-sps-plan.md) — границы
+  MVP, зависимости и спринты.
+- [Задание по нормализации документов](product/minjust-document-normalization-agent-task.md) —
+  требования и критерии приёмки нормализатора ЦБД Минюста КР.
 
 ## Решения
 
@@ -19,8 +23,8 @@
 
 ## Backend
 
-- [Выгрузка ЦБД Минюста КР](../backend/README.md) — запуск, хранение и проверка
-  загрузчика правовых документов.
+- [Выгрузка и нормализация ЦБД Минюста КР](../backend/README.md) — запуск,
+  хранение и проверка конвейера правовых документов.
 
 ## Команда
 
@@ -29,4 +33,4 @@
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.1.2 · 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.1.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

docs/operations/project-status.md

@@ -1,18 +1,19 @@
 # Статус проекта
 
-Последняя проверка: 2026-08-06
+Последняя проверка: 2026-08-14
 Назначение документа: быстро восстановить контекст проекта для участников команды и будущих агентов.
 
 - Telegram-бот: `0.2.2`
 - Telegram-бот на Synology: `0.2.1`
-- Backend: `0.1.2`
+- Backend: `0.4.1`
 - Frontend: не создан
 
 ## Краткий итог
 
-Репозиторий переориентирован с отдельного бота на весь проект юридической информационно-аналитической платформы. Telegram-бот выделен в инструмент рабочего окружения. Создана первая backend-функция: возобновляемая выгрузка документов из официального Open Data API ЦБД Минюста КР.
+Репозиторий переориентирован с отдельного бота на весь проект юридической информационно-аналитической платформы. Telegram-бот выделен в инструмент рабочего окружения. Реализованы возобновляемая выгрузка документов из официального Open Data API ЦБД Минюста КР и их локальная воспроизводимая нормализация.
 
-Ближайшая цель — расширить пилотную выборку ЦБД, затем добавить инкрементальную проверку sitemap при создании backend-планировщика.
+Ближайшая цель — оценить полный локальный индекс на 50–100 запросах RU/KY и
+настроить ранжирование до начала разработки поискового API.
 
 ## Уже сделано
 
@@ -67,9 +68,15 @@
 - Локальный Git-репозиторий восстановлен и привязан к Gitea.
 - Репозиторий организован как основа всего проекта, а не отдельного бота.
 - Бот развёрнут в Container Manager на Synology; автозапуск после перезапуска менеджера проверен.
-- Реализован backend-загрузчик ЦБД Минюста КР версии `0.1.2` без внешних зависимостей.
+- Реализован backend-загрузчик ЦБД Минюста КР версии `0.2.2` без внешних зависимостей.
 - Загрузчик сохраняет метаданные, редакции RU/KY и изображения, а прогресс — в SQLite.
 - Пилотная выгрузка двух документов и возобновление без повторного скачивания проверены на живом API.
+- Реализован backend-нормализатор версии `0.2.2` без внешних зависимостей.
+- Нормализатор создаёт канонические метаданные, безопасный HTML, чистый текст и адресуемые фрагменты RU/KY.
+- SQLite-манифест обеспечивает возобновление, повтор ошибок и пропуск неизменившихся документов.
+- Полный проход завершён: 209 958 документов нормализованы без ошибок.
+- Контрольная выборка RU/KY прошла проверки текста, фрагментов, ID и SHA-256.
+- Добавлены строгий mapping и атомарный Bulk NDJSON-экспорт для OpenSearch.
 
 ### Развёртывание
 
@@ -128,6 +135,44 @@
 
 ## История изменений статуса
 
+### 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 загруженных документов.
+- Единственный отсутствующий документ `6` повторно не отдан API Минюста.
+- Добавлены mapping фрагментов и потоковый экспорт для OpenSearch Bulk API.
+- Версия backend обновлена до `0.3.0`.
+
+### 2026-08-12
+
+- Нормализатор запрещает пересекающиеся каталоги источника и результата,
+  восстанавливает прерванную публикацию и отбрасывает некорректные URL.
+- Версия backend обновлена до `0.2.2`.
+- Загрузчик пересоздаёт временный список документов на текущей странице не
+  только после HTTP 404, но и после исчерпания повторов запроса списка.
+- Версия backend обновлена до `0.2.1`.
+
+### 2026-08-10
+
+- Добавлена первая версия воспроизводимой нормализации локального архива ЦБД.
+- Добавлены атомарная публикация результатов, контрольные суммы, карантин ошибок и терминальный прогресс.
+- Версия backend обновлена до `0.2.0`.
+
 ### 2026-08-06
 
 - Добавлен терминальный прогрессбар со скоростью и расчётным временем завершения.
@@ -178,4 +223,4 @@
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.1.2 · 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.1.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

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

@@ -0,0 +1,261 @@
+# План реализации frontend поисковой СПС
+
+Основание: `docs/Функциональные_возможности_поисковой_СПС.docx`.
+
+Документ с требованиями описывает не только frontend, но и поиск, юридическую
+обработку, персональные данные, уведомления и внешние источники. Поэтому
+frontend следует начинать после появления нормализованной базы, поискового API
+и API документов.
+
+Требования необходимо адаптировать под Кыргызскую Республику: в исходном
+документе используются примеры ТК РФ и деление
+«федеральный/региональный/муниципальный», которое нельзя переносить без
+изменений.
+
+## Задачи до начала frontend-разработки
+
+### Обязательные для MVP
+
+1. Нормализовать архив Минюста:
+   - очистить HTML;
+   - выделить структуру документа и редакций;
+   - унифицировать статусы, органы, виды документов и даты;
+   - сохранить ссылки на официальный источник и дату получения;
+   - определить правила отображения документов без текста.
+2. Подготовить backend API:
+   - `GET /search`;
+   - `GET /search/filters`;
+   - `GET /documents/{code}`;
+   - `GET /documents/{code}/editions`;
+   - `GET /documents/{code}/editions/{edition}`;
+   - описание API в OpenAPI;
+   - серверную пагинацию, фильтрацию и сортировку.
+3. Развернуть поисковый сервис. Рекомендуемый вариант — self-hosted OpenSearch:
+   - отдельные поля и анализаторы для русского и кыргызского текстов;
+   - русский морфологический анализ;
+   - ICU-нормализация кыргызского текста;
+   - словари синонимов и сокращений;
+   - подсветка совпадений;
+   - индексирование всех редакций.
+4. Подготовить тестовый набор из 50–100 реальных запросов на русском и
+   кыргызском языках и вручную отметить ожидаемые результаты.
+5. Утвердить справочники:
+   - виды документов;
+   - органы принятия;
+   - статусы;
+   - уровни действия;
+   - тематический классификатор первой версии.
+
+OpenSearch имеет встроенный
+[русский морфологический анализатор](https://docs.opensearch.org/latest/analyzers/language-analyzers/russian/),
+поддерживает [синонимы и нечёткий поиск](https://docs.opensearch.org/latest/query-dsl/full-text/match/)
+и [подсветку результатов](https://docs.opensearch.org/latest/search-plugins/searching-data/highlight).
+Встроенного кыргызского морфологического анализатора в перечне нет, поэтому
+нужно отдельно проверить ICU и словари на реальных запросах.
+[ICU-анализатор](https://docs.opensearch.org/latest/analyzers/language-analyzers/icu/)
+обеспечивает Unicode-нормализацию, но сам по себе не гарантирует кыргызскую
+морфологию.
+
+### Сторонние сервисы, не нужные для MVP
+
+Их не следует подключать заранее:
+
+- Keycloak или другой OIDC-провайдер — перед закладками, папками и ролями;
+- SMTP, Telegram или Web Push — перед «документами на контроле»;
+- LibreOffice или Gotenberg — перед экспортом в Word, PDF и RTF;
+- поставщики судебной практики и экспертных комментариев — после проверки
+  лицензий;
+- источники курсов, календарей и справочных данных — перед соответствующим
+  разделом;
+- Sentry или аналог — опционально перед публичным запуском.
+
+## Граница MVP
+
+MVP — публичная справочно-поисковая система без регистрации и персональных
+функций.
+
+В MVP входят:
+
+- интерфейс на русском и кыргызском языках;
+- строка полнотекстового поиска;
+- исправление распространённых опечаток;
+- базовые синонимы и сокращения;
+- список результатов с подсвеченными фрагментами;
+- фильтры по языку, виду документа, органу, статусу и дате;
+- сортировка по релевантности и дате;
+- пагинация;
+- карточка документа;
+- актуальная редакция, статус и дата актуальности;
+- переключение между доступными языками;
+- поиск внутри открытого документа;
+- список редакций и открытие выбранной редакции;
+- ссылка на официальный источник и сведения о происхождении данных;
+- адаптивность, доступность, состояния загрузки и ошибок.
+
+Сравнение редакций, аккаунты, заметки, уведомления, RAG и судебная практика в
+MVP не входят.
+
+Интерфейс не должен предполагать наличие обоих языков. На момент полного
+скачивания архива распределение следующее:
+
+- только русский язык — 29 433 документа;
+- только кыргызский язык — 98 905 документов;
+- оба языка — 80 930 документов;
+- нет HTML-текста — 690 документов.
+
+## Рекомендуемая основа frontend
+
+- Next.js App Router и TypeScript;
+- CSS Modules с BEM-именованием;
+- дизайн-токены для цветов, отступов, типографики и состояний;
+- серверный `fetch` и URL-параметры вместо отдельного глобального хранилища;
+- Playwright для основных пользовательских сценариев;
+- адаптивный web-интерфейс без отдельного мобильного приложения.
+
+Next.js App Router поддерживает серверные компоненты, маршрутизацию и
+TypeScript в стандартной конфигурации. См.
+[официальную документацию](https://nextjs.org/docs/app).
+
+## Дизайн-процесс и внешние ориентиры
+
+При проектировании и проверке интерфейса используются следующие источники:
+
+- [jakubkrehel/skills](https://github.com/jakubkrehel/skills) — обязательная
+  комплексная проверка интерфейса через `better-interface`, включая UI,
+  типографику, цвета, доступность, layout и тексты;
+- [UI Skills](https://www.ui-skills.com/) — каталог практик и узких skills,
+  которые подключаются только под конкретную задачу после проверки их
+  содержания и лицензии;
+- [Refero Styles](https://styles.refero.design/) — библиотека визуальных
+  направлений и примеров `DESIGN.md` для поиска референсов.
+
+Правила применения:
+
+1. До разработки экранов выбрать в Refero не более трёх подходящих направлений
+   и на их основе утвердить одно собственное направление Акылдаша.
+2. Не копировать чужую дизайн-систему целиком. Цвета, типографика, плотность и
+   компоненты должны учитывать длинные юридические тексты, два языка и
+   доступность.
+3. Зафиксировать утверждённое направление в `frontend/DESIGN.md` и перенести
+   значения в дизайн-токены проекта.
+4. Дизайн-токены и компоненты Акылдаша являются источником истины. Внешние
+   рекомендации не могут отменять BEM, доступность, требования безопасности и
+   продуктовые ограничения проекта.
+5. Каждый завершённый пользовательский сценарий проходит `better-interface`
+   review. Перед выпуском MVP выполняется полный review поиска, фильтров и
+   просмотра документа.
+6. UI Skills используется для точечного поиска решения, а не для одновременного
+   смешивания нескольких визуальных стилей.
+
+Эти ресурсы используются на этапе проектирования и review и не являются
+runtime-зависимостями frontend. Регистрация в стороннем SaaS для MVP не нужна.
+
+## План спринтов MVP
+
+### Спринт 0 — фундамент, 1 неделя
+
+- создать `frontend/`;
+- настроить Next.js, TypeScript, lint и сборку;
+- выбрать до трёх референсов в Refero Styles и утвердить одно визуальное
+  направление;
+- создать `frontend/DESIGN.md` с правилами выбранного направления;
+- установить полный набор `jakubkrehel/skills` для проектных design review;
+- определить маршруты и типы API;
+- создать дизайн-токены;
+- реализовать базовые компоненты: кнопка, поле, селект, статус, карточка,
+  пагинация;
+- создать общий layout и двуязычную навигацию;
+- добавить footer с версиями frontend и backend;
+- подготовить макеты поиска, результатов и документа;
+- провести первый `better-interface` review макетов;
+- настроить CI.
+
+Результат: интерфейсный каркас работает на mock-ответах API.
+
+### Спринт 1 — быстрый поиск, 2 недели
+
+- главная страница с поиском;
+- интеграция с `/search`;
+- список результатов;
+- подсветка совпадений;
+- URL, которым можно поделиться;
+- переключение RU/KY;
+- состояния загрузки, отсутствия результатов и ошибки API;
+- базовая мобильная версия.
+
+Результат: пользователь может найти документ и открыть результат.
+
+### Спринт 2 — точный отбор, 2 недели
+
+- фильтры по реквизитам;
+- сортировка;
+- пагинация;
+- отображение числа результатов;
+- сброс отдельных и всех фильтров;
+- сохранение состояния в URL;
+- доступное управление с клавиатуры;
+- адаптивная панель фильтров.
+
+Результат: поддерживается быстрый и реквизитный поиск.
+
+### Спринт 3 — просмотр документа, 2 недели
+
+- заголовок, реквизиты, статус и дата актуальности;
+- безопасное отображение очищенного HTML;
+- переключение языка;
+- поиск внутри документа;
+- навигация по найденным фрагментам;
+- список редакций;
+- открытие предыдущей редакции;
+- ссылка на ЦБД Минюста;
+- печать средствами браузера.
+
+Результат: пользователь может проверить текст и его происхождение.
+
+### Спринт 4 — стабилизация и выпуск, 2 недели
+
+- сквозные тесты поиска и просмотра;
+- проверка русских, кыргызских и одноязычных документов;
+- соответствие WCAG 2.2 AA;
+- защита от внедрения небезопасного HTML;
+- проверка производительности;
+- корректные метаданные страниц;
+- обработка недоступности API;
+- production-сборка и развёртывание;
+- пользовательское тестирование на 10–15 реальных юридических задачах.
+
+Результат: публичный MVP.
+
+Оценка frontend-части после готовности API: **9 недель**.
+
+## Спринты после MVP
+
+### Спринт 5 — персональный кабинет
+
+Авторизация, закладки, заметки, подборки и сохранённые фильтры.
+
+### Спринт 6 — контроль изменений
+
+Документы на контроле, подписки на редакции и уведомления.
+
+### Спринт 7 — юридические связи
+
+Сравнение редакций, прямые и обратные ссылки, утратившие силу фрагменты.
+
+### Спринт 8 — практические материалы
+
+Формы, образцы, инструкции, чек-листы, календари и справочные данные.
+
+### Спринт 9 — расширенный анализ
+
+Судебная практика, экспертные комментарии, дерево связей и RAG с обязательными
+ссылками на источники.
+
+### Спринт 10 — корпоративные функции
+
+Роли, журналирование, API, интеграция с СЭД, расширенный экспорт и
+персонализация.
+
+---
+
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

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

@@ -0,0 +1,217 @@
+# Задание агенту: нормализация документов ЦБД Минюста КР
+
+## Цель
+
+Реализовать первую рабочую версию воспроизводимого нормализатора локального
+архива `data/minjust-cbd`. Нормализованные данные должны быть пригодны для
+последующей загрузки в OpenSearch, backend API и RAG, но подключение этих
+сервисов в текущую задачу не входит.
+
+## Текущее состояние
+
+- загрузчик находится в `backend/ingestion/minjust_cbd.py`;
+- сырой архив хранится в `data/minjust-cbd` и исключён из Git;
+- в манифесте 209 811 успешно загруженных документов;
+- 13 кодов остаются в таблице `errors` и отсутствуют в таблице `documents`;
+- документ содержит `metadata.json` и каталог `editions`;
+- редакция содержит `metadata.json`, `ru.html` и/или `ky.html`, иногда
+  изображения;
+- HTML создан Microsoft Word, может содержать некорректный
+  `<meta charset=unicode>`, служебные стили и неполную разметку;
+- файлы архива записаны загрузчиком в UTF-8;
+- часть документов одноязычная, а часть не содержит HTML-текста.
+
+## Обязательные ограничения
+
+1. Не изменять и не перезаписывать `data/minjust-cbd`.
+2. Не запускать полный проход по архиву во время автоматических тестов.
+3. Не подключать PostgreSQL, OpenSearch, OCR, embeddings, машинный перевод и
+   сетевые API.
+4. Сначала использовать стандартную библиотеку. Новая зависимость допустима
+   только если на реальных образцах доказано, что стандартный HTML-парсер не
+   обеспечивает корректность или безопасность.
+5. Все записи выполнять атомарно.
+6. Ошибка одного документа не должна останавливать длительный прогон.
+7. Повторный запуск должен пропускать неизменившиеся документы.
+8. Не изменять пользовательские файлы `logs/`, исходный DOCX и несвязанные
+   незакоммиченные изменения.
+
+## Размещение
+
+Использовать существующую backend-структуру:
+
+```text
+backend/
+  normalization/
+    minjust_cbd.py
+```
+
+Результат по умолчанию:
+
+```text
+data/minjust-normalized/
+  manifest.sqlite3
+  documents/<document_code>/document.json
+  documents/<document_code>/editions/<edition_code>/edition.json
+  documents/<document_code>/editions/<edition_code>/<lang>/content.html
+  documents/<document_code>/editions/<edition_code>/<lang>/content.txt
+  documents/<document_code>/editions/<edition_code>/<lang>/fragments.json
+```
+
+Каталог уже покрывается правилом игнорирования `data/`.
+
+## Канонические данные
+
+### `document.json`
+
+Сохранить как минимум:
+
+- `schema_version`;
+- `source_code`;
+- двуязычные `class`, `type`, `title`, `name`, `status`;
+- номера и даты без юридически неподтверждённых выводов;
+- флаги публичности;
+- органы, публикации, ключевые слова и классификаторы с сохранением дерева;
+- пути листьев иерархий для будущих фильтров;
+- ссылки из `References` без выдумывания связей;
+- `available_languages`;
+- список редакций;
+- путь к источнику и SHA-256 исходных файлов;
+- версию нормализатора и время обработки.
+
+Пустые строки привести к `null`, но не переводить значения и не заменять
+официальные формулировки собственными.
+
+### `edition.json`
+
+Сохранить:
+
+- исходный код редакции;
+- двуязычное название;
+- исходный тип;
+- доступные языки;
+- изображения без бинарных данных;
+- контрольные суммы источников;
+- признаки качества.
+
+Не считать дату из `Name` датой вступления редакции в силу и не назначать
+актуальную редакцию без подтверждённого правила источника.
+
+### Языковой вариант
+
+Для каждого имеющегося `ru.html` или `ky.html` сформировать:
+
+- `content.html` — безопасный HTML для frontend;
+- `content.txt` — извлечённый текст с сохранением смысловых переносов;
+- `fragments.json` — упорядоченные адресуемые блоки.
+
+Сырой HTML всегда читать как UTF-8, не доверяя его meta charset.
+
+## Очистка HTML
+
+- удалить `script`, `style`, `meta`, `link`, комментарии и служебные элементы;
+- удалить обработчики событий, inline-стили и опасные URL;
+- разрешить минимальный набор структурных тегов: заголовки, абзацы, `pre`,
+  списки, таблицы, безопасные ссылки, изображения и базовое текстовое
+  выделение;
+- нормализовать Unicode в NFC;
+- преобразовать неразрывные пробелы и избыточные пробелы только в
+  `content.txt`, не искажая отображаемый юридический текст;
+- не загружать внешние ресурсы;
+- относительные изображения связывать только с файлами внутри редакции;
+- неизвестную или сломанную разметку сохранять как текст, а не терять молча.
+
+## Фрагменты
+
+Минимальная версия должна создавать фрагмент для каждого содержательного
+блочного элемента. Распознавание статей и пунктов допускается только простыми
+проверяемыми правилами RU/KY; обычный абзац является fallback.
+
+Каждый фрагмент содержит:
+
+- стабильный `id`;
+- `document_code`, `edition_code`, `language`;
+- порядковую позицию;
+- тип блока;
+- чистый текст;
+- SHA-256 текста.
+
+Идентификатор должен быть детерминированным и включать документ, редакцию,
+язык и позицию. Не добавлять сложное сопоставление фрагментов между
+редакциями — это отдельная будущая задача.
+
+## Манифест и возобновление
+
+SQLite-манифест должен хранить:
+
+- код документа;
+- SHA-256 набора исходных файлов;
+- версию схемы и нормализатора;
+- время успешной обработки;
+- состояние и текст последней ошибки.
+
+Если checksum и версия нормализатора не изменились, документ пропускается.
+После успешной повторной обработки ошибка удаляется. Добавить `--limit` и
+понятный терминальный прогресс, пригодный для долгого запуска.
+
+## CLI и импорт из будущего backend
+
+CLI запускается из корня репозитория:
+
+```bash
+python3 backend/normalization/minjust_cbd.py
+```
+
+Предусмотреть параметры:
+
+- `--input`;
+- `--output`;
+- `--limit`;
+- `--refresh`;
+- `--log-level`.
+
+Основную функцию можно импортировать без запуска CLI. Не создавать
+планировщик, очередь задач или framework-интеграцию.
+
+## Проверки
+
+Добавить один компактный тестовый модуль, который проверяет:
+
+1. русскую редакцию;
+2. кыргызскую редакцию;
+3. двуязычную редакцию;
+4. Word HTML с опасным `script`, inline-стилем и `javascript:` URL;
+5. документ без HTML;
+6. повторный запуск и пропуск неизменившегося документа;
+7. продолжение после ошибки одного документа;
+8. детерминированные fragment ID и checksums.
+
+Провести пилотный read-only запуск на небольшой реальной выборке через
+`--limit`, не нормализовать весь архив в рамках разработки.
+
+## Документация и версия
+
+- описать запуск, структуру результата и ограничения в `backend/README.md`;
+- отметить реализацию в `docs/operations/project-status.md`;
+- это новая обратно совместимая backend-функция: увеличить minor-версию
+  backend по SemVer;
+- обновить все отображаемые backend-версии и footer, не меняя версию
+  Telegram-бота;
+- frontend по-прежнему помечать как не созданный.
+
+## Критерии приёмки
+
+- сырой архив не изменён;
+- тесты проходят;
+- `git diff --check` проходит;
+- пилотный запуск завершается без остановки на отдельных ошибках;
+- повторный пилотный запуск пропускает неизменившиеся документы;
+- unsafe HTML не попадает в `content.html`;
+- каждый фрагмент прослеживается до документа, редакции, языка и исходного
+  файла;
+- отсутствуют молча потерянные HTML или ошибки;
+- реализация не содержит PostgreSQL/OpenSearch/RAG-кода «на будущее».
+
+---
+
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан

docs/product/project-overview.md

@@ -44,4 +44,4 @@
 
 ---
 
-Акылдаш · Telegram-бот v0.2.2 · Backend v0.1.2 · 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.1.2 · 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.1.2 · Frontend — не создан
+Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан