feat: шаг 2 — сегменты, повторная шапка, листы-продолжения и сопоставление колонок с первым сегментом
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W
This commit is contained in:
Andrey Epifancev
parent
995a778fa4
commit
ed85759b9b
252
arm_statements/segments.py
Normal file
252
arm_statements/segments.py
Normal file
@ -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))
|
||||||
115
arm_statements/tests/test_segments.py
Normal file
115
arm_statements/tests/test_segments.py
Normal file
@ -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)
|
||||||
Reference in New Issue
Block a user