AURORA-1734: split_by_form — разрез кандидата по смене формы (#5)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

arm_statements/__init__.py

@@ -53,6 +53,7 @@ from arm_statements.recompute import (
     recompute,
 )
 from arm_statements.serialize import candidate_from_dict, to_jsonable
+from arm_statements.split import split_by_form
 from arm_statements.status import (
     StatusInput,
     StatusLine,
@@ -129,6 +130,7 @@ __all__ = [
     "recompute",
     "resolve_relative",
     "sheet_joins",
+    "split_by_form",
     "statements_version",
     "status_input",
     "status_input_from_contract",

arm_statements/segments.py

@@ -180,6 +180,29 @@ def _split_and_analyse(candidate: FormCandidate, rules: Rules) -> tuple[list[_An
     return result, section
 
 
+@dataclass(frozen=True, slots=True)
+class SegmentForm:
+    """Сегмент для разреза по смене формы: первая строка, своя шапка, коды столбца кодов."""
+
+    first: int
+    has_header: bool
+    codes: tuple[str, ...]
+
+
+def split_segments(candidate: FormCandidate, rules: Rules) -> list[SegmentForm]:
+    """Сегменты кандидата по правилу шага 2 (тот же разбор, что у раскладки); сегменты без столбца кодов пропущены."""
+    segments, _ = _split_and_analyse(candidate, rules)
+    return [
+        SegmentForm(
+            seg.first,
+            bool(seg.header_rows),
+            tuple(_segment_codes(candidate.rows, seg.first, seg.last, seg.code_col, rules)),
+        )
+        for seg in segments
+        if seg.code_col is not None
+    ]
+
+
 def _map_by_date(
     drafts: Sequence[ColumnDraft], columns: Sequence[StatementColumn]
 ) -> tuple[tuple[int, int], ...] | None:

arm_statements/split.py

@@ -0,0 +1,70 @@
+"""Разрез кандидата по смене формы до шагов 2–8 (спека «Шаг 2 — разрез кандидата по смене формы», AURORA-1734).
+
+ИС Сканирование может вернуть ББ и ОФР одной таблицей: шапка ОФР внутри таблицы — повторная шапка, и без разреза строки
+ОФР попадают в ББ под его колонками. Часть начинается с сегмента со своей шапкой, первые `min_codes` кодов которого —
+одной формы, другой, чем у текущей части (тип части — по большинству кодов её первого сегмента с кодами).
+"""
+
+from dataclasses import replace
+
+from arm_statements.codes import document_type_of, majority_section, section_of
+from arm_statements.loader import Rules, default_rules
+from arm_statements.model import DocumentType, Evidence, FormCandidate
+from arm_statements.segments import split_segments
+
+
+def _opening_type(codes: tuple[str, ...], rules: Rules) -> DocumentType | None:
+    """Тип формы, которой сегмент открывается: первые `min_codes` кодов одной формы; иначе `None`."""
+    head = codes[: rules.code_ranges.min_codes]
+    if len(head) < rules.code_ranges.min_codes:
+        return None
+    kinds = {None if (s := section_of(code, rules)) is None else document_type_of(s) for code in head}
+    return kinds.pop() if len(kinds) == 1 else None
+
+
+def _cuts(candidate: FormCandidate, rules: Rules) -> list[int]:
+    """Строки сетки, с которых начинаются части после первой."""
+    cuts: list[int] = []
+    current: DocumentType | None = None
+    for segment in split_segments(candidate, rules):
+        if current is None:
+            section = majority_section(list(segment.codes), rules)
+            current = None if section is None else document_type_of(section)
+            continue
+        opening = _opening_type(segment.codes, rules)
+        if segment.has_header and opening is not None and opening is not current and segment.first > 0:
+            cuts.append(segment.first)
+            current = opening
+    return cuts
+
+
+def split_by_form(candidate: FormCandidate, rules: Rules | None = None) -> tuple[FormCandidate, ...]:
+    """Части кандидата по смене формы; без смены — `(candidate,)` тем же объектом.
+
+    Первая часть сохраняет поля исходного кандидата (кроме строк и границ листов); часть `k` (с 1) — `locator`
+    с хвостом `;form:<k>`, `type_hint = None`, пустые `evidence` и `warnings`: тип — только по кодам на шаге 6.
+    """
+    rules = rules if rules is not None else default_rules()
+    cuts = _cuts(candidate, rules)
+    if not cuts:
+        return (candidate,)
+    bounds = [0, *cuts, len(candidate.rows)]
+    parts: list[FormCandidate] = []
+    for k, (start, end) in enumerate(zip(bounds, bounds[1:], strict=False)):
+        breaks = tuple(b - start for b in candidate.segment_breaks if start <= b < end)
+        rows = candidate.rows[start:end]
+        if k == 0:
+            parts.append(replace(candidate, rows=rows, segment_breaks=breaks))
+        else:
+            parts.append(
+                replace(
+                    candidate,
+                    locator=f"{candidate.locator};form:{k}",
+                    type_hint=None,
+                    evidence=Evidence(okud=None, title=None),
+                    rows=rows,
+                    segment_breaks=breaks,
+                    warnings=(),
+                )
+            )
+    return tuple(parts)

tests/test_model.py

@@ -77,6 +77,7 @@ PUBLIC = {
     "type_hint",
     "continuation_type",
     "sheet_joins",
+    "split_by_form",
     "parse_code",
     "parse_amount",
     "parse_date",

tests/test_split.py

@@ -0,0 +1,104 @@
+"""Разрез кандидата по смене формы (AURORA-1734, решение владельца 06.11.2026)."""
+
+from datetime import date
+
+from arm_statements import normalize_candidate, split_by_form
+from arm_statements.loader import Rules
+from arm_statements.model import DocumentType, Evidence
+from tests.builders import excel, hdr, ocr
+
+BAL_HEAD = ["Наименование показателя", "Коды", "На 30 июня 2026 года", "На 31 декабря 2025 года"]
+INC_HEAD = ["Наименование показателя", "Коды", "За I полугодие 2026 года", "За I полугодие 2025 года"]
+HEADER = hdr(organization_name="КФХ «АК-ЮЛ»", inn="5634000539", report_date="30.06.2026", units="тыс. руб.")
+LOCATOR = "file:x.json;table:0"
+
+
+def balance_rows() -> list[list[str | None]]:
+    """ББ на две страницы: актив, повторная шапка, пассив."""
+    return [
+        BAL_HEAD,
+        ["НМА", "1110", "10", "10"],
+        ["Основные средства", "1150", "90", "90"],
+        ["Итого по разделу I", "1100", "100", "100"],
+        ["Запасы", "1210", "50", "50"],
+        ["Итого по разделу II", "1200", "50", "50"],
+        ["БАЛАНС", "1600", "150", "150"],
+        BAL_HEAD,
+        ["Уставный капитал", "1310", "150", "150"],
+        ["Итого по разделу III", "1300", "150", "150"],
+        ["БАЛАНС", "1700", "150", "150"],
+    ]
+
+
+def income_rows() -> list[list[str | None]]:
+    return [
+        INC_HEAD,
+        ["Выручка", "2110", "4451", "2162"],
+        ["Себестоимость продаж", "2120", "-2700", "-1332"],
+        ["Валовая прибыль (убыток)", "2100", "1751", "830"],
+        ["Чистая прибыль (убыток)", "2400", "1751", "830"],
+    ]
+
+
+def test_balance_and_income_in_one_table_split(rules: Rules) -> None:
+    rows = [*balance_rows(), *income_rows()]
+    cand = ocr(rows, header=HEADER, type_hint=DocumentType.BALANCE, title="Бухгалтерский баланс", locator=LOCATOR)
+    parts = split_by_form(cand, rules)
+    assert len(parts) == 2  # noqa: PLR2004 - две формы
+    bal, inc = parts
+    assert bal.locator == LOCATOR and bal.type_hint is DocumentType.BALANCE and bal.evidence == cand.evidence
+    assert len(bal.rows) == len(balance_rows())
+    assert inc.locator == f"{LOCATOR};form:1"
+    assert inc.type_hint is None and inc.evidence == Evidence(okud=None, title=None) and inc.warnings == ()
+    assert inc.rows[0] == cand.rows[len(balance_rows())]
+    assert inc.header == cand.header and inc.source is cand.source
+
+    bal_doc = normalize_candidate(bal, rules).document
+    inc_doc = normalize_candidate(inc, rules).document
+    assert bal_doc is not None and inc_doc is not None
+    assert bal_doc.document_type is DocumentType.BALANCE
+    assert not [line.line_code for line in bal_doc.lines if (line.line_code or "").startswith("2")]
+    assert inc_doc.document_type is DocumentType.INCOME
+    assert [c.period_date for c in inc_doc.columns] == [date(2026, 6, 30), date(2025, 6, 30)]
+    assert "TYPE_CONFLICT" not in [w.code for w in inc_doc.warnings]
+
+
+def test_same_form_repeat_header_not_split(rules: Rules) -> None:
+    cand = ocr(balance_rows(), header=HEADER, type_hint=DocumentType.BALANCE, locator=LOCATOR)
+    result = split_by_form(cand, rules)
+    assert result == (cand,)
+    assert result[0] is cand
+
+
+def test_income_tail_25xx_not_split(rules: Rules) -> None:
+    tail = [
+        INC_HEAD,
+        ["Результат от переоценки", "2510", "0", "0"],
+        ["Результат от прочих операций", "2520", "0", "0"],
+        ["Совокупный финансовый результат", "2500", "1751", "830"],
+    ]
+    cand = ocr([*income_rows(), *tail], locator=LOCATOR)
+    assert split_by_form(cand, rules) == (cand,)
+
+
+def test_two_foreign_codes_not_split(rules: Rules) -> None:
+    foreign = [INC_HEAD, ["Выручка", "2110", "4451", "2162"], ["Чистая прибыль (убыток)", "2400", "1751", "830"]]
+    cand = ocr([*balance_rows(), *foreign], locator=LOCATOR)
+    assert split_by_form(cand, rules) == (cand,)
+
+
+def test_foreign_segment_without_header_not_split(rules: Rules) -> None:
+    cand = ocr([*balance_rows(), *income_rows()[1:]], locator=LOCATOR)
+    assert split_by_form(cand, rules) == (cand,)
+
+
+def test_excel_breaks_shifted(rules: Rules) -> None:
+    continuation = [["Прочие доходы", "2340", "1", "1"], ["Прочие расходы", "2350", "-1", "-1"]]
+    cand = excel([("Б", balance_rows()), ("О", income_rows()), ("О2", continuation)], header=HEADER)
+    cut = len(balance_rows())
+    assert cand.segment_breaks == (cut, cut + len(income_rows()))
+    bal, inc = split_by_form(cand, rules)
+    assert bal.segment_breaks == ()
+    assert inc.segment_breaks == (0, len(income_rows()))
+    assert inc.locator == f"{cand.locator};form:1"
+    assert len(inc.rows) == len(income_rows()) + len(continuation)