feat(normalizer): адаптер шага 1 — таблица json ИС → FormCandidate (#46)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
This commit is contained in:
Andrey Epifancev
parent
43f74fbd50
commit
fa0f3e7a7f
100
api/src/normalizer/ocr/adapter.py
Normal file
100
api/src/normalizer/ocr/adapter.py
Normal file
@ -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,
|
||||
)
|
||||
98
api/tests/unit/normalizer/test_adapter.py
Normal file
98
api/tests/unit/normalizer/test_adapter.py
Normal file
@ -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)
|
||||
Loading…
x
Reference in New Issue
Block a user