From 6357648c9cee591a5c542e493a2f003963d36dd1 Mon Sep 17 00:00:00 2001 From: Codex Agent Date: Tue, 20 Oct 2026 01:39:36 +0300 Subject: [PATCH 1/5] Add exact current MinJust API staging adapter --- README.md | 6 +- backend/README.md | 42 +- backend/ingestion/minjust_current.py | 612 +++++++++++++++++++++++ backend/normalization/minjust_cbd.py | 18 +- backend/search/minjust_opensearch.py | 9 +- backend/search/review.html | 4 +- backend/test_minjust_current.py | 326 ++++++++++++ backend/test_minjust_normalization.py | 41 ++ backend/test_minjust_opensearch.py | 44 ++ docs/README.md | 2 +- docs/operations/CORPUS_HANDOFF.md | 51 ++ docs/operations/CURRENT_STATE.md | 52 +- docs/operations/project-status.md | 4 +- docs/product/frontend-search-sps-plan.md | 2 +- docs/product/project-overview.md | 2 +- infra/deploy/test_verify_corpus.py | 7 +- 16 files changed, 1199 insertions(+), 23 deletions(-) create mode 100644 backend/ingestion/minjust_current.py create mode 100644 backend/test_minjust_current.py diff --git a/README.md b/README.md index 58ab1d4..1387146 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ Telegram-бот — только часть рабочего окружения ## Текущее состояние Сейчас реализованы Telegram-бот-секретарь версии `0.2.2`, backend версии -`0.11.3` и внутренний frontend лаборатории `0.1.2`. Авторизация хранится в +`0.12.0` и внутренний frontend лаборатории `0.1.2`. Авторизация хранится в SQLite; первый пользователь `admin` создаётся интерактивной CLI-командой. Поиск расширяет аббревиатуры организационно-правовых форм и поднимает действующий Налоговый кодекс для запросов о налоге на прибыль. @@ -18,7 +18,7 @@ SQLite; первый пользователь `admin` создаётся инт | Компонент | Версия | Состояние | |---|---:|---| | Telegram-бот | `0.2.2` | на Synology работает `0.2.1`; обновление после слияния | -| Backend | `0.11.3` | авторизация и поиск с фильтрами; тематический поиск ОсОО/ЖЧК | +| Backend | `0.12.0` | авторизация, поиск с фильтрами и ограниченный staging текущего API Минюста | | Frontend | `0.1.2` | лаборатория поиска, личный кабинет и управление учетными записями | | Сбор и обработка правовых данных | `0.8.0` | добавлены relevance set и оценка выдачи | | RAG и база знаний | — | ещё не созданы | @@ -62,4 +62,4 @@ python3 -m unittest discover -s tools/telegram-bot -v --- -Акылдаш · Telegram-бот v0.2.2 · Backend v0.11.3 · Frontend v0.1.2 +Акылдаш · Telegram-бот v0.2.2 · Backend v0.12.0 · Frontend v0.1.2 diff --git a/backend/README.md b/backend/README.md index 7d03fb0..32303b7 100644 --- a/backend/README.md +++ b/backend/README.md @@ -1,6 +1,6 @@ # Backend Акылдаш -Версия: `0.11.3` +Версия: `0.12.0` Первая backend-область проекта — загрузка правовых документов из официального Open Data API ЦБД Минюста Кыргызской Республики. Код расположен в @@ -54,6 +54,44 @@ result = sync_archive(output_path) Планировщик, очередь задач и PostgreSQL пока не добавлены: модуль не зависит от выбора будущего backend-фреймворка. +## Текущий API Минюста и отдельный staging + +Старый OpenData endpoint может возвращать `404`; для отдельной ограниченной +выгрузки официальный сайт предоставляет `/api/v1/GetDocument` и +`/api/v1/GetEdition`. Адаптер `ingestion.minjust_current` получает один +указанный документ и сохраняет его в отдельное поколение staging: + +```bash +PYTHONPATH=backend python3 -m ingestion.minjust_current \ + --output /tmp/minjust-current-code6 \ + --document-code 6 +``` + +Продолжайте прерванную выгрузку только с тем же каталогом и `--resume`. По +умолчанию лимит скачанного текста одного документа — 4 GiB, лимит запросов — +один в секунду; каталог не должен совпадать или пересекаться с +`data/minjust-cbd` и +`data/minjust-normalized`. Адаптер сохраняет исходные ответы и checkpoints, +публикует папку документа атомарно и не заменяет существующий код в staging. +Для `/GetEdition?lang=kg` официальный ответ помещает кыргызский HTML в +`contentRu` при `contentKg=null`; архивный язык определяется запрошенным +`lang`, а не именем поля ответа. + +Исторические редакции запрашиваются только с `exact=true`; без этого параметра +официальный endpoint возвращал ID последней редакции даже для другого +запрошенного ID. Адаптер требует совпадения response ID и запрошенного ID, +сверяет названия RU/KY с metadata `GetDocument` и сохраняет request contract в +checkpoint. Старые staging generations без этого contract нельзя возобновить. +`exact=true` и совпадение ID проверены на двух контрольных редакциях; это не +подтверждает полноту всего источника. + +Порядок редакций сохраняется по ответу `/GetDocument` и используется как +официальный порядок при нормализации и выборе текущей редакции в поисковом +индексе. Вложения и встроенные изображения этим адаптером не подтверждаются: +staging остаётся неполным до отдельной проверки медиа и не заменяет корпус. +Получение полного списка документов не является частью этой ограниченной +процедуры. + ## Нормализация архива Нормализатор читает исходный архив без изменений и создаёт отдельный набор @@ -253,4 +291,4 @@ PYTHONPATH=backend python3 -m search.query "ЖЧК ачуу тартиби" --la --- -Акылдаш · Backend v0.11.3 · Frontend v0.1.2 +Акылдаш · Backend v0.12.0 · Frontend v0.1.2 diff --git a/backend/ingestion/minjust_current.py b/backend/ingestion/minjust_current.py new file mode 100644 index 0000000..cb53966 --- /dev/null +++ b/backend/ingestion/minjust_current.py @@ -0,0 +1,612 @@ +"""Read-only staging client for the current MinJust CBD API.""" + +from __future__ import annotations + +import argparse +import hashlib +import http.client +import json +import logging +import os +import sqlite3 +import time +import urllib.error +import urllib.parse +import urllib.request +from dataclasses import dataclass +from datetime import datetime, timezone +from email.utils import parsedate_to_datetime +from pathlib import Path +from typing import Callable + +from ingestion.minjust_cbd import atomic_write, connect_manifest, json_bytes + +API_ROOT = "https://cbd.minjust.gov.kg/api/v1" +APP_VERSION = "0.12.0" +REQUEST_CONTRACT = { + "version": 2, + "endpoint": "/GetEdition", + "edition_id_parameter": "editionId", + "exact_parameter": "true", + "language_parameter": "lang", + "response_id_must_match": True, +} +MAX_RESPONSE_BYTES = 16 * 1024 * 1024 +DEFAULT_DOCUMENT_BUDGET = 4 * 1024 * 1024 * 1024 +LANGUAGES = (("ru", "ru", "textRusType"), ("kg", "ky", "textKyrType")) +LOGGER = logging.getLogger(__name__) + + +class SourceError(RuntimeError): + pass + + +class SourceChanged(SourceError): + pass + + +@dataclass(frozen=True) +class ApiResponse: + value: dict + body: bytes + sha256: str + + +class CurrentCbdClient: + def __init__( + self, + requests_per_second: float = 1, + timeout: int = 60, + retries: int = 5, + max_response_bytes: int = MAX_RESPONSE_BYTES, + opener: Callable = urllib.request.urlopen, + sleep: Callable[[float], None] = time.sleep, + clock: Callable[[], float] = time.monotonic, + ) -> None: + if requests_per_second <= 0 or timeout <= 0 or retries <= 0 or max_response_bytes <= 0: + raise ValueError("rate, timeout, retries, and response limit must be positive") + self.interval = 1 / requests_per_second + self.timeout = timeout + self.retries = retries + self.max_response_bytes = max_response_bytes + self.opener = opener + self.sleep = sleep + self.clock = clock + self.last_request = 0.0 + + def request_json( + self, + path: str, + *, + method: str = "GET", + parameters: tuple[tuple[str, object], ...] = (), + payload: dict | None = None, + ) -> ApiResponse: + if not path.startswith("/") or ".." in path.split("/"): + raise ValueError("API path must be an absolute endpoint path") + query = urllib.parse.urlencode(parameters, doseq=True) + url = API_ROOT + path + (f"?{query}" if query else "") + body = json.dumps(payload or {}, ensure_ascii=False, separators=(",", ":")).encode() if method == "POST" else None + for attempt in range(self.retries): + wait = self.interval - (self.clock() - self.last_request) + if wait > 0: + self.sleep(wait) + request = urllib.request.Request( + url, + data=body, + method=method, + headers={ + "Accept": "application/json", + **({"Content-Type": "application/json"} if body is not None else {}), + "User-Agent": f"Akyldash/{APP_VERSION} (+https://cbd.minjust.gov.kg)", + }, + ) + try: + self.last_request = self.clock() + with self.opener(request, timeout=self.timeout) as response: + raw = response.read(self.max_response_bytes + 1) + if len(raw) > self.max_response_bytes: + raise SourceError(f"MinJust response exceeds {self.max_response_bytes} bytes") + value = json.loads(raw) + if not isinstance(value, dict): + raise SourceError("MinJust response must be a JSON object") + return ApiResponse(value, raw, hashlib.sha256(raw).hexdigest()) + except urllib.error.HTTPError as error: + if error.code != 429 and error.code < 500: + raise SourceError(f"MinJust returned HTTP {error.code} for {path}") from error + delay = 2**attempt + if error.code == 429: + delay = max(delay, retry_after_seconds(error.headers.get("Retry-After"))) + except (TimeoutError, urllib.error.URLError, http.client.IncompleteRead, json.JSONDecodeError) as error: + delay = 2**attempt + LOGGER.warning("MinJust request failed for %s (attempt %s/%s)", path, attempt + 1, self.retries) + last_error = error + if attempt + 1 < self.retries: + self.sleep(delay) + raise SourceError(f"MinJust request failed after {self.retries} attempts: {path}") from locals().get("last_error") + + def document(self, alias: str) -> ApiResponse: + response = self.request_json("/GetDocument", parameters=(("DocumentCode", alias),)) + value = response.value + canonical = value.get("documentCode") + editions = value.get("editions") + if not isinstance(canonical, int) or isinstance(canonical, bool) or canonical < 1: + raise SourceError("MinJust document has an invalid canonical documentCode") + if not isinstance(editions, list): + raise SourceError("MinJust document is missing editions") + ids = [edition_id(item) for item in editions] + if len(ids) != len(set(ids)): + raise SourceError(f"MinJust document {canonical} contains duplicate edition IDs") + return response + + def edition(self, identifier: int, language: str) -> ApiResponse: + if identifier < 1 or language not in {"ru", "kg"}: + raise ValueError("edition ID or requested language is invalid") + response = self.request_json( + "/GetEdition", + parameters=(("editionId", identifier), ("lang", language), ("exact", "true")), + ) + returned_id = response.value.get("id") + if not isinstance(returned_id, int) or isinstance(returned_id, bool) or returned_id < 1: + raise SourceError(f"MinJust returned an invalid edition response ID for {identifier}") + if returned_id != identifier: + raise SourceChanged(f"MinJust exact edition ID {returned_id} does not match requested ID {identifier}") + return response + + +def retry_after_seconds(value: str | None) -> float: + if not value: + return 0 + try: + return max(0, float(value)) + except ValueError: + try: + return max(0, (parsedate_to_datetime(value) - datetime.now(timezone.utc)).total_seconds()) + except (TypeError, ValueError, OverflowError): + return 0 + + +def edition_id(value: object) -> int: + if not isinstance(value, dict): + raise SourceError("MinJust document contains an invalid edition") + identifier = value.get("id") + if not isinstance(identifier, int) or isinstance(identifier, bool) or identifier < 1: + raise SourceError("MinJust edition has an invalid id") + return identifier + + +def open_manifest(path: Path) -> sqlite3.Connection: + connection = sqlite3.connect(path) + connection.execute("PRAGMA journal_mode=DELETE") + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS documents ( + alias TEXT PRIMARY KEY, + canonical_id INTEGER, + response_sha256 TEXT, + expected_editions INTEGER, + state TEXT NOT NULL, + error TEXT, + updated_at TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS editions ( + canonical_id INTEGER NOT NULL, + edition_id INTEGER NOT NULL, + source_order INTEGER NOT NULL, + response_sha256 TEXT NOT NULL, + PRIMARY KEY (canonical_id, edition_id) + ); + CREATE TABLE IF NOT EXISTS languages ( + canonical_id INTEGER NOT NULL, + edition_id INTEGER NOT NULL, + request_language TEXT NOT NULL, + archive_language TEXT NOT NULL, + response_sha256 TEXT, + content_sha256 TEXT, + byte_count INTEGER NOT NULL DEFAULT 0, + state TEXT NOT NULL, + error TEXT, + updated_at TEXT NOT NULL, + PRIMARY KEY (canonical_id, edition_id, request_language) + ); + CREATE TABLE IF NOT EXISTS generation ( + contract_sha256 TEXT NOT NULL + ); + """ + ) + return connection + + +def validate_stage_root(output: Path, resume: bool) -> None: + destination = output.resolve() + for relative in (Path("data/minjust-cbd"), Path("data/minjust-normalized")): + legacy = relative.resolve() + if destination == legacy or destination.is_relative_to(legacy) or legacy.is_relative_to(destination): + raise ValueError("current API staging must not overlap the existing corpus") + if output.exists(): + if output.is_symlink() or not output.is_dir(): + raise ValueError("staging output must be a real directory") + if any(output.iterdir()) and not resume: + raise ValueError("staging output is not empty; use --resume for this generation") + else: + output.mkdir(mode=0o700, parents=True) + for relative in (Path(".work"), Path(".work/documents"), Path("documents")): + path = output / relative + if path.is_symlink() or (path.exists() and not path.is_dir()): + raise ValueError(f"staging path must be a real directory: {relative}") + contract_path = output / "request-contract.json" + if contract_path.is_symlink(): + raise ValueError("staging request contract must not be a symbolic link") + if contract_path.exists(): + try: + contract = json.loads(contract_path.read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError) as error: + raise SourceChanged("staging generation request contract is invalid") from error + if contract != REQUEST_CONTRACT: + raise SourceChanged("staging generation uses a different request contract") + else: + if any(output.iterdir()): + raise SourceChanged("staging generation has no exact-edition request contract; start a new generation") + atomic_write(contract_path, json_bytes(REQUEST_CONTRACT)) + + +def stage_document( + output: Path, + alias: str, + client: CurrentCbdClient, + *, + resume: bool = False, + max_document_bytes: int = DEFAULT_DOCUMENT_BUDGET, +) -> dict: + """Stage one document and its official editions without touching legacy roots.""" + if not alias or len(alias) > 64 or "/" in alias or "\\" in alias: + raise ValueError("document code is invalid") + if max_document_bytes <= 0: + raise ValueError("document byte budget must be positive") + validate_stage_root(output, resume) + work_root = output / ".work" / "documents" + target_root = output / "documents" + connection = open_manifest(output / "current-api.sqlite3") + raw_manifest = connect_manifest(output / "manifest.sqlite3") + try: + contract_sha = hashlib.sha256(json.dumps(REQUEST_CONTRACT, sort_keys=True, separators=(",", ":")).encode()).hexdigest() + saved_contract = connection.execute("SELECT contract_sha256 FROM generation").fetchone() + if saved_contract and saved_contract[0] != contract_sha: + raise SourceChanged("checkpoint request contract changed; start a new staging generation") + if not saved_contract: + if connection.execute("SELECT COUNT(*) FROM documents").fetchone()[0]: + raise SourceChanged("existing checkpoints have no request contract; start a new staging generation") + with connection: + connection.execute("INSERT INTO generation (contract_sha256) VALUES (?)", (contract_sha,)) + response = client.document(alias) + source = response.value + canonical = source["documentCode"] + record_sha = response.sha256 + editions = source["editions"] + previous = connection.execute( + "SELECT canonical_id, response_sha256, expected_editions FROM documents WHERE alias = ?", + (alias,), + ).fetchone() + if previous and previous[1] is not None and ( + previous[0] != canonical or previous[1] != record_sha or previous[2] != len(editions) + ): + raise SourceChanged("document metadata changed; start a new staging generation") + working_document = work_root / str(canonical) + final_document = target_root / str(canonical) + if final_document.is_symlink(): + raise ValueError("staging target must not be a symbolic link") + if final_document.exists() and not previous: + raise ValueError("staging target already exists without a matching checkpoint") + if final_document.exists(): + metadata = json.loads((final_document / "metadata.json").read_text(encoding="utf-8")) + if ( + metadata.get("SourceApi", {}).get("response_sha256") != record_sha + or hashlib.sha256((final_document / "source-document.json").read_bytes()).hexdigest() != record_sha + ): + raise SourceChanged("published staging document does not match the current source record") + for edition in editions: + identifier = edition_id(edition) + for request_language, archive_language, type_field in LANGUAGES: + checkpoint = connection.execute( + "SELECT state, content_sha256 FROM languages WHERE canonical_id = ? AND edition_id = ? AND request_language = ?", + (canonical, identifier, request_language), + ).fetchone() + if not edition.get(type_field): + if not checkpoint or checkpoint[0] != "unavailable": + raise SourceError(f"published staging document has no checkpoint for edition {identifier} {request_language}") + elif ( + not checkpoint + or checkpoint[0] != "complete" + or not (final_document / "editions" / str(identifier) / f"{archive_language}.html").is_file() + or sha256_file(final_document / "editions" / str(identifier) / f"{archive_language}.html") != checkpoint[1] + ): + raise SourceError(f"published staging document failed verification for edition {identifier} {request_language}") + document_sha = hash_tree(final_document) + fetched_at = utc_now() + manifest_row = raw_manifest.execute( + "SELECT source_sha256 FROM documents WHERE code = ?", (canonical,) + ).fetchone() + if manifest_row and manifest_row[0] != document_sha: + raise SourceError("published staging document does not match its raw manifest") + with raw_manifest: + raw_manifest.execute( + "INSERT OR IGNORE INTO documents (code, source_sha256, fetched_at) VALUES (?, ?, ?)", + (canonical, document_sha, fetched_at), + ) + raw_manifest.execute( + "INSERT INTO errors (code, message, failed_at) VALUES (?, ?, ?) " + "ON CONFLICT(code) DO UPDATE SET message=excluded.message, failed_at=excluded.failed_at", + (canonical, "attachments and embedded media are not independently verified", fetched_at), + ) + with connection: + connection.execute( + "UPDATE documents SET state='text_complete_incomplete_attachments', error=NULL, updated_at=? WHERE alias=?", + (fetched_at, alias), + ) + return { + "alias": alias, + "canonical_id": canonical, + "editions": len(editions), + "source_sha256": document_sha, + "state": "text_complete_incomplete_attachments", + "resumed": True, + } + working_document.mkdir(parents=True, exist_ok=True) + document_metadata = map_document_metadata(source, alias, response.sha256) + atomic_write(working_document / "metadata.json", json_bytes(document_metadata)) + atomic_write(working_document / "source-document.json", response.body) + with connection: + connection.execute( + "INSERT INTO documents (alias, canonical_id, response_sha256, expected_editions, state, updated_at) VALUES (?, ?, ?, ?, 'fetching', ?) " + "ON CONFLICT(alias) DO UPDATE SET canonical_id=excluded.canonical_id, response_sha256=excluded.response_sha256, " + "expected_editions=excluded.expected_editions, state='fetching', error=NULL, updated_at=excluded.updated_at", + (alias, canonical, record_sha, len(editions), utc_now()), + ) + for order, edition in enumerate(editions): + identifier = edition_id(edition) + existing = connection.execute( + "SELECT source_order, response_sha256 FROM editions WHERE canonical_id = ? AND edition_id = ?", + (canonical, identifier), + ).fetchone() + if existing and existing != (order, hashlib.sha256(json_bytes(edition)).hexdigest()): + raise SourceChanged(f"edition metadata changed for {identifier}") + connection.execute( + "INSERT OR IGNORE INTO editions (canonical_id, edition_id, source_order, response_sha256) VALUES (?, ?, ?, ?)", + (canonical, identifier, order, hashlib.sha256(json_bytes(edition)).hexdigest()), + ) + + incomplete = [] + for order, edition in enumerate(editions): + identifier = edition_id(edition) + directory = working_document / "editions" / str(identifier) + directory.mkdir(parents=True, exist_ok=True) + atomic_write(directory / "metadata.json", json_bytes(map_edition_metadata(edition, order))) + for request_language, archive_language, type_field in LANGUAGES: + html_path = directory / f"{archive_language}.html" + checkpoint = connection.execute( + "SELECT state, content_sha256, byte_count FROM languages WHERE canonical_id = ? AND edition_id = ? AND request_language = ?", + (canonical, identifier, request_language), + ).fetchone() + if checkpoint and checkpoint[0] == "complete": + if not html_path.is_file() or sha256_file(html_path) != checkpoint[1]: + raise SourceError(f"staged content hash mismatch for edition {identifier} {request_language}") + continue + if not edition.get(type_field): + save_language_checkpoint( + connection, canonical, identifier, request_language, archive_language, + None, None, 0, "unavailable", "source metadata reports no language file", + ) + continue + try: + language_response = client.edition(identifier, request_language) + response_id = language_response.value.get("id") + if isinstance(response_id, bool) or not isinstance(response_id, int) or response_id != identifier: + raise SourceChanged(f"MinJust exact edition ID mismatch for {identifier}") + for field in ("nameRus", "nameKyr"): + if edition.get(field) is not None and language_response.value.get(field) != edition[field]: + raise SourceChanged(f"MinJust exact edition metadata mismatch for {identifier}: {field}") + content = language_response.value.get("contentRu") + if not isinstance(content, str) or not content: + raise SourceError(f"MinJust returned no content for edition {identifier} language {request_language}") + encoded = content.encode("utf-8") + completed_bytes = connection.execute( + "SELECT COALESCE(SUM(byte_count), 0) FROM languages WHERE canonical_id = ? AND state = 'complete'", + (canonical,), + ).fetchone()[0] + if completed_bytes + len(encoded) > max_document_bytes: + raise SourceError(f"document {canonical} exceeds its {max_document_bytes}-byte staging budget") + response_metadata = { + key: value + for key, value in language_response.value.items() + if key not in {"contentRu", "contentKg"} + } + response_metadata["requested_language"] = request_language + response_metadata["archive_language"] = archive_language + response_metadata["requested_edition_id"] = identifier + response_metadata["exact"] = True + response_metadata["response_sha256"] = language_response.sha256 + atomic_write( + directory / "source" / f"{request_language}.json", + json_bytes(response_metadata), + ) + atomic_write(html_path, encoded) + save_language_checkpoint( + connection, canonical, identifier, request_language, archive_language, + language_response.sha256, hashlib.sha256(encoded).hexdigest(), len(encoded), "complete", None, + ) + except Exception as error: + message = str(error)[:500] + save_language_checkpoint( + connection, canonical, identifier, request_language, archive_language, + None, None, 0, "error", message, + ) + incomplete.append({"edition_id": identifier, "language": request_language, "error": message}) + if isinstance(error, SourceChanged) or (isinstance(error, SourceError) and "budget" in message): + raise + + if incomplete: + with connection: + connection.execute( + "UPDATE documents SET state='incomplete', error=?, updated_at=? WHERE alias=?", + (json.dumps(incomplete, ensure_ascii=False), utc_now(), alias), + ) + return {"alias": alias, "canonical_id": canonical, "editions": len(editions), "incomplete": incomplete} + + metadata = json.loads((working_document / "metadata.json").read_text(encoding="utf-8")) + metadata["SourceCompleteness"] = { + "text": "complete", + "attachments": "not independently verified", + } + atomic_write(working_document / "metadata.json", json_bytes(metadata)) + target_root.mkdir(parents=True, exist_ok=True) + if final_document.exists(): + raise ValueError("refusing to replace a document in this staging generation") + os.replace(working_document, final_document) + document_sha = hash_tree(final_document) + fetched_at = utc_now() + with raw_manifest: + raw_manifest.execute( + "INSERT INTO documents (code, source_sha256, fetched_at) VALUES (?, ?, ?)", + (canonical, document_sha, fetched_at), + ) + raw_manifest.execute( + "INSERT INTO errors (code, message, failed_at) VALUES (?, ?, ?)", + (canonical, "attachments and embedded media are not independently verified", fetched_at), + ) + with connection: + connection.execute( + "UPDATE documents SET state='text_complete_incomplete_attachments', updated_at=? WHERE alias=?", + (fetched_at, alias), + ) + return { + "alias": alias, + "canonical_id": canonical, + "editions": len(editions), + "source_sha256": document_sha, + "state": "text_complete_incomplete_attachments", + } + except Exception as error: + with connection: + connection.execute( + "INSERT INTO documents (alias, state, error, updated_at) VALUES (?, 'error', ?, ?) " + "ON CONFLICT(alias) DO UPDATE SET state='error', error=excluded.error, updated_at=excluded.updated_at", + (alias, str(error)[:500], utc_now()), + ) + raise + finally: + raw_manifest.close() + connection.close() + + +def map_document_metadata(source: dict, alias: str, response_sha256: str) -> dict: + canonical = source["documentCode"] + return { + "Code": canonical, + "Name": {"Rus": source.get("nameRus"), "Kyr": source.get("nameKyr")}, + "Title": {"Rus": source.get("headerRu"), "Kyr": source.get("headerKg")}, + "Number": source.get("number"), + "DateAdopted": source.get("dateAdopted"), + "DateInclusion": source.get("dateInclusion"), + "DateOfEntry": source.get("dateOfEntry"), + "DatePublication": source.get("datePublication"), + "DateRegistration": source.get("dateRegistration"), + "IsPublicInRegister": source.get("isPublicInRegister"), + "SourceStatus": source.get("status"), + "SourceClassId": source.get("classId"), + "SourceRefTypeId": source.get("refTypeId"), + "SourceAuthorityIds": source.get("authoritiesId"), + "SourcePublicationIds": source.get("sourcePublicationsId"), + "SourceMetadata": {key: value for key, value in source.items() if key != "editions"}, + "SourceDocumentCode": alias, + "SourceMediaStatus": "not independently verified", + "SourceApi": {"endpoint": "/GetDocument", "response_sha256": response_sha256}, + } + + +def map_edition_metadata(source: dict, source_order: int) -> dict: + return { + "Code": edition_id(source), + "Name": {"Rus": source.get("nameRus"), "Kyr": source.get("nameKyr")}, + "SourceOrder": source_order, + "SourceTextTypes": { + "ru": source.get("textRusType"), + "ky": source.get("textKyrType"), + }, + "SourceMediaStatus": "not independently verified", + "SourceMetadata": source, + } + + +def save_language_checkpoint( + connection: sqlite3.Connection, + canonical: int, + edition: int, + request_language: str, + archive_language: str, + response_sha256: str | None, + content_sha256: str | None, + byte_count: int, + state: str, + error: str | None, +) -> None: + with connection: + connection.execute( + "INSERT INTO languages (canonical_id, edition_id, request_language, archive_language, response_sha256, content_sha256, byte_count, state, error, updated_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(canonical_id, edition_id, request_language) DO UPDATE SET " + "archive_language=excluded.archive_language, response_sha256=excluded.response_sha256, content_sha256=excluded.content_sha256, " + "byte_count=excluded.byte_count, state=excluded.state, error=excluded.error, updated_at=excluded.updated_at", + (canonical, edition, request_language, archive_language, response_sha256, content_sha256, byte_count, state, error, utc_now()), + ) + + +def sha256_file(path: Path) -> str: + digest = hashlib.sha256() + with path.open("rb") as source: + for block in iter(lambda: source.read(1024 * 1024), b""): + digest.update(block) + return digest.hexdigest() + + +def hash_tree(root: Path) -> str: + digest = hashlib.sha256() + for path in sorted(item for item in root.rglob("*") if item.is_file()): + digest.update(path.relative_to(root).as_posix().encode()) + digest.update(b"\0") + digest.update(sha256_file(path).encode("ascii")) + digest.update(b"\0") + return digest.hexdigest() + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--output", type=Path, required=True, help="new, separate staging generation") + parser.add_argument("--document-code", required=True, help="one official document alias or canonical code") + parser.add_argument("--resume", action="store_true", help="resume only this staging generation") + parser.add_argument("--requests-per-second", type=float, default=1) + parser.add_argument("--timeout", type=int, default=60) + parser.add_argument("--retries", type=int, default=5) + parser.add_argument("--max-document-bytes", type=int, default=DEFAULT_DOCUMENT_BUDGET) + parser.add_argument("--version", action="version", version=APP_VERSION) + return parser.parse_args() + + +def main() -> int: + arguments = parse_args() + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + result = stage_document( + arguments.output, + arguments.document_code, + CurrentCbdClient(arguments.requests_per_second, arguments.timeout, arguments.retries), + resume=arguments.resume, + max_document_bytes=arguments.max_document_bytes, + ) + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return int(result.get("state") != "text_complete_incomplete_attachments") + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backend/normalization/minjust_cbd.py b/backend/normalization/minjust_cbd.py index 97552cf..0cd3869 100644 --- a/backend/normalization/minjust_cbd.py +++ b/backend/normalization/minjust_cbd.py @@ -23,9 +23,9 @@ from pathlib import Path from typing import Callable from urllib.parse import urlsplit -APP_VERSION = "0.9.3" +APP_VERSION = "0.10.0" SCHEMA_VERSION = "1" -NORMALIZER_VERSION = "1.0.0" +NORMALIZER_VERSION = "1.1.0" LANGUAGES = ("ru", "ky") LOGGER = logging.getLogger(__name__) @@ -376,13 +376,16 @@ def load_json(path: Path) -> dict: 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 { + summary = { "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(), } + if "SourceOrder" in metadata: + summary["source_order"] = metadata["SourceOrder"] + return summary def normalize_document( @@ -406,6 +409,15 @@ def normalize_document( 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] + source_orders = [item.get("source_order") for item in summaries] + if any(order is not None for order in source_orders): + if any(not isinstance(order, int) or isinstance(order, bool) or order < 0 for order in source_orders): + raise ValueError(f"Invalid source edition order in {document_directory}") + if len(set(source_orders)) != len(source_orders): + raise ValueError(f"Duplicate source edition order in {document_directory}") + ordered = sorted(zip(source_orders, edition_directories, summaries), key=lambda item: item[0]) + edition_directories = [item[1] for item in ordered] + summaries = [item[2] for item in ordered] available_languages = [language for language in LANGUAGES if any(language in item["available_languages"] for item in summaries)] normalized_metadata = clean_value(metadata) document = { diff --git a/backend/search/minjust_opensearch.py b/backend/search/minjust_opensearch.py index f62a681..d1547b2 100644 --- a/backend/search/minjust_opensearch.py +++ b/backend/search/minjust_opensearch.py @@ -17,7 +17,7 @@ from typing import Iterator from search.catalog import authority_codes, legal_force_code, source_code -APP_VERSION = "0.11.3" +APP_VERSION = "0.12.0" LANGUAGES = {"ru", "ky"} DEFAULT_MAPPING = Path(__file__).with_name("minjust-fragments-index.json") @@ -129,7 +129,12 @@ def bulk_pairs( editions = [path.name for path in edition_root.iterdir() if path.is_dir()] if edition_root.is_dir() else [] if not editions: continue - current_edition = max(editions, key=lambda value: (not value.isdigit(), int(value) if value.isdigit() else value)) + official_editions = document.get("editions") or [] + current_edition = ( + str(official_editions[-1]["source_code"]) + if official_editions + else max(editions, key=lambda value: (not value.isdigit(), int(value) if value.isdigit() else value)) + ) for fragment_path in sorted(directory.glob("editions/*/*/fragments.json")): edition_code, language = fragment_path.parts[-3:-1] values = read_json(fragment_path) diff --git a/backend/search/review.html b/backend/search/review.html index 5817921..f0eed37 100644 --- a/backend/search/review.html +++ b/backend/search/review.html @@ -176,7 +176,7 @@

Управление пользователями

Пароль — от 12 символов. Новый пользователь сменит его при первом входе. Передайте временный пароль отдельно.

    Документ

    Короткое знакомство

     - +