diff --git a/README.md b/README.md index a82f802..dc37a6a 100644 --- a/README.md +++ b/README.md @@ -10,15 +10,15 @@ Telegram-бот — только часть рабочего окружения ## Текущее состояние Сейчас реализованы Telegram-бот-секретарь версии `0.2.2` и backend версии -`0.3.0`: возобновляемая выгрузка, нормализация и подготовка индекса документов +`0.4.1`: возобновляемая загрузка индекса документов Министерства юстиции ЦБД Минюста КР. | Компонент | Версия | Состояние | |---|---:|---| | Telegram-бот | `0.2.2` | на Synology работает `0.2.1`; обновление после слияния | -| Backend | `0.3.0` | реализованы выгрузка, нормализация и экспорт для OpenSearch | +| Backend | `0.4.1` | добавлено продолжение прерванной Bulk-загрузки | | Frontend | — | ещё не создан | -| Сбор и обработка правовых данных | `0.3.0` | добавлены mapping и Bulk NDJSON для OpenSearch | +| Сбор и обработка правовых данных | `0.4.1` | добавлено продолжение загрузки существующего индекса | | RAG и база знаний | — | ещё не созданы | ## Структура репозитория @@ -59,4 +59,4 @@ python3 -m unittest discover -s tools/telegram-bot -v --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан +Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан diff --git a/backend/README.md b/backend/README.md index 9dba1da..8a13be4 100644 --- a/backend/README.md +++ b/backend/README.md @@ -1,6 +1,6 @@ # Backend Акылдаш -Версия: `0.3.0` +Версия: `0.4.1` Первая backend-область проекта — загрузка правовых документов из официального Open Data API ЦБД Минюста Кыргызской Республики. Код расположен в @@ -114,10 +114,57 @@ python3 backend/search/minjust_opensearch.py По умолчанию создаётся `data/opensearch/minjust-fragments.ndjson`. Экспорт атомарный и детерминированный; для проверки можно передать `--limit 1`. -Полный файл может быть большим, поэтому перед запуском нужно проверить -свободное место. Загружать данные следует в новый версионный индекс и после -проверки переключать alias, чтобы удалённые фрагменты не оставались в поиске. +Для прямой загрузки без большого промежуточного файла используется `--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/.checkpoint.json`; путь можно изменить через +`--checkpoint`. Checkpoint привязан к URL, cluster UUID, index UUID, `--limit` +и SHA-256 нормализованного manifest. Resume отклоняется при любом несовпадении: +для обновлённого корпуса или пересозданного индекса нужно создать новый +версионный индекс, проверить его и переключить alias. Это не оставляет +удалённые trailing-фрагменты старых документов. + +## Локальный OpenSearch + +Стенд использует один узел OpenSearch без Dashboards, устанавливает +`analysis-icu`, выделяет JVM 8 ГБ и доступен только на `127.0.0.1:9200`. +Индекс хранится в `data/opensearch-node` на диске проекта. + +```bash +sudo sysctl -w vm.max_map_count=262144 +docker compose -f deploy/local-opensearch/compose.yaml up -d --build +curl http://127.0.0.1:9200/_cluster/health +``` + +Security plugin отключён только для локальной разработки; этот compose нельзя +публиковать в сеть или использовать в production. Mapping локального стенда +также задаёт одну shard и ноль replicas; для production число shard следует +рассчитать по размеру корпуса и настроить не менее одной replica. --- -Акылдаш · Backend v0.3.0 · Frontend — не создан +Акылдаш · Backend v0.4.1 · Frontend — не создан diff --git a/backend/ingestion/minjust_cbd.py b/backend/ingestion/minjust_cbd.py index 39ed986..d4122d4 100644 --- a/backend/ingestion/minjust_cbd.py +++ b/backend/ingestion/minjust_cbd.py @@ -21,7 +21,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Callable, Iterable -APP_VERSION = "0.3.0" +APP_VERSION = "0.4.1" API_BASE_URL = "https://cbd.minjust.gov.kg/api/v1/OpenData/" LANGUAGES = {"Rus": "ru", "Kyr": "ky"} IMAGE_LANGUAGES = {"Russian": "ru", "Kyrgyz": "ky"} diff --git a/backend/normalization/minjust_cbd.py b/backend/normalization/minjust_cbd.py index f1bad20..a41ca9c 100644 --- a/backend/normalization/minjust_cbd.py +++ b/backend/normalization/minjust_cbd.py @@ -23,7 +23,7 @@ from pathlib import Path from typing import Callable from urllib.parse import urlsplit -APP_VERSION = "0.3.0" +APP_VERSION = "0.4.1" SCHEMA_VERSION = "1" NORMALIZER_VERSION = "1.0.0" LANGUAGES = ("ru", "ky") diff --git a/backend/search/minjust-fragments-index.json b/backend/search/minjust-fragments-index.json index f7f1ca6..daa9381 100644 --- a/backend/search/minjust-fragments-index.json +++ b/backend/search/minjust-fragments-index.json @@ -1,4 +1,11 @@ { + "settings": { + "index": { + "number_of_shards": 1, + "number_of_replicas": 0, + "refresh_interval": "30s" + } + }, "mappings": { "dynamic": "strict", "properties": { diff --git a/backend/search/minjust_opensearch.py b/backend/search/minjust_opensearch.py index 8158696..d6d3ae0 100644 --- a/backend/search/minjust_opensearch.py +++ b/backend/search/minjust_opensearch.py @@ -8,18 +8,30 @@ import json import os import sqlite3 import tempfile +import time +import urllib.error +import urllib.parse +import urllib.request from pathlib import Path +from typing import Iterator -APP_VERSION = "0.3.0" +APP_VERSION = "0.4.1" LANGUAGES = {"ru", "ky"} +DEFAULT_MAPPING = Path(__file__).with_name("minjust-fragments-index.json") def read_json(path: Path): def reject_constant(value: str): raise ValueError(f"Invalid JSON constant: {value}") - with path.open(encoding="utf-8") as source: - return json.load(source, parse_constant=reject_constant) + 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): @@ -73,49 +85,286 @@ def search_document(document: dict, fragment: dict, expected: tuple[str, str, st return {key: value for key, value in result.items() if value is not None} +def document_codes(input_root: Path, limit: int | None = None, start_at: str | None = None) -> list[str]: + connection = sqlite3.connect(f"file:{input_root / 'manifest.sqlite3'}?mode=ro", uri=True) + try: + query = "SELECT code FROM documents WHERE state = 'success' ORDER BY CAST(code AS INTEGER), code" + parameters: list[int] = [] + if limit is not None: + query += " LIMIT ?" + parameters.append(limit) + codes = [str(code) for (code,) in connection.execute(query, parameters)] + finally: + connection.close() + if start_at is None: + return codes + try: + return codes[codes.index(start_at):] + except ValueError as error: + raise ValueError(f"Resume document not found in selected range: {start_at}") from error + + +def bulk_pairs( + input_root: Path, + index: str, + limit: int | None = None, + start_at: str | None = None, +) -> Iterator[tuple[str, bytes]]: + document_root = input_root / "documents" + if not document_root.is_dir(): + raise FileNotFoundError(f"Document directory not found: {document_root}") + for code in document_codes(input_root, limit, start_at): + directory = document_root / code + document = read_json(directory / "document.json") + if document.get("source_code") != directory.name: + raise ValueError(f"Document identity does not match its path: {directory}") + for fragment_path in sorted(directory.glob("editions/*/*/fragments.json")): + edition_code, language = fragment_path.parts[-3:-1] + values = read_json(fragment_path) + if not isinstance(values, list): + raise ValueError(f"Fragments must be a list: {fragment_path}") + for position, fragment in enumerate(values, 1): + source = search_document(document, fragment, (directory.name, edition_code, language, position)) + action = json.dumps({"index": {"_index": index, "_id": fragment["id"]}}, ensure_ascii=False, allow_nan=False) + body = json.dumps(source, ensure_ascii=False, allow_nan=False) + yield directory.name, f"{action}\n{body}\n".encode() + + +def document_count(input_root: Path, limit: int | None, start_at: str | None = None) -> int: + return len(document_codes(input_root, limit, start_at)) + + +def bulk_batches(pairs: Iterator[tuple[str, bytes]], maximum_bytes: int) -> Iterator[tuple[str, bytes]]: + batch = bytearray() + last_code = "" + for code, pair in pairs: + if len(pair) > maximum_bytes: + raise ValueError(f"One Bulk pair exceeds the {maximum_bytes}-byte batch limit") + if batch and len(batch) + len(pair) > maximum_bytes: + yield last_code, bytes(batch) + batch.clear() + last_code = code + batch.extend(pair) + if batch: + yield last_code, bytes(batch) + + def export_bulk(input_root: Path, output: Path, index: str, limit: int | None = None) -> tuple[int, int]: source = input_root.resolve() destination = output.resolve() if destination == source or destination.is_relative_to(source): raise ValueError("--output must not be inside --input") - document_root = input_root / "documents" - if not document_root.is_dir(): - raise FileNotFoundError(f"Document directory not found: {document_root}") - connection = sqlite3.connect(f"file:{input_root / 'manifest.sqlite3'}?mode=ro", uri=True) - query = "SELECT code FROM documents WHERE state = 'success' ORDER BY CAST(code AS INTEGER), code" - parameters: tuple[int, ...] = () - if limit is not None: - query += " LIMIT ?" - parameters = (limit,) output.parent.mkdir(parents=True, exist_ok=True) - documents = fragments = 0 - with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=output.parent, delete=False) as target: + fragments = 0 + with tempfile.NamedTemporaryFile("wb", dir=output.parent, delete=False) as target: temporary = Path(target.name) try: - for (code,) in connection.execute(query, parameters): - directory = document_root / str(code) - document = read_json(directory / "document.json") - if document.get("source_code") != directory.name: - raise ValueError(f"Document identity does not match its path: {directory}") - documents += 1 - for fragment_path in sorted(directory.glob("editions/*/*/fragments.json")): - edition_code, language = fragment_path.parts[-3:-1] - values = read_json(fragment_path) - if not isinstance(values, list): - raise ValueError(f"Fragments must be a list: {fragment_path}") - for position, fragment in enumerate(values, 1): - source = search_document(document, fragment, (directory.name, edition_code, language, position)) - target.write(json.dumps({"index": {"_index": index, "_id": fragment["id"]}}, ensure_ascii=False, allow_nan=False) + "\n") - target.write(json.dumps(source, ensure_ascii=False, allow_nan=False) + "\n") - fragments += 1 + 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 - finally: - connection.close() + 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', '')}: {item['error'].get('type', 'error')}: " + f"{item['error'].get('reason', '')}" + for item in failures[:5] + ) + raise RuntimeError( + f"OpenSearch Bulk API failed for {len(failures)} item(s); " + f"resume from {checkpoint}: {details}" + ) + if len(items) != expected: + raise RuntimeError( + f"OpenSearch Bulk API returned {len(items)} of {expected} item result(s); " + f"resume from {checkpoint}" + ) + fragments += len(items) + state["last_document_code"] = last_code + write_json_atomic(checkpoint, state) + print(f"checkpoint={last_code} fragments={fragments}", flush=True) + state["complete"] = True + write_json_atomic(checkpoint, state) return documents, fragments @@ -125,12 +374,38 @@ def main() -> int: parser.add_argument("--output", type=Path, default=Path("data/opensearch/minjust-fragments.ndjson")) parser.add_argument("--index", default="akyldash-fragments-v1") parser.add_argument("--limit", type=int) + parser.add_argument("--url", help="create the index and stream bounded Bulk requests instead of writing a file") + parser.add_argument("--mapping", type=Path, default=DEFAULT_MAPPING) + parser.add_argument("--batch-mb", type=int, default=25) + parser.add_argument("--resume", action="store_true", help="load into an existing index") + parser.add_argument("--checkpoint", type=Path, help="persistent resume checkpoint path") parser.add_argument("--version", action="version", version=APP_VERSION) arguments = parser.parse_args() if arguments.limit is not None and arguments.limit <= 0: raise SystemExit("--limit must be greater than zero") - documents, fragments = export_bulk(arguments.input, arguments.output, arguments.index, arguments.limit) - print(f"documents={documents} fragments={fragments} output={arguments.output}\nAkyldash Backend v{APP_VERSION} · Frontend — not created") + 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 diff --git a/backend/test_minjust_opensearch.py b/backend/test_minjust_opensearch.py index f29d167..db76c75 100644 --- a/backend/test_minjust_opensearch.py +++ b/backend/test_minjust_opensearch.py @@ -1,11 +1,15 @@ import hashlib +import io import json import sqlite3 import tempfile import unittest +import urllib.error +from contextlib import closing from pathlib import Path +from unittest.mock import patch -from search.minjust_opensearch import export_bulk +from search.minjust_opensearch import bulk_batches, document_codes, export_bulk, load_bulk, request_json class MinjustOpenSearchTest(unittest.TestCase): @@ -14,9 +18,11 @@ class MinjustOpenSearchTest(unittest.TestCase): root = Path(temporary) document_root = root / "normalized/documents/7" document_root.mkdir(parents=True) - with sqlite3.connect(root / "normalized/manifest.sqlite3") as connection: - connection.execute("CREATE TABLE documents (code TEXT, state TEXT)") - connection.execute("INSERT INTO documents VALUES ('7', 'success')") + 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( { @@ -33,6 +39,11 @@ class MinjustOpenSearchTest(unittest.TestCase): ), encoding="utf-8", ) + empty = root / "normalized/documents/8" + empty.mkdir() + (empty / "document.json").write_text( + json.dumps({"schema_version": "1", "source_code": "8"}), encoding="utf-8" + ) for language, text in (("ru", "Текст \"RU\"\nстрока"), ("ky", "Кыргызча текст")): path = document_root / f"editions/10/{language}/fragments.json" path.parent.mkdir(parents=True) @@ -56,7 +67,7 @@ class MinjustOpenSearchTest(unittest.TestCase): ) output = root / "bulk.ndjson" - self.assertEqual(export_bulk(root / "normalized", output, "test-index"), (1, 2)) + self.assertEqual(export_bulk(root / "normalized", output, "test-index"), (2, 2)) content = output.read_bytes() self.assertTrue(content.endswith(b"\n")) lines = [json.loads(line) for line in content.splitlines()] @@ -65,6 +76,186 @@ class MinjustOpenSearchTest(unittest.TestCase): self.assertIn("text_ky", lines[1]) self.assertNotIn("text_ru", lines[1]) self.assertEqual(lines[3]["text_ru"], "Текст \"RU\"\nстрока") + self.assertEqual( + list(bulk_batches(iter((("7", b"a\nb\n"), ("8", b"c\nd\n"))), 4)), + [("7", b"a\nb\n"), ("8", b"c\nd\n")], + ) + self.assertEqual(document_codes(root / "normalized", 2, "8"), ["8"]) + with self.assertRaisesRegex(ValueError, "selected range"): + document_codes(root / "normalized", 1, "8") + + checkpoint = root / "checkpoint.json" + with patch("search.minjust_opensearch.request_json") as request: + request.side_effect = [ + {}, + {"cluster_uuid": "cluster-1"}, + {"test-index": {"settings": {"index": {"uuid": "index-1"}}}}, + {"errors": False, "items": [{"index": {}}, {"index": {}}]}, + ] + self.assertEqual( + load_bulk( + root / "normalized", + "http://127.0.0.1:9200", + "test-index", + maximum_bytes=4096, + checkpoint=checkpoint, + ), + (2, 2), + ) + self.assertEqual(request.call_args_list[-1].args[3], "application/x-ndjson") + state = json.loads(checkpoint.read_text(encoding="utf-8")) + self.assertEqual(state["last_document_code"], "7") + self.assertTrue(state["complete"]) + + state["last_document_code"] = "8" + state["complete"] = False + checkpoint.write_text(json.dumps(state), encoding="utf-8") + with patch("search.minjust_opensearch.request_json") as request: + request.side_effect = [ + {"cluster_uuid": "cluster-2"}, + {"test-index": {"settings": {"index": {"uuid": "index-1"}}}}, + ] + with self.assertRaisesRegex(ValueError, "does not match"): + load_bulk( + root / "normalized", + "http://127.0.0.1:9200", + "test-index", + maximum_bytes=4096, + resume=True, + checkpoint=checkpoint, + ) + with patch("search.minjust_opensearch.request_json") as request: + request.side_effect = [ + {"cluster_uuid": "cluster-1"}, + {"test-index": {"settings": {"index": {"uuid": "index-1"}}}}, + ] + with self.assertRaisesRegex(ValueError, "limit"): + load_bulk( + root / "normalized", + "http://127.0.0.1:9200", + "test-index", + maximum_bytes=4096, + limit=1, + resume=True, + checkpoint=checkpoint, + ) + with patch("search.minjust_opensearch.request_json") as request: + request.side_effect = [ + {"cluster_uuid": "cluster-1"}, + {"test-index": {"settings": {"index": {"uuid": "index-1"}}}}, + ] + self.assertEqual( + load_bulk( + root / "normalized", + "http://127.0.0.1:9200", + "test-index", + maximum_bytes=4096, + resume=True, + checkpoint=checkpoint, + ), + (1, 0), + ) + self.assertTrue(all(call.args[1] == "GET" for call in request.call_args_list)) + + state["last_document_code"] = "9" + state["complete"] = False + checkpoint.write_text(json.dumps(state), encoding="utf-8") + with patch("search.minjust_opensearch.request_json") as request: + request.side_effect = [ + {"cluster_uuid": "cluster-1"}, + {"test-index": {"settings": {"index": {"uuid": "index-1"}}}}, + ] + with self.assertRaisesRegex(ValueError, "Resume document not found"): + load_bulk( + root / "normalized", + "http://127.0.0.1:9200", + "test-index", + maximum_bytes=4096, + resume=True, + checkpoint=checkpoint, + ) + + failed_checkpoint = root / "failed-checkpoint.json" + failure = { + "errors": True, + "items": [{"index": {"_id": "bad-id", "error": {"type": "mapper", "reason": "bad value"}}}], + } + failed_requests = [ + {}, + {"cluster_uuid": "cluster-1"}, + {"failed-index": {"settings": {"index": {"uuid": "index-2"}}}}, + failure, + ] + with patch("search.minjust_opensearch.request_json", side_effect=failed_requests): + with self.assertRaisesRegex(RuntimeError, "bad-id: mapper: bad value"): + load_bulk( + root / "normalized", + "http://127.0.0.1:9200", + "failed-index", + maximum_bytes=4096, + checkpoint=failed_checkpoint, + ) + failed_state = json.loads(failed_checkpoint.read_text(encoding="utf-8")) + self.assertIsNone(failed_state["last_document_code"]) + self.assertFalse(failed_state["complete"]) + + state["last_document_code"] = "8" + state["complete"] = False + checkpoint.write_text(json.dumps(state), encoding="utf-8") + with closing(sqlite3.connect(root / "normalized/manifest.sqlite3")) as connection: + with connection: + connection.execute("INSERT INTO documents VALUES ('9', 'success')") + with patch("search.minjust_opensearch.request_json") as request: + request.side_effect = [ + {"cluster_uuid": "cluster-1"}, + {"test-index": {"settings": {"index": {"uuid": "index-1"}}}}, + ] + with self.assertRaisesRegex(ValueError, "manifest changed"): + load_bulk( + root / "normalized", + "http://127.0.0.1:9200", + "test-index", + maximum_bytes=4096, + resume=True, + checkpoint=checkpoint, + ) + + http_error = urllib.error.HTTPError( + "http://127.0.0.1:9200/test", + 429, + "busy", + {}, + io.BytesIO(b"busy"), + ) + with ( + patch("search.minjust_opensearch.urllib.request.urlopen", side_effect=[http_error, io.BytesIO(b"{}")]), + patch("search.minjust_opensearch.time.sleep") as sleep, + ): + self.assertEqual( + request_json("http://127.0.0.1:9200/test", "GET", None, "application/json", attempts=2), + {}, + ) + sleep.assert_called_once_with(1) + + with ( + patch( + "search.minjust_opensearch.urllib.request.urlopen", + side_effect=[io.BytesIO(b"{"), io.BytesIO(b"{}")], + ), + patch("search.minjust_opensearch.time.sleep") as sleep, + ): + self.assertEqual( + request_json( + "http://127.0.0.1:9200/_bulk", + "POST", + b"{}\n{}\n", + "application/x-ndjson", + attempts=2, + retry_invalid_json=True, + ), + {}, + ) + sleep.assert_called_once_with(1) bad = document_root / "editions/10/ru/fragments.json" fragments = json.loads(bad.read_text(encoding="utf-8")) @@ -83,9 +274,17 @@ class MinjustOpenSearchTest(unittest.TestCase): export_bulk(root / "normalized", root / "normalized/manifest.sqlite3", "test-index") self.assertEqual(output.read_bytes(), content) - mapping = json.loads( + bad.write_text("", encoding="utf-8") + with self.assertRaisesRegex(ValueError, "fragments.json"): + export_bulk(root / "normalized", output, "test-index") + self.assertEqual(output.read_bytes(), content) + + definition = json.loads( (Path(__file__).parent / "search/minjust-fragments-index.json").read_text(encoding="utf-8") - )["mappings"] + ) + mapping = definition["mappings"] + self.assertEqual(definition["settings"]["index"]["number_of_shards"], 1) + self.assertEqual(definition["settings"]["index"]["number_of_replicas"], 0) self.assertEqual(mapping["dynamic"], "strict") self.assertEqual(mapping["properties"]["position"]["type"], "integer") self.assertEqual(mapping["properties"]["text_ru"]["analyzer"], "russian") diff --git a/deploy/local-opensearch/Dockerfile b/deploy/local-opensearch/Dockerfile new file mode 100644 index 0000000..d1f050b --- /dev/null +++ b/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 diff --git a/deploy/local-opensearch/compose.yaml b/deploy/local-opensearch/compose.yaml new file mode 100644 index 0000000..eb322a3 --- /dev/null +++ b/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 diff --git a/docs/README.md b/docs/README.md index 1443fbc..b054064 100644 --- a/docs/README.md +++ b/docs/README.md @@ -33,4 +33,4 @@ --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан +Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан diff --git a/docs/decisions/001-telegram-workspace-mvp.md b/docs/decisions/001-telegram-workspace-mvp.md index 38b12dc..5d3632c 100644 --- a/docs/decisions/001-telegram-workspace-mvp.md +++ b/docs/decisions/001-telegram-workspace-mvp.md @@ -78,4 +78,4 @@ Telegram позволяет запретить пользователям отп --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан +Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан diff --git a/docs/operations/project-status.md b/docs/operations/project-status.md index 69112b0..3711e22 100644 --- a/docs/operations/project-status.md +++ b/docs/operations/project-status.md @@ -1,19 +1,19 @@ # Статус проекта -Последняя проверка: 2026-08-13 +Последняя проверка: 2026-08-14 Назначение документа: быстро восстановить контекст проекта для участников команды и будущих агентов. - Telegram-бот: `0.2.2` - Telegram-бот на Synology: `0.2.1` -- Backend: `0.3.0` +- Backend: `0.4.1` - Frontend: не создан ## Краткий итог Репозиторий переориентирован с отдельного бота на весь проект юридической информационно-аналитической платформы. Telegram-бот выделен в инструмент рабочего окружения. Реализованы возобновляемая выгрузка документов из официального Open Data API ЦБД Минюста КР и их локальная воспроизводимая нормализация. -Ближайшая цель — развернуть проверяемый OpenSearch с `analysis-icu`, загрузить -подготовленный Bulk NDJSON и оценить поиск на 50–100 запросах RU/KY. +Ближайшая цель — оценить полный локальный индекс на 50–100 запросах RU/KY и +настроить ранжирование до начала разработки поискового API. ## Уже сделано @@ -135,6 +135,22 @@ ## История изменений статуса +### 2026-08-14 + +- Полный локальный индекс содержит 56 295 965 фрагментов и успешно отвечает на + RU/KY-запросы. +- Добавлены атомарный checkpoint и безопасное продолжение прерванной загрузки + существующего индекса. +- Добавлены retry/backoff для временных HTTP-сбоев и подробные ошибки Bulk API. +- Ошибки чтения JSON теперь содержат точный путь и повторяются при временном сбое. +- Версия backend обновлена до `0.4.1`. + +### 2026-08-13 + +- Добавлен локальный одноузловой OpenSearch с `analysis-icu` без Dashboards. +- Добавлена прямая потоковая загрузка корпуса пакетами до 25 МБ. +- Версия backend обновлена до `0.4.0`. + ### 2026-08-13 - Подтверждена полная успешная нормализация 209 958 загруженных документов. @@ -207,4 +223,4 @@ --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан +Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан diff --git a/docs/operations/telegram-workspace-plan.md b/docs/operations/telegram-workspace-plan.md index 0afb44b..57be505 100644 --- a/docs/operations/telegram-workspace-plan.md +++ b/docs/operations/telegram-workspace-plan.md @@ -398,4 +398,4 @@ Git сохраняет актуальную версию --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан +Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан diff --git a/docs/product/frontend-search-sps-plan.md b/docs/product/frontend-search-sps-plan.md index 3e92a4a..392c2d1 100644 --- a/docs/product/frontend-search-sps-plan.md +++ b/docs/product/frontend-search-sps-plan.md @@ -258,4 +258,4 @@ runtime-зависимостями frontend. Регистрация в стор --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан +Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан diff --git a/docs/product/minjust-document-normalization-agent-task.md b/docs/product/minjust-document-normalization-agent-task.md index b93bc66..7d71fa2 100644 --- a/docs/product/minjust-document-normalization-agent-task.md +++ b/docs/product/minjust-document-normalization-agent-task.md @@ -214,4 +214,4 @@ python3 backend/normalization/minjust_cbd.py --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан +Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан diff --git a/docs/product/project-overview.md b/docs/product/project-overview.md index e3d9313..2c6e54b 100644 --- a/docs/product/project-overview.md +++ b/docs/product/project-overview.md @@ -44,4 +44,4 @@ --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан +Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан diff --git a/docs/team/ai-skills-for-beginners.md b/docs/team/ai-skills-for-beginners.md index a231dc4..a615a97 100644 --- a/docs/team/ai-skills-for-beginners.md +++ b/docs/team/ai-skills-for-beginners.md @@ -180,4 +180,4 @@ --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан +Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан diff --git a/tools/telegram-bot/README.md b/tools/telegram-bot/README.md index 686c3e9..720903c 100644 --- a/tools/telegram-bot/README.md +++ b/tools/telegram-bot/README.md @@ -48,4 +48,4 @@ python3 -m unittest -v --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.3.0 · Frontend — не создан +Акылдаш · Telegram-бот v0.2.2 · Backend v0.4.1 · Frontend — не создан