diff --git a/api/src/normalizer/ocr/rows.py b/api/src/normalizer/ocr/rows.py new file mode 100644 index 0000000..f684b04 --- /dev/null +++ b/api/src/normalizer/ocr/rows.py @@ -0,0 +1,129 @@ +"""Строки таблицы OCR (§11.1 п. 3–5). Перенос `_classify_row`, `_parse_table_rows`, `_synthetic_header_cells`, +`_resolve_table_type` старого ядра: шапка — строка с «Код» (повторная перезаписывает раскладку), без шапки раскладка +выводится по данным и шапка синтезируется; строки до первой шапки — по выведенной раскладке.""" + +from collections.abc import Collection, Mapping, Sequence +from dataclasses import dataclass +from enum import StrEnum +from typing import Literal + +from src.normalizer.ocr.layout import Cell, Cells, ColumnLayout, build_layout, cell_text, cells_by_index, infer_layout + +__all__ = ["ParsedRow", "ParsedTable", "RowKind", "classify_row", "parse_table_rows", "table_type_by_codes"] + +HEADER_CODE = "Код" +HEADER_LABEL = "Наименование показателя" +FULL_CONFIDENCE = 100.0 + + +class RowKind(StrEnum): + HEADER = "HEADER" + ROW = "ROW" + TOTAL = "TOTAL" + BALANCE = "BALANCE" + + +@dataclass(frozen=True) +class ParsedRow: + cells: Cells + layout: ColumnLayout + kind: RowKind + ordinal: int + + +@dataclass(frozen=True) +class ParsedTable: + """Разобранные строки: `rows[0]` — шапка (настоящая или синтетическая), дальше строки данных по порядку.""" + + rows: list[ParsedRow] + synthetic_header: bool + + +def classify_row(cells: Cells, layout: ColumnLayout) -> RowKind | None: + code = cell_text(cells, layout.code_idx).strip() + if not code: + return None + try: + int(code) + except (TypeError, ValueError): + return RowKind.HEADER + label = cell_text(cells, layout.label_idx) + if label == "Баланс": + return RowKind.BALANCE + if label.lower().startswith("итого"): + return RowKind.TOTAL + return RowKind.ROW + + +def _data_row(cells: Cells, layout: ColumnLayout, ordinal: int) -> ParsedRow | None: + kind = classify_row(cells, layout) + if kind is None or kind is RowKind.HEADER: + return None + return ParsedRow(cells, layout, kind, ordinal) + + +def _synthetic_header_cells(layout: ColumnLayout) -> Cells: + """Ячейки восстановленного заголовка: код и наименование, тексты периодов пустые (ячеек под значениями нет).""" + cells: Cells = {layout.code_idx: Cell(HEADER_CODE, FULL_CONFIDENCE)} + if layout.label_idx is not None: + cells[layout.label_idx] = Cell(HEADER_LABEL, FULL_CONFIDENCE) + return cells + + +def _ordinal(row: Mapping[str, object], fallback: int) -> int: + index = row.get("row_index") + return index if isinstance(index, int) else fallback + + +def parse_table_rows(table_rows: Sequence[Mapping[str, object]], known_codes: Collection[str]) -> ParsedTable | None: + rows_cells = [(_ordinal(row, position), cells_by_index(row)) for position, row in enumerate(table_rows)] + first_header = next( + (pos for pos, (_, cells) in enumerate(rows_cells) if build_layout(cells) is not None), len(rows_cells) + ) + before_header = rows_cells[:first_header] + inferred = infer_layout([cells for _, cells in before_header], known_codes) if before_header else None + leading_rows: list[ParsedRow] = [] + if inferred is not None: + for ordinal, cells in before_header: + parsed = _data_row(cells, inferred, ordinal) + if parsed is not None: + leading_rows.append(parsed) + + if first_header == len(rows_cells): + if not leading_rows or inferred is None: + return None + header_ordinal = min(p.ordinal for p in leading_rows) - 1 + header = ParsedRow(_synthetic_header_cells(inferred), inferred, RowKind.HEADER, header_ordinal) + return ParsedTable([header, *leading_rows], synthetic_header=True) + + header_row_index, header_cells = rows_cells[first_header] + layout = build_layout(header_cells) + if layout is None: # исключено выбором first_header + return None + header_ordinal = min([header_row_index, *(p.ordinal - 1 for p in leading_rows)]) + parsed_rows = [ParsedRow(header_cells, layout, RowKind.HEADER, header_ordinal), *leading_rows] + for ordinal, cells in rows_cells[first_header + 1 :]: + header_layout = build_layout(cells) + if header_layout is not None: + layout = header_layout + continue + parsed = _data_row(cells, layout, ordinal) + if parsed is not None: + parsed_rows.append(parsed) + return ParsedTable(parsed_rows, synthetic_header=False) + + +def table_type_by_codes(parsed: ParsedTable) -> Literal["BALANCE", "INCOME"] | None: + """Тип таблицы по первой цифре кодов строк данных: больше `1xxx` → BALANCE, больше `2xxx` → INCOME.""" + counts = {"1": 0, "2": 0} + for row in parsed.rows: + if row.kind is RowKind.HEADER: + continue + code = cell_text(row.cells, row.layout.code_idx).strip() + if code: + counts[code[0]] = counts.get(code[0], 0) + 1 + if counts["1"] > counts["2"]: + return "BALANCE" + if counts["2"] > counts["1"]: + return "INCOME" + return None diff --git a/api/tests/unit/normalizer/test_rows.py b/api/tests/unit/normalizer/test_rows.py new file mode 100644 index 0000000..d8b8d6a --- /dev/null +++ b/api/tests/unit/normalizer/test_rows.py @@ -0,0 +1,141 @@ +"""Строки таблицы OCR (§11.1 п. 3–5): шапка настоящая/повторная/синтетическая, классы строк, тип по кодам.""" + +import json +from pathlib import Path + +import pytest + +from src.normalizer.ocr.layout import ColumnLayout, cells_by_index +from src.normalizer.ocr.rows import ParsedTable, RowKind, classify_row, parse_table_rows, table_type_by_codes + +API_DIR = Path(__file__).resolve().parents[3] +REAL_FIXTURES = API_DIR.parent.parent / "fixtures" / "ocr" +KNOWN = frozenset({"1105", "1110", "1120", "1130", "1150", "1600", "2110", "2120", "2100"}) + + +def _row(index: int, *cells: tuple[int, str, float]) -> dict: + return {"row_index": index, "cells": [{"column_index": i, "value": v, "confidence": c} for i, v, c in cells]} + + +HEADER = _row( + 0, + (0, "Наименование показателя", 100), + (1, "Код", 100), + (2, "На 30 июня\n2025 г", 100), + (4, "На 31 декабря\n2024 г.", 100), +) + + +def _real_table(name: str, index: int) -> list[dict]: + if not REAL_FIXTURES.is_dir(): + pytest.skip("fixtures недоступен: положите репозиторий fixtures рядом с workspace") + data = json.loads((REAL_FIXTURES / name).read_text(encoding="utf-8")) + return list(data["data_ocr"]["table_structure"][index]["rows"]) + + +class TestClassifyRow: + layout = ColumnLayout(label_idx=0, code_idx=1, value_idxs=(2,)) + + @pytest.mark.parametrize( + ("label", "code", "expected"), + [ + ("Выручка", "2110", RowKind.ROW), + ("Итого по разделу I", "1100", RowKind.TOTAL), + ("Баланс", "1600", RowKind.BALANCE), + ("Наименование", "Код", RowKind.HEADER), + ("в том числе:", "", None), + ], + ) + def test_kinds(self, label: str, code: str, expected: RowKind | None) -> None: + cells = cells_by_index(_row(1, (0, label, 100), (1, code, 100), (2, "1", 100))) + assert classify_row(cells, self.layout) is expected + + +class TestParseTableRows: + def test_header_then_rows(self) -> None: + rows = [ + HEADER, + _row(1, (0, "Выручка", 100), (1, "2110", 100), (2, "13 070", 100), (4, "", 100)), + _row(2, (0, "в том числе:", 93), (1, "", 100), (2, "13 070", 100), (4, "", 100)), + ] + parsed = parse_table_rows(rows, KNOWN) + assert isinstance(parsed, ParsedTable) and not parsed.synthetic_header + assert [r.kind for r in parsed.rows] == [RowKind.HEADER, RowKind.ROW] + assert parsed.rows[0].layout == ColumnLayout(label_idx=0, code_idx=1, value_idxs=(2, 4)) + assert parsed.rows[1].ordinal == 1 + + def test_second_header_overrides_layout(self) -> None: + """Review Focus 1: после повторной шапки с другим набором колонок строки идут по новой раскладке.""" + rows = [ + HEADER, + _row(1, (0, "Выручка", 100), (1, "2110", 100), (2, "1", 100), (4, "2", 100)), + _row(2, (0, "Наименование показателя", 100), (1, "Код", 100), (3, "За 2024 г.", 100)), + _row(3, (0, "Себестоимость", 100), (1, "2120", 100), (3, "(5)", 100)), + ] + parsed = parse_table_rows(rows, KNOWN) + assert parsed is not None + kinds = [r.kind for r in parsed.rows] + assert kinds == [RowKind.HEADER, RowKind.ROW, RowKind.ROW] # повторная шапка не становится строкой + assert parsed.rows[1].layout.value_idxs == (2, 4) + assert parsed.rows[2].layout.value_idxs == (3,) + + def test_rows_before_header_use_inferred_layout(self) -> None: + rows = [ + _row(0, (1, "1110", 100), (0, "Нематериальные активы", 100), (2, "5", 100)), + _row(1, (0, "Наименование показателя", 100), (1, "Код", 100), (2, "2024", 100)), + _row(2, (0, "Основные средства", 100), (1, "1150", 100), (2, "7", 100)), + ] + parsed = parse_table_rows(rows, KNOWN) + assert parsed is not None and not parsed.synthetic_header + assert [r.ordinal for r in parsed.rows] == [-1, 0, 2] # шапка встаёт раньше первой строки данных + assert parsed.rows[0].kind is RowKind.HEADER + + def test_synthetic_header_when_none(self) -> None: + rows = [ + _row(0, (1, "1110", 100), (0, "Нематериальные активы", 100), (2, "5", 100)), + _row(1, (1, "1150", 100), (0, "Основные средства", 100), (2, "7", 100)), + ] + parsed = parse_table_rows(rows, KNOWN) + assert parsed is not None and parsed.synthetic_header + header = parsed.rows[0] + assert header.kind is RowKind.HEADER and header.ordinal == -1 + assert header.cells[1].value == "Код" and header.cells[0].value == "Наименование показателя" + assert header.layout.value_idxs == (2,) and header.cells.get(2) is None # тексты периодов пустые + + def test_no_header_no_codes_is_none(self) -> None: + assert parse_table_rows([_row(0, (0, "текст", 100), (1, "мусор", 100))], KNOWN) is None + assert parse_table_rows([], KNOWN) is None + + @pytest.mark.fixtures + def test_aep3_balance_without_header(self) -> None: + parsed = parse_table_rows(_real_table("aep_3.json", 0), KNOWN) + assert parsed is not None and parsed.synthetic_header + assert parsed.rows[0].layout.code_idx == 1 + assert any(r.cells[1].value.strip() == "1120" for r in parsed.rows[1:]) + + @pytest.mark.fixtures + def test_aep6_balance_with_header(self) -> None: + parsed = parse_table_rows(_real_table("aep_6.json", 0), KNOWN) + assert parsed is not None and not parsed.synthetic_header + assert parsed.rows[0].layout == ColumnLayout(label_idx=0, code_idx=1, value_idxs=(2, 4, 5)) + assert table_type_by_codes(parsed) == "BALANCE" + + +class TestTableType: + def test_by_first_digit(self) -> None: + rows = [ + HEADER, + _row(1, (0, "Выручка", 100), (1, "2110", 100), (2, "1", 100)), + _row(2, (0, "Себестоимость", 100), (1, "2120", 100), (2, "1", 100)), + _row(3, (0, "ОС", 100), (1, "1150", 100), (2, "1", 100)), + ] + assert table_type_by_codes(parse_table_rows(rows, KNOWN)) == "INCOME" + + def test_tie_or_empty_is_none(self) -> None: + rows = [ + HEADER, + _row(1, (0, "a", 100), (1, "2110", 100), (2, "1", 100)), + _row(2, (0, "b", 100), (1, "1150", 100), (2, "1", 100)), + ] + assert table_type_by_codes(parse_table_rows(rows, KNOWN)) is None + assert table_type_by_codes(ParsedTable(rows=[], synthetic_header=False)) is None