Add bounded current MinJust API staging #51

Merged
agent merged 5 commits from fix/current-minjust-api-ingestion into main 2026-10-21 05:44:35 +00:00
18 changed files with 1421 additions and 25 deletions

No files matched your search

+5
View File
@@ -2,6 +2,11 @@
## Не выпущено
- Backend `0.12.1`: добавлен ограниченный возобновляемый staging редакций из
текущего API Минюста с проверкой точных ID и метаданных, изоляцией от
legacy-корпуса и сохранением исходного порядка редакций при нормализации и
экспорте. Тест overlap guard задаёт безопасные права fixture независимо от
`umask` CI; сам guard не ослаблен.
- Rollback preflight проверяет digest-pinned image references по сохранённому
image ID и останавливается до изменений deployment; переключение production
на digest и полноценный digest rollback пока не реализованы.
+3 -3
View File
@@ -10,7 +10,7 @@ Telegram-бот — только часть рабочего окружения
## Текущее состояние
Сейчас реализованы Telegram-бот-секретарь версии `0.2.2`, backend версии
`0.11.3` и внутренний frontend лаборатории `0.1.2`. Авторизация хранится в
`0.12.1` и внутренний 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.1` | авторизация, поиск с фильтрами и ограниченный 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.1 · Frontend v0.1.2
+45 -2
View File
@@ -1,6 +1,6 @@
# Backend Акылдаш
Версия: `0.11.3`
Версия: `0.12.1`
Первая backend-область проекта — загрузка правовых документов из официального
Open Data API ЦБД Минюста Кыргызской Республики. Код расположен в
@@ -54,6 +54,49 @@ 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.
Каталог staging должен быть отдельной директорией режима 0700, принадлежащей
текущему пользователю; его родитель должен быть существующим доверенным
каталогом. Символические ссылки в generation, checkpoints и каталогах
документов отклоняются. Legacy-корни вычисляются от корня репозитория, а не
от текущего рабочего каталога.
Для `/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 +296,4 @@ PYTHONPATH=backend python3 -m search.query "ЖЧК ачуу тартиби" --la
---
Акылдаш · Backend v0.11.3 · Frontend v0.1.2
Акылдаш · Backend v0.12.1 · Frontend v0.1.2
+720
View File
@@ -0,0 +1,720 @@
"""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 stat
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.1"
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__)
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
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 reject_symlink_components(path: Path) -> None:
current = Path(path.anchor)
for part in path.parts[1:]:
current /= part
try:
mode = current.lstat().st_mode
except FileNotFoundError:
continue
if stat.S_ISLNK(mode):
raise ValueError(f"staging path must not contain a symbolic link: {current}")
if current != path and not stat.S_ISDIR(mode):
raise ValueError(f"staging parent must be a directory: {current}")
def ensure_stage_path(root: Path, path: Path) -> Path:
root = root.absolute()
path = path.absolute()
try:
relative = path.relative_to(root)
except ValueError as error:
raise ValueError("staging file escaped its private generation") from error
current = root
if current.is_symlink():
raise ValueError("staging root must not be a symbolic link")
for part in relative.parts:
current /= part
try:
mode = current.lstat().st_mode
except FileNotFoundError:
continue
if stat.S_ISLNK(mode):
raise ValueError(f"staging path must not contain a symbolic link: {relative}")
if current != path and not stat.S_ISDIR(mode):
raise ValueError(f"staging parent must be a directory: {relative}")
return path
def stage_write(root: Path, path: Path, content: bytes) -> None:
path = ensure_stage_path(root, path)
atomic_write(path, content)
ensure_stage_path(root, path)
def stage_mkdir(root: Path, path: Path) -> None:
path = ensure_stage_path(root, path)
path.mkdir(parents=True, exist_ok=True)
ensure_stage_path(root, path)
if not path.is_dir():
raise ValueError("staging path must be a directory")
def validate_stage_tree(output: Path) -> None:
for relative in (
Path("current-api.sqlite3"), Path("current-api.sqlite3-journal"),
Path("current-api.sqlite3-wal"), Path("current-api.sqlite3-shm"),
Path("manifest.sqlite3"), Path("manifest.sqlite3-journal"),
Path("manifest.sqlite3-wal"), Path("manifest.sqlite3-shm"),
Path("request-contract.json"),
):
path = ensure_stage_path(output, output / relative)
if path.exists() and not path.is_file():
raise ValueError(f"staging checkpoint must be a regular file: {relative}")
for relative in (Path(".work"), Path("documents")):
root = ensure_stage_path(output, output / relative)
if root.exists() and not root.is_dir():
raise ValueError(f"staging path must be a directory: {relative}")
if root.is_dir():
for directory, directories, files in os.walk(root, followlinks=False):
for name in directories + files:
ensure_stage_path(output, Path(directory) / name)
def validate_stage_root(output: Path, resume: bool) -> Path:
requested = Path(output)
if ".." in requested.parts:
raise ValueError("staging output path must not contain '..'")
destination = Path(os.path.abspath(requested))
reject_symlink_components(destination)
parent = destination.parent
if not parent.is_dir():
raise ValueError("staging output parent must already exist")
for ancestor in (parent, *parent.parents):
info = ancestor.lstat()
mode = stat.S_IMODE(info.st_mode)
if stat.S_ISLNK(info.st_mode) or not stat.S_ISDIR(info.st_mode):
raise ValueError("staging output ancestors must be real directories")
sticky_parent = bool(mode & stat.S_ISVTX)
if mode & 0o022 and not sticky_parent:
raise ValueError("staging output parent is not trusted")
for relative in (Path("data/minjust-cbd"), Path("data/minjust-normalized")):
legacy = (REPOSITORY_ROOT / 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 destination.exists():
if destination.is_symlink() or not destination.is_dir():
raise ValueError("staging output must be a real directory")
else:
destination.mkdir(mode=0o700)
root_info = destination.lstat()
root_mode = stat.S_IMODE(root_info.st_mode)
if (
stat.S_ISLNK(root_info.st_mode)
or not stat.S_ISDIR(root_info.st_mode)
or root_info.st_uid != os.geteuid()
or root_mode & 0o077
or (root_mode & 0o700) != 0o700
):
raise ValueError("staging output must be a private directory owned by the current user")
validate_stage_tree(destination)
if any(destination.iterdir()) and not resume:
raise ValueError("staging output is not empty; use --resume for this generation")
contract_path = destination / "request-contract.json"
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(destination.iterdir()):
raise SourceChanged("staging generation has no exact-edition request contract; start a new generation")
stage_write(destination, contract_path, json_bytes(REQUEST_CONTRACT))
return destination
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")
output = 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)
ensure_stage_path(output, working_document)
ensure_stage_path(output, final_document)
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():
final_metadata = ensure_stage_path(output, final_document / "metadata.json")
source_document = ensure_stage_path(output, final_document / "source-document.json")
metadata = json.loads(final_metadata.read_text(encoding="utf-8"))
if (
metadata.get("SourceApi", {}).get("response_sha256") != record_sha
or hashlib.sha256(source_document.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:
html_path = ensure_stage_path(
output,
final_document / "editions" / str(identifier) / f"{archive_language}.html",
)
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 html_path.is_file()
or sha256_file(html_path) != 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,
}
stage_mkdir(output, working_document)
document_metadata = map_document_metadata(source, alias, response.sha256)
stage_write(output, working_document / "metadata.json", json_bytes(document_metadata))
stage_write(output, 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)
stage_mkdir(output, directory)
stage_write(output, directory / "metadata.json", json_bytes(map_edition_metadata(edition, order)))
for request_language, archive_language, type_field in LANGUAGES:
html_path = ensure_stage_path(output, 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
stage_write(
output,
directory / "source" / f"{request_language}.json",
json_bytes(response_metadata),
)
stage_write(output, 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}
working_metadata = ensure_stage_path(output, working_document / "metadata.json")
metadata = json.loads(working_metadata.read_text(encoding="utf-8"))
metadata["SourceCompleteness"] = {
"text": "complete",
"attachments": "not independently verified",
}
stage_write(output, working_metadata, json_bytes(metadata))
stage_mkdir(output, target_root)
ensure_stage_path(output, working_document)
ensure_stage_path(output, final_document)
if final_document.exists():
raise ValueError("refusing to replace a document in this staging generation")
os.replace(working_document, final_document)
validate_stage_tree(output)
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())
+15 -3
View File
@@ -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 = {
+7 -2
View File
@@ -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.1"
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)
+2 -2
View File
@@ -176,7 +176,7 @@
<dialog id="users-dialog" aria-labelledby="users-heading"><section class="review__page"><h2 id="users-heading">Управление пользователями</h2><form id="create-user-form"><label>Логин<input id="new-username" autocomplete="off" minlength="3" maxlength="64" required></label><label>Временный пароль<input id="temporary-password" type="password" autocomplete="new-password" minlength="12" maxlength="256" required></label><p>Пароль — от 12 символов. Новый пользователь сменит его при первом входе. Передайте временный пароль отдельно.</p><p id="users-error" role="alert" class="review__status review__status--error"></p><button class="review__button--primary" type="submit">Создать пользователя</button></form><ul id="user-list" class="review__user-list"></ul><button type="button" data-close="users-dialog">Закрыть</button></section></dialog>
<dialog id="document-dialog"><button class="review__dialog-close" id="dialog-close" type="button">Закрыть</button><h2 id="dialog-heading">Документ</h2><div class="review__document-body" id="dialog-body"></div></dialog>
<div class="review__tour-spotlight" id="tutorial-spotlight" hidden aria-hidden="true"></div><dialog class="review__tour" id="tutorial-dialog" aria-labelledby="tutorial-heading" aria-describedby="tutorial-text"><div class="review__tour-content"><p class="review__tour-kicker">Короткое знакомство</p><h2 id="tutorial-heading"></h2><p id="tutorial-text"></p><div class="review__tour-footer"><span class="review__tour-progress" id="tutorial-progress"></span><div class="review__tour-actions"><button id="tutorial-skip" type="button">Пропустить</button><button class="review__button--primary" id="tutorial-next" type="button">Далее</button></div></div></div></dialog>
<footer class="review__header"><div class="review__header-content">Акылдаш · Frontend v0.1.2 · Backend v0.11.3 · Внутренняя лаборатория релевантности</div></footer>
<footer class="review__header"><div class="review__header-content">Акылдаш · Frontend v0.1.2 · Backend v0.12.1 · Внутренняя лаборатория релевантности</div></footer>
<script>
document.body.classList.add('review__locked');
const DRAFT_KEY = 'akyldash-search-review-draft';
@@ -193,7 +193,7 @@
let account = null;
const reviewRoute = location.pathname === '/review';
$('review').hidden = !reviewRoute; $('search-page').hidden = reviewRoute; $('review-nav').classList.toggle('is-active', reviewRoute); $('search-nav').classList.toggle('is-active', !reviewRoute);
if (!reviewRoute) { document.title = 'Поиск по правовой базе · Акылдаш'; document.querySelector('.review__header-content').hidden = true; $('tutorial-open').hidden = true; document.querySelector('.review__skip').href = '#public-results'; document.querySelector('footer .review__header-content').textContent = 'Акылдаш · Frontend v0.1.2 · Backend v0.11.3'; }
if (!reviewRoute) { document.title = 'Поиск по правовой базе · Акылдаш'; document.querySelector('.review__header-content').hidden = true; $('tutorial-open').hidden = true; document.querySelector('.review__skip').href = '#public-results'; document.querySelector('footer .review__header-content').textContent = 'Акылдаш · Frontend v0.1.2 · Backend v0.12.1'; }
async function apiFetch(path, options = {}) { const response = await fetch(path, {credentials: 'same-origin', ...options}); if (response.status === 401 && path !== '/auth/login') { document.body.classList.add('review__locked'); if (!$('login-dialog').open) $('login-dialog').showModal(); $('login-username').focus(); } return response; }
const showLogin = () => { document.body.classList.add('review__locked'); if (!$('login-dialog').open) $('login-dialog').showModal(); $('login-username').focus(); };
const enterWorkspace = (user) => { account = user; $('signed-in-user').textContent = `${user.username} · ${user.role}`; $('reviewer').value = user.username; $('profile-username').textContent = user.username; $('profile-role').textContent = user.role; $('profile-created').textContent = new Date(user.created_at * 1000).toLocaleDateString('ru-RU'); $('users-open').hidden = user.username.toLowerCase() !== 'admin'; document.body.classList.remove('review__locked'); if (!reviewRoute) initSearchPage(); else if (!hasSeenTour()) setTimeout(openTutorial, 350); };
+405
View File
@@ -0,0 +1,405 @@
import json
import os
import sqlite3
import tempfile
import unittest
import urllib.error
from email.message import Message
from pathlib import Path
from unittest.mock import Mock, patch
from ingestion import minjust_current
from ingestion.minjust_current import (
ApiResponse,
CurrentCbdClient,
SourceChanged,
SourceError,
stage_document,
)
def api_response(value):
body = json.dumps(value, ensure_ascii=False).encode()
import hashlib
return ApiResponse(value, body, hashlib.sha256(body).hexdigest())
def document_record(alias="6", canonical=6, editions=None):
return api_response(
{
"documentCode": canonical,
"nameRus": "Закон",
"nameKyr": "Мыйзам",
"headerRu": "Заголовок",
"headerKg": "Баш сөз",
"number": 114,
"dateAdopted": "1998-08-04",
"classId": 1,
"refTypeId": 10,
"status": 1,
"editions": editions if editions is not None else [
{"id": 920492, "nameRus": "Редакция A", "nameKyr": "Редакция A", "textRusType": "docx", "textKyrType": "docx"},
{"id": 920292, "nameRus": "Редакция B", "nameKyr": "Редакция B", "textRusType": "docx", "textKyrType": "docx"},
],
}
)
class CurrentCbdClientTests(unittest.TestCase):
def test_rejects_legacy_roots_when_called_from_an_unrelated_working_directory(self):
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
repository = base / "repository"
elsewhere = base / "elsewhere"
elsewhere.mkdir()
legacy_roots = [
repository / "data/minjust-cbd",
repository / "data/minjust-normalized",
]
for legacy in legacy_roots:
legacy.mkdir(parents=True)
# Root CI uses umask 000, so make the trusted fixture's parent permissions explicit.
for directory in (repository, repository / "data", *legacy_roots):
directory.chmod(0o755)
original_cwd = Path.cwd()
client = Mock()
try:
os.chdir(elsewhere)
with patch.object(minjust_current, "REPOSITORY_ROOT", repository):
for legacy in legacy_roots:
with self.subTest(legacy=legacy), self.assertRaisesRegex(ValueError, "must not overlap"):
stage_document(legacy / "staging", "6", client)
self.assertFalse((legacy / "staging").exists())
finally:
os.chdir(original_cwd)
client.document.assert_not_called()
def test_requires_private_owned_staging_root(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary) / "shared"
output.mkdir(mode=0o755)
output.chmod(0o755)
with self.assertRaisesRegex(ValueError, "private directory"):
stage_document(output, "6", Mock())
def test_resume_rejects_symlinked_checkpoints_and_document_paths(self):
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
targets = [
("request-contract.json", False),
("current-api.sqlite3", False),
("manifest.sqlite3", False),
(".work/documents/6", True),
(".work/documents/6/editions/920492/ru.html", False),
("documents/6", True),
]
client = Mock()
for index, (relative, directory) in enumerate(targets):
with self.subTest(relative=relative):
output = base / f"generation-{index}"
output.mkdir(mode=0o700)
contract = output / "request-contract.json"
if relative != "request-contract.json":
contract.write_text(json.dumps(minjust_current.REQUEST_CONTRACT), encoding="utf-8")
outside = base / f"outside-{index}"
if directory:
outside.mkdir()
else:
outside.write_text("preserve", encoding="utf-8")
link = output / relative
link.parent.mkdir(parents=True, exist_ok=True)
try:
link.symlink_to(outside, target_is_directory=directory)
except (NotImplementedError, OSError):
self.skipTest("symlinks are unavailable")
with self.assertRaisesRegex(ValueError, "symbolic link"):
stage_document(output, "6", client, resume=True)
client.document.assert_not_called()
if not directory:
self.assertEqual(outside.read_text(encoding="utf-8"), "preserve")
else:
self.assertEqual(list(outside.iterdir()), [])
def test_respects_retry_after_and_retries_server_errors(self):
headers = Message()
headers["Retry-After"] = "7"
response = Mock()
response.read.return_value = b"{}"
response.__enter__ = Mock(return_value=response)
response.__exit__ = Mock(return_value=False)
opener = Mock(side_effect=[
urllib.error.HTTPError("https://example.test", 429, "", headers, None),
urllib.error.HTTPError("https://example.test", 503, "", Message(), None),
response,
])
delays = []
CurrentCbdClient(opener=opener, sleep=delays.append, clock=lambda: 0, retries=3).request_json("/GetEdition")
self.assertIn(7.0, delays)
self.assertIn(2, delays)
self.assertEqual(opener.call_count, 3)
def test_enforces_request_interval(self):
now = [0.0]
starts = []
response = Mock()
response.read.return_value = b"{}"
response.__enter__ = Mock(return_value=response)
response.__exit__ = Mock(return_value=False)
def opener(_request, timeout):
starts.append(now[0])
return response
client = CurrentCbdClient(
opener=opener,
clock=lambda: now[0],
sleep=lambda delay: now.__setitem__(0, now[0] + delay),
)
client.request_json("/GetEdition")
client.request_json("/GetEdition")
self.assertGreaterEqual(starts[1] - starts[0], 1.0)
def test_rejects_oversized_response(self):
response = Mock()
response.read.return_value = b"x" * 11
response.__enter__ = Mock(return_value=response)
response.__exit__ = Mock(return_value=False)
with self.assertRaisesRegex(SourceError, "exceeds 10 bytes"):
CurrentCbdClient(opener=lambda *_args, **_kwargs: response, max_response_bytes=10).request_json("/GetEdition")
def test_get_edition_uses_exact_query_and_checks_requested_id(self):
response = Mock()
response.read.return_value = json.dumps({"id": 920292, "contentRu": "text", "contentKg": None}).encode()
response.__enter__ = Mock(return_value=response)
response.__exit__ = Mock(return_value=False)
requests = []
def opener(request, timeout):
requests.append(request)
return response
result = CurrentCbdClient(opener=opener, sleep=lambda _delay: None).edition(920292, "kg")
self.assertEqual(requests[0].full_url, "https://cbd.minjust.gov.kg/api/v1/GetEdition?editionId=920292&lang=kg&exact=true")
self.assertEqual(requests[0].method, "GET")
self.assertEqual(result.value["id"], 920292)
def test_rejects_response_id_that_does_not_match_exact_request(self):
response = Mock()
response.read.return_value = json.dumps({"id": 920492, "contentRu": "text", "contentKg": None}).encode()
response.__enter__ = Mock(return_value=response)
response.__exit__ = Mock(return_value=False)
with self.assertRaisesRegex(SourceChanged, "does not match requested ID"):
CurrentCbdClient(opener=lambda *_args, **_kwargs: response).edition(920292, "ru")
def test_stages_alias_mapping_language_quirk_and_official_edition_order(self):
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
output = base / "generation"
old_raw = base / "legacy" / "documents" / "6" / "sentinel"
old_raw.parent.mkdir(parents=True)
old_raw.write_text("preserve", encoding="utf-8")
class Client:
def document(self, alias):
self.alias = alias
return document_record(alias=alias, canonical=6)
def edition(self, identifier, language):
text = "<p>Russian</p>" if language == "ru" else "<p>Кыргызча</p>"
name = "Редакция A" if identifier == 920492 else "Редакция B"
return api_response({
"id": identifier, "nameRus": name, "nameKyr": name,
"contentRu": text, "contentKg": None,
})
client = Client()
result = stage_document(output, "6", client)
raw = output / "documents" / "6"
self.assertEqual(result["state"], "text_complete_incomplete_attachments")
self.assertEqual((raw / "editions/920492/ru.html").read_text(), "<p>Russian</p>")
self.assertEqual((raw / "editions/920492/ky.html").read_text(), "<p>Кыргызча</p>")
first = json.loads((raw / "editions/920492/metadata.json").read_text())
second = json.loads((raw / "editions/920292/metadata.json").read_text())
self.assertEqual((first["SourceOrder"], second["SourceOrder"]), (0, 1))
language_meta = json.loads((raw / "editions/920292/source/ru.json").read_text())
self.assertEqual((language_meta["requested_edition_id"], language_meta["exact"]), (920292, True))
with sqlite3.connect(output / "current-api.sqlite3") as connection:
self.assertTrue(connection.execute("SELECT contract_sha256 FROM generation").fetchone()[0])
self.assertEqual(old_raw.read_text(), "preserve")
with sqlite3.connect(output / "current-api.sqlite3") as connection:
rows = connection.execute(
"SELECT request_language, archive_language, state FROM languages ORDER BY request_language, edition_id"
).fetchall()
self.assertEqual(rows, [("kg", "ky", "complete")] * 2 + [("ru", "ru", "complete")] * 2)
def test_resolves_official_alias_to_canonical_staging_directory(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary) / "generation"
class Client:
def document(self, alias):
return document_record(alias=alias, canonical=205064, editions=[])
result = stage_document(output, "1-4", Client())
self.assertEqual(result["canonical_id"], 205064)
self.assertTrue((output / "documents/205064/metadata.json").is_file())
self.assertFalse((output / "documents/1-4").exists())
def test_rejects_non_exact_edition_response_id_without_publishing(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary) / "generation"
class Client:
def document(self, _alias):
return document_record(editions=[{"id": 920292, "textRusType": "docx", "textKyrType": "docx"}])
def edition(self, _identifier, language):
return api_response({"id": 920492, "contentRu": f"<{language} edition>", "contentKg": None})
with self.assertRaisesRegex(SourceChanged, "exact edition ID mismatch"):
stage_document(output, "6", Client())
self.assertFalse((output / "documents/6").exists())
with sqlite3.connect(output / "current-api.sqlite3") as connection:
self.assertEqual(connection.execute("SELECT state FROM languages").fetchone()[0], "error")
def test_rejects_exact_edition_metadata_contradiction(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary) / "generation"
client = CurrentCbdClient()
client.document = lambda _alias: document_record(editions=[{
"id": 920292, "nameRus": "04.08.1998", "nameKyr": "04.08.1998",
"textRusType": "docx", "textKyrType": "docx",
}])
client.request_json = lambda *_args, **_kwargs: api_response({
"id": 920292, "nameRus": "13.04.2017", "nameKyr": "13.04.2017",
"contentRu": "<wrong edition>", "contentKg": None,
})
with self.assertRaisesRegex(SourceChanged, "metadata mismatch"):
stage_document(output, "6", client)
self.assertFalse((output / "documents/6").exists())
def test_recovers_atomic_publish_before_manifest_checkpoint(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary) / "generation"
class Client:
def document(self, _alias):
return document_record(editions=[{"id": 920492, "textRusType": "docx", "textKyrType": "docx"}])
def edition(self, identifier, language):
return api_response({"id": identifier, "contentRu": f"<{language}>", "contentKg": None})
client = Client()
first = stage_document(output, "6", client)
with sqlite3.connect(output / "current-api.sqlite3") as connection:
connection.execute("UPDATE documents SET state='fetching' WHERE alias='6'")
with sqlite3.connect(output / "manifest.sqlite3") as connection:
connection.execute("DELETE FROM documents WHERE code=6")
connection.execute("DELETE FROM errors WHERE code=6")
resumed = stage_document(output, "6", client, resume=True)
self.assertEqual(resumed["source_sha256"], first["source_sha256"])
self.assertTrue(resumed["resumed"])
with sqlite3.connect(output / "manifest.sqlite3") as connection:
self.assertEqual(connection.execute("SELECT COUNT(*) FROM documents WHERE code=6").fetchone()[0], 1)
self.assertEqual(connection.execute("SELECT COUNT(*) FROM errors WHERE code=6").fetchone()[0], 1)
def test_retries_after_document_metadata_request_failed_before_checkpoint(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary) / "generation"
class Client:
failed = False
def document(self, _alias):
if not self.failed:
self.failed = True
raise SourceError("temporary DNS failure")
return document_record(editions=[])
client = Client()
with self.assertRaisesRegex(SourceError, "DNS"):
stage_document(output, "6", client)
result = stage_document(output, "6", client, resume=True)
self.assertEqual(result["canonical_id"], 6)
with sqlite3.connect(output / "current-api.sqlite3") as connection:
row = connection.execute(
"SELECT canonical_id, response_sha256, expected_editions FROM documents WHERE alias='6'"
).fetchone()
self.assertEqual(row, (6, document_record(editions=[]).sha256, 0))
def test_does_not_resume_generation_without_exact_request_contract(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary) / "legacy-generation"
output.mkdir(mode=0o700)
sentinel = output / "old-response.json"
sentinel.write_text("preserve", encoding="utf-8")
client = Mock()
with self.assertRaisesRegex(SourceChanged, "no exact-edition request contract"):
stage_document(output, "6", client, resume=True)
client.document.assert_not_called()
self.assertEqual(sentinel.read_text(encoding="utf-8"), "preserve")
def test_incomplete_translation_resumes_without_rewriting_successful_language(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary) / "generation"
class Client:
failed = False
def document(self, _alias):
return document_record(editions=[{"id": 920492, "textRusType": "docx", "textKyrType": "docx"}])
def edition(self, identifier, language):
if language == "kg" and not self.failed:
self.failed = True
raise SourceError("simulated interrupted response")
return api_response({"id": identifier, "contentRu": f"<{language}>", "contentKg": None})
client = Client()
first = stage_document(output, "6", client)
self.assertIn("incomplete", first)
self.assertFalse((output / "documents/6").exists())
second = stage_document(output, "6", client, resume=True)
self.assertEqual(second["state"], "text_complete_incomplete_attachments")
self.assertEqual((output / "documents/6/editions/920492/ru.html").read_text(), "<ru>")
self.assertEqual((output / "documents/6/editions/920492/ky.html").read_text(), "<kg>")
self.assertEqual(client.failed, True)
def test_refuses_budget_overrun_without_publishing_partial_document(self):
with tempfile.TemporaryDirectory() as temporary:
output = Path(temporary) / "generation"
class Client:
def document(self, _alias):
return document_record(editions=[{"id": 920492, "textRusType": "docx", "textKyrType": "docx"}])
def edition(self, identifier, _language):
return api_response({"id": identifier, "contentRu": "123456", "contentKg": None})
with self.assertRaisesRegex(SourceError, "budget"):
stage_document(output, "6", Client(), max_document_bytes=5)
self.assertFalse((output / "documents/6").exists())
if __name__ == "__main__":
unittest.main()
+41
View File
@@ -86,6 +86,47 @@ class MinjustNormalizationTest(unittest.TestCase):
empty = json.loads((output / "documents/1/editions/40/edition.json").read_text(encoding="utf-8"))
self.assertFalse(empty["quality"]["has_html"])
def test_preserves_official_edition_order_instead_of_sorting_ids(self):
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
source = base / "source"
output = base / "normalized"
document = self.document(source, 6)
self.edition(document, 920492, {"ru": "<p>Первая редакция</p>"})
self.edition(document, 920292, {"ru": "<p>Последняя редакция</p>"})
for code, order in ((920492, 0), (920292, 1)):
metadata_path = document / "editions" / str(code) / "metadata.json"
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
metadata["SourceOrder"] = order
metadata_path.write_text(json.dumps(metadata, ensure_ascii=False), encoding="utf-8")
normalize_archive(source, output)
normalized = json.loads((output / "documents/6/document.json").read_text(encoding="utf-8"))
self.assertEqual([item["source_code"] for item in normalized["editions"]], ["920492", "920292"])
self.assertEqual([item["source_order"] for item in normalized["editions"]], [0, 1])
def test_rejects_partial_source_edition_order(self):
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
source = base / "source"
output = base / "normalized"
document = self.document(source, 6)
self.edition(document, 10, {"ru": "<p>Один</p>"})
self.edition(document, 20, {"ru": "<p>Два</p>"})
metadata_path = document / "editions/10/metadata.json"
metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
metadata["SourceOrder"] = 0
metadata_path.write_text(json.dumps(metadata), encoding="utf-8")
with self.assertLogs("normalization.minjust_cbd", level="ERROR"):
result = normalize_archive(source, output)
self.assertEqual((result.failed, result.normalized), (1, 0))
with sqlite3.connect(output / "manifest.sqlite3") as connection:
error = connection.execute("SELECT error FROM documents WHERE code='6'").fetchone()[0]
self.assertIn("Invalid source edition order", error)
def test_continues_after_bad_document_and_clears_repaired_error(self):
with tempfile.TemporaryDirectory() as temporary:
base = Path(temporary)
+44
View File
@@ -25,6 +25,50 @@ class MinjustOpenSearchTest(unittest.TestCase):
self.assertEqual(legal_force_code({"ru": "Указ", "ky": "Жарлыгы"}), "subordinate")
self.assertEqual(source_code("document_type", {"ru": "устав"}), "charter")
def test_export_marks_last_official_edition_current_even_when_ids_are_out_of_order(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
input_root = root / "normalized"
document_root = input_root / "documents/6"
document_root.mkdir(parents=True)
with closing(sqlite3.connect(input_root / "manifest.sqlite3")) as connection:
with connection:
connection.execute("CREATE TABLE documents (code TEXT, state TEXT)")
connection.execute("INSERT INTO documents VALUES ('6', 'success')")
(document_root / "document.json").write_text(
json.dumps({"schema_version": "1", "source_code": "6", "editions": [
{"source_code": "920492", "source_order": 0},
{"source_code": "920292", "source_order": 1},
]}),
encoding="utf-8",
)
for edition in ("920292", "920492"):
text = f"Текст {edition}"
fragment_path = document_root / f"editions/{edition}/ru/fragments.json"
fragment_path.parent.mkdir(parents=True)
fragment_path.write_text(json.dumps([{
"id": f"document:6:edition:{edition}:lang:ru:fragment:1",
"document_code": "6",
"edition_code": edition,
"language": "ru",
"position": 1,
"type": "paragraph",
"text": text,
"text_sha256": hashlib.sha256(text.encode()).hexdigest(),
"source_path": f"documents/6/editions/{edition}/ru.html",
"source_sha256": "b" * 64,
}]), encoding="utf-8")
output = root / "bulk.ndjson"
self.assertEqual(export_bulk(input_root, output, "test-index"), (1, 2))
lines = [json.loads(line) for line in output.read_text(encoding="utf-8").splitlines()]
flags = {
lines[index]["index"]["_id"].split(":")[3]: lines[index + 1]["is_current_edition"]
for index in range(0, len(lines), 2)
}
self.assertEqual(flags, {"920292": True, "920492": False})
def test_switches_alias_atomically_after_successful_load(self):
with patch("search.minjust_opensearch.request_json", return_value={"acknowledged": True}) as request:
switch_alias("http://127.0.0.1:9200/", "akyldash-fragments-v2", "akyldash-fragments-current")
+1 -1
View File
@@ -45,4 +45,4 @@
---
Акылдаш · Telegram-бот v0.2.2 · Backend v0.11.3 · Frontend v0.1.2
Акылдаш · Telegram-бот v0.2.2 · Backend v0.12.1 · Frontend v0.1.2
+51
View File
@@ -96,6 +96,57 @@ From the VPS, a metadata-only HTTPS check to `CheckAvailable` failed at the
HTTP/2 transport layer (`curl` exit 92, `INTERNAL_ERROR`, HTTP status `000`,
TLS verification `0`). An HTTP/1.1 retry was not completed.
## Current API and bounded code 6 recovery (2026-10-20)
The current official site uses `POST /api/v1/GetDocuments` (page size 12),
`GET /api/v1/GetDocument?DocumentCode=...`, and
`GET /api/v1/GetEdition?editionId=...&lang=ru|kg`. A bounded first-page probe
reported 210,639 records; this recovery does not enumerate the full list. The
official alias `1-4` resolves to canonical code `205064`. Code `6` is present
and has 201 editions. The edition order in `GetDocument` is authoritative;
numeric IDs do not define currentness.
For `lang=kg`, the API returns Kyrgyz HTML in `contentRu` and `contentKg=null`.
The bounded adapter maps that response by the requested language to archive
`ky`, preserves the response and checkpoints, limits requests to one per
second, caps JSON responses at 16 MiB, and caps one document's text at 4 GiB.
It writes only to a separate generation under `/tmp`; it neither resumes nor
overwrites the existing ignored corpus.
Without `exact=true`, a request for `920292` returned response field
`id=920492` despite different text length/hash. Two exact probes for `920292`
and `920492` returned matching IDs and names/dates (`04.08.1998` and
`13.04.2017`) matching `GetDocument`; their RU text lengths/hashes were
485,487 / `2b51c2ff4e9b91cbe2a7ed052a5a9ed9150bda993fc15c120ac948580926b718`
and 1,139,908 / `e5f0fdbb47766e5a5dbb971183d765e4c1dcaf9c969a218a84206244ce9132e8`.
The adapter now requires `exact=true`, returned-ID equality, and matching RU/KY
names before a language checkpoint can complete. Its versioned request
contract is part of both staging metadata and the SQLite resume identity;
pre-contract generations cannot be resumed.
The adapter does not independently validate attachments or embedded media.
Even after the text stage completes, the raw manifest records this as an
unresolved error. Do not merge this single document into the existing corpus,
claim corpus completeness, or transfer it to production without a separate
review and comparison. The prior no-exact pass at
`/tmp/akyldash-current-code6.ZAMC37u4ex` is marked quarantined in
`qualification.json`; its 240 text checkpoints and 304,493,520 bytes are
retained as unverified, with no document directory published. Its preserved
work-tree SHA-256 is
`1e874c1d7fab1e00e7e45ff1b16422ea474e22ee13c901ac9a0612f47aa5d24a`. It must
not be reused or normalized.
The exact-only generation at `/tmp/akyldash-current-code6-exact.L4k3AvnOlP`
contains 201 editions and 402 complete RU/KY text checkpoints (604,377,885
raw bytes). The raw manifest passes `quick_check`; its stored source SHA-256
matches the staged document tree at
`81d8e1a53b2ebef61f08b56d04b3f0a80077d587a4fadae0152bb4b7f13055ea`. Separate
normalization at `/tmp/akyldash-current-code6-exact.L4k3AvnOlP-normalized`
passes `quick_check`, records normalizer `1.1.0`, and preserves first/last
edition order `920292` (`04.08.1998`) to `920492` (`13.04.2017`). The normalized
generation is about 2.05 GB. Neither temporary generation has been merged into
the existing corpus. Attachments and embedded media remain unverified.
The first full pass (run `b4a4556e30cc464d90bc8d75ef1673e9`) stopped before
validation with `Python int too large to convert to SQLite INTEGER` after
2,120,420 raw entries and 397,856 normalized entries. This was an inventory
+52 -4
View File
@@ -63,8 +63,8 @@ protected file metadata remain unverified.
unverified.
- No public application listener is observed on TCP `80`/`443`. Docker
container/volume state still requires the root audit. Repository Compose describes
Caddy `2.11.6` pinned by digest, Python Search API `0.11.3`, SQLite
account/review data, and OpenSearch `3.7.0` with `analysis-icu`.
Caddy `2.11.6` pinned by digest, Python Search API `0.12.1` in this branch,
SQLite account/review data, and OpenSearch `3.7.0` with `analysis-icu`.
- The target publishes only Caddy on TCP `80`/`443`. Backend and Caddy share a
proxy network; OpenSearch and backend share a separate internal network, and
backend has no host port. Caddy certificate storage, OpenSearch, and
@@ -127,8 +127,9 @@ protected file metadata remain unverified.
reached healthy status for OpenSearch, backend, and Caddy, and returned HTTP
200 for `/review` through Caddy. Its containers, networks, and volumes were
removed afterward. This does not verify VPS startup, DNS, or public TLS.
- The app has no migration framework or SMTP setup. Current app versions:
backend `0.11.3`, frontend `0.1.2`, Telegram bot `0.2.2` (the bot is not
- The app has no migration framework or SMTP setup. Repository versions on this
branch are:
backend `0.12.1`, frontend `0.1.2`, Telegram bot `0.2.2` (the bot is not
deployed on this VPS).
## Corpus readiness (2026-10-18)
@@ -168,6 +169,53 @@ protected file metadata remain unverified.
unknown. No corpus was transferred and no production service, Docker socket,
runner policy, or VPS configuration was changed.
## Current MinJust API staging (2026-10-20)
- The public site's current API uses `POST /api/v1/GetDocuments` with page size
12, `GET /api/v1/GetDocument?DocumentCode=...`, and
`GET /api/v1/GetEdition?editionId=...&lang=ru|kg`. A bounded page-1 probe
reported 210,639 filtered records; no full listing was enumerated for this
recovery. Official alias `1-4` resolves to canonical document code `205064`.
- Document code `6` has 201 official editions. The API's `lang=kg` response
places Kyrgyz HTML in `contentRu` while returning `contentKg=null`; staging
maps by requested language (`kg` to archive `ky`) and preserves the edition
order from the document response. Normalization and search export now retain
and use that order instead of sorting IDs.
- Without `exact=true`, one probe for edition `920292` returned response field
`id=920492` despite a different text length/hash. Two bounded `exact=true`
probes returned matching IDs and names/dates from `GetDocument`: edition
`920292` / `04.08.1998` (485,487 RU characters, SHA-256
`2b51c2ff4e9b91cbe2a7ed052a5a9ed9150bda993fc15c120ac948580926b718`) and
edition `920492` / `13.04.2017` (1,139,908 RU characters, SHA-256
`e5f0fdbb47766e5a5dbb971183d765e4c1dcaf9c969a218a84206244ce9132e8`).
Every edition request now uses `exact=true`, requires returned ID equality,
and checks RU/KY names against `GetDocument` metadata.
- `backend/ingestion/minjust_current.py` stages one document into a separate
generation, at no more than one request per second, with a 16 MiB response
limit, a 4 GiB text budget per document, contract-bound checkpoints, and
atomic publication. Staging requires a private 0700 directory owned by the
current user and rejects symlinks under checkpoints/document paths; legacy
corpus exclusions are anchored to the repository root. The corpus verifier
continues to validate unchanged normalized manifests from versions 1.0.0
and 1.1.0. The earlier non-exact pass is quarantined at
`/tmp/akyldash-current-code6.ZAMC37u4ex/qualification.json`: its 240 language
responses (304,493,520 bytes) remain unverified and are not reusable. The
retained work-tree SHA-256 is
`1e874c1d7fab1e00e7e45ff1b16422ea474e22ee13c901ac9a0612f47aa5d24a`.
- Exact-only code-6 staging is complete at
`/tmp/akyldash-current-code6-exact.L4k3AvnOlP`: 201 editions and 402
complete language checkpoints, 604,377,885 raw bytes. The raw manifest
passes `quick_check`; its document checksum matches the staged directory at
`81d8e1a53b2ebef61f08b56d04b3f0a80077d587a4fadae0152bb4b7f13055ea`.
Separate normalization at
`/tmp/akyldash-current-code6-exact.L4k3AvnOlP-normalized` contains 201
editions in official response order, passes `quick_check`, and records
normalizer `1.1.0` (2,048,797,024 bytes). The prior `data/minjust-cbd` and
`data/minjust-normalized` trees and production were not used as input or
modified. The raw staging manifest still records attachments and embedded
media as unverified; this single document does not establish corpus
completeness or production readiness.
## Checks not completed
- Docker Engine version from the server API, container and volume inventory,
+2 -2
View File
@@ -5,7 +5,7 @@
- Telegram-бот: `0.2.2`
- Telegram-бот на Synology: `0.2.1`
- Backend: `0.11.3`
- Backend: `0.12.1`
- Frontend: `0.1.2` — поиск, лаборатория релевантности, аккаунты
## Краткий итог
@@ -282,4 +282,4 @@
---
Акылдаш · Telegram-бот v0.2.2 · Backend v0.11.3 · Frontend v0.1.2
Акылдаш · Telegram-бот v0.2.2 · Backend v0.12.1 · Frontend v0.1.2
+1 -1
View File
@@ -262,4 +262,4 @@ runtime-зависимостями frontend. Регистрация в стор
---
Акылдаш · Telegram-бот v0.2.2 · Backend v0.11.3 · Frontend v0.1.2
Акылдаш · Telegram-бот v0.2.2 · Backend v0.12.1 · Frontend v0.1.2
+1 -1
View File
@@ -44,4 +44,4 @@
---
Акылдаш · Telegram-бот v0.2.2 · Backend v0.11.3 · Frontend v0.1.2
Акылдаш · Telegram-бот v0.2.2 · Backend v0.12.1 · Frontend v0.1.2
+24 -3
View File
@@ -14,6 +14,7 @@ from verify_corpus import (
prefix_upper_bound,
raw_file_rows,
combined_source_hash,
NORMALIZER_VERSION,
safe_relative_path,
validate_fragment_file,
verify_corpus,
@@ -39,7 +40,7 @@ class CorpusInventoryTests(unittest.TestCase):
)
connection.close()
def _nonempty_corpus(self):
def _nonempty_corpus(self, normalizer_version=NORMALIZER_VERSION):
document = self.raw / "documents/1"
edition = document / "editions/2"
edition.mkdir(parents=True)
@@ -79,7 +80,7 @@ class CorpusInventoryTests(unittest.TestCase):
document_json = {
"schema_version": "1",
"source_code": "1",
"normalizer": {"version": "1.0.0"},
"normalizer": {"version": normalizer_version},
"source": {"path": "documents/1", "files": source_files, "sha256": source_sha},
"editions": [{"source_code": "2"}],
}
@@ -98,7 +99,9 @@ class CorpusInventoryTests(unittest.TestCase):
normalized_manifest.execute(
"CREATE TABLE documents (code TEXT PRIMARY KEY, source_sha256 TEXT, schema_version TEXT, normalizer_version TEXT, state TEXT)"
)
normalized_manifest.execute("INSERT INTO documents VALUES ('1', ?, '1', '1.0.0', 'success')", (source_sha,))
normalized_manifest.execute(
"INSERT INTO documents VALUES ('1', ?, '1', ?, 'success')", (source_sha, normalizer_version)
)
normalized_manifest.commit()
normalized_manifest.close()
return normalized, source_sha
@@ -273,6 +276,24 @@ class CorpusInventoryTests(unittest.TestCase):
self.assertEqual(result["manifest_summary"]["raw"]["documents"], 1)
self.assertEqual(result["manifest_summary"]["normalized"]["fragments"], 1)
def test_verifies_legacy_normalizer_1_0_0_manifest(self):
normalized, _ = self._nonempty_corpus(normalizer_version="1.0.0")
result = verify_corpus(self.raw, normalized, self.base / "artifacts")
self.assertEqual(result["status"], "complete", result["finding_examples"])
self.assertNotIn(
"normalizer_version_mismatch",
{item["kind"] for item in result["finding_examples"]},
)
def test_rejects_unsupported_normalizer_version(self):
normalized, _ = self._nonempty_corpus(normalizer_version="9.9.9")
result = verify_corpus(self.raw, normalized, self.base / "artifacts")
self.assertEqual(result["status"], "failed")
self.assertIn(
"normalizer_version_mismatch",
{item["kind"] for item in result["finding_examples"]},
)
def test_raw_normalized_source_mismatch_is_reported(self):
normalized, _ = self._nonempty_corpus()
manifest = sqlite3.connect(normalized / "manifest.sqlite3")
+2 -1
View File
@@ -27,6 +27,7 @@ NORMALIZER_ROOT = Path(__file__).resolve().parents[2]
TOOL_SHA256 = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
sys.path.insert(0, str(NORMALIZER_ROOT))
from backend.normalization.minjust_cbd import NORMALIZER_VERSION, SCHEMA_VERSION as NORMALIZED_SCHEMA_VERSION # noqa: E402
SUPPORTED_NORMALIZER_VERSIONS = frozenset({"1.0.0", NORMALIZER_VERSION})
class InventoryError(RuntimeError):
@@ -552,7 +553,7 @@ def load_manifest(
finding(connection, run_id, root_id, "normalized_manifest_state", f"documents/{code}", str(state))
if root_id == "normalized" and schema_version != NORMALIZED_SCHEMA_VERSION:
finding(connection, run_id, root_id, "schema_version_mismatch", f"documents/{code}", str(schema_version))
if root_id == "normalized" and normalizer_version != NORMALIZER_VERSION:
if root_id == "normalized" and normalizer_version not in SUPPORTED_NORMALIZER_VERSIONS:
finding(connection, run_id, root_id, "normalizer_version_mismatch", f"documents/{code}", str(normalizer_version))
if root_id == "raw":
connection.execute("DELETE FROM manifest_errors WHERE root_id='raw'")