feat(statements): агрегат формы по коду — проверки, межформенная, parent_code, достройка, правка по коду

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W

api/src/domain/statement_aggregate.py

@@ -0,0 +1,87 @@
+"""Агрегат формы ББ/ОФР по коду строки (спека `arm-core`, v5.19; P3-14).
+
+Код строки в форме может повторяться (лист-продолжение, повтор в источнике, добавленная строка). Все чтения «по коду» —
+проверки §7, межформенная проверка, `parent_code` подстрок, достройка по шаблону — идут через агрегат: суммы строк
+`ok` с одним кодом складываются по колонке, пустая сумма не превращается в 0. Чистая функция, без БД.
+"""
+
+import uuid
+from collections.abc import Callable, Iterable, Mapping, Sequence
+from dataclasses import dataclass
+from decimal import Decimal
+from typing import Protocol
+
+from src.models.enums import StatementLineStatus
+
+__all__ = ["AggLine", "CodeAggregate", "aggregate_by_code", "checks_input"]
+
+
+@dataclass(frozen=True)
+class AggLine:
+    """Строка формы для агрегата: `amounts` — сумма по `column_id` (`None` — пустая ячейка или её нет)."""
+
+    id: uuid.UUID
+    row_index: int | None
+    line_status: StatementLineStatus
+    line_code: str | None
+    amounts: Mapping[uuid.UUID, Decimal | None]
+
+
+@dataclass(frozen=True)
+class CodeAggregate:
+    code: str
+    amounts: dict[uuid.UUID, Decimal | None]
+    line_ids: tuple[uuid.UUID, ...]  # по `row_index`, `NULL` (шаблонные) последними
+    reference_name: str | None
+    has_source_line: bool  # есть строка источника (`row_index IS NOT NULL`)
+
+
+def _no_name(_: str) -> str | None:
+    return None
+
+
+def aggregate_by_code(
+    lines: Iterable[AggLine], reference_name: Callable[[str], str | None] = _no_name
+) -> dict[str, CodeAggregate]:
+    """Агрегат строк `ok` по коду, по возрастанию кода. Сумма по колонке — сумма непустых; нет непустых — `None`."""
+    groups: dict[str, list[AggLine]] = {}
+    for line in lines:
+        if line.line_status is StatementLineStatus.ok and line.line_code is not None:
+            groups.setdefault(line.line_code, []).append(line)
+    result: dict[str, CodeAggregate] = {}
+    for code in sorted(groups):
+        ordered = sorted(groups[code], key=lambda item: (item.row_index is None, item.row_index or 0, str(item.id)))
+        amounts: dict[uuid.UUID, Decimal | None] = {}
+        for line in ordered:
+            for column_id, amount in line.amounts.items():
+                previous = amounts.get(column_id)
+                amounts[column_id] = amount if previous is None else previous + (amount or Decimal(0))
+        result[code] = CodeAggregate(
+            code=code,
+            amounts=amounts,
+            line_ids=tuple(line.id for line in ordered),
+            reference_name=reference_name(code),
+            has_source_line=any(line.row_index is not None for line in ordered),
+        )
+    return result
+
+
+class _Column(Protocol):
+    id: uuid.UUID
+    column_index: int
+
+
+def checks_input(
+    columns: Sequence[_Column], aggregates: Mapping[str, CodeAggregate]
+) -> tuple[list[int], list[tuple[str, dict[int, Decimal | None]]]]:
+    """Вход `run_checks`: индексы колонок (по возрастанию) и `(код, {column_index: сумма})` по возрастанию кода.
+
+    `columns` — колонки формы (`id`, `column_index`). Код без строки источника и без единой суммы (пустая шаблонная
+    строка) пропускается (design D-5 шаблонных строк)."""
+    index_by_id = {column.id: column.column_index for column in columns}
+    lines = [
+        (code, {index_by_id[cid]: amount for cid, amount in agg.amounts.items() if cid in index_by_id})
+        for code, agg in sorted(aggregates.items())
+        if agg.has_source_line or any(amount is not None for amount in agg.amounts.values())
+    ]
+    return sorted(column.column_index for column in columns), lines

api/src/domain/statement_checks.py

@@ -72,6 +72,7 @@ class FailedCheck:
     relation: str
     expected: Decimal
     actual: Decimal
+    codes: tuple[str, ...] = ()  # цель и члены правила (P3-14); у межформенной — 1370 и 2400
 
 
 @dataclass(frozen=True)
@@ -144,7 +145,7 @@ def subline_parents(codes: Sequence[str]) -> dict[str, str]:
 
     Подстрока относится к ближайшей четырёхзначной строке выше с теми же тремя цифрами, иначе — к строке с кодом из
     её первых четырёх цифр, если такая есть; без родителя в результат не попадает. Порядок строк в БД не хранится
-    (DBML порядка не знает): «выше» — порядок `codes`, и вызывающий передаёт его тот же, в каком `statement_amounts`
+    (DBML порядка не знает): «выше» — порядок `codes`, и вызывающий передаёт его тот же, в каком `checks_input`
     отдаёт коды проверке (по возрастанию кода; M-9 2b). Порядок результата — порядок подстрок в `codes`."""
     present = {code for code in codes if len(code) == _PARENT_LEN}
     result: dict[str, str] = {}
@@ -211,7 +212,8 @@ def run_checks(
             holds, expected, actual = _holds(amounts, rule, rules.tolerance)
             if holds:
                 continue
-            failed = FailedCheck(column_index, rule.relation, expected, actual)
+            codes = (rule.target, *(code for _, code in rule.terms))
+            failed = FailedCheck(column_index, rule.relation, expected, actual, codes)
             scale = max([abs(expected), *(abs(amounts.get(code, _ZERO)) for _, code in rule.terms)])
             rounding = scale * rules.rounding_tolerance_percent / _HUNDRED
             if rule.strict and abs(expected - actual) > rounding:
@@ -234,4 +236,4 @@ def cross_form_check(
     delta = current - year_start
     if abs(delta - income_2400) <= tolerance:
         return None
-    return FailedCheck(-1, CROSS_FORM_RELATION, delta, income_2400)
+    return FailedCheck(-1, CROSS_FORM_RELATION, delta, income_2400, ("1370", "2400"))

api/src/repositories/statements.py

@@ -3,15 +3,16 @@
 
 import datetime
 import uuid
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from decimal import Decimal
 from typing import Any
 
-from sqlalchemy import delete, insert, select
+from sqlalchemy import delete, func, insert, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from src.domain.statement_aggregate import AggLine, CodeAggregate, aggregate_by_code
 from src.domain.statement_mapping import StatementDraft
-from src.models.enums import DocumentStatus, DocumentType, FileStatus
+from src.models.enums import DocumentStatus, DocumentType, FileStatus, StatementLineStatus
 from src.models.files import Document, StoredFile
 from src.models.statements import (
     DocumentIssue,
@@ -25,16 +26,18 @@ __all__ = [
     "BATCH",
     "add_line",
     "add_value",
+    "code_amount",
     "column_ids",
     "delete_documents_statements",
     "delete_file_statements",
+    "form_aggregate",
     "form_lines",
     "get_meta",
+    "has_source_lines",
     "insert_statement",
     "load_form",
     "lock_cells",
     "pair_amounts",
-    "statement_amounts",
     "value_flags",
 ]
 
@@ -72,12 +75,13 @@ async def insert_statement(
         await session.execute(insert(FinancialStatementColumn).values(columns))
     line_ids: dict[str, uuid.UUID] = {}
     lines: list[dict[str, Any]] = []
-    for line in draft.lines:
+    for row_index, line in enumerate(draft.lines):
         line_ids[line.line_code] = uuid.uuid4()
         lines.append(
             {
                 "id": line_ids[line.line_code],
                 "meta_id": meta.id,
+                "row_index": row_index,  # строка источника (до маппера `2.0`: порядок черновика)
                 "line_code": line.line_code,
                 "line_code_original_value": line.line_code,
                 "line_code_confidence": line.line_code_confidence,
@@ -130,52 +134,60 @@ async def _delete_statements(session: AsyncSession, documents: Any) -> None:
     await session.execute(delete(FinancialStatementMeta).where(FinancialStatementMeta.document_id.in_(documents)))
 
 
-async def has_lines(session: AsyncSession, meta_id: uuid.UUID) -> bool:
-    """В форме есть строки `financial_statement_line` (единое определение «есть строки» для статуса D-6)."""
-    stmt = select(select(FinancialStatementLine.id).where(FinancialStatementLine.meta_id == meta_id).exists())
+async def has_source_lines(session: AsyncSession, meta_id: uuid.UUID) -> bool:
+    """В форме есть строка источника (`row_index IS NOT NULL`) — единое определение «есть строки» для статуса D-6;
+    шаблонные строки (созданные правкой) не в счёт."""
+    stmt = select(
+        select(FinancialStatementLine.id)
+        .where(FinancialStatementLine.meta_id == meta_id, FinancialStatementLine.row_index.is_not(None))
+        .exists()
+    )
     return bool((await session.execute(stmt)).scalar_one())
 
 
-async def statement_amounts(
-    session: AsyncSession, meta_id: uuid.UUID
-) -> tuple[list[int], list[tuple[str, dict[int, Decimal | None]]]]:
-    """Индексы колонок и `(код строки, {колонка: сумма})` по возрастанию кода — вход `run_checks`. Порядок строк
-    формы в БД не хранится (DBML не хранит), поэтому порядок строк — лексикографический порядок кодов.
-    Созданные правкой строки (`line_code_original_value IS NULL`) без единой суммы не участвуют
-    (spec statement-checks, design D-5)."""
-    columns = (
+async def form_aggregate(
+    session: AsyncSession,
+    meta_id: uuid.UUID,
+    reference_name: Callable[[str], str | None] = lambda _: None,
+) -> tuple[list[FinancialStatementColumn], dict[str, CodeAggregate]]:
+    """Колонки формы (по `column_index`) и агрегат строк `ok` по коду (`domain/statement_aggregate`) — вход проверок
+    §7, `parent_code` и достройки по шаблону. Порядок строк формы в БД задаёт `row_index`, но проверки §7 читают коды
+    по возрастанию (M-9 2b)."""
+    columns = list(
+        (
+            await session.execute(
+                select(FinancialStatementColumn)
+                .where(FinancialStatementColumn.meta_id == meta_id)
+                .order_by(FinancialStatementColumn.column_index)
+            )
+        ).scalars()
+    )
+    lines = (
         await session.execute(
-            select(FinancialStatementColumn.id, FinancialStatementColumn.column_index)
-            .where(FinancialStatementColumn.meta_id == meta_id)
-            .order_by(FinancialStatementColumn.column_index)
+            select(FinancialStatementLine).where(
+                FinancialStatementLine.meta_id == meta_id,
+                FinancialStatementLine.line_status == StatementLineStatus.ok,
+                FinancialStatementLine.line_code.is_not(None),
+            )
         )
-    ).all()
-    index_by_id = {column_id: index for column_id, index in columns}
+    ).scalars()
+    by_line: dict[uuid.UUID, dict[uuid.UUID, Decimal | None]] = {}
     rows = (
         await session.execute(
-            select(
-                FinancialStatementLine.line_code,
-                FinancialStatementLine.line_code_original_value,
-                FinancialStatementValue.column_id,
-                FinancialStatementValue.amount,
+            select(FinancialStatementValue.line_id, FinancialStatementValue.column_id, FinancialStatementValue.amount)
+            .join(FinancialStatementLine, FinancialStatementLine.id == FinancialStatementValue.line_id)
+            .where(
+                FinancialStatementLine.meta_id == meta_id,
+                FinancialStatementLine.line_status == StatementLineStatus.ok,
             )
-            .join(FinancialStatementValue, FinancialStatementValue.line_id == FinancialStatementLine.id)
-            .where(FinancialStatementLine.meta_id == meta_id)
-            .order_by(FinancialStatementLine.line_code)
         )
     ).all()
-    by_code: dict[str, dict[int, Decimal | None]] = {}
-    added: set[str] = set()  # строки, созданные правкой аналитика (design D-5)
-    for code, original, column_id, amount in rows:
-        by_code.setdefault(code or "", {})[index_by_id[column_id]] = amount
-        if original is None:
-            added.add(code or "")
-    lines = [
-        (code, amounts)
-        for code, amounts in by_code.items()
-        if code not in added or any(amount is not None for amount in amounts.values())
+    for line_id, column_id, amount in rows:
+        by_line.setdefault(line_id, {})[column_id] = amount
+    agg_lines = [
+        AggLine(line.id, line.row_index, line.line_status, line.line_code, by_line.get(line.id, {})) for line in lines
     ]
-    return [index for _, index in columns], lines
+    return columns, aggregate_by_code(agg_lines, reference_name)
 
 
 async def value_flags(session: AsyncSession, meta_id: uuid.UUID) -> list[tuple[bool, bool | None, Decimal | None]]:
@@ -234,7 +246,7 @@ async def column_ids(session: AsyncSession, meta_id: uuid.UUID) -> set[uuid.UUID
 
 
 async def add_line(session: AsyncSession, meta_id: uuid.UUID, code: str, name: str) -> FinancialStatementLine:
-    """Строка справочника, созданная правкой аналитика: `*_original_value = NULL` — признак «не из источника»."""
+    """Шаблонная строка, созданная правкой аналитика: `row_index = NULL` — признак «не из источника»; статус `ok`."""
     line = FinancialStatementLine(
         id=uuid.uuid4(),
         meta_id=meta_id,
@@ -246,6 +258,9 @@ async def add_line(session: AsyncSession, meta_id: uuid.UUID, code: str, name: s
         line_name_original_value=None,
         line_name_confidence=None,
         line_name_modified=False,
+        row_index=None,
+        source_ref=None,
+        line_status=StatementLineStatus.ok,
     )
     session.add(line)
     await session.flush()
@@ -283,7 +298,8 @@ async def get_meta(
 async def load_form(
     session: AsyncSession, meta_id: uuid.UUID
 ) -> tuple[list[FinancialStatementColumn], list[tuple[FinancialStatementLine, list[FinancialStatementValue]]]]:
-    """Колонки по `column_index` и строки по `line_code` с их значениями по `column_index`."""
+    """Колонки по `column_index` и строки по `line_code`, `row_index` (`NULL` последними), `id` с их значениями по
+    `column_index`."""
     columns = list(
         (
             await session.execute(
@@ -299,7 +315,11 @@ async def load_form(
             await session.execute(
                 select(FinancialStatementLine)
                 .where(FinancialStatementLine.meta_id == meta_id)
-                .order_by(FinancialStatementLine.line_code, FinancialStatementLine.id)
+                .order_by(
+                    FinancialStatementLine.line_code,
+                    FinancialStatementLine.row_index.asc().nulls_last(),
+                    FinancialStatementLine.id,
+                )
             )
         ).scalars()
     )
@@ -316,27 +336,34 @@ async def load_form(
     return columns, [(line, by_line[line.id]) for line in lines]
 
 
-async def _line_amount(
+async def code_amount(
     session: AsyncSession,
     meta_id: uuid.UUID,
-    line_code: str,
+    code: str,
     *,
     current: bool = False,
     period_date: datetime.date | None = None,
 ) -> Decimal | None:
-    """Сумма строки `line_code` в колонке `is_current` (`current`) или с датой `period_date`; нет — `None`."""
-    stmt = (
-        select(FinancialStatementValue.amount)
-        .join(FinancialStatementLine, FinancialStatementLine.id == FinancialStatementValue.line_id)
-        .join(FinancialStatementColumn, FinancialStatementColumn.id == FinancialStatementValue.column_id)
-        .where(FinancialStatementLine.meta_id == meta_id, FinancialStatementLine.line_code == line_code)
-        .order_by(FinancialStatementColumn.column_index)
-        .limit(1)
-    )
+    """`SUM(amount)` строк `ok` с кодом `code` в колонке `is_current` (`current`) или с датой `period_date`; колонки
+    нет или непустых сумм нет — `None`."""
+    column_stmt = select(FinancialStatementColumn.id).where(FinancialStatementColumn.meta_id == meta_id)
     if current:
-        stmt = stmt.where(FinancialStatementColumn.is_current.is_(True))
+        column_stmt = column_stmt.where(FinancialStatementColumn.is_current.is_(True))
     else:
-        stmt = stmt.where(FinancialStatementColumn.period_date == period_date)
+        column_stmt = column_stmt.where(FinancialStatementColumn.period_date == period_date)
+    column_id = (await session.execute(column_stmt.order_by(FinancialStatementColumn.column_index).limit(1))).scalar()
+    if column_id is None:
+        return None
+    stmt = (
+        select(func.sum(FinancialStatementValue.amount))
+        .join(FinancialStatementLine, FinancialStatementLine.id == FinancialStatementValue.line_id)
+        .where(
+            FinancialStatementLine.meta_id == meta_id,
+            FinancialStatementLine.line_code == code,
+            FinancialStatementLine.line_status == StatementLineStatus.ok,
+            FinancialStatementValue.column_id == column_id,
+        )
+    )
     return (await session.execute(stmt)).scalar_one_or_none()
 
 
@@ -376,8 +403,8 @@ async def pair_amounts(
     if balance_meta is not None:
         year_start = datetime.date(document.period_year - 1, 12, 31)
         balance = (
-            await _line_amount(session, balance_meta.id, "1370", current=True),
-            await _line_amount(session, balance_meta.id, "1370", period_date=year_start),
+            await code_amount(session, balance_meta.id, "1370", current=True),
+            await code_amount(session, balance_meta.id, "1370", period_date=year_start),
         )
-    income = await _line_amount(session, income_meta.id, "2400", current=True) if income_meta is not None else None
+    income = await code_amount(session, income_meta.id, "2400", current=True) if income_meta is not None else None
     return balance, income

api/src/schemas/statements.py

@@ -15,7 +15,7 @@ from pydantic import BaseModel, ConfigDict, Field, StringConstraints, model_vali
 
 from src.domain.statement_checks import ChecksReport, FailedCheck
 from src.domain.statement_lines import TemplateLine
-from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter
+from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter, StatementLineStatus
 from src.models.files import Document
 from src.models.statements import (
     FinancialStatementColumn,
@@ -155,8 +155,8 @@ def _value_out(value: FinancialStatementValue, column_index: dict[uuid.UUID, int
 
 
 def line_origin(line: FinancialStatementLine) -> LineOrigin:
-    """`template` — строка создана правкой аналитика (маппинг всегда пишет `line_code_original_value`)."""
-    return "template" if line.line_code_original_value is None else "source"
+    """`template` — строка создана правкой аналитика (`row_index IS NULL`); у строк источника индекс задан всегда."""
+    return "template" if line.row_index is None else "source"
 
 
 def _empty_value(column: FinancialStatementColumn) -> ValueOut:
@@ -202,7 +202,8 @@ def _lines_out(
                 values=cells,
             )
         )
-    codes = {line.line_code for line, _ in lines}
+    # достройка по шаблону — только для кода без строки `ok` (строка `unreadable` с тем же текстом кода не мешает)
+    codes = {line.line_code for line, _ in lines if line.line_status is StatementLineStatus.ok}
     for item in template:
         if item.code in codes:
             continue

api/src/services/result_mapping.py

@@ -26,6 +26,7 @@ PROCESSING` у файла, чьё ТЕКУЩЕЕ задание (`current_recogn
 
 import asyncio
 import dataclasses
+import functools
 import uuid
 from collections.abc import Awaitable, Callable, Mapping, Sequence
 from dataclasses import dataclass
@@ -49,8 +50,10 @@ from src.domain.osv_mapping import (
     document_status,
     opening_matches,
 )
+from src.domain.statement_aggregate import checks_input
 from src.domain.statement_checks import load_check_rules, run_checks
-from src.domain.statement_mapping import StatementDraft, build_statement, statement_status, values_ok
+from src.domain.statement_lines import load_line_catalog
+from src.domain.statement_mapping import StatementDraft, build_statement, edition_for, statement_status, values_ok
 from src.models.accounting import ChartOfAccount
 from src.models.consolidation import Consolidation, ConsolidationVersion
 from src.models.directories import CompanyGroup, LegalEntity
@@ -415,9 +418,15 @@ async def _insert_statements(
     if not forms:
         return
     rules = load_check_rules()
+    catalog = load_line_catalog()
     for document, draft in forms:
+        edition = edition_for(document.reporting_date)
+        form = document.document_type.value
         meta = await statements_repo.insert_statement(session, document_id=document.id, job_id=job.id, draft=draft)
-        column_indexes, amounts = await statements_repo.statement_amounts(session, meta.id)
+        columns, aggregates = await statements_repo.form_aggregate(
+            session, meta.id, functools.partial(catalog.reference_name, edition, form)
+        )
+        column_indexes, amounts = checks_input(columns, aggregates)
         report = run_checks(document.document_type.value, amounts, column_indexes, rules)
         document.status = statement_status(
             job_done=job_done,

api/src/services/statements.py

@@ -15,6 +15,7 @@ from src.common.clock import utc_now
 from src.common.exceptions import ConflictError, DomainValidationError, NotFoundError
 from src.domain.osv_mapping import counterparty_check
 from src.domain.policy import VisibilityScope
+from src.domain.statement_aggregate import CodeAggregate, checks_input
 from src.domain.statement_checks import (
     ChecksReport,
     FailedCheck,
@@ -34,9 +35,10 @@ from src.models.enums import (
     DocumentType,
     FileStatus,
     RecognitionStatus,
+    StatementLineStatus,
 )
 from src.models.files import Document, RecognitionJob
-from src.models.statements import FinancialStatementLine, FinancialStatementMeta
+from src.models.statements import FinancialStatementColumn, FinancialStatementLine, FinancialStatementMeta
 from src.repositories import files as files_repo
 from src.repositories import statements as repo
 from src.repositories.counterparties import CounterpartyIndex
@@ -56,13 +58,24 @@ INN_RE = re.compile(r"\d{10}|\d{12}", re.ASCII)
 FORM_TYPES = frozenset({DocumentType.BALANCE, DocumentType.INCOME})
 
 
+async def form_aggregate(
+    session: AsyncSession, document: Document, meta: FinancialStatementMeta
+) -> tuple[list[FinancialStatementColumn], dict[str, CodeAggregate]]:
+    """Колонки и агрегат формы по коду со справочными наименованиями редакции документа (P3-14)."""
+    catalog = load_line_catalog()
+    edition = edition_for(document.reporting_date)
+    form = document.document_type.value
+    return await repo.form_aggregate(session, meta.id, lambda code: catalog.reference_name(edition, form, code))
+
+
 async def statement_hard_failures(
     session: AsyncSession, document: Document, meta: FinancialStatementMeta
 ) -> tuple[FailedCheck, ...]:
     """Нарушенные жёсткие проверки §7 по текущим суммам формы; при акцепте — пустой кортеж (D-8 Среза 3b)."""
     if meta.checks_accepted_at is not None:
         return ()
-    column_indexes, amounts = await repo.statement_amounts(session, meta.id)
+    columns, aggregates = await form_aggregate(session, document, meta)
+    column_indexes, amounts = checks_input(columns, aggregates)
     report = run_checks(document.document_type.value, amounts, column_indexes, load_check_rules())
     return tuple(report.failed)
 
@@ -121,10 +134,16 @@ class StatementService:
         self, document: Document, meta: FinancialStatementMeta
     ) -> tuple[ChecksReport, FailedCheck | None, dict[str, str]]:
         """Проверки §7 по текущим суммам формы, мягкая межформенная проверка (D-5) и родители подстрок
-        (`parent_code` ответа; AURORA-1527). Родители считаются по тем же кодам и в том же порядке
-        (`statement_amounts`), что и проверка «сумма подстрок = родитель», — ответ и проверка не расходятся."""
+        (`parent_code` ответа; AURORA-1527). Всё считается по агрегату формы по коду (`form_aggregate`): родители — по
+        тем же кодам и в том же порядке, что и проверка «сумма подстрок = родитель» (ответ и проверка не расходятся)."""
+        columns, aggregates = await form_aggregate(self._session, document, meta)
+        return await self._report(document, columns, aggregates)
+
+    async def _report(
+        self, document: Document, columns: list[FinancialStatementColumn], aggregates: dict[str, CodeAggregate]
+    ) -> tuple[ChecksReport, FailedCheck | None, dict[str, str]]:
         rules = load_check_rules()
-        column_indexes, amounts = await repo.statement_amounts(self._session, meta.id)
+        column_indexes, amounts = checks_input(columns, aggregates)
         report = run_checks(document.document_type.value, amounts, column_indexes, rules)
         parents = subline_parents([code for code, _ in amounts])
         balance_1370, income_2400 = await repo.pair_amounts(self._session, document)
@@ -163,7 +182,7 @@ class StatementService:
             counterparty=document.counterparty_check or DocumentCheckResult.UNKNOWN,
             period_known=document.period_year is not None and document.period_quarter is not None,
             values_ok=values_ok(await repo.value_flags(self._session, meta.id)),
-            has_lines=await repo.has_lines(self._session, meta.id),
+            has_lines=await repo.has_source_lines(self._session, meta.id),
             warnings=(),
         )
         document.updated_at = utc_now()
@@ -221,7 +240,7 @@ class StatementService:
         columns = await repo.column_ids(self._session, meta.id)
         lines = await repo.form_lines(self._session, meta.id)
         by_id = {line.id: line for line in lines}
-        codes = {line.line_code for line in lines}
+        codes = {line.line_code for line in lines if line.line_status is StatementLineStatus.ok}
         edition = edition_for(document.reporting_date)
         catalog_lines = load_line_catalog().template_lines(edition, document.document_type.value)
         template = {item.code: item for item in catalog_lines}

api/tests/test_api_statements.py

@@ -1377,3 +1377,132 @@ async def test_template_line_checks_follow_filled_amounts(
     after_clear = response.json()["checks"]["failed"]
     assert not any(f["relation"].startswith("1700 =") for f in after_clear)  # очищенная 1700 снова не цель
     assert {f["column_index"] for f in after_clear if f["relation"] == "1600 = 1700"} == {0, 1, 2}
+
+
+# --- агрегат формы по коду (v5.19, A7) ---
+
+
+async def insert_line(  # noqa: PLR0913 - строка формы теста: код, статус, индекс, сумма
+    session: AsyncSession,
+    document: Document,
+    *,
+    code: str | None,
+    row_index: int | None,
+    status: str = "ok",
+    amount: Decimal | None = None,
+    current: bool = True,
+    code_original: str | None = None,
+    code_text: str | None = None,
+) -> FinancialStatementLine:
+    """Строка формы напрямую в БД; ячейка с `amount` — в колонке отчётной даты (`current`) или нулевой."""
+    meta = await _meta(session, document)
+    columns = (
+        (
+            await session.execute(
+                select(FinancialStatementColumn)
+                .where(FinancialStatementColumn.meta_id == meta.id)
+                .order_by(FinancialStatementColumn.column_index)
+            )
+        )
+        .scalars()
+        .all()
+    )
+    column = next(c for c in columns if c.is_current) if current else columns[0]
+    line = FinancialStatementLine(
+        id=uuid.uuid4(),
+        meta_id=meta.id,
+        line_code=code,
+        line_code_original_value=code_original,
+        line_status=status,
+        line_code_text=code_text,
+        line_name="тест",
+        row_index=row_index,
+    )
+    session.add(line)
+    await session.flush()
+    if amount is not None:
+        session.add(
+            FinancialStatementValue(
+                id=uuid.uuid4(), line_id=line.id, column_id=column.id, amount=amount, is_modified=False
+            )
+        )
+        await session.flush()
+    return line
+
+
+async def test_cross_form_uses_aggregate(
+    seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
+) -> None:
+    """Review Focus 2: две строки 1370 в колонке отчётной даты — 500 + 300 = 800."""
+    balance, income = await two_forms(seeded, fixtures_dir)
+    await set_amount(seeded, balance, "1370", Decimal("500"), current=True)
+    await set_amount(seeded, balance, "1370", Decimal("0"), period_date=datetime.date(2024, 12, 31))
+    await insert_line(seeded, balance, code="1370", row_index=99, amount=Decimal("300"), code_original="1370")
+    client = await auth_client(KA_RF)
+    await set_amount(seeded, income, "2400", Decimal("800"), current=True)
+    assert (await client.get(f"/documents/{income.id}/statement")).json()["checks"]["cross_form"] is None
+    await set_amount(seeded, income, "2400", Decimal("500"), current=True)
+    cross = (await client.get(f"/documents/{income.id}/statement")).json()["checks"]["cross_form"]
+    assert cross is not None and cross["expected"] == "800.00" and cross["actual"] == "500.00"
+
+
+async def test_parent_code_with_repeated_parent(
+    seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
+) -> None:
+    document = await form(seeded, fixtures_dir, UNAGRANDE_BALANCE)
+    await insert_line(seeded, document, code="1150", row_index=98, amount=Decimal("1"), code_original="1150")
+    await insert_line(seeded, document, code="11501", row_index=99, amount=Decimal("1"), code_original="11501")
+    client = await auth_client(KA_RF)
+    body = (await client.get(f"/documents/{document.id}/statement")).json()
+    sub = [line for line in body["lines"] if line["line_code"] == "11501"]
+    assert [line["parent_code"] for line in sub] == ["1150", "1150"][: len(sub)] and sub
+    assert len([line for line in body["lines"] if line["line_code"] == "1150"]) == 2  # noqa: PLR2004
+
+
+async def test_template_not_completed_for_code_with_ok_line(
+    seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
+) -> None:
+    document = await form(seeded, fixtures_dir)
+    await insert_line(seeded, document, code="1215", row_index=90, amount=Decimal("5"), code_original="1215")
+    client = await auth_client(KA_RF)
+    body = (await client.get(f"/documents/{document.id}/statement")).json()
+    lines = [line for line in body["lines"] if line["line_code"] == "1215"]
+    assert [(line["origin"], line["id"] is None) for line in lines] == [("source", False)]
+
+
+async def test_template_completed_for_code_only_unreadable(
+    seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
+) -> None:
+    document = await form(seeded, fixtures_dir)
+    bad = await insert_line(
+        seeded, document, code=None, row_index=91, status="unreadable", amount=Decimal("5"), code_text="ИЧ,215"
+    )
+    client = await auth_client(KA_RF)
+    body = (await client.get(f"/documents/{document.id}/statement")).json()
+    template = [line for line in body["lines"] if line["line_code"] == "1215"]
+    assert [(line["origin"], line["id"]) for line in template] == [("template", None)]
+    assert any(line["id"] == str(bad.id) for line in body["lines"])
+
+
+async def test_patch_by_line_code_with_ok_line_is_unknown_cell(
+    seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory
+) -> None:
+    document = await form(seeded, fixtures_dir)
+    await insert_line(seeded, document, code="1215", row_index=90, amount=Decimal("5"), code_original="1215")
+    client = await auth_client(KA_RF)
+    body = (await client.get(f"/documents/{document.id}/statement")).json()
+    column_id = body["columns"][0]["id"]
+    response = await client.patch(
+        f"/documents/{document.id}/statement",
+        json={"values": [{"line_code": "1215", "column_id": column_id, "amount": "1"}]},
+    )
+    assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY
+    assert response.json()["detail"]["code"] == "UNKNOWN_CELL"
+
+
+async def test_origin_by_row_index(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
+    document = await form(seeded, fixtures_dir)
+    line = await insert_line(seeded, document, code="1250", row_index=7, amount=Decimal("5"), code_original=None)
+    client = await auth_client(KA_RF)
+    body = (await client.get(f"/documents/{document.id}/statement")).json()
+    assert next(item for item in body["lines"] if item["id"] == str(line.id))["origin"] == "source"

api/tests/test_domain_statement_aggregate.py

@@ -0,0 +1,83 @@
+"""Агрегат формы по коду (спека `arm-core` v5.19; Review Focus 2)."""
+
+import uuid
+from dataclasses import dataclass
+from decimal import Decimal
+
+from src.domain.statement_aggregate import AggLine, aggregate_by_code, checks_input
+from src.models.enums import StatementLineStatus
+
+C1 = uuid.UUID(int=1)
+C2 = uuid.UUID(int=2)
+L1, L5 = uuid.UUID(int=101), uuid.UUID(int=105)
+
+
+def line(  # noqa: PLR0913 - поля строки теста
+    status: str,
+    code: str | None,
+    *,
+    r: int | None,
+    c1: str | None = None,
+    c2: str | None = None,
+    id_: uuid.UUID | None = None,
+) -> AggLine:
+    amounts = {C1: Decimal(c1) if c1 is not None else None, C2: Decimal(c2) if c2 is not None else None}
+    return AggLine(id_ or uuid.uuid4(), r, StatementLineStatus(status), code, amounts)
+
+
+def ok(code: str, *, r: int | None, c1: str | None = None, c2: str | None = None, id_: uuid.UUID | None = None):
+    return line("ok", code, r=r, c1=c1, c2=c2, id_=id_)
+
+
+def test_repeated_code_sums() -> None:
+    agg = aggregate_by_code([ok("2410", r=1, c1="100", id_=L1), ok("2410", r=5, c1="649", id_=L5)])
+    assert agg["2410"].amounts[C1] == Decimal("749") and agg["2410"].line_ids == (L1, L5)
+
+
+def test_empty_amounts_stay_null() -> None:
+    assert aggregate_by_code([ok("1110", r=0), ok("1110", r=3)])["1110"].amounts[C1] is None
+    assert aggregate_by_code([ok("1110", r=0), ok("1110", r=3, c1="5")])["1110"].amounts[C1] == Decimal("5")
+
+
+def test_non_ok_lines_excluded() -> None:
+    agg = aggregate_by_code(
+        [
+            line("detail", None, r=2, c1="7"),
+            line("unreadable", None, r=3, c1="9"),
+            line("missing", None, r=4, c1="1"),
+            ok("1250", r=5, c1="3"),
+        ]
+    )
+    assert list(agg) == ["1250"]
+
+
+def test_template_line_counts_as_ok() -> None:
+    agg = aggregate_by_code([ok("1250", r=None, c1="10")])
+    assert agg["1250"].amounts[C1] == Decimal("10") and agg["1250"].has_source_line is False
+
+
+def test_codes_ascending_and_template_line_ids_last() -> None:
+    agg = aggregate_by_code([ok("2400", r=None, id_=L5), ok("1600", r=2), ok("2400", r=7, id_=L1)])
+    assert list(agg) == ["1600", "2400"] and agg["2400"].line_ids == (L1, L5)
+    assert agg["2400"].has_source_line is True
+
+
+def test_reference_name_from_callback() -> None:
+    agg = aggregate_by_code([ok("1250", r=1)], reference_name=lambda code: f"имя {code}")
+    assert agg["1250"].reference_name == "имя 1250"
+
+
+@dataclass
+class Col:
+    id: uuid.UUID
+    column_index: int
+
+
+def test_checks_input_skips_empty_template_codes() -> None:
+    agg = aggregate_by_code(
+        [ok("1250", r=None), ok("1260", r=None, c1="4"), ok("1270", r=3), ok("1280", r=4, c1="1"), ok("1280", r=None)]
+    )
+    indexes, lines = checks_input([Col(C2, 1), Col(C1, 0)], agg)
+    assert indexes == [0, 1]
+    assert [code for code, _ in lines] == ["1260", "1270", "1280"]
+    assert dict(lines)["1260"] == {0: Decimal("4"), 1: None}

api/tests/test_domain_statement_checks.py

@@ -35,7 +35,7 @@ def test_hard_violation() -> None:
     report = run_checks(
         "BALANCE", _lines(("1100", {0: "300"}), ("1200", {0: "600"}), ("1600", {0: "1000"})), [0], RULES
     )
-    assert FailedCheck(0, "1600 = 1100 + 1200", D("1000"), D("900")) in report.failed
+    assert FailedCheck(0, "1600 = 1100 + 1200", D("1000"), D("900"), ("1600", "1100", "1200")) in report.failed
 
 
 def test_relation_skipped_without_target() -> None:
@@ -107,7 +107,7 @@ def test_relation_checked_per_column_and_none_is_zero() -> None:
     )
     report = run_checks("BALANCE", lines, [0, 1, 2], RULES)
     assert _by_relation(report.failed, "1600 = 1100 + 1200") == [
-        FailedCheck(1, "1600 = 1100 + 1200", D("1000"), D("900"))
+        FailedCheck(1, "1600 = 1100 + 1200", D("1000"), D("900"), ("1600", "1100", "1200"))
     ]
     assert _by_relation(report.soft_failed, "1600 = 1100 + 1200") == []
 
@@ -117,7 +117,7 @@ def test_tolerance_exceeded_by_one_unit_is_violation() -> None:
     lines = _lines(("1100", {0: "300"}), ("1200", {0: "600"}), ("1600", {0: "902"}))
     report = run_checks("BALANCE", lines, [0], RULES)
     assert _by_relation(report.failed, "1600 = 1100 + 1200") == [
-        FailedCheck(0, "1600 = 1100 + 1200", D("902"), D("900"))
+        FailedCheck(0, "1600 = 1100 + 1200", D("902"), D("900"), ("1600", "1100", "1200"))
     ]
 
 
@@ -179,7 +179,7 @@ def test_rounding_threshold_boundary() -> None:
         (
             (D("500"), D("300")),
             D("150"),
-            FailedCheck(-1, "1370 − 1370(начало года) = 2400", D("200"), D("150")),
+            FailedCheck(-1, "1370 − 1370(начало года) = 2400", D("200"), D("150"), ("1370", "2400")),
         ),
         ((None, D("300")), D("200"), None),
         ((D("500"), D("300")), None, None),