From fa0f3e7a7f1511a46c37676993baea10d4cb32e7 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 18 Oct 2026 19:56:24 +0400 Subject: [PATCH] =?UTF-8?q?feat(normalizer):=20=D0=B0=D0=B4=D0=B0=D0=BF?= =?UTF-8?q?=D1=82=D0=B5=D1=80=20=D1=88=D0=B0=D0=B3=D0=B0=201=20=E2=80=94?= =?UTF-8?q?=20=D1=82=D0=B0=D0=B1=D0=BB=D0=B8=D1=86=D0=B0=20json=20=D0=98?= =?UTF-8?q?=D0=A1=20=E2=86=92=20FormCandidate=20(#46)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W --- api/src/normalizer/ocr/adapter.py | 100 ++++++++++++++++++++++ api/tests/unit/normalizer/test_adapter.py | 98 +++++++++++++++++++++ 2 files changed, 198 insertions(+) create mode 100644 api/src/normalizer/ocr/adapter.py create mode 100644 api/tests/unit/normalizer/test_adapter.py diff --git a/api/src/normalizer/ocr/adapter.py b/api/src/normalizer/ocr/adapter.py new file mode 100644 index 0000000..0c915d1 --- /dev/null +++ b/api/src/normalizer/ocr/adapter.py @@ -0,0 +1,100 @@ +"""Адаптер шага 1: таблица json ИС Сканирование → `FormCandidate` пакета `arm_statements`. + +Спека scan-gateway «Адаптер json ИС шага 1», 05-domain-rules §11.1. Адаптер только раскладывает ячейки по сетке +(строка ИС → строка сетки, `column_index` → позиция) и копирует шапку; текст не очищается (это делает пакет). +Чистые функции: без I/O и логов. +""" + +import math +from collections.abc import Mapping +from typing import TypeGuard + +from arm_statements import Cell, Evidence, FormCandidate, HeaderCells, Rules, Source, type_hint + + +class TableStructureError(ValueError): + """Нарушена структура таблицы ответа ИС.""" + + +def confidence_of(raw: object) -> float: + """Уверенность в `[0, 100]`; не число, `bool`, NaN, бесконечность и отсутствие дают 0.0.""" + if isinstance(raw, bool) or not isinstance(raw, int | float): + return 0.0 + value = float(raw) + if not math.isfinite(value): + return 0.0 + return min(100.0, max(0.0, value)) + + +def _label_cell(raw: object, ref: str) -> Cell | None: + if not isinstance(raw, Mapping): + return None + text = raw.get("form_name") + if not isinstance(text, str) or not text.strip(): + return None + return Cell(text, None, confidence_of(raw.get("confidence")), ref) + + +def header_cells(data_ocr: Mapping[str, object]) -> HeaderCells: + """Шапка из `data_ocr`: organization, inn, units, date; остальные поля — `None`.""" + + def field(name: str) -> Cell | None: + return _label_cell(data_ocr.get(name), f"data_ocr.{name}") + + return HeaderCells( + organization_name=field("organization"), + inn=field("inn"), + report_date=field("date"), + units=field("units"), + ) + + +def _is_int(raw: object) -> TypeGuard[int]: + return isinstance(raw, int) and not isinstance(raw, bool) + + +def table_candidate(file_name: str, index: int, table: object, header: HeaderCells, rules: Rules) -> FormCandidate: + """Кандидат формы на одну таблицу json; бросает `TableStructureError` при нарушении структуры.""" + if not isinstance(table, Mapping): + raise TableStructureError(f"таблица {index}: не объект") + raw_rows = table.get("rows") + if not isinstance(raw_rows, list): + raise TableStructureError(f"таблица {index}: rows не список") + + parsed: dict[int, dict[int, Cell]] = {} + for raw_row in raw_rows: + if not isinstance(raw_row, Mapping): + raise TableStructureError(f"таблица {index}: строка не объект") + row_index = raw_row.get("row_index") + if not _is_int(row_index): + raise TableStructureError(f"таблица {index}: row_index не целое") + if row_index in parsed: + raise TableStructureError(f"таблица {index}: row_index {row_index} повторяется") + raw_cells = raw_row.get("cells") + if not isinstance(raw_cells, list): + raise TableStructureError(f"таблица {index}, строка {row_index}: cells не список") + cells: dict[int, Cell] = {} + for raw_cell in raw_cells: + if not isinstance(raw_cell, Mapping): + raise TableStructureError(f"таблица {index}, строка {row_index}: ячейка не объект") + column = raw_cell.get("column_index") + if not _is_int(column) or column < 0: + raise TableStructureError(f"таблица {index}, строка {row_index}: неверный column_index") + if column in cells: + raise TableStructureError(f"таблица {index}, строка {row_index}: column_index {column} повторяется") + value = raw_cell.get("value") + text = "" if value is None else value if isinstance(value, str) else str(value) + cells[column] = Cell(text, None, confidence_of(raw_cell.get("confidence")), f"row:{row_index}") + parsed[row_index] = cells + + width = max((max(cells) + 1 for cells in parsed.values() if cells), default=0) + rows = tuple(tuple(parsed[row_index].get(column) for column in range(width)) for row_index in sorted(parsed)) + title = _label_cell(table.get("label"), f"table:{index}.label") + return FormCandidate( + source=Source.OCR, + locator=f"file:{file_name};table:{index}", + type_hint=type_hint(None, title.text if title else None, rules), + evidence=Evidence(okud=None, title=title), + header=header, + rows=rows, + ) diff --git a/api/tests/unit/normalizer/test_adapter.py b/api/tests/unit/normalizer/test_adapter.py new file mode 100644 index 0000000..646778b --- /dev/null +++ b/api/tests/unit/normalizer/test_adapter.py @@ -0,0 +1,98 @@ +import math + +import pytest + +from arm_statements import Cell, DocumentType, Source, default_rules +from src.normalizer.ocr.adapter import TableStructureError, confidence_of, header_cells, table_candidate + +RULES = default_rules() +DATA_OCR = { + "label": {"form_name": "Бухгалтерская отчетность", "confidence": 100.0}, + "organization": {"form_name": "ООО «Пример»", "confidence": 100.0}, + "inn": {"form_name": "3|1 21 2|", "confidence": 61.538464}, + "units": {"form_name": "тыс. руб.", "confidence": 100.0}, + "date": {"form_name": "", "confidence": 0.0}, +} + + +def row(index: object, *cells: tuple[object, object, object]) -> dict[str, object]: + return {"row_index": index, "cells": [{"column_index": c, "value": v, "confidence": p} for c, v, p in cells]} + + +def table(label: str, *rows: dict[str, object]) -> dict[str, object]: + return {"label": {"form_name": label, "confidence": 100.0}, "rows": list(rows)} + + +def test_header_is_copied_from_json() -> None: + h = header_cells(DATA_OCR) + assert h.organization_name == Cell("ООО «Пример»", None, 100.0, "data_ocr.organization") + assert h.inn == Cell("3|1 21 2|", None, 61.538464, "data_ocr.inn") + assert h.units == Cell("тыс. руб.", None, 100.0, "data_ocr.units") + assert (h.report_date, h.okud, h.okpo, h.okopf, h.okfs, h.okei, h.address) == (None,) * 7 + + +def test_grid_by_row_and_column_index() -> None: + t = table( + "Бухгалтерский баланс", + row(9, (0, "Итого", 100.0)), + row(8, (1, "[190 :", 25.0), (0, "Прочие", 100.0), (4, "", 0.0), (5, None, 100.0)), + ) + cand = table_candidate("j.json", 0, t, header_cells(DATA_OCR), RULES) + assert (cand.source, cand.locator, cand.type_hint) == (Source.OCR, "file:j.json;table:0", DocumentType.BALANCE) + assert cand.segment_breaks == () and cand.warnings == () + assert cand.evidence.okud is None + assert cand.evidence.title == Cell("Бухгалтерский баланс", None, 100.0, "table:0.label") + assert cand.rows == ( + ( + Cell("Прочие", None, 100.0, "row:8"), + Cell("[190 :", None, 25.0, "row:8"), + None, + None, + Cell("", None, 0.0, "row:8"), + Cell("", None, 100.0, "row:8"), + ), + (Cell("Итого", None, 100.0, "row:9"), None, None, None, None, None), + ) + assert cand.header == header_cells(DATA_OCR) + + +def test_multiline_text_is_kept() -> None: + t = table("Бухгалтерский баланс", row(93, (1, "1500\n1700", 78.0), (2, "917 775\n2 371 333", 100.0))) + cand = table_candidate("j.json", 0, t, header_cells(DATA_OCR), RULES) + assert cand.rows[0][1] == Cell("1500\n1700", None, 78.0, "row:93") + + +def test_income_label_and_unknown_label() -> None: + income = table_candidate("j.json", 1, table("Отчет о финансовых результатах"), header_cells(DATA_OCR), RULES) + assert income.type_hint is DocumentType.INCOME and income.rows == () + blank = table_candidate( + "j.json", 2, {"label": {"form_name": " ", "confidence": 0.0}, "rows": []}, header_cells(DATA_OCR), RULES + ) + assert (blank.evidence.title, blank.type_hint) == (None, None) + + +@pytest.mark.parametrize( + ("raw", "expected"), + [(25, 25.0), (150.0, 100.0), (-3, 0.0), (math.nan, 0.0), (math.inf, 0.0), ("100", 0.0), (True, 0.0), (None, 0.0)], +) +def test_confidence_of(raw: object, expected: float) -> None: + assert confidence_of(raw) == expected + + +@pytest.mark.parametrize( + "broken", + [ + "не объект", + {"label": {}, "rows": "x"}, + table("ББ", {"cells": []}), + table("ББ", row(True, (0, "a", 100.0))), + table("ББ", row(1, (0, "a", 100.0)), row(1, (0, "b", 100.0))), + table("ББ", {"row_index": 0, "cells": "x"}), + table("ББ", row(0, ("1", "a", 100.0))), + table("ББ", row(0, (-1, "a", 100.0))), + table("ББ", row(0, (1, "a", 100.0), (1, "b", 100.0))), + ], +) +def test_broken_structure_raises(broken: object) -> None: + with pytest.raises(TableStructureError): + table_candidate("j.json", 0, broken, header_cells(DATA_OCR), RULES)