feat(normalizer): строки таблицы OCR — шапки, классы строк, тип по кодам (§11.1 п. 3–5)
Перенос _classify_row, _parse_table_rows, _synthetic_header_cells, _resolve_table_type из reference-old-arm с адаптацией к новой архитектуре: - Cell dataclass вместо dict - ParsedTable вместо list возврата - Fallback для row_index если не int - Безопасная обработка пустых кодов в table_type_by_codes 14 тестов пройдено (2 на реальных данных fixtures). Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev
parent
81081a90c7
commit
462ea9ee32
129
api/src/normalizer/ocr/rows.py
Normal file
129
api/src/normalizer/ocr/rows.py
Normal file
@ -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
|
||||
141
api/tests/unit/normalizer/test_rows.py
Normal file
141
api/tests/unit/normalizer/test_rows.py
Normal file
@ -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
|
||||
Loading…
x
Reference in New Issue
Block a user