730 lines
28 KiB
Python
730 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
"""Normalize the local Ministry of Justice CBD archive."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import html
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import unicodedata
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
from urllib.parse import urlsplit
|
|
|
|
APP_VERSION = "0.9.0"
|
|
SCHEMA_VERSION = "1"
|
|
NORMALIZER_VERSION = "1.0.0"
|
|
LANGUAGES = ("ru", "ky")
|
|
LOGGER = logging.getLogger(__name__)
|
|
|
|
ALLOWED_TAGS = {
|
|
"a", "b", "blockquote", "br", "div", "em", "h1", "h2", "h3", "h4",
|
|
"h5", "h6", "i", "img", "li", "ol", "p", "pre", "span", "strong",
|
|
"sub", "sup", "table", "tbody", "td", "tfoot", "th", "thead", "tr",
|
|
"u", "ul",
|
|
}
|
|
VOID_TAGS = {"br", "img"}
|
|
DROP_CONTENT_TAGS = {"applet", "iframe", "noscript", "object", "script", "style", "svg"}
|
|
DROP_ELEMENT_TAGS = {"link", "meta"}
|
|
BLOCK_TAGS = {"blockquote", "h1", "h2", "h3", "h4", "h5", "h6", "li", "p", "pre", "td", "th"}
|
|
AUTO_CLOSE = {
|
|
"li": {"li"},
|
|
"p": {"blockquote", "div", "h1", "h2", "h3", "h4", "h5", "h6", "li", "ol", "p", "pre", "table", "ul"},
|
|
"td": {"td", "th"},
|
|
"th": {"td", "th"},
|
|
"tr": {"tr"},
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class NormalizeResult:
|
|
discovered: int = 0
|
|
normalized: int = 0
|
|
skipped: int = 0
|
|
failed: int = 0
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def atomic_write(path: Path, content: bytes) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with tempfile.NamedTemporaryFile(dir=path.parent, delete=False) as temporary:
|
|
temporary.write(content)
|
|
temporary_path = Path(temporary.name)
|
|
os.replace(temporary_path, path)
|
|
|
|
|
|
def json_bytes(value: object) -> bytes:
|
|
return (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
|
|
|
|
|
|
def sha256_bytes(content: bytes) -> str:
|
|
return hashlib.sha256(content).hexdigest()
|
|
|
|
|
|
def source_inventory(document_directory: Path, input_root: Path) -> tuple[list[dict], str]:
|
|
files = []
|
|
combined = hashlib.sha256()
|
|
for path in sorted(item for item in document_directory.rglob("*") if item.is_file()):
|
|
relative = path.relative_to(input_root).as_posix()
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as source:
|
|
for chunk in iter(lambda: source.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
checksum = digest.hexdigest()
|
|
files.append({"path": relative, "sha256": checksum})
|
|
combined.update(relative.encode("utf-8"))
|
|
combined.update(b"\0")
|
|
combined.update(checksum.encode("ascii"))
|
|
combined.update(b"\0")
|
|
return files, combined.hexdigest()
|
|
|
|
|
|
def clean_value(value):
|
|
if isinstance(value, str):
|
|
normalized = unicodedata.normalize("NFC", value)
|
|
return normalized if normalized.strip() else None
|
|
if isinstance(value, dict):
|
|
return {key: clean_value(item) for key, item in value.items()}
|
|
if isinstance(value, list):
|
|
return [clean_value(item) for item in value]
|
|
return value
|
|
|
|
|
|
def bilingual(value) -> dict[str, object]:
|
|
value = value if isinstance(value, dict) else {}
|
|
return {"ru": clean_value(value.get("Rus")), "ky": clean_value(value.get("Kyr"))}
|
|
|
|
|
|
def hierarchy_paths(items: object, child_key: str) -> list[dict]:
|
|
paths: list[dict] = []
|
|
|
|
def visit(nodes: object, ancestors: dict[str, list[str]]) -> None:
|
|
for node in nodes if isinstance(nodes, list) else []:
|
|
if not isinstance(node, dict):
|
|
continue
|
|
names = bilingual(node.get("Name"))
|
|
current = {language: list(ancestors[language]) for language in LANGUAGES}
|
|
for language in LANGUAGES:
|
|
name = names[language]
|
|
if name:
|
|
current[language].append(str(name))
|
|
children = node.get(child_key)
|
|
if children:
|
|
visit(children, current)
|
|
else:
|
|
paths.append(current)
|
|
|
|
visit(items, {"ru": [], "ky": []})
|
|
return paths
|
|
|
|
|
|
class SafeHtmlParser(HTMLParser):
|
|
def __init__(self, edition_directory: Path) -> None:
|
|
super().__init__(convert_charrefs=True)
|
|
self.edition_directory = edition_directory.resolve()
|
|
self.parts: list[str] = []
|
|
self.stack: list[str] = []
|
|
self.drop_depth = 0
|
|
self.removed_elements = 0
|
|
self.removed_attributes = 0
|
|
self.removed_images = 0
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
tag = tag.lower()
|
|
if self.drop_depth:
|
|
if tag in DROP_CONTENT_TAGS:
|
|
self.drop_depth += 1
|
|
return
|
|
if tag in DROP_CONTENT_TAGS:
|
|
self.drop_depth = 1
|
|
self.removed_elements += 1
|
|
return
|
|
if tag in DROP_ELEMENT_TAGS:
|
|
self.removed_elements += 1
|
|
return
|
|
if tag not in ALLOWED_TAGS:
|
|
self.removed_elements += 1
|
|
return
|
|
for open_tag, closing_tags in AUTO_CLOSE.items():
|
|
if tag in closing_tags and open_tag in self.stack:
|
|
self._close(open_tag)
|
|
safe_attrs = self._attributes(tag, attrs)
|
|
if tag == "img" and not any(name == "src" for name, _ in safe_attrs):
|
|
self.removed_images += 1
|
|
return
|
|
rendered = "".join(
|
|
f' {name}="{html.escape(value, quote=True)}"' for name, value in safe_attrs
|
|
)
|
|
self.parts.append(f"<{tag}{rendered}>")
|
|
if tag not in VOID_TAGS:
|
|
self.stack.append(tag)
|
|
|
|
def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
self.handle_starttag(tag, attrs)
|
|
if tag.lower() not in VOID_TAGS:
|
|
self.handle_endtag(tag)
|
|
|
|
def handle_endtag(self, tag: str) -> None:
|
|
tag = tag.lower()
|
|
if self.drop_depth:
|
|
if tag in DROP_CONTENT_TAGS:
|
|
self.drop_depth -= 1
|
|
return
|
|
if tag in self.stack:
|
|
self._close(tag)
|
|
|
|
def handle_data(self, data: str) -> None:
|
|
if not self.drop_depth:
|
|
self.parts.append(html.escape(unicodedata.normalize("NFC", data), quote=False))
|
|
|
|
def close(self) -> None:
|
|
super().close()
|
|
while self.stack:
|
|
self.parts.append(f"</{self.stack.pop()}>")
|
|
|
|
def _close(self, tag: str) -> None:
|
|
while self.stack:
|
|
current = self.stack.pop()
|
|
self.parts.append(f"</{current}>")
|
|
if current == tag:
|
|
break
|
|
|
|
def _attributes(self, tag: str, attrs: list[tuple[str, str | None]]) -> list[tuple[str, str]]:
|
|
allowed = {"title"}
|
|
if tag == "a":
|
|
allowed |= {"href"}
|
|
elif tag == "img":
|
|
allowed |= {"alt", "src"}
|
|
elif tag in {"td", "th"}:
|
|
allowed |= {"colspan", "rowspan"}
|
|
safe = []
|
|
for raw_name, raw_value in attrs:
|
|
name = raw_name.lower()
|
|
value = unicodedata.normalize("NFC", raw_value or "")
|
|
if name not in allowed:
|
|
self.removed_attributes += 1
|
|
continue
|
|
if name == "href" and not safe_link(value):
|
|
self.removed_attributes += 1
|
|
continue
|
|
if name == "src" and not self.safe_image(value):
|
|
self.removed_attributes += 1
|
|
continue
|
|
if name in {"colspan", "rowspan"} and not value.isdigit():
|
|
self.removed_attributes += 1
|
|
continue
|
|
safe.append((name, value))
|
|
if tag == "a" and any(name == "href" and urlsplit(value).scheme in {"http", "https"} for name, value in safe):
|
|
safe.append(("rel", "noopener noreferrer"))
|
|
return safe
|
|
|
|
def safe_image(self, value: str) -> bool:
|
|
try:
|
|
parsed = urlsplit(value)
|
|
except ValueError:
|
|
return False
|
|
if parsed.scheme or parsed.netloc or not parsed.path or parsed.path.startswith(("/", "\\")):
|
|
return False
|
|
candidate = (self.edition_directory / parsed.path.replace("\\", "/")).resolve()
|
|
try:
|
|
candidate.relative_to(self.edition_directory)
|
|
except ValueError:
|
|
return False
|
|
return candidate.is_file()
|
|
|
|
|
|
def safe_link(value: str) -> bool:
|
|
value = value.strip()
|
|
if not value or value.startswith(("//", "\\\\")):
|
|
return False
|
|
try:
|
|
parsed = urlsplit(value)
|
|
except ValueError:
|
|
return False
|
|
return parsed.scheme.lower() in {"", "http", "https", "mailto"} and not (
|
|
not parsed.scheme and parsed.netloc
|
|
)
|
|
|
|
|
|
def sanitize_html(source: str, edition_directory: Path) -> tuple[str, dict]:
|
|
parser = SafeHtmlParser(edition_directory)
|
|
parser.feed(source)
|
|
parser.close()
|
|
return unicodedata.normalize("NFC", "".join(parser.parts)), {
|
|
"removed_elements": parser.removed_elements,
|
|
"removed_attributes": parser.removed_attributes,
|
|
"removed_images": parser.removed_images,
|
|
}
|
|
|
|
|
|
class TextBlockParser(HTMLParser):
|
|
def __init__(self) -> None:
|
|
super().__init__(convert_charrefs=True)
|
|
self.blocks: list[tuple[str, str]] = []
|
|
self.active_tag: str | None = None
|
|
self.active: list[str] = []
|
|
self.loose: list[str] = []
|
|
|
|
def handle_starttag(self, tag: str, attrs) -> None:
|
|
if tag in BLOCK_TAGS:
|
|
self._flush_active()
|
|
self._flush_loose()
|
|
self.active_tag = tag
|
|
elif tag == "br":
|
|
(self.active if self.active_tag else self.loose).append("\n")
|
|
|
|
def handle_endtag(self, tag: str) -> None:
|
|
if tag == self.active_tag:
|
|
self._flush_active()
|
|
|
|
def handle_data(self, data: str) -> None:
|
|
(self.active if self.active_tag else self.loose).append(data)
|
|
|
|
def close(self) -> None:
|
|
super().close()
|
|
self._flush_active()
|
|
self._flush_loose()
|
|
|
|
def _flush_active(self) -> None:
|
|
if self.active_tag:
|
|
text = clean_text("".join(self.active))
|
|
if text:
|
|
self.blocks.append((self.active_tag, text))
|
|
self.active_tag = None
|
|
self.active = []
|
|
|
|
def _flush_loose(self) -> None:
|
|
text = clean_text("".join(self.loose))
|
|
if text:
|
|
self.blocks.append(("p", text))
|
|
self.loose = []
|
|
|
|
|
|
def clean_text(value: str) -> str:
|
|
lines = []
|
|
for line in unicodedata.normalize("NFC", value).replace("\xa0", " ").splitlines():
|
|
line = re.sub(r"[ \t\f\v]+", " ", line).strip()
|
|
if line:
|
|
lines.append(line)
|
|
return "\n".join(lines)
|
|
|
|
|
|
def fragment_type(tag: str, text: str, language: str) -> str:
|
|
lowered = text.casefold()
|
|
article_words = ("статья", "ст.") if language == "ru" else ("берене", "статья")
|
|
if any(re.match(rf"^{re.escape(word)}\s*\d", lowered) for word in article_words):
|
|
return "article"
|
|
if re.match(r"^\d+(?:\.\d+)*[.)]?\s+", text):
|
|
return "point"
|
|
if tag.startswith("h"):
|
|
return "heading"
|
|
return {"li": "list_item", "td": "table_cell", "th": "table_header"}.get(tag, "paragraph")
|
|
|
|
|
|
def extract_text_and_fragments(
|
|
sanitized: str,
|
|
document_code: str,
|
|
edition_code: str,
|
|
language: str,
|
|
source_path: str,
|
|
source_sha256: str,
|
|
) -> tuple[str, list[dict]]:
|
|
parser = TextBlockParser()
|
|
parser.feed(sanitized)
|
|
parser.close()
|
|
fragments = []
|
|
for position, (tag, text) in enumerate(parser.blocks, 1):
|
|
fragments.append(
|
|
{
|
|
"id": f"document:{document_code}:edition:{edition_code}:lang:{language}:fragment:{position}",
|
|
"document_code": document_code,
|
|
"edition_code": edition_code,
|
|
"language": language,
|
|
"position": position,
|
|
"type": fragment_type(tag, text, language),
|
|
"text": text,
|
|
"text_sha256": sha256_bytes(text.encode("utf-8")),
|
|
"source_path": source_path,
|
|
"source_sha256": source_sha256,
|
|
}
|
|
)
|
|
return "\n\n".join(fragment["text"] for fragment in fragments), fragments
|
|
|
|
|
|
def load_json(path: Path) -> dict:
|
|
value = json.loads(path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"Expected JSON object: {path}")
|
|
return value
|
|
|
|
|
|
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 {
|
|
"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(),
|
|
}
|
|
|
|
|
|
def normalize_document(
|
|
document_directory: Path,
|
|
input_root: Path,
|
|
destination: Path,
|
|
files: list[dict] | None = None,
|
|
source_checksum: str | None = None,
|
|
) -> None:
|
|
metadata_path = document_directory / "metadata.json"
|
|
metadata = load_json(metadata_path)
|
|
document_code = str(metadata.get("Code", document_directory.name))
|
|
if document_code != document_directory.name:
|
|
raise ValueError(f"Document code mismatch in {metadata_path}")
|
|
if files is None or source_checksum is None:
|
|
files, source_checksum = source_inventory(document_directory, input_root)
|
|
file_checksums = {item["path"]: item["sha256"] for item in files}
|
|
edition_root = document_directory / "editions"
|
|
edition_directories = sorted(
|
|
(path for path in edition_root.iterdir() if path.is_dir()),
|
|
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]
|
|
available_languages = [language for language in LANGUAGES if any(language in item["available_languages"] for item in summaries)]
|
|
normalized_metadata = clean_value(metadata)
|
|
document = {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"source_code": document_code,
|
|
"class": bilingual(metadata.get("Class")),
|
|
"type": bilingual(metadata.get("Type")),
|
|
"title": bilingual(metadata.get("Title")),
|
|
"name": bilingual(metadata.get("Name")),
|
|
"status": bilingual(metadata.get("Status")),
|
|
"number": clean_value(metadata.get("Number")),
|
|
"dates": {key: value for key, value in normalized_metadata.items() if key.startswith("Date")},
|
|
"registration_number": clean_value(metadata.get("NumberRegistration")),
|
|
"publication_number": clean_value(metadata.get("NumberPublication")),
|
|
"is_public_in_cdb": metadata.get("IsPublicInCdb"),
|
|
"is_public_in_register": metadata.get("IsPublicInRegister"),
|
|
"authorities": normalized_metadata.get("Authorities") or [],
|
|
"authority_paths": hierarchy_paths(metadata.get("Authorities"), "Authorities"),
|
|
"source_publications": normalized_metadata.get("SourcePublications") or [],
|
|
"source_publication_paths": hierarchy_paths(metadata.get("SourcePublications"), "SourcePublications"),
|
|
"keywords": normalized_metadata.get("Keywords") or [],
|
|
"keyword_paths": hierarchy_paths(metadata.get("Keywords"), "Keywords"),
|
|
"general_classifiers": normalized_metadata.get("GeneralClassifiers") or [],
|
|
"general_classifier_paths": hierarchy_paths(metadata.get("GeneralClassifiers"), "GeneralClassifiers"),
|
|
"references": normalized_metadata.get("References") or [],
|
|
"source_metadata": normalized_metadata,
|
|
"available_languages": available_languages,
|
|
"editions": summaries,
|
|
"source": {
|
|
"path": document_directory.relative_to(input_root).as_posix(),
|
|
"files": files,
|
|
"sha256": source_checksum,
|
|
},
|
|
"normalizer": {"version": NORMALIZER_VERSION, "processed_at": utc_now()},
|
|
}
|
|
atomic_write(destination / "document.json", json_bytes(document))
|
|
|
|
for edition_directory, summary in zip(edition_directories, summaries):
|
|
edition_code = summary["source_code"]
|
|
edition_metadata = load_json(edition_directory / "metadata.json")
|
|
edition_destination = destination / "editions" / edition_directory.name
|
|
image_records = []
|
|
for image in edition_metadata.get("Images") or []:
|
|
language = {"Russian": "ru", "Kyrgyz": "ky"}.get(image.get("Lang"), "unknown")
|
|
name = Path(str(image.get("Name") or "").replace("\\", "/")).name
|
|
source_path = edition_directory / "images" / language / name
|
|
relative = source_path.relative_to(input_root).as_posix()
|
|
image_records.append(
|
|
{
|
|
"language": language,
|
|
"name": clean_value(image.get("Name")),
|
|
"source_path": relative if source_path.is_file() else None,
|
|
"source_sha256": file_checksums.get(relative),
|
|
"source_metadata": clean_value(image),
|
|
}
|
|
)
|
|
quality = {"has_html": bool(summary["available_languages"]), "languages": {}}
|
|
for language in summary["available_languages"]:
|
|
html_path = edition_directory / f"{language}.html"
|
|
relative = html_path.relative_to(input_root).as_posix()
|
|
raw = html_path.read_text(encoding="utf-8")
|
|
sanitized, sanitizer_quality = sanitize_html(raw, edition_directory)
|
|
text, fragments = extract_text_and_fragments(
|
|
sanitized, document_code, edition_code, language, relative, file_checksums[relative]
|
|
)
|
|
language_destination = edition_destination / language
|
|
atomic_write(language_destination / "content.html", sanitized.encode("utf-8"))
|
|
atomic_write(language_destination / "content.txt", (text + ("\n" if text else "")).encode("utf-8"))
|
|
atomic_write(language_destination / "fragments.json", json_bytes(fragments))
|
|
quality["languages"][language] = {
|
|
**sanitizer_quality,
|
|
"empty_text": not bool(text),
|
|
"fragment_count": len(fragments),
|
|
}
|
|
edition = {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"source_code": edition_code,
|
|
"name": summary["name"],
|
|
"source_type": summary["source_type"],
|
|
"available_languages": summary["available_languages"],
|
|
"images": image_records,
|
|
"source_metadata": clean_value(edition_metadata),
|
|
"source": {
|
|
"path": summary["source_path"],
|
|
"files": [item for item in files if item["path"].startswith(summary["source_path"] + "/")],
|
|
},
|
|
"quality": quality,
|
|
}
|
|
atomic_write(edition_destination / "edition.json", json_bytes(edition))
|
|
|
|
|
|
def connect_manifest(path: Path) -> sqlite3.Connection:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
connection = sqlite3.connect(path)
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS documents (
|
|
code TEXT PRIMARY KEY,
|
|
source_sha256 TEXT,
|
|
schema_version TEXT NOT NULL,
|
|
normalizer_version TEXT NOT NULL,
|
|
processed_at TEXT,
|
|
state TEXT NOT NULL,
|
|
error TEXT,
|
|
failed_at TEXT
|
|
)
|
|
"""
|
|
)
|
|
columns = {row[1] for row in connection.execute("PRAGMA table_info(documents)")}
|
|
if "failed_at" not in columns:
|
|
connection.execute("ALTER TABLE documents ADD COLUMN failed_at TEXT")
|
|
return connection
|
|
|
|
|
|
def publish_directory(staged: Path, target: Path) -> None:
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
backup = target.parent / f".{target.name}.previous"
|
|
if backup.exists():
|
|
shutil.rmtree(backup)
|
|
if target.exists():
|
|
os.replace(target, backup)
|
|
try:
|
|
os.replace(staged, target)
|
|
except Exception:
|
|
if backup.exists():
|
|
os.replace(backup, target)
|
|
raise
|
|
if backup.exists():
|
|
shutil.rmtree(backup)
|
|
|
|
|
|
def recover_directory(target: Path) -> None:
|
|
backup = target.parent / f".{target.name}.previous"
|
|
if not backup.exists():
|
|
return
|
|
if target.exists():
|
|
shutil.rmtree(backup)
|
|
else:
|
|
os.replace(backup, target)
|
|
|
|
|
|
def validate_roots(input_root: Path, output: Path) -> None:
|
|
source = input_root.resolve()
|
|
destination = output.resolve()
|
|
if source == destination or source.is_relative_to(destination) or destination.is_relative_to(source):
|
|
raise ValueError("--input and --output must not overlap")
|
|
|
|
|
|
def normalize_archive(
|
|
input_root: Path = Path("data/minjust-cbd"),
|
|
output: Path = Path("data/minjust-normalized"),
|
|
limit: int | None = None,
|
|
refresh: bool = False,
|
|
progress: Callable[[NormalizeResult, int], None] | None = None,
|
|
) -> NormalizeResult:
|
|
validate_roots(input_root, output)
|
|
document_root = input_root / "documents"
|
|
if not document_root.is_dir():
|
|
raise FileNotFoundError(f"Document directory not found: {document_root}")
|
|
output.mkdir(parents=True, exist_ok=True)
|
|
connection = connect_manifest(output / "manifest.sqlite3")
|
|
known = {
|
|
row[0]: (row[1], row[2], row[3], row[4])
|
|
for row in connection.execute(
|
|
"SELECT code, source_sha256, schema_version, normalizer_version, state FROM documents"
|
|
)
|
|
}
|
|
directories = sorted(
|
|
(path for path in document_root.iterdir() if path.is_dir()),
|
|
key=lambda path: (not path.name.isdigit(), int(path.name) if path.name.isdigit() else path.name),
|
|
)
|
|
if limit is not None:
|
|
directories = directories[:limit]
|
|
total = len(directories)
|
|
discovered = normalized = skipped = failed = 0
|
|
staging_root = output / ".staging"
|
|
staging_root.mkdir(exist_ok=True)
|
|
try:
|
|
for source_directory in directories:
|
|
discovered += 1
|
|
code = source_directory.name
|
|
target = output / "documents" / code
|
|
checksum = None
|
|
try:
|
|
recover_directory(target)
|
|
files, checksum = source_inventory(source_directory, input_root)
|
|
if target.is_dir() and not refresh and known.get(code) == (
|
|
checksum, SCHEMA_VERSION, NORMALIZER_VERSION, "success"
|
|
):
|
|
skipped += 1
|
|
else:
|
|
with tempfile.TemporaryDirectory(dir=staging_root) as temporary:
|
|
staged = Path(temporary) / code
|
|
normalize_document(source_directory, input_root, staged, files, checksum)
|
|
publish_directory(staged, target)
|
|
with connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO documents (
|
|
code, source_sha256, schema_version,
|
|
normalizer_version, processed_at, state, error,
|
|
failed_at
|
|
) VALUES (?, ?, ?, ?, ?, 'success', NULL, NULL)
|
|
ON CONFLICT(code) DO UPDATE SET
|
|
source_sha256=excluded.source_sha256,
|
|
schema_version=excluded.schema_version,
|
|
normalizer_version=excluded.normalizer_version,
|
|
processed_at=excluded.processed_at,
|
|
state='success', error=NULL, failed_at=NULL
|
|
""",
|
|
(code, checksum, SCHEMA_VERSION, NORMALIZER_VERSION, utc_now()),
|
|
)
|
|
normalized += 1
|
|
except Exception as error: # Keep a corpus run alive after one malformed record.
|
|
LOGGER.exception("Failed to normalize document %s", code)
|
|
with connection:
|
|
connection.execute(
|
|
"""
|
|
INSERT INTO documents (
|
|
code, source_sha256, schema_version,
|
|
normalizer_version, processed_at, state, error,
|
|
failed_at
|
|
) VALUES (?, ?, ?, ?, NULL, 'error', ?, ?)
|
|
ON CONFLICT(code) DO UPDATE SET
|
|
source_sha256=excluded.source_sha256,
|
|
schema_version=excluded.schema_version,
|
|
normalizer_version=excluded.normalizer_version,
|
|
state='error', error=excluded.error,
|
|
failed_at=excluded.failed_at
|
|
""",
|
|
(
|
|
code,
|
|
checksum,
|
|
SCHEMA_VERSION,
|
|
NORMALIZER_VERSION,
|
|
str(error),
|
|
utc_now(),
|
|
),
|
|
)
|
|
failed += 1
|
|
result = NormalizeResult(discovered, normalized, skipped, failed)
|
|
if progress:
|
|
progress(result, total)
|
|
elif discovered % 100 == 0:
|
|
LOGGER.info("discovered=%s normalized=%s skipped=%s failed=%s", discovered, normalized, skipped, failed)
|
|
finally:
|
|
connection.close()
|
|
try:
|
|
staging_root.rmdir()
|
|
except OSError:
|
|
pass
|
|
return NormalizeResult(discovered, normalized, skipped, failed)
|
|
|
|
|
|
def format_duration(seconds: float) -> str:
|
|
seconds = max(0, round(seconds))
|
|
hours, seconds = divmod(seconds, 3600)
|
|
minutes, seconds = divmod(seconds, 60)
|
|
return f"{hours:02d}:{minutes:02d}:{seconds:02d}"
|
|
|
|
|
|
def progress_line(result: NormalizeResult, total: int, elapsed: float, width: int = 24) -> str:
|
|
fraction = result.discovered / total if total else 0
|
|
filled = min(width, round(width * fraction))
|
|
rate = result.discovered / elapsed if elapsed > 0 else 0
|
|
eta = (total - result.discovered) / rate if rate else 0
|
|
return (
|
|
f"[{'#' * filled}{'-' * (width - filled)}] {fraction:6.2%} "
|
|
f"{result.discovered}/{total} normalized={result.normalized} "
|
|
f"skipped={result.skipped} failed={result.failed} "
|
|
f"rate={rate:.2f}/s ETA={format_duration(eta)}"
|
|
)
|
|
|
|
|
|
def terminal_progress() -> Callable[[NormalizeResult, int], None]:
|
|
started = time.monotonic()
|
|
|
|
def update(result: NormalizeResult, total: int) -> None:
|
|
print(
|
|
f"\r{progress_line(result, total, time.monotonic() - started)}",
|
|
end="\n" if result.discovered >= total else "",
|
|
file=sys.stderr,
|
|
flush=True,
|
|
)
|
|
|
|
return update
|
|
|
|
|
|
def parse_args() -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--input", type=Path, default=Path("data/minjust-cbd"))
|
|
parser.add_argument("--output", type=Path, default=Path("data/minjust-normalized"))
|
|
parser.add_argument("--limit", type=int, help="normalize only the first N documents")
|
|
parser.add_argument("--refresh", action="store_true", help="renormalize unchanged documents")
|
|
parser.add_argument("--log-level", choices=("DEBUG", "INFO", "WARNING", "ERROR"), default="INFO")
|
|
parser.add_argument("--version", action="version", version=APP_VERSION)
|
|
return parser.parse_args()
|
|
|
|
|
|
def main() -> int:
|
|
arguments = parse_args()
|
|
if arguments.limit is not None and arguments.limit <= 0:
|
|
raise SystemExit("--limit must be greater than zero")
|
|
logging.basicConfig(level=arguments.log_level, format="%(asctime)s %(levelname)s %(message)s")
|
|
result = normalize_archive(
|
|
arguments.input,
|
|
arguments.output,
|
|
arguments.limit,
|
|
arguments.refresh,
|
|
terminal_progress() if sys.stderr.isatty() else None,
|
|
)
|
|
print(
|
|
f"discovered={result.discovered} normalized={result.normalized} "
|
|
f"skipped={result.skipped} failed={result.failed}\n"
|
|
f"Akyldash Backend v{APP_VERSION} · Frontend — not created"
|
|
)
|
|
return int(result.failed > 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|