"""Export normalized Ministry fragments for the OpenSearch Bulk API.""" from __future__ import annotations import argparse import hashlib import json import os import sqlite3 import tempfile from pathlib import Path APP_VERSION = "0.3.0" LANGUAGES = {"ru", "ky"} def read_json(path: Path): def reject_constant(value: str): raise ValueError(f"Invalid JSON constant: {value}") with path.open(encoding="utf-8") as source: return json.load(source, parse_constant=reject_constant) def localized(value: object, language: str): return value.get(language) if isinstance(value, dict) else None def paths(document: dict, field: str, language: str) -> list[str]: return [" > ".join(item[language]) for item in document.get(field, []) if item.get(language)] def search_document(document: dict, fragment: dict, expected: tuple[str, str, str, int]) -> dict: document_code, edition_code, language, position = expected fragment_id = f"document:{document_code}:edition:{edition_code}:lang:{language}:fragment:{position}" required = ("id", "document_code", "edition_code", "language", "position", "type", "text", "text_sha256", "source_path", "source_sha256") if not isinstance(fragment, dict) or any(key not in fragment for key in required): raise ValueError(f"Incomplete fragment {fragment_id}") if (fragment["document_code"], fragment["edition_code"], fragment["language"], fragment["position"], fragment["id"]) != (*expected, fragment_id): raise ValueError(f"Fragment identity does not match its path: {fragment_id}") if language not in LANGUAGES or not isinstance(fragment["text"], str) or not fragment["text"]: raise ValueError(f"Invalid fragment content: {fragment_id}") for key in ("text_sha256", "source_sha256"): value = fragment[key] if not isinstance(value, str) or len(value) != 64 or any(char not in "0123456789abcdef" for char in value): raise ValueError(f"Invalid {key}: {fragment_id}") if hashlib.sha256(fragment["text"].encode()).hexdigest() != fragment["text_sha256"]: raise ValueError(f"Text checksum mismatch: {fragment_id}") dates = document.get("dates") or {} result = { "schema_version": document["schema_version"], "document_code": document_code, "edition_code": edition_code, "language": language, "position": position, "fragment_type": fragment["type"], f"text_{language}": fragment["text"], "document_name_ru": localized(document.get("name"), "ru"), "document_name_ky": localized(document.get("name"), "ky"), "document_type_ru": localized(document.get("type"), "ru"), "document_type_ky": localized(document.get("type"), "ky"), "status_ru": localized(document.get("status"), "ru"), "status_ky": localized(document.get("status"), "ky"), "number": document.get("number"), "date_adopted": dates.get("DateAdopted"), "authority_paths_ru": paths(document, "authority_paths", "ru"), "authority_paths_ky": paths(document, "authority_paths", "ky"), "source_path": fragment["source_path"], "source_sha256": fragment["source_sha256"], "text_sha256": fragment["text_sha256"], } return {key: value for key, value in result.items() if value is not None} def export_bulk(input_root: Path, output: Path, index: str, limit: int | None = None) -> tuple[int, int]: source = input_root.resolve() destination = output.resolve() if destination == source or destination.is_relative_to(source): raise ValueError("--output must not be inside --input") document_root = input_root / "documents" if not document_root.is_dir(): raise FileNotFoundError(f"Document directory not found: {document_root}") connection = sqlite3.connect(f"file:{input_root / 'manifest.sqlite3'}?mode=ro", uri=True) query = "SELECT code FROM documents WHERE state = 'success' ORDER BY CAST(code AS INTEGER), code" parameters: tuple[int, ...] = () if limit is not None: query += " LIMIT ?" parameters = (limit,) output.parent.mkdir(parents=True, exist_ok=True) documents = fragments = 0 with tempfile.NamedTemporaryFile("w", encoding="utf-8", dir=output.parent, delete=False) as target: temporary = Path(target.name) try: for (code,) in connection.execute(query, parameters): directory = document_root / str(code) document = read_json(directory / "document.json") if document.get("source_code") != directory.name: raise ValueError(f"Document identity does not match its path: {directory}") documents += 1 for fragment_path in sorted(directory.glob("editions/*/*/fragments.json")): edition_code, language = fragment_path.parts[-3:-1] values = read_json(fragment_path) if not isinstance(values, list): raise ValueError(f"Fragments must be a list: {fragment_path}") for position, fragment in enumerate(values, 1): source = search_document(document, fragment, (directory.name, edition_code, language, position)) target.write(json.dumps({"index": {"_index": index, "_id": fragment["id"]}}, ensure_ascii=False, allow_nan=False) + "\n") target.write(json.dumps(source, ensure_ascii=False, allow_nan=False) + "\n") fragments += 1 target.flush() os.fsync(target.fileno()) os.replace(temporary, output) except BaseException: temporary.unlink(missing_ok=True) raise finally: connection.close() return documents, fragments def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input", type=Path, default=Path("data/minjust-normalized")) parser.add_argument("--output", type=Path, default=Path("data/opensearch/minjust-fragments.ndjson")) parser.add_argument("--index", default="akyldash-fragments-v1") parser.add_argument("--limit", type=int) parser.add_argument("--version", action="version", version=APP_VERSION) arguments = parser.parse_args() if arguments.limit is not None and arguments.limit <= 0: raise SystemExit("--limit must be greater than zero") documents, fragments = export_bulk(arguments.input, arguments.output, arguments.index, arguments.limit) print(f"documents={documents} fragments={fragments} output={arguments.output}\nAkyldash Backend v{APP_VERSION} · Frontend — not created") return 0 if __name__ == "__main__": raise SystemExit(main())