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