From 91e96795d5814a9ba359357f664d7a3d50327fea Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Sun, 18 Oct 2026 13:44:47 +0400 Subject: [PATCH] =?UTF-8?q?feat:=20=D1=88=D0=B0=D0=B3=203=20=E2=80=94=20?= =?UTF-8?q?=D1=81=D1=82=D1=80=D0=BE=D0=BA=D0=B8=20=D0=B4=D0=B0=D0=BD=D0=BD?= =?UTF-8?q?=D1=8B=D1=85=20=D1=81=D0=BE=20=D1=81=D1=82=D0=B0=D1=82=D1=83?= =?UTF-8?q?=D1=81=D0=B0=D0=BC=D0=B8=20ok/unreadable/detail/missing,=20ref,?= =?UTF-8?q?=20=D1=80=D0=B0=D0=B7=D0=BB=D0=B5=D0=BF=D0=BB=D0=B5=D0=BD=D0=B8?= =?UTF-8?q?=D0=B5=20=D1=81=D0=BA=D0=BB=D0=B5=D0=B5=D0=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W --- arm_statements/rows.py | 253 ++++++++++++++++++++++++++++++ arm_statements/tests/test_rows.py | 115 ++++++++++++++ 2 files changed, 368 insertions(+) create mode 100644 arm_statements/rows.py create mode 100644 arm_statements/tests/test_rows.py diff --git a/arm_statements/rows.py b/arm_statements/rows.py new file mode 100644 index 0000000..cf05c63 --- /dev/null +++ b/arm_statements/rows.py @@ -0,0 +1,253 @@ +"""Шаг 3: строки данных, их статусы, `parent_ref` и разлепление склеенных строк.""" + +from collections.abc import Sequence +from dataclasses import dataclass + +from arm_statements.amounts import parse_amount +from arm_statements.codes import code_parts, parse_code, section_of +from arm_statements.issues import WarningCode, make_warning +from arm_statements.layout import Grid, cell_at, name_cell +from arm_statements.loader import Rules +from arm_statements.model import ( + Cell, + FormCandidate, + LineStatus, + RowRole, + Source, + StatementWarning, +) +from arm_statements.segments import Layout, SegmentLayout +from arm_statements.text import has_digit, is_disguised_digit + +_FULL_CONFIDENCE = 100.0 +_MIN_SPLIT_PARTS = 2 + +CellsByColumn = tuple[tuple[int, Cell | None], ...] + + +@dataclass(frozen=True, slots=True) +class RawLine: + """Строка данных до разбора значений: ячейки и статус; `cells` — по всем колонкам документа по порядку.""" + + row_index: int + ref: str + grid_row: int + code_cell: Cell | None + code: str | None + code_fixed: bool + line_status: LineStatus + name: Cell | None + parent_ref: str | None + cells: CellsByColumn + split: bool + + +@dataclass(slots=True) +class _State: + """Сквозное состояние документа: счётчик строк и ближайшая строка `ok`.""" + + next_index: int = 0 + last_ok_ref: str | None = None + + +def _blank(cell: Cell | None) -> bool: + return cell is None or not cell.text.strip() + + +def _is_content(cell: Cell | None, source: Source, rules: Rules) -> bool: + """Содержимое ячейки значения: сумма, неуверенный непустой текст или «цифра под видом буквы».""" + if cell is None: + return False + if cell.value is not None: + return True + text = cell.text + if not text.strip(): + return False + if parse_amount(text, source, rules).amount is not None: + return True + return cell.confidence < _FULL_CONFIDENCE or is_disguised_digit(text, rules.lookalikes) + + +def _first_ref(row: Sequence[Cell | None], code_cell: Cell | None) -> str: + if code_cell is not None: + return code_cell.ref + for cell in row: + if not _blank(cell): + return cell.ref if cell is not None else "" + return next((c.ref for c in row if c is not None), "") + + +def _grid_cols(seg: SegmentLayout, layout: Layout) -> dict[int, int]: + """`column_index` колонки документа → столбец сетки сегмента.""" + by_column = {column_index: grid_col for grid_col, column_index in seg.value_cols} + return {c.column_index: by_column[c.column_index] for c in layout.columns if c.column_index in by_column} + + +def _segment_cells(rows: Grid, row: int, mapping: dict[int, int], layout: Layout) -> CellsByColumn: + return tuple( + (c.column_index, cell_at(rows, row, mapping[c.column_index]) if c.column_index in mapping else None) + for c in layout.columns + ) + + +def _plain_name(cells: Sequence[Cell | None]) -> tuple[Cell | None, int]: + """Сегмент без столбца кодов: наименование — левая ячейка с буквами; второй элемент — её столбец.""" + for col, cell in enumerate(cells): + if cell is not None and any(ch.isalpha() for ch in cell.text): + return cell, col + return None, -1 + + +def _has_value_content(cells: CellsByColumn, source: Source, rules: Rules) -> bool: + return any(_is_content(cell, source, rules) for _, cell in cells) + + +def _part(cell: Cell, k: int, text: str) -> Cell: + return Cell(text, None, cell.confidence, f"{cell.ref}#{k}") + + +def _split_parts(code_cell: Cell | None, cells: CellsByColumn, layout: Layout, rules: Rules) -> tuple[str, ...] | None: + """Части кода склеенной строки или `None`, если разлеплять нельзя.""" + if code_cell is None or layout.column_section is None: + return None + parts = code_parts(code_cell.text) + count = len(parts) + if count < _MIN_SPLIT_PARTS: + return None + if any(not _blank(cell) and len(code_parts(cell.text if cell else None)) != count for _, cell in cells): + return None + in_range = ( + (code := parse_code(part, rules).code) is not None and section_of(code, rules) is layout.column_section + for part in parts + ) + return parts if any(in_range) else None + + +def _status(code_cell: Cell | None, code: str | None, *, has_content: bool, state: _State) -> LineStatus | None: + """Статус строки без разлепления; `None` — строка данных без содержимого.""" + if code_cell is not None and has_digit(code_cell.text): + return LineStatus.OK if code is not None else LineStatus.UNREADABLE + if not has_content: + return None + return LineStatus.DETAIL if state.last_ok_ref is not None else LineStatus.MISSING + + +def _emit( # noqa: PLR0913 - сборка одной строки + state: _State, + *, + ref: str, + grid_row: int, + code_cell: Cell | None, + code: str | None, + fixed: bool, + status: LineStatus, + name: Cell | None, + cells: CellsByColumn, + split: bool, +) -> RawLine: + parent = state.last_ok_ref if status is LineStatus.DETAIL else None + line = RawLine(state.next_index, ref, grid_row, code_cell, code, fixed, status, name, parent, cells, split) + state.next_index += 1 + if status is LineStatus.OK: + state.last_ok_ref = ref + return line + + +def _split_row( # noqa: PLR0913, PLR0917 - разлепление одной строки сетки + state: _State, + grid_row: int, + base_ref: str, + parts: tuple[str, ...], + code_cell: Cell, + name: Cell | None, + cells: CellsByColumn, + rules: Rules, +) -> list[RawLine]: + count = len(parts) + name_parts = code_parts(name.text) if name is not None else () + lines: list[RawLine] = [] + for k, part in enumerate(parts, start=1): + parsed = parse_code(part, rules) + derived_cells: CellsByColumn = tuple( + (index, cell if _blank(cell) or cell is None else _part(cell, k, code_parts(cell.text)[k - 1])) + for index, cell in cells + ) + derived_name = name + if name is not None and len(name_parts) == count: + derived_name = _part(name, k, name_parts[k - 1]) + lines.append( + _emit( + state, + ref=f"{base_ref}#{k}", + grid_row=grid_row, + code_cell=_part(code_cell, k, part), + code=parsed.code, + fixed=parsed.fixed, + status=LineStatus.OK if parsed.code is not None else LineStatus.UNREADABLE, + name=derived_name, + cells=derived_cells, + split=True, + ) + ) + return lines + + +def _row_lines( # noqa: PLR0913, PLR0917 - разбор одной строки сетки + candidate: FormCandidate, layout: Layout, seg: SegmentLayout, grid_row: int, state: _State, rules: Rules +) -> list[RawLine]: + rows = candidate.rows + row = rows[grid_row] + cells = _segment_cells(rows, grid_row, _grid_cols(seg, layout), layout) + code_cell = cell_at(rows, grid_row, seg.code_col) if seg.code_col is not None else None + if seg.code_col is not None: + name = name_cell(row, seg.name_col, seg.code_col, seg.explanations_col) + content_cells = cells + else: + name, name_col = _plain_name(row) + content_cells = ( + cells if seg.value_cols else tuple((col, cell) for col, cell in enumerate(row) if col > name_col) + ) + base_ref = _first_ref(row, code_cell) + parts = _split_parts(code_cell, cells, layout, rules) + if parts is not None and code_cell is not None: + return _split_row(state, grid_row, base_ref, parts, code_cell, name, cells, rules) + parsed = parse_code(code_cell.text if code_cell is not None else None, rules) + has_content = _has_value_content(content_cells, candidate.source, rules) + status = _status(code_cell, parsed.code, has_content=has_content, state=state) + if status is None: + return [] + ref = base_ref or _first_ref(row, None) + return [ + _emit( + state, + ref=ref, + grid_row=grid_row, + code_cell=code_cell, + code=parsed.code, + fixed=parsed.fixed, + status=status, + name=name, + cells=cells, + split=False, + ) + ] + + +def collect_raw_lines( + candidate: FormCandidate, layout: Layout, rules: Rules +) -> tuple[list[RawLine], list[StatementWarning]]: + """Строки данных по сегментам и предупреждения `ROW_SPLIT` (`info`) на каждую разлепленную строку сетки.""" + state = _State() + lines: list[RawLine] = [] + warnings: list[StatementWarning] = [] + for seg in layout.segments: + for grid_row in range(seg.first_row, seg.last_row + 1): + if grid_row >= len(candidate.rows) or layout.roles[grid_row] is not RowRole.DATA: + continue + produced = _row_lines(candidate, layout, seg, grid_row, state, rules) + lines.extend(produced) + if produced and produced[0].split: + message = f"склеенная строка сетки {grid_row} разделена на {len(produced)} строк" + refs = tuple(line.ref for line in produced) + warnings.append(make_warning(WarningCode.ROW_SPLIT, message, rules, line_refs=refs)) + return lines, warnings diff --git a/arm_statements/tests/test_rows.py b/arm_statements/tests/test_rows.py new file mode 100644 index 0000000..2c33bff --- /dev/null +++ b/arm_statements/tests/test_rows.py @@ -0,0 +1,115 @@ +from typing import Any + +from arm_statements.header import parse_header +from arm_statements.loader import Rules +from arm_statements.model import LineStatus, StatementWarning +from arm_statements.rows import RawLine, collect_raw_lines +from arm_statements.segments import find_layout +from arm_statements.tests.builders import excel, hdr, ocr + +HEAD = ["Наименование показателя", "Код", "На 30.09.2025", "На 31.12.2024"] +OK, UNREADABLE, DETAIL, MISSING = LineStatus.OK, LineStatus.UNREADABLE, LineStatus.DETAIL, LineStatus.MISSING + + +def lines_of(rows: Any, rules: Rules, source: str = "ocr") -> tuple[list[RawLine], list[StatementWarning]]: + cand = ocr(rows) if source == "ocr" else excel([("Баланс", rows)]) + layout = find_layout(cand, parse_header(hdr(), rules), rules) + return collect_raw_lines(cand, layout, rules) + + +def test_junk_code_lines_survive(rules: Rules) -> None: + lines, _ = lines_of( + [ + HEAD, + ["НМА", "1110", "5", "4"], + ["Запасы", "1150", "5", "4"], + ["Мусор", "ИЧ,250", "7", "6"], + ["в том числе: прочее", "", "1", ""], + ["", "44 30\nI |", "", ""], + ["Итого", "1100", "10", "8"], + ], + rules, + ) + assert [(ln.line_status, ln.code) for ln in lines] == [ + (OK, "1110"), + (OK, "1150"), + (UNREADABLE, None), + (DETAIL, None), + (UNREADABLE, None), + (OK, "1100"), + ] + assert lines[3].parent_ref == lines[1].ref == "row:2" + assert [ln.row_index for ln in lines] == [0, 1, 2, 3, 4, 5] + + +def test_lookalike_only_code_cells_are_empty(rules: Rules) -> None: + lines, _ = lines_of( + [ + HEAD, + ["НМА", "1110", "5", "4"], + ["Запасы", "1150", "5", "4"], + ["", "изо :", ("3", 0.0), ""], + ["", "НИ", "I", ""], + ["", "| 140", "2", "1"], + ["Итого", "1100", "12", "9"], + ], + rules, + ) + assert [(ln.line_status, ln.code, ln.code_fixed) for ln in lines] == [ + (OK, "1110", False), + (OK, "1150", False), + (DETAIL, None, False), + (DETAIL, None, False), + (OK, "1140", True), + (OK, "1100", False), + ] + assert [ln.parent_ref for ln in lines[2:4]] == ["row:2", "row:2"] + assert len({ln.ref for ln in lines}) == len(lines) + + +def test_missing_before_first_code_and_name_only_row(rules: Rules) -> None: + lines, _ = lines_of( + [HEAD, ["Выручка", "", "100", "90"], ["Раздел I. Активы", "", "", ""], ["Запасы", "1150", "5", "4"]], rules + ) + assert [ln.line_status for ln in lines] == [MISSING, OK] + + +def test_duplicate_code_both_ok(rules: Rules) -> None: + lines, _ = lines_of([HEAD, ["Налог", "2410", "100", "1"], ["Налог УСН", "2410", "649", "2"]], rules) + assert [(ln.line_status, ln.code) for ln in lines] == [(OK, "2410"), (OK, "2410")] + + +def test_split_glued_rows(rules: Rules) -> None: + lines, warnings = lines_of( + [HEAD, ["Запасы", "1150", "5", "4"], ["Займы\nКредиторы", "1500\n1700", "917 775\n2 371 333", "1\n2"]], rules + ) + split = lines[1:] + assert [(ln.ref, ln.code, ln.split) for ln in split] == [("row:2#1", "1500", True), ("row:2#2", "1700", True)] + assert [ln.cells[0][1].text for ln in split if ln.cells[0][1] is not None] == ["917 775", "2 371 333"] + assert [ln.name.text for ln in split if ln.name is not None] == ["Займы", "Кредиторы"] + assert [(w.code, w.line_refs) for w in warnings] == [("ROW_SPLIT", ("row:2#1", "row:2#2"))] + + +def test_not_split_when_no_part_in_column_range(rules: Rules) -> None: + lines, warnings = lines_of( + [ + HEAD, + ["НМА", "1110", "5", "4"], + ["Запасы", "1150", "5", "4"], + ["", "44 30\nI |", "1\n2", ""], + ["Итого", "1100", "10", "8"], + ], + rules, + ) + assert [ln.line_status for ln in lines] == [OK, OK, UNREADABLE, OK] and warnings == [] + + +def test_excel_signature_text_is_not_content(rules: Rules) -> None: + rows = [ + ["Наименование", "", "Код", "На 30.09.2025"], + ["Запасы", "", "1150", "5"], + ["Руководитель", "", "", "Косарев"], + ["", "", "", "(расшифровка подписи)"], + ] + lines, _ = lines_of(rows, rules, source="excel") + assert [ln.ref for ln in lines] == ["Баланс!C2"]