Codex Agent
07381ad70f
Add a resumable standard-library normalization pipeline for the downloaded CBD archive. It produces canonical bilingual metadata, sanitized HTML, plain text, deterministic fragments, checksums, quality markers, and an SQLite processing manifest while preserving the raw source. Recover document-list pagination when the Ministry API exhausts request retries, and cover that scenario with a regression test. Document the normalization workflow and frontend-search MVP plan, include the source functional specification, ignore local runtime logs, and bump the backend version to 0.2.1.
122 lines
5.6 KiB
Python
122 lines
5.6 KiB
Python
import json
|
|
import sqlite3
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from normalization.minjust_cbd import normalize_archive
|
|
|
|
|
|
class MinjustNormalizationTest(unittest.TestCase):
|
|
def write_json(self, path: Path, value: object) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8")
|
|
|
|
def edition(self, root: Path, code: int, languages: dict[str, str]) -> None:
|
|
directory = root / "editions" / str(code)
|
|
self.write_json(
|
|
directory / "metadata.json",
|
|
{"Code": code, "Name": {"Rus": "Редакция", "Kyr": "Редакция"}, "Type": "edition", "Images": []},
|
|
)
|
|
for language, content in languages.items():
|
|
(directory / f"{language}.html").write_text(content, encoding="utf-8")
|
|
|
|
def document(self, root: Path, code: int = 1) -> Path:
|
|
directory = root / "documents" / str(code)
|
|
self.write_json(
|
|
directory / "metadata.json",
|
|
{
|
|
"Code": code,
|
|
"Class": {"Rus": "Акты", "Kyr": "Актылар"},
|
|
"Type": {"Rus": "Закон", "Kyr": "Мыйзам"},
|
|
"Title": {"Rus": " ", "Kyr": None},
|
|
"Name": {"Rus": "Документ", "Kyr": "Документ"},
|
|
"Status": {"Rus": "Действует", "Kyr": "Күчүндө"},
|
|
"Number": "1",
|
|
"DateAdopted": "2026-01-01",
|
|
"IsPublicInCdb": True,
|
|
"IsPublicInRegister": True,
|
|
"Authorities": [],
|
|
"SourcePublications": [],
|
|
"Keywords": [],
|
|
"GeneralClassifiers": [],
|
|
"References": [],
|
|
},
|
|
)
|
|
return directory
|
|
|
|
def test_languages_safety_empty_document_and_deterministic_fragments(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
base = Path(temporary)
|
|
source = base / "source"
|
|
output = base / "normalized"
|
|
document = self.document(source)
|
|
self.edition(
|
|
document,
|
|
10,
|
|
{
|
|
"ru": '<meta charset=unicode><style>x</style><p style="color:red">Статья 1 Закон<script>bad()</script></p><a href="javascript:bad">ссылка</a>',
|
|
},
|
|
)
|
|
self.edition(document, 20, {"ky": "<p>1. Кыргызча жобо</p>"})
|
|
self.edition(document, 30, {"ru": "<p>Русский</p>", "ky": "<p>Кыргызча</p>"})
|
|
self.edition(document, 40, {})
|
|
|
|
first = normalize_archive(source, output)
|
|
fragments_path = output / "documents/1/editions/10/ru/fragments.json"
|
|
fragments = fragments_path.read_bytes()
|
|
second = normalize_archive(source, output)
|
|
|
|
self.assertEqual((first.normalized, second.skipped), (1, 1))
|
|
self.assertEqual(fragments, fragments_path.read_bytes())
|
|
safe_html = (output / "documents/1/editions/10/ru/content.html").read_text(encoding="utf-8")
|
|
self.assertNotIn("script", safe_html)
|
|
self.assertNotIn("style=", safe_html)
|
|
self.assertNotIn("javascript:", safe_html)
|
|
self.assertIn("Статья 1 Закон", safe_html)
|
|
parsed = json.loads(fragments)
|
|
self.assertEqual(parsed[0]["type"], "article")
|
|
self.assertEqual(parsed[0]["id"], "document:1:edition:10:lang:ru:fragment:1")
|
|
self.assertEqual(len(parsed[0]["text_sha256"]), 64)
|
|
canonical = json.loads((output / "documents/1/document.json").read_text(encoding="utf-8"))
|
|
self.assertEqual(canonical["available_languages"], ["ru", "ky"])
|
|
self.assertIsNone(canonical["title"]["ru"])
|
|
empty = json.loads((output / "documents/1/editions/40/edition.json").read_text(encoding="utf-8"))
|
|
self.assertFalse(empty["quality"]["has_html"])
|
|
|
|
def test_continues_after_bad_document_and_clears_repaired_error(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
base = Path(temporary)
|
|
source = base / "source"
|
|
output = base / "normalized"
|
|
bad = source / "documents/1"
|
|
bad.mkdir(parents=True)
|
|
(bad / "metadata.json").write_text("not json", encoding="utf-8")
|
|
good = self.document(source, 2)
|
|
self.edition(good, 10, {"ky": "<p>Берене 1 Текст</p>"})
|
|
|
|
with self.assertLogs("normalization.minjust_cbd", level="ERROR"):
|
|
failed = normalize_archive(source, output)
|
|
with sqlite3.connect(output / "manifest.sqlite3") as connection:
|
|
state, failed_at = connection.execute(
|
|
"SELECT state, failed_at FROM documents WHERE code='1'"
|
|
).fetchone()
|
|
self.assertEqual(state, "error")
|
|
self.assertIsNotNone(failed_at)
|
|
self.write_json(bad / "metadata.json", {"Code": 1, "Name": {"Rus": "Исправлен", "Kyr": None}})
|
|
repaired = normalize_archive(source, output)
|
|
|
|
self.assertEqual((failed.failed, failed.normalized), (1, 1))
|
|
self.assertEqual((repaired.normalized, repaired.skipped, repaired.failed), (1, 1, 0))
|
|
with sqlite3.connect(output / "manifest.sqlite3") as connection:
|
|
self.assertEqual(
|
|
connection.execute(
|
|
"SELECT state, error, failed_at FROM documents WHERE code='1'"
|
|
).fetchone(),
|
|
("success", None, None),
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|