#!/usr/bin/env python3 """Read-only reconciliation of the V4 workbook, canonical labs, staging and documents. Default mode never writes SQLite or the workbook. Detailed reports and staging plans are private artifacts (0600) and must stay outside Git. Aggregate output contains counts, contract names and source digests only. """ from __future__ import annotations import argparse import hashlib import json import os import re import shutil import sqlite3 import stat import sys import tempfile import unicodedata from collections import Counter, defaultdict from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from typing import Any HERE = Path(__file__).resolve().parent if str(HERE) not in sys.path: sys.path.insert(0, str(HERE)) from dashboard_v5.lab_registry import LAB_ALLOWLIST # noqa: E402 from dashboard_v5.read_api import _lab_key, _verified_lab_rows, probe_original, table_exists # noqa: E402 from health_pipeline import best_reference_xlsx, get_lab_matrix, parse_reference_xlsx # noqa: E402 BASE = Path.home() / ".hermes" / "assets" / "Gesundheit" DEFAULT_DB = BASE / "health_data.db" PRIVATE_REPORT_DIR = BASE / "reports" / "private" CLASSIFICATIONS = ( "exact_match", "workbook_only", "database_only", "staging_only", "value_conflict", "unit_conflict", "missing_unit", "missing_reference", "duplicate_same_day", "verified_but_not_catalogued", "catalogued_but_no_verified_data", "linked_reviewed_document", "linked_unreviewed_document", "missing_document_link", "ambiguous_document_link", ) def _text(value: Any) -> str: return " ".join(str(value or "").strip().split()) def normalize_parameter(value: Any) -> str: text = unicodedata.normalize("NFKD", _text(value).casefold()) text = "".join(char for char in text if not unicodedata.combining(char)) return re.sub(r"[^a-z0-9]+", "_", text).strip("_") def normalize_date(value: Any) -> str | None: text = _text(value)[:10] try: return datetime.strptime(text, "%Y-%m-%d").date().isoformat() except ValueError: return None def normalize_unit(value: Any) -> str | None: text = _text(value).replace("μ", "µ").replace(" ", "") return text.casefold() or None def normalize_value(value: Any) -> str | None: text = _text(value).replace("'", "").replace(" ", "").replace(",", ".") match = re.fullmatch(r"([<>]=?|=)?([-+]?\d+(?:\.\d+)?)", text) if not match: return None try: number = Decimal(match.group(2)).normalize() except InvalidOperation: return None return f"{match.group(1) or '='}{format(number, 'f')}" def _digest(records: list[dict[str, Any]]) -> str: raw = json.dumps(records, ensure_ascii=False, sort_keys=True, separators=(",", ":")) return hashlib.sha256(raw.encode("utf-8")).hexdigest() def _columns(connection: sqlite3.Connection, table: str) -> set[str]: if not table_exists(connection, table): return set() return {str(row[1]) for row in connection.execute(f"PRAGMA table_info({table})")} def _row_dicts(connection: sqlite3.Connection, table: str) -> list[dict[str, Any]]: if not table_exists(connection, table): return [] return [dict(row) for row in connection.execute(f"SELECT * FROM {table} ORDER BY id")] def _lab_record(row: dict[str, Any], source: str) -> dict[str, Any]: if source == "staging": value = row.get("wert_final") or row.get("wert_korrigiert") or row.get("wert_extrahiert") reference = row.get("referenzbereich") or row.get("reference_min") or row.get("reference_max") else: value = row.get("wert") reference = row.get("reference_min") or row.get("reference_max") day = normalize_date(row.get("abnahme_datum") or row.get("befund_datum")) return { "source": source, "row_id": row.get("id"), "parameter": normalize_parameter(row.get("parameter_name")), "date": day, "unit": normalize_unit(row.get("einheit")), "value": normalize_value(value), "reference_present": bool(_text(reference)), } def _workbook_records(path: Path | None) -> tuple[list[dict[str, Any]], dict[str, Any]]: selected = path or best_reference_xlsx() values = parse_reference_xlsx(selected) dates, parameters, matrix, _categories, _references = get_lab_matrix(values) records = [ { "source": "workbook", "row_id": None, "parameter": normalize_parameter(item.parameter), "date": normalize_date(item.date), "unit": normalize_unit(item.unit), "value": normalize_value(item.value), "reference_present": bool(_text(item.reference)), } for item in values ] contract = { "best_reference_xlsx": path is None, "parse_reference_xlsx": True, "get_lab_matrix": True, "matrix_dates": len(dates), "matrix_parameters": len(parameters), "matrix_cells": len(matrix), } return records, contract def _key(record: dict[str, Any]) -> tuple[str, str | None]: return str(record["parameter"]), record["date"] def _catalogued(record: dict[str, Any]) -> bool: return _lab_key(record.get("parameter"), record.get("unit")) in LAB_ALLOWLIST def _canonical_identity(record: dict[str, Any]) -> tuple[str, str | None] | None: public = LAB_ALLOWLIST.get(_lab_key(record.get("parameter"), record.get("unit"))) if not public: return None return normalize_parameter(public[0]), normalize_unit(public[1]) def _canonical_tuple(record: dict[str, Any]) -> tuple[str, str | None, str | None, str | None] | None: identity = _canonical_identity(record) if identity is None or record.get("date") is None or record.get("value") is None: return None return identity[0], identity[1], record["date"], record["value"] def _original_status(row: dict[str, Any]) -> str: reviewed = row.get("review_status") == "geprueft" probe = probe_original( row.get("local_original_path") or row.get("dateipfad"), reviewed=reviewed ) try: return probe.status finally: probe.close() def run_audit(db_path: Path, workbook_path: Path | None = None) -> dict[str, Any]: workbook, v4_contract = _workbook_records(workbook_path) uri = f"file:{db_path.resolve()}?mode=ro" connection = sqlite3.connect(uri, uri=True) connection.row_factory = sqlite3.Row try: connection.execute("PRAGMA query_only=ON") if connection.execute("PRAGMA query_only").fetchone()[0] != 1: raise RuntimeError("read-only query contract unavailable") db_raw = _row_dicts(connection, "laborwerte") staging_raw = _row_dicts(connection, "laborwerte_staging") documents = _row_dicts(connection, "dokumente") db = [_lab_record(row, "database") for row in db_raw] staging = [_lab_record(row, "staging") for row in staging_raw] canonical_rows = _verified_lab_rows( connection, include_missing_reference=True ) canonical_records = [ { "source": "canonical_v5", "row_id": None, "parameter": normalize_parameter(row["parameter"]), "date": row["date"], "unit": normalize_unit(row["unit"]), "value": normalize_value(row["value"]), "reference_present": row.get("reference_status") != "missing_reference", } for row in canonical_rows ] counts: Counter[str] = Counter({name: 0 for name in CLASSIFICATIONS}) details: dict[str, list[dict[str, Any]]] = {name: [] for name in CLASSIFICATIONS} by_source: dict[str, dict[tuple[str, str | None], list[dict[str, Any]]]] = {} for name, records in (("workbook", workbook), ("database", db), ("staging", staging)): grouped: dict[tuple[str, str | None], list[dict[str, Any]]] = defaultdict(list) for record in records: grouped[_key(record)].append(record) by_source[name] = grouped for record in workbook: same = by_source["database"].get(_key(record), []) if not same: counts["workbook_only"] += 1 details["workbook_only"].append(record) continue exact = [ candidate for candidate in same if record["value"] is not None and candidate["value"] is not None and candidate["value"] == record["value"] and candidate["unit"] == record["unit"] ] if exact: counts["exact_match"] += 1 else: if record["value"] is None or any( candidate["value"] is None or candidate["value"] != record["value"] for candidate in same ): counts["value_conflict"] += 1 details["value_conflict"].append({"workbook": record, "database": same}) if any(candidate["unit"] != record["unit"] for candidate in same): counts["unit_conflict"] += 1 details["unit_conflict"].append({"workbook": record, "database": same}) for record in db: if _key(record) not in by_source["workbook"]: counts["database_only"] += 1 details["database_only"].append(record) for record in staging: if _key(record) not in by_source["workbook"] and _key(record) not in by_source["database"]: counts["staging_only"] += 1 details["staging_only"].append(record) for record in [*workbook, *db, *staging]: if record["unit"] is None: counts["missing_unit"] += 1 details["missing_unit"].append(record) for record, raw in zip(db, db_raw, strict=True): if not record["reference_present"]: counts["missing_reference"] += 1 details["missing_reference"].append(record) verified = ( _text(raw.get("validierungsstatus")).casefold() == "validiert" and int(raw.get("verified_against_original") or 0) == 1 and _text(raw.get("reference_range_source")).casefold() == "scanned_original" ) if verified and not _catalogued(record): counts["verified_but_not_catalogued"] += 1 details["verified_but_not_catalogued"].append(record) for key, rows in by_source["database"].items(): if len(rows) > 1: counts["duplicate_same_day"] += 1 details["duplicate_same_day"].append({"key": key, "count": len(rows)}) verified_pairs = { LAB_ALLOWLIST[_lab_key(record["parameter"], record["unit"])] for record, raw in zip(db, db_raw, strict=True) if _text(raw.get("validierungsstatus")).casefold() == "validiert" and int(raw.get("verified_against_original") or 0) == 1 and _text(raw.get("reference_range_source")).casefold() == "scanned_original" and _lab_key(record["parameter"], record["unit"]) in LAB_ALLOWLIST } for pair in sorted(set(LAB_ALLOWLIST.values())): if pair not in verified_pairs: counts["catalogued_but_no_verified_data"] += 1 details["catalogued_but_no_verified_data"].append({"catalogue_contract": pair}) document_by_id = {row.get("id"): row for row in documents} exact_backfills: list[dict[str, int]] = [] for row in db_raw: legacy = row.get("dokument_id") canonical_document = row.get("canonical_document_id") if legacy and canonical_document and legacy != canonical_document: counts["ambiguous_document_link"] += 1 details["ambiguous_document_link"].append({"lab_id": row.get("id"), "legacy": legacy, "canonical": canonical_document}) continue linked_id = canonical_document or legacy document = document_by_id.get(linked_id) if document is None: counts["missing_document_link"] += 1 details["missing_document_link"].append({"lab_id": row.get("id")}) continue if canonical_document is None and legacy is not None: exact_backfills.append({"lab_id": int(row["id"]), "document_id": int(legacy)}) classification = "linked_reviewed_document" if document.get("review_status") == "geprueft" else "linked_unreviewed_document" counts[classification] += 1 details[classification].append({"lab_id": row.get("id"), "document_id": linked_id}) document_statuses = [_original_status(row) for row in documents] extracted_ids = {int(row["id"]) for row in documents if _text(row.get("extrahierte_inhalte"))} reviewed_ids = {int(row["id"]) for row in documents if row.get("review_status") == "geprueft"} fts_ids: set[int] = set() if table_exists(connection, "health_document_fts"): fts_ids = { int(row[0]) for row in connection.execute("SELECT DISTINCT document_id FROM health_document_fts") if str(row[0]).isdigit() } safe_records = { "workbook": sorted([{k: row[k] for k in ("parameter", "date", "unit", "value")} for row in workbook], key=lambda row: tuple(str(row[k]) for k in row)), "database": sorted([{k: row[k] for k in ("parameter", "date", "unit", "value")} for row in db], key=lambda row: tuple(str(row[k]) for k in row)), "staging": sorted([{k: row[k] for k in ("parameter", "date", "unit", "value")} for row in staging], key=lambda row: tuple(str(row[k]) for k in row)), "canonical_v5": sorted([{k: row[k] for k in ("parameter", "date", "unit", "value")} for row in canonical_records], key=lambda row: tuple(str(row[k]) for k in row)), } canonical_exact = { identity for row in canonical_records if (identity := _canonical_tuple(row)) } workbook_not_canonical = [ row for row in workbook if _canonical_tuple(row) not in canonical_exact ] actionable_staging_identities = { _digest([{key: row.get(key) for key in ("parameter", "date", "unit", "value")}]) for row in workbook_not_canonical if row.get("date") is not None and row.get("value") is not None } workbook_canonical_exact = sum( _canonical_tuple(row) in canonical_exact for row in workbook ) aggregate = { "schema_version": 1, "mode": "read_only_fail_closed", "contracts": ["best_reference_xlsx", "parse_reference_xlsx", "get_lab_matrix", "laborwerte", "laborwerte_staging", "dokumente", "health_document_fts", "exact_document_relation_only"], "v4_contract": v4_contract, "source_counts": {"workbook_values": len(workbook), "laborwerte_rows": len(db), "laborwerte_staging_rows": len(staging), "canonical_v5_rows": len(canonical_records)}, "parity": { "workbook_exact_in_raw_database": counts["exact_match"], "workbook_only_raw": counts["workbook_only"], "workbook_exact_in_canonical_v5": workbook_canonical_exact, "workbook_not_in_canonical_v5": len(workbook) - workbook_canonical_exact, "staging_review_candidates": len(workbook_not_canonical), "staging_plan_unique_actionable": len(actionable_staging_identities), }, "classifications": dict(sorted(counts.items())), "document_coverage": { "registered": len(documents), "original_available": sum(status == "available" for status in document_statuses), "text_extracted": len(extracted_ids), "content_reviewed": len(reviewed_ids), "fulltext_searchable": len(fts_ids & reviewed_ids), }, "linkage": { "linked": counts["linked_reviewed_document"] + counts["linked_unreviewed_document"], "missing": counts["missing_document_link"], "ambiguous": counts["ambiguous_document_link"], "safe_exact_backfill_candidates": len(exact_backfills), "manual_review_required": counts["missing_document_link"] + counts["ambiguous_document_link"], }, "digests": {name: _digest(records) for name, records in safe_records.items()}, } return { "aggregate": aggregate, "details": details, "exact_backfills": exact_backfills, "staging_candidates": workbook_not_canonical, } finally: connection.close() def staging_plan(audit: dict[str, Any]) -> dict[str, Any]: candidates = [] excluded = [] seen: set[str] = set() for row in audit["staging_candidates"]: identity = _digest([{key: row.get(key) for key in ("parameter", "date", "unit", "value")}]) if row.get("date") is None or row.get("value") is None: excluded.append({"candidate_id": identity, "reason": "incomplete_date_or_numeric_value"}) continue if identity in seen: excluded.append({"candidate_id": identity, "reason": "duplicate_normalized_candidate"}) continue seen.add(identity) candidates.append({ "candidate_id": identity, "parameter": row["parameter"], "date": row["date"], "unit": row["unit"], "value": row["value"], "status": "zur_pruefung", "provenance": "legacy_reference_xlsx", "verified_against_original": 0, "reference_range_source": "legacy_reference_xlsx_unverified", }) return { "mode": "plan_only_no_database_write", "idempotency_key": "candidate_id", "discrepancy_count": len(audit["staging_candidates"]), "covered_count": len(candidates) + len(excluded), "candidates": candidates, "excluded": excluded, } def _reject_symlink_components(path: Path) -> None: current = Path(path.anchor) for part in path.parts[1:]: current /= part try: mode = os.lstat(current).st_mode except FileNotFoundError: continue if stat.S_ISLNK(mode): raise RuntimeError("private path components must not be symlinks") def _private_destination(path: Path) -> Path: root_candidate = Path(os.path.abspath(PRIVATE_REPORT_DIR.expanduser())) _reject_symlink_components(root_candidate) root_candidate.mkdir(parents=True, exist_ok=True, mode=0o700) _reject_symlink_components(root_candidate) os.chmod(root_candidate, 0o700) root = root_candidate.resolve(strict=True) candidate = path.expanduser() if not candidate.is_absolute(): candidate = Path.cwd() / candidate candidate = Path(os.path.abspath(candidate)) if not candidate.is_relative_to(root): raise RuntimeError("private artifact must stay below the private report root") _reject_symlink_components(candidate.parent) candidate.parent.mkdir(parents=True, exist_ok=True, mode=0o700) _reject_symlink_components(candidate.parent) if candidate.is_symlink() or (candidate.exists() and not candidate.is_file()): raise RuntimeError("private artifact target must be a regular non-symlink file") return candidate def _private_directory(path: Path) -> Path: marker = _private_destination(path / ".private-directory-contract") directory = marker.parent os.chmod(directory, 0o700) return directory def _write_private(path: Path, payload: Any) -> None: destination = _private_destination(path) descriptor, temporary_name = tempfile.mkstemp( prefix=f".{destination.name}.", dir=destination.parent ) temporary = Path(temporary_name) try: os.fchmod(descriptor, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True) handle.write("\n") handle.flush() os.fsync(handle.fileno()) os.replace(temporary, destination) os.chmod(destination, 0o600) status = destination.stat(follow_symlinks=False) if status.st_uid != os.getuid() or stat.S_IMODE(status.st_mode) != 0o600: raise RuntimeError("private artifact ownership or mode contract failed") finally: if temporary.exists(): temporary.unlink() def _validation_digest(connection: sqlite3.Connection) -> str: rows = [dict(row) for row in connection.execute("SELECT id,validierungsstatus,verified_against_original,reference_range_source FROM laborwerte ORDER BY id")] return _digest(rows) def _digest_value(value: Any) -> Any: if isinstance(value, bytes): return {"blob_sha256": hashlib.sha256(value).hexdigest(), "bytes": len(value)} return value def _database_state_digest( connection: sqlite3.Connection, allowed_links: dict[int, int] ) -> str: schema = [ tuple(row) for row in connection.execute( """SELECT type,name,tbl_name,COALESCE(sql,'') FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' ORDER BY type,name""" ) ] tables = [ str(row[0]) for row in connection.execute( "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name" ) ] content: dict[str, list[list[Any]]] = {} for table in tables: escaped = table.replace('"', '""') columns = [ str(row[1]) for row in connection.execute(f'PRAGMA table_info("{escaped}")') ] rows: list[list[Any]] = [] for raw in connection.execute(f'SELECT * FROM "{escaped}"'): values = [_digest_value(value) for value in raw] if table == "laborwerte" and "id" in columns and "canonical_document_id" in columns: row_id = int(raw[columns.index("id")]) expected = allowed_links.get(row_id) link_index = columns.index("canonical_document_id") if expected is not None and raw[link_index] in (None, expected): values[link_index] = {"intended_link": expected} rows.append(values) rows.sort(key=lambda row: json.dumps(row, ensure_ascii=False, sort_keys=True, separators=(",", ":"))) content[table] = rows return _digest([{"schema": schema, "tables": content}]) def _apply_verified_link_transaction( connection: sqlite3.Connection, candidates: list[dict[str, int]] ) -> tuple[int, int]: validation_before = _validation_digest(connection) allowed_links = {int(item["lab_id"]): int(item["document_id"]) for item in candidates} protected_before = _database_state_digest(connection, allowed_links) changed = 0 expected_changes = 0 connection.execute("BEGIN IMMEDIATE") try: for candidate in candidates: row = connection.execute( "SELECT dokument_id,canonical_document_id FROM laborwerte WHERE id=?", (candidate["lab_id"],), ).fetchone() document_exists = connection.execute( "SELECT 1 FROM dokumente WHERE id=?", (candidate["document_id"],) ).fetchone() if row is None or document_exists is None or row["dokument_id"] != candidate["document_id"]: raise RuntimeError("exact backfill relation changed since dry-run") if row["canonical_document_id"] == candidate["document_id"]: continue if row["canonical_document_id"] is not None: raise RuntimeError("canonical document relation is no longer empty") expected_changes += 1 changed += connection.execute( "UPDATE laborwerte SET canonical_document_id=? WHERE id=? AND canonical_document_id IS NULL AND dokument_id=?", (candidate["document_id"], candidate["lab_id"], candidate["document_id"]), ).rowcount second = sum( connection.execute( "UPDATE laborwerte SET canonical_document_id=? WHERE id=? AND canonical_document_id IS NULL AND dokument_id=?", (candidate["document_id"], candidate["lab_id"], candidate["document_id"]), ).rowcount for candidate in candidates ) if changed != expected_changes or second != 0: raise RuntimeError("exact backfill row-count or idempotency contract failed") if connection.execute("PRAGMA integrity_check").fetchone()[0] != "ok": raise RuntimeError("post-migration integrity check failed") if connection.execute("PRAGMA foreign_key_check").fetchall(): raise RuntimeError("post-migration foreign-key check failed") if _validation_digest(connection) != validation_before: raise RuntimeError("validation fields changed") if _database_state_digest(connection, allowed_links) != protected_before: raise RuntimeError("database changed outside intended canonical links") connection.commit() return changed, second except Exception: connection.rollback() raise def apply_exact_backfills(db_path: Path, candidates: list[dict[str, int]], backup_dir: Path, expected_digest: str) -> dict[str, Any]: candidate_digest = _digest(candidates) if not candidates or not expected_digest or candidate_digest != expected_digest: raise RuntimeError("exact backfill confirmation digest mismatch") private_backup_dir = _private_directory(backup_dir) backup = private_backup_dir / f"health_data.before_lab_links.{datetime.now(timezone.utc):%Y%m%dT%H%M%S%fZ}.sqlite" source = sqlite3.connect(db_path) source.row_factory = sqlite3.Row try: if source.execute("PRAGMA integrity_check").fetchone()[0] != "ok": raise RuntimeError("pre-migration integrity check failed") if source.execute("PRAGMA foreign_key_check").fetchall(): raise RuntimeError("pre-migration foreign-key check failed") validation_before = _validation_digest(source) allowed_links = {int(item["lab_id"]): int(item["document_id"]) for item in candidates} protected_before = _database_state_digest(source, allowed_links) descriptor = os.open(backup, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) os.close(descriptor) target = sqlite3.connect(backup) try: source.backup(target) finally: target.close() os.chmod(backup, 0o600) with tempfile.TemporaryDirectory(prefix="lab-link-validation-") as validation_dir: validation_path = Path(validation_dir) / "candidate.sqlite" shutil.copy2(backup, validation_path) validation = sqlite3.connect(validation_path) validation.row_factory = sqlite3.Row try: _apply_verified_link_transaction(validation, candidates) finally: validation.close() changed, second = _apply_verified_link_transaction(source, candidates) with tempfile.TemporaryDirectory(prefix="lab-link-restore-") as restore_dir: restored_path = Path(restore_dir) / "restored.sqlite" shutil.copy2(backup, restored_path) restored = sqlite3.connect(restored_path) restored.row_factory = sqlite3.Row try: restore_ok = ( restored.execute("PRAGMA integrity_check").fetchone()[0] == "ok" and not restored.execute("PRAGMA foreign_key_check").fetchall() and _validation_digest(restored) == validation_before and _database_state_digest(restored, allowed_links) == protected_before ) finally: restored.close() if not restore_ok: raise RuntimeError("backup restore verification failed") return {"changed": changed, "idempotent_second_run_changes": second, "integrity": "ok", "restore_test": True, "candidate_digest": candidate_digest, "backup_created": True} finally: source.close() def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--db", type=Path, default=Path(os.environ.get("HEALTH_DASHBOARD_DB", DEFAULT_DB))) parser.add_argument("--xlsx", type=Path) parser.add_argument("--private-report", type=Path) parser.add_argument("--private-staging-plan", type=Path) parser.add_argument("--aggregate-json", type=Path) parser.add_argument("--apply-exact-links", action="store_true") parser.add_argument("--confirm-digest", default="") parser.add_argument("--backup-dir", type=Path, default=PRIVATE_REPORT_DIR / "backups") args = parser.parse_args() audit = run_audit(args.db, args.xlsx) timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") private_report = args.private_report or PRIVATE_REPORT_DIR / f"lab-reconciliation-{timestamp}.json" _write_private( private_report, { "aggregate": audit["aggregate"], "details": audit["details"], "staging_plan": staging_plan(audit), }, ) if args.private_staging_plan: _write_private(args.private_staging_plan, staging_plan(audit)) if args.aggregate_json: args.aggregate_json.parent.mkdir(parents=True, exist_ok=True) args.aggregate_json.write_text(json.dumps(audit["aggregate"], ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") result: dict[str, Any] = {"aggregate": audit["aggregate"], "private_report_written": True} if args.apply_exact_links: result["migration"] = apply_exact_backfills(args.db, audit["exact_backfills"], args.backup_dir, args.confirm_digest) else: result["exact_backfill_candidate_digest"] = _digest(audit["exact_backfills"]) print(json.dumps(result, ensure_ascii=False, sort_keys=True)) return 0 if __name__ == "__main__": raise SystemExit(main())