feat: шаг 3 — строки данных со статусами ok/unreadable/detail/missing, ref, разлепление склеек
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
This commit is contained in:
Andrey Epifancev
parent
ed85759b9b
commit
91e96795d5
253
arm_statements/rows.py
Normal file
253
arm_statements/rows.py
Normal file
@ -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
|
||||
115
arm_statements/tests/test_rows.py
Normal file
115
arm_statements/tests/test_rows.py
Normal file
@ -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"]
|
||||
Reference in New Issue
Block a user