arm-core/api/arm_statements/rows.py

"""Шаг 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