"""Шаг 2: сегменты сетки, повторные шапки, границы листов и сопоставление колонок с первым сегментом.""" from collections.abc import Sequence from dataclasses import dataclass from typing import NamedTuple from arm_statements.codes import code_parts, is_code_cell, majority_section, parse_code from arm_statements.columns import ColumnDraft, DocumentDates, document_dates, finalize_columns, segment_columns from arm_statements.issues import WarningCode, make_warning from arm_statements.layout import ( Grid, cell_at, find_code_column, find_header_rows, find_name_column, first_code_row, is_numbering_row, segment_roles, ) from arm_statements.loader import CodeSection, Rules from arm_statements.model import ( FormCandidate, RowRole, StatementColumn, StatementHeader, StatementWarning, ) from arm_statements.text import normalize_caption @dataclass(frozen=True, slots=True) class SegmentLayout: """Раскладка сегмента; `value_cols` — (столбец сетки сегмента, `column_index` колонки документа).""" first_row: int last_row: int header_rows: tuple[int, ...] code_col: int | None name_col: int | None explanations_col: int | None value_cols: tuple[tuple[int, int], ...] extra_cols: tuple[int, ...] @dataclass(frozen=True, slots=True) class Layout: """Раскладка кандидата; `roles` — по всем строкам сетки.""" code_found: bool column_section: CodeSection | None columns: tuple[StatementColumn, ...] segments: tuple[SegmentLayout, ...] roles: tuple[RowRole, ...] dates: DocumentDates warnings: tuple[StatementWarning, ...] @dataclass(frozen=True, slots=True) class _Analysed: """Сегмент до сопоставления колонок.""" first: int last: int header_rows: tuple[int, ...] code_col: int | None name_col: int | None explanations_col: int | None roles: tuple[RowRole, ...] drafts: tuple[ColumnDraft, ...] draft_warnings: tuple[StatementWarning, ...] has_content: bool class _Bounds(NamedTuple): last: int reference: frozenset[str] def _text(rows: Grid, row: int, col: int) -> str: cell = cell_at(rows, row, col) return "" if cell is None else cell.text.strip() def _chunks(rows: Grid, breaks: Sequence[int]) -> list[tuple[int, int]]: """Куски сетки по границам листов; границы вне сетки и повторы отбрасываются.""" cuts = sorted({b for b in breaks if 0 < b < len(rows)}) starts = [0, *cuts] ends = [*(c - 1 for c in cuts), len(rows) - 1] return list(zip(starts, ends, strict=True)) def _caption_cells(rows: Grid, header_rows: Sequence[int]) -> frozenset[str]: cells = (normalize_caption(_text(rows, r, c)) for r in header_rows for c in range(len(rows[r]))) return frozenset(c for c in cells if c) def _is_repeat_header(rows: Grid, row: int, code_col: int | None, reference: frozenset[str], rules: Rules) -> bool: """Строка-шапка: ячейка подписи кода или не менее двух ячеек, равных ячейкам шапки первого сегмента.""" if is_numbering_row(rows[row], rules): return False if code_col is not None and is_code_cell(_text(rows, row, code_col), rules): return False normalized = [normalize_caption(c.text) for c in rows[row] if c is not None and c.text.strip()] if any(rules.headers.code_caption.search(text) for text in normalized): return True return sum(1 for text in normalized if text and text in reference) >= _MIN_SAME_CELLS _MIN_SAME_CELLS = 2 def _segment_bounds(rows: Grid, start: int, chunk_last: int, reference: frozenset[str], rules: Rules) -> _Bounds: """Последняя строка сегмента: до следующей шапки, стоящей после первой строки с кодом.""" code_col = find_code_column(rows, start, chunk_last, rules) first_code = None if code_col is None else first_code_row(rows, start, chunk_last, code_col, rules) if code_col is None or first_code is None: return _Bounds(chunk_last, reference) ref = reference or _caption_cells(rows, find_header_rows(rows, start, first_code, code_col, rules)) for row in range(first_code + 1, chunk_last + 1): if _is_repeat_header(rows, row, code_col, ref, rules): return _Bounds(row - 1, ref) return _Bounds(chunk_last, ref) def _segment_codes(rows: Grid, first: int, last: int, code_col: int, rules: Rules) -> list[str]: codes: list[str] = [] for row in range(first, last + 1): text = _text(rows, row, code_col) if is_code_cell(text, rules): codes.extend(c for part in code_parts(text) if (c := parse_code(part, rules).code) is not None) return codes def _analyse( # noqa: PLR0913, PLR0917 - внутренняя сборка сегмента candidate: FormCandidate, first: int, last: int, section: CodeSection | None, rules: Rules ) -> _Analysed: rows = candidate.rows code_col = find_code_column(rows, first, last, rules) if code_col is None and section is not None: code_col = find_code_column(rows, first, last, rules, relaxed_section=section) has_content = any(c is not None and c.text.strip() for r in range(first, last + 1) for c in rows[r]) first_code = None if code_col is None else first_code_row(rows, first, last, code_col, rules) if code_col is None or first_code is None: return _Analysed(first, last, (), None, None, None, (RowRole.DATA,) * (last - first + 1), (), (), has_content) header_rows = find_header_rows(rows, first, first_code, code_col, rules) roles = tuple(segment_roles(rows, first, last, header_rows, rules)) data_rows = [r for r, role in zip(range(first, last + 1), roles, strict=True) if role is RowRole.DATA] name_col, notes_col = find_name_column(rows, header_rows, first, last, code_col, rules) drafts, warnings = segment_columns(rows, header_rows, data_rows, code_col, candidate.source, rules) return _Analysed( first, last, header_rows, code_col, name_col, notes_col, roles, tuple(drafts), tuple(warnings), has_content ) def _split_and_analyse(candidate: FormCandidate, rules: Rules) -> tuple[list[_Analysed], CodeSection | None]: rows = candidate.rows result: list[_Analysed] = [] reference: frozenset[str] = frozenset() section: CodeSection | None = None for chunk_first, chunk_last in _chunks(rows, candidate.segment_breaks): start = chunk_first while start <= chunk_last: bounds = _segment_bounds(rows, start, chunk_last, reference, rules) reference = reference or bounds.reference seg = _analyse(candidate, start, bounds.last, section, rules) if section is None and seg.code_col is not None: section = majority_section(_segment_codes(rows, seg.first, seg.last, seg.code_col, rules), rules) result.append(seg) start = bounds.last + 1 return result, section def _map_by_date( drafts: Sequence[ColumnDraft], columns: Sequence[StatementColumn] ) -> tuple[tuple[int, int], ...] | None: """Сопоставление по `period_date`: у каждой колонки дата есть и совпадает ровно с одной колонкой документа.""" if not drafts or any(d.date is None for d in drafts): return None pairs: list[tuple[int, int]] = [] for draft in drafts: found = [c.column_index for c in columns if c.period_date == draft.date] if len(found) != 1: return None pairs.append((draft.grid_col, found[0])) if len({target for _, target in pairs}) != len(pairs): return None return tuple(pairs) def _map_segment( seg: _Analysed, columns: tuple[StatementColumn, ...], rules: Rules ) -> tuple[SegmentLayout, StatementWarning | None]: """Колонки сегмента → колонки документа; по порядку и разное число — `SEGMENT_COLUMNS_MISMATCH`.""" drafts = seg.drafts mapped = _map_by_date(drafts, columns) if seg.header_rows else None warning: StatementWarning | None = None extra: tuple[int, ...] = () if mapped is None: size = min(len(drafts), len(columns)) mapped = tuple((d.grid_col, c.column_index) for d, c in zip(drafts[:size], columns[:size], strict=False)) extra = tuple(d.grid_col for d in drafts[size:]) if len(drafts) != len(columns): tail = f"; лишние столбцы сегмента: {', '.join(str(c) for c in extra)}" if extra else "" message = ( f"в сегменте со строки {seg.first} колонок значений {len(drafts)}, " f"в первом сегменте {len(columns)}{tail}" ) warning = make_warning(WarningCode.SEGMENT_COLUMNS_MISMATCH, message, rules, locator=None) layout = SegmentLayout( seg.first, seg.last, seg.header_rows, seg.code_col, seg.name_col, seg.explanations_col, mapped, extra ) return layout, warning def find_layout(candidate: FormCandidate, header: StatementHeader, rules: Rules) -> Layout: """Раскладка кандидата: сегменты, колонки документа, роли всех строк и предупреждения шага 2.""" segments, section = _split_and_analyse(candidate, rules) coded = [s for s in segments if s.code_col is not None] defining = next((s for s in coded if s.header_rows), coded[0] if coded else None) warnings: list[StatementWarning] = [] columns: tuple[StatementColumn, ...] = () dates = document_dates([], header) if defining is not None: dates = document_dates(defining.drafts, header) columns, finalize_warnings = finalize_columns(defining.drafts, dates, rules) warnings.extend(defining.draft_warnings) warnings.extend(finalize_warnings) if not any(s.header_rows for s in coded): warnings.append( make_warning(WarningCode.HEADER_ROW_NOT_FOUND, "строка заголовка таблицы не найдена", rules) ) laid: list[SegmentLayout] = [] for seg in segments: if seg.code_col is None: laid.append(SegmentLayout(seg.first, seg.last, (), None, None, None, (), ())) if coded and seg.has_content: message = f"в сегменте со строки {seg.first} не найден столбец кодов" warnings.append(make_warning(WarningCode.SEGMENT_WITHOUT_CODES, message, rules)) elif seg is defining: own = tuple((c.column_index, c.column_index) for c in columns) laid.append( SegmentLayout( seg.first, seg.last, seg.header_rows, seg.code_col, seg.name_col, seg.explanations_col, own, () ) ) else: segment_layout, mismatch = _map_segment(seg, columns, rules) laid.append(segment_layout) if mismatch is not None: warnings.append(mismatch) roles = tuple(role for seg in segments for role in seg.roles) return Layout(bool(coded), section, columns, tuple(laid), roles, dates, tuple(warnings))