diff --git a/arm_statements/segments.py b/arm_statements/segments.py new file mode 100644 index 0000000..9cbedec --- /dev/null +++ b/arm_statements/segments.py @@ -0,0 +1,252 @@ +"""Шаг 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)) diff --git a/arm_statements/tests/test_segments.py b/arm_statements/tests/test_segments.py new file mode 100644 index 0000000..fe86101 --- /dev/null +++ b/arm_statements/tests/test_segments.py @@ -0,0 +1,115 @@ +from decimal import Decimal as D + +from arm_statements.header import parse_header +from arm_statements.loader import Rules +from arm_statements.model import RowRole +from arm_statements.segments import find_layout +from arm_statements.tests.builders import excel, hdr, ocr, place + +DATES3 = ["На 30 сентября 2025 г.", "На 31 декабря 2024 г.", "На 31 декабря 2023 г."] + + +def test_npz_sheet_with_own_header_maps_by_date(rules: Rules) -> None: + first = [ + place({1: "Наименование показателя", 12: "Код", 16: DATES3[0], 21: DATES3[1], 24: DATES3[2]}), + place({1: "Нематериальные активы", 12: "1110", 16: D("10"), 21: D("9"), 24: D("8")}), + place({1: "Итого по разделу I", 12: "1100", 16: D("10"), 21: D("9"), 24: D("8")}), + ] + sheet4 = [ + place({0: "Наименование показателя", 6: "Код", 8: DATES3[0], 9: DATES3[1], 10: DATES3[2]}), + place({0: "Заемные средства", 6: "1510", 8: D("3"), 9: D("2"), 10: D("1")}), + ] + cand = excel([("Бухгалтерский баланс", first), ("Бухгалтерский баланс. Лист 4", sheet4)]) + layout = find_layout(cand, parse_header(hdr(report_date="30.09.2025"), rules), rules) + assert [c.column_index for c in layout.columns] == [16, 21, 24] + assert layout.segments[1].value_cols == ((8, 16), (9, 21), (10, 24)) + assert not [w for w in layout.warnings if w.code.startswith("SEGMENT_")] + + +def test_repeat_header_mid_table(rules: Rules) -> None: + rows = [ + ["Наименование показателя", "Код", "На 30.09.2025", "На 31.12.2024"], + ["Запасы", "1210", "5", "4"], + ["Наименование показателя", "Код", "На 30.09.2025", "На 31.12.2024"], + ["Итого", "1200", "5", "4"], + ] + layout = find_layout(ocr(rows), parse_header(hdr(), rules), rules) + assert [s.first_row for s in layout.segments] == [0, 2] + assert layout.roles[2] is RowRole.HEADER + + +def test_repeat_header_without_dates_maps_by_order(rules: Rules) -> None: + rows = [ + ["Наименование", "Код", "На 30.09.2025", "На 31.12.2024"], + ["Запасы", "1210", "5", "4"], + ["Наименование", "Код", "отчетн", "прошл"], + ["Итого", "1200", "5", "4"], + ] + layout = find_layout(ocr(rows), parse_header(hdr(), rules), rules) + assert layout.segments[1].value_cols == ((2, 2), (3, 3)) + assert layout.warnings == () + + +def test_wider_segment_is_mismatch(rules: Rules) -> None: + rows = [ + ["Наименование", "Код", "На 30.09.2025", "На 31.12.2024", "На 31.12.2023"], + ["Запасы", "1210", "5", "4", "3"], + ["Наименование", "Код", "графа 3", "графа 4", "графа 5", "графа 6"], + ["Итого", "1200", "5", "4", "3", "2"], + ] + layout = find_layout(ocr(rows), parse_header(hdr(), rules), rules) + (warning,) = [w for w in layout.warnings if w.code == "SEGMENT_COLUMNS_MISMATCH"] + assert warning.severity.value == "error" and "5" in warning.message + assert layout.segments[1].value_cols == ((2, 2), (3, 3), (4, 4)) and layout.segments[1].extra_cols == (5,) + + +def test_second_sheet_single_code_in_other_column(rules: Rules) -> None: + first = [ + place({0: "Наименование", 3: "Код", 4: "За 9 месяцев 2025 г.", 5: "За 9 месяцев 2024 г."}), + place({0: "Выручка", 3: "2110", 4: D("100"), 5: D("90")}), + place({0: "Себестоимость", 3: "2120", 4: D("-60"), 5: D("-50")}), + place({0: "Валовая прибыль", 3: "2100", 4: D("40"), 5: D("40")}), + ] + second = [place({0: "Чистая прибыль", 1: "2400", 2: D("7"), 3: D("6")})] + cand = excel([("ОФР", first), ("ОФР_2", second)]) + layout = find_layout(cand, parse_header(hdr(), rules), rules) + assert layout.segments[1].code_col == 1 + assert layout.segments[1].value_cols == ((2, 4), (3, 5)) + assert not [w for w in layout.warnings if w.code.startswith(("SEGMENT_", "HEADER_"))] + + +def test_header_not_found_anywhere(rules: Rules) -> None: + rows = [["Запасы", "1210", "5", "4"], ["Дебиторы", "1230", "3", "2"], ["Итого", "1200", "8", "6"]] + layout = find_layout(ocr(rows), parse_header(hdr(), rules), rules) + assert [w.code for w in layout.warnings] == ["HEADER_ROW_NOT_FOUND"] + assert [(c.column_index, c.label) for c in layout.columns] == [(2, ""), (3, "")] + + +def test_segment_without_codes(rules: Rules) -> None: + first = [["Наименование", "Код", "На 30.09.2025"], ["Запасы", "1210", "5"], ["Итого", "1200", "5"]] + tail = [["Примечание к форме", "", "7"]] + layout = find_layout(excel([("Баланс", first), ("Хвост", tail)]), parse_header(hdr(), rules), rules) + assert [w.code for w in layout.warnings] == ["SEGMENT_WITHOUT_CODES"] + assert layout.roles[3] is RowRole.DATA and layout.code_found + + +def test_first_segment_without_header_columns_from_first_headed(rules: Rules) -> None: + """P1-18: колонки документа задаёт первый сегмент с заголовком, у первого значения — по порядку.""" + rows = [ + ["Запасы", "1210", "5", "4"], + ["Дебиторы", "1230", "3", "2"], + ["Итого", "1200", "8", "6"], + ["Наименование", "Код", "На 30.09.2025", "На 31.12.2024"], + ["Прочее", "1250", "1", "1"], + ] + layout = find_layout(ocr(rows), parse_header(hdr(), rules), rules) + assert [s.first_row for s in layout.segments] == [0, 3] + assert [c.column_index for c in layout.columns] == [2, 3] + assert layout.segments[0].value_cols == ((2, 2), (3, 3)) + assert layout.warnings == () + + +def test_no_codes_anywhere_has_no_segment_warnings(rules: Rules) -> None: + layout = find_layout(ocr([["Просто", "текст"], ["ещё", "строка"]]), parse_header(hdr(), rules), rules) + assert not layout.code_found and layout.columns == () and layout.warnings == () + assert layout.roles == (RowRole.DATA, RowRole.DATA)