diff --git a/api/src/normalizer/ocr/layout.py b/api/src/normalizer/ocr/layout.py new file mode 100644 index 0000000..79eccbf --- /dev/null +++ b/api/src/normalizer/ocr/layout.py @@ -0,0 +1,100 @@ +"""Раскладка колонок таблицы OCR (§11.1 п. 3). Перенос `_build_layout`, `_infer_layout`, `_cells_by_index`, +`_cell_value` старого ядра: справочник кодов приходит параметром, ячейки — `Cell` вместо словарей.""" + +from collections.abc import Collection, Mapping, Sequence +from dataclasses import dataclass + +from src.normalizer.ocr.amounts import is_blank, parse_amount +from src.normalizer.ocr.text import is_code_header, is_value_header, normalize_header + +__all__ = ["Cell", "Cells", "ColumnLayout", "build_layout", "cell_text", "cells_by_index", "infer_layout"] + + +@dataclass(frozen=True) +class Cell: + value: str + confidence: float + + +Cells = dict[int, Cell] + + +@dataclass(frozen=True) +class ColumnLayout: + """Индексы значимых столбцов таблицы OCR: из строки заголовка или выведенные по строкам данных.""" + + label_idx: int | None + code_idx: int + value_idxs: tuple[int, ...] + + +def cells_by_index(row: Mapping[str, object]) -> Cells: + """Нормализует набор ячеек из распознанной строки в словарь по индексам колонок. + + value → str(value or ""), confidence → float(confidence or 0); + ячейка без column_index пропускается. + """ + cells: Cells = {} + raw_cells = row.get("cells") + if not isinstance(raw_cells, list): + return cells + for raw in raw_cells: + if not isinstance(raw, Mapping) or not isinstance(raw.get("column_index"), int): + continue + confidence = raw.get("confidence") + cells[int(raw["column_index"])] = Cell( + str(raw.get("value") or ""), float(confidence) if isinstance(confidence, int | float) else 0.0 + ) + return cells + + +def cell_text(cells: Cells, idx: int | None) -> str: + """Возвращает текст ячейки по индексу; отсутствующей или None индекс — пустая строка.""" + if idx is None or idx not in cells: + return "" + return cells[idx].value + + +def build_layout(cells: Cells) -> ColumnLayout | None: + """Раскладка, если строка является заголовком (есть ячейка «Код…»): наименование — ячейка с «наименование», + иначе последняя заполненная левее кода; значения — правее кода, где год или «на »/«за ».""" + headers = {idx: normalize_header(cells[idx].value) for idx in sorted(cells)} + code_idx = next((idx for idx, text in headers.items() if is_code_header(text)), None) + if code_idx is None: + return None + label_idx = next((idx for idx, text in headers.items() if "наименование" in text), None) + if label_idx is None: + left_filled = [idx for idx, text in headers.items() if idx < code_idx and text] + label_idx = left_filled[-1] if left_filled else None + value_idxs = tuple(idx for idx, text in headers.items() if idx > code_idx and is_value_header(text)) + return ColumnLayout(label_idx=label_idx, code_idx=code_idx, value_idxs=value_idxs) + + +def infer_layout(rows_cells: Sequence[Cells], known_codes: Collection[str]) -> ColumnLayout | None: + """Раскладка по строкам данных, когда заголовка нет: код — столбец с наибольшим числом + известных кодов; наименование — левее кода с наибольшим числом текстовых ячеек; значения — + правее кода, где суммы не реже мусора или столбец пуст. Столбец оценивается по всем + строкам сразу, поэтому пустые ячейки отдельных строк не влияют.""" + columns: dict[int, list[str]] = {} + for cells in rows_cells: + for idx, cell in cells.items(): + columns.setdefault(idx, []).append(cell.value.strip()) + code_hits = {idx: sum(1 for v in values if v in known_codes) for idx, values in columns.items()} + code_idx = min(code_hits, key=lambda idx: (-code_hits[idx], idx), default=None) + if code_idx is None or code_hits[code_idx] == 0: + return None + text_hits = { + idx: sum(1 for v in values if any(c.isalpha() for c in v)) for idx, values in columns.items() if idx < code_idx + } + label_idx = min(text_hits, key=lambda idx: (-text_hits[idx], -idx), default=None) + if label_idx is not None and text_hits[label_idx] == 0: + label_idx = None + value_idxs = [] + for idx in sorted(columns): + if idx <= code_idx: + continue + filled = [v for v in columns[idx] if not is_blank(v)] + amounts = sum(1 for v in filled if parse_amount(v).amount is not None) + if amounts >= len(filled) - amounts: + value_idxs.append(idx) + return ColumnLayout(label_idx=label_idx, code_idx=code_idx, value_idxs=tuple(value_idxs)) diff --git a/api/tests/unit/normalizer/test_layout.py b/api/tests/unit/normalizer/test_layout.py new file mode 100644 index 0000000..91020d9 --- /dev/null +++ b/api/tests/unit/normalizer/test_layout.py @@ -0,0 +1,96 @@ +"""Раскладка колонок таблицы OCR (§11.1 п. 3): шапка «Код» и вывод по данным. + +Перенос `_build_layout`/`_infer_layout` из старого ядра.""" + +from src.normalizer.ocr.layout import Cell, ColumnLayout, build_layout, cell_text, cells_by_index, infer_layout + +KNOWN = frozenset({"1110", "1120", "1130", "1150", "2110", "2120"}) + + +def _row(*cells: tuple[int, str, float]) -> dict: + return {"row_index": 0, "cells": [{"column_index": i, "value": v, "confidence": c} for i, v, c in cells]} + + +def test_cells_by_index_normalizes_types() -> None: + cells = cells_by_index(_row((0, "Выручка", 71.4), (1, None, None))) + assert cells == {0: Cell("Выручка", 71.4), 1: Cell("", 0.0)} + assert cell_text(cells, 0) == "Выручка" + assert cell_text(cells, 7) == "" + assert cell_text(cells, None) == "" + + +class TestBuildLayout: + def test_header_with_code_and_years(self) -> None: + cells = cells_by_index( + _row( + (0, "Наименование показателя", 100), + (1, "Код", 100), + (2, "На 30 июня\n2025 г", 100), + (4, "На 31 декабря\n2024 г.", 100), + (5, "На 31 декабря\n2023 г.", 100), + ) + ) + assert build_layout(cells) == ColumnLayout(label_idx=0, code_idx=1, value_idxs=(2, 4, 5)) + + def test_label_defaults_to_last_filled_left_of_code(self) -> None: + cells = cells_by_index(_row((0, "Пояснения", 100), (1, "Показатель", 100), (2, "Код", 100), (3, "2024", 100))) + assert build_layout(cells) == ColumnLayout(label_idx=1, code_idx=2, value_idxs=(3,)) + + def test_no_code_cell_is_not_a_header(self) -> None: + assert build_layout(cells_by_index(_row((0, "Выручка", 100), (1, "2110", 100), (2, "13 070", 100)))) is None + + +class TestInferLayout: + def test_aep3_like_rows_without_header(self) -> None: + rows = [ + cells_by_index( + _row( + (1, "НИ", 0), + (0, "Нематериальные активы", 100), + (2, "30127", 60), + (4, "| | г", 16), + (5, "", 100), + ) + ), + cells_by_index( + _row( + (1, "1120", 50), + (0, "Результаты исследований и", 94), + (2, ":", 0), + (4, "--", 0), + (5, ":", 0), + ) + ), + cells_by_index( + _row( + (1, "1130", 80), + (0, "Нематериальные поисковые", 87), + (2, "не ****", 5), + (4, "3", 0), + (5, "+ :", 0), + ) + ), + cells_by_index( + _row( + (1, "1150", 100), + (0, "Основные средства", 100), + (2, "1 234", 100), + (4, "5 678", 100), + (5, "", 100), + ) + ), + ] + layout = infer_layout(rows, KNOWN) + assert layout is not None + assert layout.code_idx == 1 + assert layout.label_idx == 0 + # R-1: column 5 contains only garbage (empty, ':', '+ :'), no actual amounts + # column 4 has both garbage and amounts ('--', '3', '5 678') but amounts >= garbage + assert layout.value_idxs == (2, 4) + + def test_no_known_codes_is_none(self) -> None: + rows = [cells_by_index(_row((0, "текст", 100), (1, "мусор", 100)))] + assert infer_layout(rows, KNOWN) is None + + def test_empty_rows_is_none(self) -> None: + assert infer_layout([], KNOWN) is None