254 lines
9.6 KiB
Python
254 lines
9.6 KiB
Python
"""Шаг 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
|