2026-10-18 19:56:24 +04:00

101 lines
4.7 KiB
Python

"""Адаптер шага 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,
)