Andrey Epifancev 80b8635633 plan: Срез 3b — план реализации (16 задач): формы ББ/ОФР в ядре, API экрана проверки, мелкие дыры API

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

2026-10-07 10:51:08 +04:00

150 KiB

Raw Blame History

Срез 3b — формы ББ/ОФР в ядре и API до сводной ОСВ: план реализации

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: формы ББ/ОФР из tabular-parser и scan-gateway попадают в financial_statement_*, у фронта есть API экрана «Проверка распознавания» (форма, правка, проверки §7, акцепт, подтверждение) и закрыты мелкие дыры API пути до сводной ОСВ (исходный файл, комментарии, имена пользователей, documents_count, CSV чек-листа).

Architecture: маппинг форм — ветка существующего services/result_mapping.map_file с чистыми функциями domain/statement_mapping.py; проверки §7 — чистый domain/statement_checks.py (перенос tabular-parser) со словарём api/rules/statement_checks.yaml; API формы — services/statements.py + маршруты в api/endpoints/files/statements.py; мелкие дыры — точечные правки существующих сервисов и маршрутов. Контракт, DBML и миграции не меняются.

Tech Stack: Python 3.11, FastAPI, SQLAlchemy 2 async, PostgreSQL 16, pydantic v2, miniopy-async, PyYAML, pytest (asyncio_mode=auto, маркеры integration, fixtures), ruff, mypy.

Spec: openspec/changes/slice-3b-statements-api/ (proposal.md, design.md D-1…D-14 и решения №1–10, specs/osv-result-mapping, specs/statement-checks, specs/documents-api, specs/consolidations-api, specs/directories-api, specs/users-admin). Правила: ../docs/context/05-domain-rules.md §7, §9.3, §9.5, §11, §12.

Global Constraints

Репозиторий arm-core, ветка feat/36-slice-3b-statements-api, worktree /Users/andrey.epifancev/projects/awp-ca/.wt/slice-3b/arm-core; рядом симлинки ../docs, ../fixtures, ../tabular-parser, ../scan-gateway. make — из корня worktree, pytest — из api/ через .venv/bin/pytest. Интеграционные тесты требуют make up (compose-проект arm-ka: PostgreSQL, NATS, MinIO).
Никаких cd в основные checkout'ы /Users/andrey.epifancev/projects/awp-ca/{docs,arm-core,tabular-parser,scan-gateway,fixtures} (они на main; хук guard-main блокирует записи и ломает сабагентов); читать — абсолютными путями.
DBML v5.16 не меняется, миграций нет; новых значений enum нет. Поля — только из DBML (financial_statement_*, comment).
Контракт api/contracts/* не меняется (копия v1.2 уже в ветке).
Деньги — Decimal, numeric(18,2); никаких float в суммах. Ответы API отдают суммы строкой (как EntryOut).
Язык комментариев, docstring'ов, сообщений и коммитов — русский; идентификаторы — английские; строка ≤ 120 символов; make check (ruff check, ruff format --check, mypy src scripts, unit-тесты) зелёный после каждой задачи.
Права — через require("<CODE>") на маршруте, видимость — в сервисе (чужой = 404, как у документов): чтение формы, комментариев, файла — DOCUMENT_VIEW; правка, акцепт, комментарий, подтверждение — DOCUMENT_EDIT; CSV чек-листа — CONSOLIDATION_VIEW; /users/brief — current_user без права.
Аудит — только через RequestAudit.record (append-only); правка формы и акцепт — AuditAction.RECOGNIZED_DATA_EDIT, AuditSection.RECOGNITION, object_type = "document".
Порядок блокировок правки и акцепта — как у подтверждения: files_repo.lock_file → files_repo.lock_document.
Коммит после каждой задачи; сообщения feat(...), test(...), docs(...) на русском; в конце строк — атрибуция из системного напоминания сессии (Co-Authored-By: …, Claude-Session: …).

Review Focus

Одна ячейка дважды в теле PATCH (values с повтором (line_id, column_id)): ожидается 422 DUPLICATE_CELL, а не «побеждает последняя». Тест — Task 6 (test_patch_duplicate_cell_rejected).
Сумма с тремя знаками после запятой или больше 16 цифр до запятой в PATCH: 422 (валидация схемы), данные не меняются. Тест — Task 6 (test_patch_amount_precision_rejected).
Значение результата с column_index вне колонок формы (контракт не запрещает): пропускается с предупреждением маппинга, остальное маппится. Тест — Task 2 (test_value_outside_columns_skipped).
Прерванное скачивание большого файла: ответ S3 закрывается и отпускается (close + release) и при обрыве клиента. Тест — Task 9 (test_stream_releases_on_error).
Правка даты отчёта на дату не конец квартала (2025-10-31): квартал NULL, документ NEEDS_CHECK, подтверждение — 409 PERIOD_REQUIRED. Тест — Task 7 (test_confirm_after_non_quarter_date_edit).

Структура файлов

Путь (от `api/`)	Ответственность
`rules/statement_checks.yaml`	копия словаря §7 tabular-parser (D-4)
`src/domain/statement_checks.py`	`load_check_rules`, `run_checks`, `cross_form_check` — чистые функции §7
`src/domain/statement_mapping.py`	`statement_period`, `quarter_of`, `edition_for`, `build_statement`, `statement_status`
`src/domain/osv_mapping.py`	`counterparty_check` обобщается: `(inn, name, owner_inn, owner_names)`
`src/services/result_mapping.py`	ветка форм в `map_file`, `_load_result` без требования ОСВ, очистка форм файла
`src/repositories/statements.py`	загрузка формы документа, ячеек, пары для межформенной проверки, вставка строк
`src/schemas/statements.py`	`StatementOut`, `StatementPatch`, `ChecksAcceptIn`, вложенные модели
`src/services/statements.py`	`StatementService`: `get`, `patch`, `accept_checks`; правило статуса после правки
`src/api/endpoints/files/statements.py`	`GET/PATCH /documents/{id}/statement`, `POST …/checks/accept`
`src/services/documents.py`	условия подтверждения форм, `reviewed_by/at`, `comments_count`, комментарии
`src/api/endpoints/files/comments.py`	`POST/GET /documents/{id}/comments`
`src/core/storage.py`, `src/core/clients.py`	`ObjectStorage.stream` — поток объекта частями
`src/api/endpoints/files/read.py`	`GET /files/{id}/content`
`src/api/endpoints/users.py`	`GET /users/brief`
`src/services/legal_entities.py`, `src/repositories/directories.py`, `src/schemas/directories.py`	`documents_count`
`src/domain/checklist_csv.py`, `src/api/endpoints/consolidations/checklist.py`	CSV чек-листа
`tests/…`	по задаче; golden форм — `tests/test_mapping_statements.py`

Task 0: Подготовка окружения

Files: нет изменений.

Step 1: Окружение и базовая проверка

cd /Users/andrey.epifancev/projects/awp-ca/.wt/slice-3b/arm-core
make venv
make up
make check
cd api && .venv/bin/pytest -m integration -q -p no:cacheprovider tests/test_api_documents.py tests/test_result_mapping.py

Ожидание: make check зелёный; интеграционные тесты документов и маппинга зелёные. Если make up занят другим проектом — compose-проект общий (arm-ka), достаточно, чтобы контейнеры postgres/nats/minio были healthy (docker ps).

Step 2: Коммита нет — задача только проверяет окружение; в отчёте указать число тестов.

Task 1: Проверки §7 — словарь и чистые функции

Files:

Create: api/rules/statement_checks.yaml, api/src/domain/statement_checks.py
Test: api/tests/test_domain_statement_checks.py

Interfaces:

Produces:
- @dataclass(frozen=True) RelationRule(form: Literal["BALANCE","INCOME"], relation: str, target: str, terms: tuple[tuple[int, str], ...], strict: bool, abs_terms: frozenset[str]).
- @dataclass(frozen=True) CheckRules(tolerance: Decimal, rounding_tolerance_percent: Decimal, subline_check: bool, relations: tuple[RelationRule, ...]); метод relations_for(form) -> tuple[RelationRule, ...].
- load_check_rules(path: Path | None = None) -> CheckRules (по умолчанию api/rules/statement_checks.yaml), кэш functools.lru_cache на путь; битый файл — ValueError с именем файла.
- @dataclass(frozen=True) FailedCheck(column_index: int, relation: str, expected: Decimal, actual: Decimal).
- @dataclass(frozen=True) ChecksReport(failed: tuple[FailedCheck, ...], soft_failed: tuple[FailedCheck, ...]).
- run_checks(form: str, lines: Sequence[tuple[str, Mapping[int, Decimal | None]]], column_indexes: Sequence[int], rules: CheckRules) -> ChecksReport — lines — (line_code, {column_index: amount}) в порядке строк документа.
- cross_form_check(balance_1370: tuple[Decimal | None, Decimal | None], income_2400: Decimal | None, tolerance: Decimal) -> FailedCheck | None — (1370 на отчётную дату, 1370 на начало года); любой None → None (не проверялось); FailedCheck(column_index=-1, relation="1370 − 1370(начало года) = 2400", expected=Δ1370, actual=2400).
Step 1: Скопировать словарь

cp /Users/andrey.epifancev/projects/awp-ca/tabular-parser/api/rules/statement_checks.yaml api/rules/statement_checks.yaml

Дописать в начало файла две строки комментария (данные не трогать):

# КОПИЯ tabular-parser/api/rules/statement_checks.yaml (Срез 3b, design D-4): меняется в обоих репозиториях одновременно.
# Межрепозиторного теста идентичности нет сознательно (решение №10); расхождение копий — известный риск (03-architecture).

Step 2: Падающие тесты — api/tests/test_domain_statement_checks.py:

"""Проверки §7 ядра (spec statement-checks; design D-4, D-5): перенос проверок tabular-parser."""

from decimal import Decimal

import pytest

from src.domain.statement_checks import (
    ChecksReport,
    FailedCheck,
    cross_form_check,
    load_check_rules,
    run_checks,
)

D = Decimal
RULES = load_check_rules()


def _lines(*items: tuple[str, dict[int, str | None]]) -> list[tuple[str, dict[int, Decimal | None]]]:
    return [(code, {i: (D(v) if v is not None else None) for i, v in values.items()}) for code, values in items]


def test_rules_loaded_from_copy() -> None:
    assert RULES.tolerance == D("1")
    assert RULES.rounding_tolerance_percent == D("0.001")
    relations = [rule.relation for rule in RULES.relations_for("BALANCE")]
    assert "1600 = 1100 + 1200" in relations and "1600 = 1700" in relations
    income = {rule.relation: rule for rule in RULES.relations_for("INCOME")}
    assert income["2100 = 2110 - 2120"].abs_terms == frozenset({"2120"})
    assert income["2400 = 2300 - 2410 + 2420"].strict is False


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


def test_relation_skipped_without_target() -> None:
    report = run_checks("BALANCE", _lines(("1110", {0: "5"}), ("1150", {0: "7"})), [0], RULES)
    assert report == ChecksReport(failed=(), soft_failed=())


def test_tolerance_one_unit() -> None:
    report = run_checks("BALANCE", _lines(("1100", {0: "300"}), ("1200", {0: "600"}), ("1600", {0: "901"})), [0], RULES)
    assert all(item.relation != "1600 = 1100 + 1200" for item in report.failed)


def test_abs_terms_minus_sign() -> None:
    report = run_checks("INCOME", _lines(("2110", {0: "100"}), ("2120", {0: "-60"}), ("2100", {0: "40"})), [0], RULES)
    assert all(item.relation != "2100 = 2110 - 2120" for item in report.failed)


def test_rounding_goes_soft() -> None:
    # расхождение 5 при числах ~1 000 000: 5 ≤ 1 000 000 × 0,001 % = 10 → мягкое «в пределах округления»
    report = run_checks(
        "BALANCE", _lines(("1100", {0: "500000"}), ("1200", {0: "500000"}), ("1600", {0: "1000005"})), [0], RULES
    )
    assert all(item.relation != "1600 = 1100 + 1200" for item in report.failed)
    assert any(item.relation == "1600 = 1100 + 1200" for item in report.soft_failed)


def test_soft_relation_not_hard() -> None:
    report = run_checks("INCOME", _lines(("2300", {0: "100"}), ("2410", {0: "20"}), ("2400", {0: "50"})), [0], RULES)
    assert all(item.relation != "2400 = 2300 - 2410 + 2420" for item in report.failed)
    assert any(item.relation == "2400 = 2300 - 2410 + 2420" for item in report.soft_failed)


def test_sublines_soft() -> None:
    report = run_checks("BALANCE", _lines(("1150", {0: "10"}), ("11501", {0: "6"}), ("11502", {0: "2"})), [0], RULES)
    assert any(item.relation == "1150 = 11501 + 11502" for item in report.soft_failed)
    assert report.failed == ()


def test_per_column_and_null_as_zero() -> None:
    report = run_checks(
        "BALANCE",
        _lines(("1100", {0: "300", 1: None}), ("1200", {0: "600", 1: "5"}), ("1600", {0: "900", 1: "5"})),
        [0, 1],
        RULES,
    )
    assert report.failed == ()


@pytest.mark.parametrize(
    ("b1370", "i2400", "expected"),
    [
        ((D("500"), D("300")), D("200"), None),
        ((D("500"), D("300")), D("150"), FailedCheck(-1, "1370 − 1370(начало года) = 2400", D("200"), D("150"))),
        ((None, D("300")), D("200"), None),
        ((D("500"), D("300")), None, None),
    ],
)
def test_cross_form(b1370: tuple[Decimal | None, Decimal | None], i2400: Decimal | None, expected: FailedCheck | None) -> None:
    assert cross_form_check(b1370, i2400, D("1")) == expected


def test_broken_rules_file(tmp_path) -> None:
    bad = tmp_path / "statement_checks.yaml"
    bad.write_text("tolerance: 1\nrelations:\n  - {form: BALANCE, relation: '16A0 = 1100', strict: true}\n", "utf-8")
    with pytest.raises(ValueError, match="statement_checks.yaml"):
        load_check_rules(bad)

Step 3: Убедиться, что падают — cd api && .venv/bin/pytest tests/test_domain_statement_checks.py -q → ModuleNotFoundError.
Step 4: Реализация — api/src/domain/statement_checks.py:

"""Логические проверки форм ББ/ОФР §7 в ядре (spec statement-checks; design D-4, D-5).

Перенос `tabular-parser/api/src/parser/statements/checks.py` по семантике: соотношение по каждой колонке, только если в
колонке есть его цель; отсутствующий код и пустая сумма — 0; правая часть считается как записано и с `abs_terms` по
модулю, выполнено, если хоть один вариант в пределах `tolerance`; жёсткое нарушение не больше
`rounding_tolerance_percent` от наибольшего по модулю числа соотношения — мягкое; «сумма пятизначных подстрок =
четырёхзначный родитель» — мягкая. Результат не хранится (DBML).
"""

import functools
import re
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Literal

import yaml

__all__ = [
    "CROSS_FORM_RELATION",
    "CheckRules",
    "ChecksReport",
    "FailedCheck",
    "RelationRule",
    "cross_form_check",
    "load_check_rules",
    "run_checks",
]

Form = Literal["BALANCE", "INCOME"]
RULES_PATH = Path(__file__).resolve().parents[2] / "rules" / "statement_checks.yaml"
CROSS_FORM_RELATION = "1370 − 1370(начало года) = 2400"
# диапазоны четырёхзначных кодов форм для генерации подстрок (§12 п. 2; в tabular-parser — statement_headers.yaml)
CODE_RANGES: dict[str, tuple[int, int]] = {"BALANCE": (1100, 1799), "INCOME": (2100, 2999)}
_ZERO = Decimal("0")
_HUNDRED = Decimal("100")
_PARENT_LEN = 4
_GROUP_LEN = 3
_RELATION = re.compile(r"^\s*(?P<target>\d{4})\s*=\s*(?P<rhs>\d{4,5}(\s*[+\-]\s*\d{4,5})*)\s*$")
_TERM = re.compile(r"([+\-])?\s*(\d{4,5})")


@dataclass(frozen=True)
class RelationRule:
    form: Form
    relation: str
    target: str
    terms: tuple[tuple[int, str], ...]
    strict: bool
    abs_terms: frozenset[str] = frozenset()


@dataclass(frozen=True)
class CheckRules:
    tolerance: Decimal
    rounding_tolerance_percent: Decimal
    subline_check: bool
    relations: tuple[RelationRule, ...]

    def relations_for(self, form: str) -> tuple[RelationRule, ...]:
        return tuple(rule for rule in self.relations if rule.form == form)


@dataclass(frozen=True)
class FailedCheck:
    column_index: int
    relation: str
    expected: Decimal
    actual: Decimal


@dataclass(frozen=True)
class ChecksReport:
    failed: tuple[FailedCheck, ...]
    soft_failed: tuple[FailedCheck, ...]


def _parse_relation(form: Form, relation: str, strict: bool, abs_terms: Sequence[str]) -> RelationRule:
    found = _RELATION.match(relation)
    if found is None:
        raise ValueError(f"соотношение «{relation}»: ожидается «<код> = <код> [+|-] <код> …»")
    terms = tuple((-1 if sign == "-" else 1, code) for sign, code in _TERM.findall(found.group("rhs")))
    unknown = set(abs_terms) - {code for _, code in terms}
    if unknown:
        raise ValueError(f"abs_terms {sorted(unknown)} нет в правой части «{relation}»")
    return RelationRule(form, relation, found.group("target"), terms, strict, frozenset(abs_terms))


@functools.lru_cache(maxsize=4)
def load_check_rules(path: Path | None = None) -> CheckRules:
    """Словарь §7 из yaml; любая ошибка — `ValueError` с именем файла."""
    source = path or RULES_PATH
    try:
        data = yaml.safe_load(source.read_text(encoding="utf-8"))
        relations = tuple(
            _parse_relation(item["form"], item["relation"], bool(item["strict"]), item.get("abs_terms", ()))
            for item in data["relations"]
        )
        for rule in relations:
            if rule.form not in CODE_RANGES:
                raise ValueError(f"форма {rule.form!r}: ожидается BALANCE или INCOME")
        return CheckRules(
            tolerance=Decimal(str(data["tolerance"])),
            rounding_tolerance_percent=Decimal(str(data.get("rounding_tolerance_percent", 0))),
            subline_check=bool(data.get("subline_check", True)),
            relations=relations,
        )
    except (OSError, KeyError, TypeError, ValueError, InvalidOperation, yaml.YAMLError) as exc:
        raise ValueError(f"словарь проверок {source.name}: {exc}") from exc


def _rhs(amounts: Mapping[str, Decimal], rule: RelationRule, *, absolute: bool) -> Decimal:
    total = _ZERO
    for sign, code in rule.terms:
        value = amounts.get(code, _ZERO)
        if absolute and code in rule.abs_terms:
            value = abs(value)
        total += sign * value
    return total


def _holds(amounts: Mapping[str, Decimal], rule: RelationRule, tolerance: Decimal) -> tuple[bool, Decimal, Decimal]:
    expected = amounts.get(rule.target, _ZERO)
    plain = _rhs(amounts, rule, absolute=False)
    if abs(expected - plain) <= tolerance:
        return True, expected, plain
    if rule.abs_terms:
        with_abs = _rhs(amounts, rule, absolute=True)
        if abs(expected - with_abs) <= tolerance:
            return True, expected, with_abs
        if abs(expected - with_abs) < abs(expected - plain):
            return False, expected, with_abs
    return False, expected, plain


def _subline_rules(codes: Sequence[str], form: str) -> list[RelationRule]:
    """«1110 = 11101 + …» для четырёхзначных кодов формы с пятизначными подстроками (по порядку строк, M-9 2b)."""
    low, high = CODE_RANGES[form]
    parents = {code for code in codes if len(code) == _PARENT_LEN}
    children: dict[str, list[str]] = {}
    last_parent: str | None = None
    for code in codes:
        if len(code) == _PARENT_LEN:
            last_parent = code
            continue
        if len(code) != _PARENT_LEN + 1:
            continue
        if last_parent is not None and last_parent[:_GROUP_LEN] == code[:_GROUP_LEN]:
            parent = last_parent
        elif code[:_PARENT_LEN] in parents:
            parent = code[:_PARENT_LEN]
        else:
            continue
        children.setdefault(parent, []).append(code)
    rules: list[RelationRule] = []
    for parent, kids in children.items():
        if parent.isdigit() and low <= int(parent) <= high:
            relation = f"{parent} = {' + '.join(kids)}"
            rules.append(RelationRule(form, relation, parent, tuple((1, kid) for kid in kids), False))  # type: ignore[arg-type]
    return rules


def run_checks(
    form: str,
    lines: Sequence[tuple[str, Mapping[int, Decimal | None]]],
    column_indexes: Sequence[int],
    rules: CheckRules,
) -> ChecksReport:
    """Проверки §7 формы `form` по колонкам `column_indexes`; `lines` — `(код, {колонка: сумма})` по порядку строк."""
    explicit = list(rules.relations_for(form))
    sublines = _subline_rules([code for code, _ in lines], form) if rules.subline_check else []
    hard: list[FailedCheck] = []
    soft: list[FailedCheck] = []
    for column_index in column_indexes:
        amounts = {code: (values.get(column_index) or _ZERO) for code, values in lines}
        for rule in [*explicit, *sublines]:
            if rule not in sublines and rule.target not in amounts:
                continue
            holds, expected, actual = _holds(amounts, rule, rules.tolerance)
            if holds:
                continue
            failed = FailedCheck(column_index, rule.relation, expected, actual)
            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:
                hard.append(failed)
            else:
                soft.append(failed)
    return ChecksReport(failed=tuple(hard), soft_failed=tuple(soft))


def cross_form_check(
    balance_1370: tuple[Decimal | None, Decimal | None], income_2400: Decimal | None, tolerance: Decimal
) -> FailedCheck | None:
    """Межформенная §7 (мягкая, D-5): `1370(отчётная дата) − 1370(начало года) = 2400(период)`; неполные данные — None."""
    current, year_start = balance_1370
    if current is None or year_start is None or income_2400 is None:
        return None
    delta = current - year_start
    if abs(delta - income_2400) <= tolerance:
        return None
    return FailedCheck(-1, CROSS_FORM_RELATION, delta, income_2400)

Примечание: RelationRule у подстрок строится с form типа str; если mypy ругается — привести form к Form через cast(Form, form) вместо # type: ignore.

Step 5: Тесты и линт — cd api && .venv/bin/pytest tests/test_domain_statement_checks.py -q && cd .. && make lint → PASS.
Step 6: Commit

git add api/rules/statement_checks.yaml api/src/domain/statement_checks.py api/tests/test_domain_statement_checks.py
git commit -m "feat(statements): проверки §7 в ядре — перенос tabular-parser, словарь statement_checks.yaml (D-4, D-5)"

Task 2: Чистые функции маппинга формы

Files:

Create: api/src/domain/statement_mapping.py
Modify: api/src/domain/osv_mapping.py (обобщить counterparty_check)
Test: api/tests/test_domain_statement_mapping.py, api/tests/test_domain_osv_mapping.py (адаптация вызовов)

Interfaces:

Consumes: src.contract.result.StatementDocument (header, columns, lines), ChecksReport, run_checks, load_check_rules (Task 1).
Produces (в src/domain/statement_mapping.py):
- quarter_of(day: date) -> PeriodQuarter | None — Q1..Q4 только для 31.03, 30.06, 30.09, 31.12.
- statement_period(report_date: date | None, column_dates: Sequence[date | None]) -> date | None — дата отчёта, иначе максимальная дата колонок, иначе None.
- edition_for(period_date: date | None) -> str | None — "66n" при ≤ 2024-12-31, "fsbu4" позже, None без даты.
- @dataclass ValueDraft(line_code: str, column_index: int, amount: Decimal | None, string_value: str | None, confidence: Decimal | None, is_recognized: bool | None).
- @dataclass LineDraft(line_code: str, line_code_confidence: Decimal | None, line_name: str, line_name_confidence: Decimal | None).
- @dataclass StatementDraft(document_type: DocumentType, statement_type: StatementType, locator: str, period_date: date | None, period_year: int | None, period_quarter: PeriodQuarter | None, header: dict[str, Any], columns: list[tuple[int, str, date | None, bool | None]], lines: list[LineDraft], values: list[ValueDraft], warnings: list[str]) — header — поля financial_statement_meta (ключи — имена колонок ORM), columns — (column_index, label, period_date, is_current).
- build_statement(doc: StatementDocument) -> StatementDraft — D-2 (значение на каждую пару строка × колонка, повтор кода, значение вне колонок, период, is_current).
- statement_status(*, job_done: bool, checks_passed: bool, counterparty: DocumentCheckResult, period_known: bool, values_ok: bool, has_lines: bool, warnings: Sequence[str]) -> DocumentStatus — D-6.
- values_ok(values: Iterable[tuple[bool, bool | None, Decimal | None]]) -> bool — по кортежам (is_modified, is_recognized, confidence).
Produces (в src/domain/osv_mapping.py): counterparty_check(inn: str | None, name: str | None, owner_inn: str, owner_names_normalized: frozenset[str]) -> DocumentCheckResult; вызовы для ОСВ передают header.organization_inn, header.organization_name.
Step 1: Падающие тесты — api/tests/test_domain_statement_mapping.py:

"""Чистые функции маппинга формы (spec osv-result-mapping «Документ формы ББ/ОФР»; design D-2, D-3, D-6, D-14)."""

import datetime
from decimal import Decimal

import pytest

from src.contract.result import StatementDocument
from src.domain.statement_mapping import (
    build_statement,
    edition_for,
    quarter_of,
    statement_period,
    statement_status,
    values_ok,
)
from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter, StatementType

d = datetime.date


@pytest.mark.parametrize(
    ("day", "quarter"),
    [(d(2025, 3, 31), PeriodQuarter.Q1), (d(2025, 6, 30), PeriodQuarter.Q2), (d(2025, 9, 30), PeriodQuarter.Q3),
     (d(2024, 12, 31), PeriodQuarter.Q4), (d(2025, 10, 31), None), (d(2025, 6, 29), None)],
)
def test_quarter_of(day: datetime.date, quarter: PeriodQuarter | None) -> None:
    assert quarter_of(day) is quarter


def test_period_and_edition() -> None:
    assert statement_period(d(2025, 9, 30), [d(2024, 12, 31)]) == d(2025, 9, 30)
    assert statement_period(None, [d(2024, 12, 31), d(2025, 6, 30), None]) == d(2025, 6, 30)
    assert statement_period(None, [None]) is None
    assert edition_for(d(2024, 12, 31)) == "66n" and edition_for(d(2025, 3, 31)) == "fsbu4" and edition_for(None) is None


def _doc(**overrides: object) -> StatementDocument:
    data: dict[str, object] = {
        "document_type": "BALANCE",
        "locator": "sheet:ББ",
        "header": {"form_name": "Бухгалтерский баланс", "inn": "7700000000", "inn_confidence": 100,
                   "report_date": "2025-09-30", "report_date_text": "на 30 сентября 2025 г.", "units": "тыс. руб.",
                   "organization_name": "ООО Пример", "okud": "0710001"},
        "columns": [{"column_index": 0, "label": "На 30.09.2025", "period_date": "2025-09-30"},
                    {"column_index": 1, "label": "На 31.12.2024", "period_date": "2024-12-31"}],
        "lines": [
            {"line_code": "1150", "line_code_confidence": 100, "line_name": "ОС", "line_name_confidence": 100,
             "values": [{"column_index": 0, "amount": "10", "confidence": 100, "is_recognized": True}]},
            {"line_code": "1150", "line_name": "дубль", "values": []},
            {"line_code": "1600", "line_name": "БАЛАНС",
             "values": [{"column_index": 5, "amount": "1", "confidence": 100, "is_recognized": True}]},
        ],
    }
    data.update(overrides)
    return StatementDocument.model_validate(data)


def test_build_statement_cells_and_header() -> None:
    draft = build_statement(_doc())
    assert draft.document_type is DocumentType.BALANCE and draft.statement_type is StatementType.balance
    assert (draft.period_year, draft.period_quarter, draft.period_date) == (2025, PeriodQuarter.Q3, d(2025, 9, 30))
    assert draft.header["inn"] == "7700000000" and draft.header["inn_original_value"] == "7700000000"
    assert draft.header["inn_verified"] is True
    assert draft.header["report_date_original_value"] == "на 30 сентября 2025 г."
    assert draft.header["units"] == draft.header["units_original_value"] == "тыс. руб."
    assert draft.columns == [(0, "На 30.09.2025", d(2025, 9, 30), True), (1, "На 31.12.2024", d(2024, 12, 31), False)]
    assert [line.line_code for line in draft.lines] == ["1150", "1600"]  # дубль 1150 отброшен
    cells = {(v.line_code, v.column_index): v for v in draft.values}
    assert len(cells) == 4  # 2 строки × 2 колонки
    assert cells[("1150", 0)].amount == Decimal("10") and cells[("1150", 1)].amount is None
    assert cells[("1150", 1)].is_recognized is None and cells[("1600", 0)].confidence is None
    assert any("1150" in w and "повтор" in w for w in draft.warnings)
    assert any("колонк" in w and "5" in w for w in draft.warnings)


def test_value_outside_columns_skipped() -> None:
    """Review Focus 3: значение с column_index вне колонок — пропущено с предупреждением, остальное смаплено."""
    draft = build_statement(_doc())
    assert all(v.column_index in (0, 1) for v in draft.values)


def test_period_from_columns_and_non_quarter() -> None:
    no_date = _doc(header={"form_name": "ББ", "report_date": None})
    draft = build_statement(no_date)
    assert (draft.period_year, draft.period_quarter) == (2025, PeriodQuarter.Q3)
    assert draft.header["report_date"] is None
    october = _doc(header={"form_name": "ББ", "report_date": "2025-10-31"})
    draft = build_statement(october)
    assert (draft.period_year, draft.period_quarter) == (2025, None)
    assert any("конец квартала" in w for w in draft.warnings)


def test_no_dates_at_all() -> None:
    draft = build_statement(_doc(header={"form_name": "ББ"}, columns=[{"column_index": 0, "label": ""}]))
    assert (draft.period_year, draft.period_quarter, draft.period_date) == (None, None, None)
    assert draft.columns == [(0, "", None, None)]


def test_status_rules() -> None:
    ok = dict(job_done=True, checks_passed=True, counterparty=DocumentCheckResult.MATCH, period_known=True,
              values_ok=True, has_lines=True, warnings=[])
    assert statement_status(**ok) is DocumentStatus.RECOGNIZED  # type: ignore[arg-type]
    for key, bad in [("job_done", False), ("checks_passed", False), ("counterparty", DocumentCheckResult.MISMATCH),
                     ("period_known", False), ("values_ok", False), ("has_lines", False), ("warnings", ["x"])]:
        assert statement_status(**{**ok, key: bad}) is DocumentStatus.NEEDS_CHECK  # type: ignore[arg-type]
    assert statement_status(**{**ok, "counterparty": DocumentCheckResult.UNKNOWN}) is DocumentStatus.RECOGNIZED  # type: ignore[arg-type]


def test_values_ok() -> None:
    hundred = Decimal("100")
    assert values_ok([(False, True, hundred), (False, None, None)])
    assert not values_ok([(False, False, hundred)])
    assert not values_ok([(False, True, Decimal("99"))])
    assert values_ok([(True, False, Decimal("10"))])  # правленое аналитиком значение не тянет статус вниз

В api/tests/test_domain_osv_mapping.py найти вызовы counterparty_check(header, …) (grep -n counterparty_check) и переписать на counterparty_check(header.organization_inn, header.organization_name, …); добавить тест:

def test_counterparty_check_plain_values() -> None:
    names = frozenset({normalize_alias("ООО Ромашка")})
    assert counterparty_check("7700000000", None, "7700000000", names) is DocumentCheckResult.MATCH
    assert counterparty_check(None, "ООО «Ромашка»", "7700000001", names) is DocumentCheckResult.MATCH
    assert counterparty_check("3121200351", "мусор", "7700000000", names) is DocumentCheckResult.MISMATCH
    assert counterparty_check(None, None, "7700000000", names) is DocumentCheckResult.UNKNOWN

Step 2: Убедиться, что падают — cd api && .venv/bin/pytest tests/test_domain_statement_mapping.py tests/test_domain_osv_mapping.py -q.
Step 3: counterparty_check в src/domain/osv_mapping.py:

def counterparty_check(
    inn: str | None, name: str | None, owner_inn: str, owner_names_normalized: frozenset[str]
) -> DocumentCheckResult:
    """D-8 Среза 2c, общий для ОСВ и форм (Срез 3b): организация в шапке — владелец файла?

    ИНН шапки есть → `MATCH`, если его цифры равны ИНН владельца, иначе `MISMATCH`. ИНН нет →
    `normalize_alias(name)` среди нормализованных имени и алиасов владельца → `MATCH`, иначе `MISMATCH`.
    Нет ни ИНН, ни имени → `UNKNOWN`.
    """
    header_inn = _NON_DIGITS_RE.sub("", inn or "")
    if header_inn:
        return (
            DocumentCheckResult.MATCH
            if header_inn == _NON_DIGITS_RE.sub("", owner_inn)
            else DocumentCheckResult.MISMATCH
        )
    normalized = normalize_alias(name)
    if not normalized:
        return DocumentCheckResult.UNKNOWN
    return DocumentCheckResult.MATCH if normalized in owner_names_normalized else DocumentCheckResult.MISMATCH

В src/services/result_mapping.py вызов заменить на counterparty_check(sheet.header.organization_inn, sheet.header.organization_name, owner.inn, owner_names); убрать неиспользуемый импорт OsvHeader из osv_mapping.py, если он больше не нужен.

Step 4: statement_mapping.py:

"""Чистые функции маппинга формы ББ/ОФР (`statementDocument` v1.2) в черновики `financial_statement_*`
(spec osv-result-mapping «Документ формы ББ/ОФР», «Статус документа формы»; design D-2, D-3, D-6, D-14)."""

import datetime
from collections.abc import Iterable, Sequence
from dataclasses import dataclass, field
from decimal import Decimal
from typing import Any

from src.contract.result import StatementDocument
from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, PeriodQuarter, StatementType

__all__ = [
    "LineDraft",
    "StatementDraft",
    "ValueDraft",
    "build_statement",
    "edition_for",
    "quarter_of",
    "statement_period",
    "statement_status",
    "values_ok",
]

FULL_CONFIDENCE = Decimal("100")
EDITION_BOUNDARY = datetime.date(2024, 12, 31)
_QUARTER_ENDS = {(3, 31): PeriodQuarter.Q1, (6, 30): PeriodQuarter.Q2, (9, 30): PeriodQuarter.Q3, (12, 31): PeriodQuarter.Q4}
_STATEMENT_TYPES = {DocumentType.BALANCE: StatementType.balance, DocumentType.INCOME: StatementType.income}


@dataclass
class ValueDraft:
    line_code: str
    column_index: int
    amount: Decimal | None
    string_value: str | None
    confidence: Decimal | None
    is_recognized: bool | None


@dataclass
class LineDraft:
    line_code: str
    line_code_confidence: Decimal | None
    line_name: str
    line_name_confidence: Decimal | None


@dataclass
class StatementDraft:
    document_type: DocumentType
    statement_type: StatementType
    locator: str
    period_date: datetime.date | None
    period_year: int | None
    period_quarter: PeriodQuarter | None
    header: dict[str, Any]
    columns: list[tuple[int, str, datetime.date | None, bool | None]]
    lines: list[LineDraft]
    values: list[ValueDraft]
    warnings: list[str] = field(default_factory=list)


def quarter_of(day: datetime.date) -> PeriodQuarter | None:
    return _QUARTER_ENDS.get((day.month, day.day))


def statement_period(report_date: datetime.date | None, column_dates: Sequence[datetime.date | None]) -> datetime.date | None:
    """D-2: дата отчёта, иначе самая поздняя дата колонок."""
    if report_date is not None:
        return report_date
    dates = [day for day in column_dates if day is not None]
    return max(dates) if dates else None


def edition_for(period_date: datetime.date | None) -> str | None:
    """D-14: редакция формы для шаблона фронта."""
    if period_date is None:
        return None
    return "66n" if period_date <= EDITION_BOUNDARY else "fsbu4"


def _decimal(value: float | None) -> Decimal | None:
    return Decimal(str(value)) if value is not None else None


def build_statement(doc: StatementDocument) -> StatementDraft:
    """Черновик формы `BALANCE`/`INCOME` (D-2): шапка, колонки, строки (повтор кода — первая), значение на каждую ячейку."""
    document_type = DocumentType(doc.document_type)
    header = doc.header
    warnings: list[str] = []
    column_dates = [column.period_date for column in doc.columns]
    period_date = statement_period(header.report_date, column_dates)
    period_year = period_date.year if period_date is not None else None
    period_quarter = quarter_of(period_date) if period_date is not None else None
    if period_date is None:
        warnings.append(f"{doc.locator}: период формы не определён (нет даты отчёта и дат колонок)")
    elif period_quarter is None:
        warnings.append(f"{doc.locator}: дата {period_date.isoformat()} не конец квартала — квартал не определён")
    latest = max((day for day in column_dates if day is not None), default=None)
    columns = [
        (
            column.column_index,
            column.label,
            column.period_date,
            None if latest is None else column.period_date == latest
            and column.column_index == min(c.column_index for c in doc.columns if c.period_date == latest),
        )
        for column in doc.columns
    ]
    known_columns = {column.column_index for column in doc.columns}
    lines: list[LineDraft] = []
    values: list[ValueDraft] = []
    seen: set[str] = set()
    for line in doc.lines:
        if line.line_code in seen:
            warnings.append(f"{doc.locator}: код {line.line_code} повторяется — оставлена первая строка")
            continue
        seen.add(line.line_code)
        lines.append(
            LineDraft(line.line_code, _decimal(line.line_code_confidence), line.line_name, _decimal(line.line_name_confidence))
        )
        by_column = {}
        for value in line.values:
            if value.column_index not in known_columns:
                warnings.append(
                    f"{doc.locator}: строка {line.line_code}: значение колонки {value.column_index} вне колонок формы пропущено"
                )
                continue
            by_column.setdefault(value.column_index, value)
        for column in doc.columns:
            value = by_column.get(column.column_index)
            values.append(
                ValueDraft(
                    line.line_code,
                    column.column_index,
                    value.amount if value is not None else None,
                    value.string_value if value is not None else None,
                    _decimal(value.confidence) if value is not None else None,
                    value.is_recognized if value is not None else None,
                )
            )
    inn_confidence = _decimal(header.inn_confidence)
    meta = {
        "inn": header.inn,
        "inn_original_value": header.inn,
        "inn_confidence": inn_confidence,
        "inn_verified": inn_confidence == FULL_CONFIDENCE if header.inn is not None else None,
        "inn_modified": False,
        "organization_name": header.organization_name,
        "report_date": header.report_date,
        "report_date_original_value": header.report_date_text,
        "report_date_verified": header.report_date is not None,
        "report_date_modified": False,
        "units": header.units,
        "units_original_value": header.units,
        "units_verified": header.units is not None,
        "units_modified": False,
        "okud": header.okud,
        "okpo": header.okpo,
        "okopf": header.okopf,
        "okfs": header.okfs,
        "address": header.address,
    }
    return StatementDraft(
        document_type=document_type,
        statement_type=_STATEMENT_TYPES[document_type],
        locator=doc.locator,
        period_date=period_date,
        period_year=period_year,
        period_quarter=period_quarter,
        header=meta,
        columns=columns,
        lines=lines,
        values=values,
        warnings=warnings,
    )


def values_ok(values: Iterable[tuple[bool, bool | None, Decimal | None]]) -> bool:
    """D-6: каждое значение правлено аналитиком или распознано без сомнений (`is_recognized` не false, уверенность 100/NULL)."""
    return all(
        modified or (recognized is not False and (confidence is None or confidence == FULL_CONFIDENCE))
        for modified, recognized, confidence in values
    )


def statement_status(  # noqa: PLR0913 - условия D-6 перечислены явно
    *,
    job_done: bool,
    checks_passed: bool,
    counterparty: DocumentCheckResult,
    period_known: bool,
    values_ok: bool,
    has_lines: bool,
    warnings: Sequence[str],
) -> DocumentStatus:
    """D-6: `RECOGNIZED` только при всех условиях; иначе `NEEDS_CHECK`."""
    recognized = (
        job_done and checks_passed and counterparty != DocumentCheckResult.MISMATCH and period_known
        and values_ok and has_lines and not warnings
    )
    return DocumentStatus.RECOGNIZED if recognized else DocumentStatus.NEEDS_CHECK

Упростить вычисление is_current при ревью можно, но семантика фиксирована: True у колонки с самой поздней датой (при равенстве — с меньшим column_index), False у остальных при наличии дат, None у всех без дат. Тип confidence в контракте (Confidence = float | None) — поэтому _decimal через str.

Step 5: Тесты и линт — cd api && .venv/bin/pytest tests/test_domain_statement_mapping.py tests/test_domain_osv_mapping.py -q && cd .. && make lint → PASS.
Step 6: Commit

git add api/src/domain/statement_mapping.py api/src/domain/osv_mapping.py api/src/services/result_mapping.py api/tests/test_domain_statement_mapping.py api/tests/test_domain_osv_mapping.py
git commit -m "feat(statements): чистые функции маппинга формы — период, черновик, статус; counterparty_check общий (D-2, D-6)"

Task 3: Помощники golden-тестов форм

Files:

Modify: api/tests/fixtures_helpers.py
Test: api/tests/test_fixtures_marker.py (дописать проверки помощников)

Interfaces:

Produces:
- ocr_result(path: Path, *, job_id: uuid.UUID, file_id: uuid.UUID) -> RecognitionResult — эталон нормализатора fixtures/ocr/expected_normalize/<name>.json ({status, warnings, metrics, error, documents}) как RecognitionResult: schema_version = "1.2", service = {"name": "scan-gateway", "version": "golden"}, status, documents, error и metrics как есть (если не None).
- map_golden_snapshot(..., result_builder=golden_result) — новый необязательный именованный параметр result_builder: Callable[..., RecognitionResult]; возвращает все документы файла, отсортированные по (document_type, osv_account_code) (было — только по osv_account_code).
- Константы: GOLDEN_STATEMENTS = Path("golden") / "expected_statement_parse", OCR_EXPECTED = Path("ocr") / "expected_normalize".
Step 1: Падающий тест — в api/tests/test_fixtures_marker.py дописать:

@pytest.mark.fixtures
def test_ocr_result_wraps_normalizer_golden(fixtures_dir: Path) -> None:
    import uuid

    from tests.fixtures_helpers import OCR_EXPECTED, ocr_result

    job_id, file_id = uuid.uuid4(), uuid.uuid4()
    result = ocr_result(fixtures_dir / OCR_EXPECTED / "aep_6.json", job_id=job_id, file_id=file_id)
    assert result.job_id == job_id and result.file_id == file_id
    assert result.service.name == "scan-gateway"
    assert [doc.document_type for doc in result.documents] == ["BALANCE", "INCOME"]

Step 2: Убедиться, что падает — cd api && .venv/bin/pytest tests/test_fixtures_marker.py -q -k ocr_result.
Step 3: Реализация в api/tests/fixtures_helpers.py:

GOLDEN_STATEMENTS = Path("golden") / "expected_statement_parse"
OCR_EXPECTED = Path("ocr") / "expected_normalize"


def ocr_result(path: Path, *, job_id: uuid.UUID, file_id: uuid.UUID) -> "RecognitionResult":
    """Эталон нормализатора scan-gateway (`ocr/expected_normalize/<name>.json`) как `RecognitionResult` контракта v1.2."""
    from src.contract.result import RecognitionResult  # noqa: PLC0415 - цикл с conftest

    data = json.loads(path.read_text("utf-8"))
    payload: dict[str, object] = {
        "schema_version": "1.2",
        "job_id": str(job_id),
        "file_id": str(file_id),
        "service": {"name": "scan-gateway", "version": "golden"},
        "status": data["status"],
        "documents": data["documents"],
    }
    if data.get("error"):
        payload["error"] = data["error"]
    if data.get("metrics"):
        payload["metrics"] = data["metrics"]
    return RecognitionResult.model_validate(payload)

В map_golden_snapshot добавить параметр result_builder: "Callable[..., RecognitionResult] | None" = None (по умолчанию golden_result), использовать (result_builder or golden_result)(path, job_id=job.id, file_id=file.id); сортировку документов заменить на .order_by(Document.document_type, Document.osv_account_code). Существующие вызовы не меняются (ОСВ-документы — те же; сортировка по типу для ОСВ ничего не меняет).

Step 4: Тесты — cd api && .venv/bin/pytest tests/test_fixtures_marker.py -q && cd .. && make lint → PASS.
Step 5: Commit

git add api/tests/fixtures_helpers.py api/tests/test_fixtures_marker.py
git commit -m "test(statements): помощники golden — ocr_result, все документы файла в map_golden_snapshot"

Task 4: Маппинг форм в `map_file`

Files:

Create: api/src/repositories/statements.py
Modify: api/src/services/result_mapping.py (_load_result, _clear_file, map_file)
Test: api/tests/test_mapping_statements.py (новый), api/tests/test_result_mapping.py (сценарий «нет документов ОСВ» → «нет документов»)

Interfaces:

Consumes: build_statement, statement_status, values_ok (Task 2); run_checks, load_check_rules (Task 1); counterparty_check (Task 2); ocr_result, map_golden_snapshot, GOLDEN_STATEMENTS, OCR_EXPECTED (Task 3).
Produces:
- repositories/statements.py:
  - async def insert_statement(session, *, document_id: uuid.UUID, job_id: uuid.UUID, draft: StatementDraft) -> FinancialStatementMeta — вставляет meta, колонки, строки и значения (пакетами по 1000), возвращает meta.
  - async def delete_file_statements(session, file_id: uuid.UUID) -> None — удаляет financial_statement_value/line/column/meta документов файла (в порядке FK).
  - async def statement_amounts(session, meta_id: uuid.UUID) -> tuple[list[int], list[tuple[str, dict[int, Decimal | None]]]] — индексы колонок и (line_code, {column_index: amount}) по порядку строк (по line_code) — вход run_checks.
  - async def value_flags(session, meta_id: uuid.UUID) -> list[tuple[bool, bool | None, Decimal | None]] — (is_modified, is_recognized, confidence) всех значений — вход values_ok.
- services/result_mapping.py: константа NO_DOCUMENTS = "в результате нет документов"; MappingFatal(NO_DOCUMENTS) вместо прежнего текста; формы маппятся после листов ОСВ.
- Статус формы при маппинге: checks_passed = not run_checks(...).failed (акцепта при маппинге нет).
Step 1: Падающие тесты — api/tests/test_mapping_statements.py:

"""Маппинг форм ББ/ОФР (spec osv-result-mapping «Цикл маппинга…», «Документ формы ББ/ОФР», «Повтор формы в файле и
прочие документы», «Статус документа формы»; design D-1…D-3, D-6) на golden-снимках fixtures."""

import uuid
from decimal import Decimal
from pathlib import Path

import pytest
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from src.models.enums import DocumentCheckResult, DocumentStatus, DocumentType, FileStatus, PeriodQuarter
from src.models.files import Document, StoredFile
from src.models.statements import (
    FinancialStatementColumn,
    FinancialStatementLine,
    FinancialStatementMeta,
    FinancialStatementValue,
)
from tests.factories import make_legal_entity
from tests.fixtures_helpers import GOLDEN_STATEMENTS, OCR_EXPECTED, map_golden_snapshot, ocr_result

pytestmark = [pytest.mark.integration, pytest.mark.fixtures]

MEZH = GOLDEN_STATEMENTS / "mezheninovskaya" / "2025-Q3"
NPZ = GOLDEN_STATEMENTS / "npz" / "2025-Q3"
SYNTH = GOLDEN_STATEMENTS / "synthetic"


async def _meta(session: AsyncSession, document: Document) -> FinancialStatementMeta:
    return (
        await session.execute(select(FinancialStatementMeta).where(FinancialStatementMeta.document_id == document.id))
    ).scalar_one()


async def _count(session: AsyncSession, model: type, meta_id: uuid.UUID) -> int:
    if model is FinancialStatementValue:
        stmt = (
            select(func.count())
            .select_from(FinancialStatementValue)
            .join(FinancialStatementLine, FinancialStatementLine.id == FinancialStatementValue.line_id)
            .where(FinancialStatementLine.meta_id == meta_id)
        )
    else:
        stmt = select(func.count()).select_from(model).where(model.meta_id == meta_id)
    return (await session.execute(stmt)).scalar_one()


async def _owner(session: AsyncSession, path: Path) -> object:
    """Владелец файла с ИНН шапки формы (MATCH), чтобы чистые формы могли быть RECOGNIZED."""
    import json

    data = json.loads(path.read_text("utf-8"))
    inn = next((d["header"].get("inn") for d in data["documents"] if d.get("header", {}).get("inn")), None)
    return await make_legal_entity(session, name="ООО Владелец", inn=inn)


async def test_file_with_forms_only(seeded: AsyncSession, fixtures_dir: Path) -> None:
    path = fixtures_dir / MEZH / "1_Форма1_бух_09.json"
    owner = await _owner(seeded, path)
    documents = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
    assert [d.document_type for d in documents] == [DocumentType.BALANCE]
    balance = documents[0]
    assert (balance.period_year, balance.period_quarter) == (2025, PeriodQuarter.Q3)
    assert balance.reporting_date.isoformat() == "2025-09-30"
    assert balance.osv_account_code is None and balance.is_cumulative is None
    meta = await _meta(seeded, balance)
    lines = await _count(seeded, FinancialStatementLine, meta.id)
    columns = await _count(seeded, FinancialStatementColumn, meta.id)
    assert lines == 37
    assert await _count(seeded, FinancialStatementValue, meta.id) == lines * columns
    file = await seeded.get(StoredFile, balance.stored_file_id)
    assert file.status is FileStatus.PROCESSED


async def test_clean_income_recognized(seeded: AsyncSession, fixtures_dir: Path) -> None:
    path = fixtures_dir / MEZH / "1_Форма2_бух_09.json"
    owner = await _owner(seeded, path)
    (income,) = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
    assert income.document_type is DocumentType.INCOME
    assert income.counterparty_check is DocumentCheckResult.MATCH
    assert income.status is DocumentStatus.RECOGNIZED


async def test_book_with_two_forms(seeded: AsyncSession, fixtures_dir: Path) -> None:
    path = next((fixtures_dir / NPZ).glob("*.json"))
    owner = await _owner(seeded, path)
    documents = await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
    by_type = {d.document_type: d for d in documents}
    assert set(by_type) == {DocumentType.BALANCE, DocumentType.INCOME}
    assert await _count(seeded, FinancialStatementLine, (await _meta(seeded, by_type[DocumentType.BALANCE])).id) == 132
    assert await _count(seeded, FinancialStatementLine, (await _meta(seeded, by_type[DocumentType.INCOME])).id) == 43


async def test_disclosures_merged(seeded: AsyncSession, fixtures_dir: Path) -> None:
    owner = await make_legal_entity(seeded, name="ООО Синтетика")
    documents = await map_golden_snapshot(seeded, entity=owner, path=fixtures_dir / SYNTH / "st06_header_variants.json", uploaded_by=None)
    types = [d.document_type for d in documents]
    assert types.count(DocumentType.DISCLOSURE) == 1 and DocumentType.INCOME in types
    disclosure = next(d for d in documents if d.document_type is DocumentType.DISCLOSURE)
    assert disclosure.status is DocumentStatus.RECOGNIZED and disclosure.locator.count(",") == 4


async def test_form_without_codes(seeded: AsyncSession, fixtures_dir: Path) -> None:
    owner = await make_legal_entity(seeded, name="ООО Без кодов")
    (balance,) = await map_golden_snapshot(seeded, entity=owner, path=fixtures_dir / SYNTH / "st03_no_codes.json", uploaded_by=None)
    assert balance.status is DocumentStatus.NEEDS_CHECK
    meta = await _meta(seeded, balance)
    assert await _count(seeded, FinancialStatementLine, meta.id) == 0
    assert await _count(seeded, FinancialStatementColumn, meta.id) >= 1


async def test_mixed_book_osv_and_balance(seeded: AsyncSession, fixtures_dir: Path) -> None:
    owner = await make_legal_entity(seeded, name="ООО Смешанная")
    documents = await map_golden_snapshot(seeded, entity=owner, path=fixtures_dir / SYNTH / "st04_mixed_book.json", uploaded_by=None)
    assert {d.document_type for d in documents} == {DocumentType.OSV, DocumentType.BALANCE}


async def test_ocr_form_without_report_date(seeded: AsyncSession, fixtures_dir: Path) -> None:
    owner = await make_legal_entity(seeded, name="ООО Дорстроймонтаж")
    documents = await map_golden_snapshot(
        seeded, entity=owner, path=fixtures_dir / OCR_EXPECTED / "aep_6.json", uploaded_by=None, result_builder=ocr_result
    )
    balance = next(d for d in documents if d.document_type is DocumentType.BALANCE)
    assert (balance.period_year, balance.period_quarter) == (2025, PeriodQuarter.Q2)
    meta = await _meta(seeded, balance)
    assert meta.report_date is None and meta.inn is None
    current = (
        await seeded.execute(
            select(FinancialStatementColumn.period_date).where(
                FinancialStatementColumn.meta_id == meta.id, FinancialStatementColumn.is_current.is_(True)
            )
        )
    ).scalar_one()
    assert current.isoformat() == "2025-06-30"
    assert balance.status is DocumentStatus.NEEDS_CHECK


async def test_ocr_noisy_form_mismatch(seeded: AsyncSession, fixtures_dir: Path) -> None:
    owner = await make_legal_entity(seeded, name="ООО Другое", inn="7700000000")
    documents = await map_golden_snapshot(
        seeded, entity=owner, path=fixtures_dir / OCR_EXPECTED / "aep_3.json", uploaded_by=None, result_builder=ocr_result
    )
    balance = next(d for d in documents if d.document_type is DocumentType.BALANCE)
    assert balance.counterparty_check is DocumentCheckResult.MISMATCH and balance.status is DocumentStatus.NEEDS_CHECK


async def test_all_statement_snapshots_map(seeded: AsyncSession, fixtures_dir: Path) -> None:
    """Все golden форм (Excel и OCR, кроме ответов-ошибок) маппятся без исключений; у каждой формы значения на все ячейки."""
    owner = await make_legal_entity(seeded, name="ООО Все формы")
    excel = sorted((fixtures_dir / GOLDEN_STATEMENTS).rglob("*.json"))
    ocr = sorted(p for p in (fixtures_dir / OCR_EXPECTED).glob("*.json"))
    assert len(excel) >= 25
    mapped = 0
    for path in excel:
        await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None)
        mapped += 1
    import json

    for path in ocr:
        if json.loads(path.read_text("utf-8"))["status"] == "error":
            continue
        await map_golden_snapshot(seeded, entity=owner, path=path, uploaded_by=None, result_builder=ocr_result)
        mapped += 1
    assert mapped >= 27
    metas = (await seeded.execute(select(FinancialStatementMeta.id))).scalars().all()
    for meta_id in metas:
        lines = await _count(seeded, FinancialStatementLine, meta_id)
        columns = await _count(seeded, FinancialStatementColumn, meta_id)
        assert await _count(seeded, FinancialStatementValue, meta_id) == lines * columns

В api/tests/test_result_mapping.py найти сценарий с текстом «нет документов ОСВ» (grep -n "нет документов ОСВ"): для результата без документов ожидается "в результате нет документов"; результат только с BALANCE больше не ошибка — если в файле есть тест «результат без ОСВ → ERROR» с формой, переписать его на documents = [].

Step 2: Убедиться, что падают — cd api && .venv/bin/pytest tests/test_mapping_statements.py -q -p no:cacheprovider (нужен make up).
Step 3: Репозиторий — api/src/repositories/statements.py:

"""Таблицы `financial_statement_*` (Срез 3b): вставка формы при маппинге, очистка файла, чтение сумм и флагов значений.
Только SQL, без правил: правила — `domain/statement_mapping.py`, `domain/statement_checks.py`."""

import uuid
from collections.abc import Sequence
from decimal import Decimal

from sqlalchemy import delete, insert, select
from sqlalchemy.ext.asyncio import AsyncSession

from src.domain.statement_mapping import StatementDraft
from src.models.files import Document
from src.models.statements import (
    FinancialStatementColumn,
    FinancialStatementLine,
    FinancialStatementMeta,
    FinancialStatementValue,
)

BATCH = 1000


async def insert_statement(
    session: AsyncSession, *, document_id: uuid.UUID, job_id: uuid.UUID, draft: StatementDraft
) -> FinancialStatementMeta:
    meta = FinancialStatementMeta(
        id=uuid.uuid4(),
        document_id=document_id,
        recognition_job_id=job_id,
        statement_type=draft.statement_type,
        **draft.header,
    )
    session.add(meta)
    await session.flush()
    column_ids: dict[int, uuid.UUID] = {}
    columns = []
    for column_index, label, period_date, is_current in draft.columns:
        column_ids[column_index] = uuid.uuid4()
        columns.append(
            {"id": column_ids[column_index], "meta_id": meta.id, "column_index": column_index, "label": label,
             "period_date": period_date, "is_current": is_current}
        )
    if columns:
        await session.execute(insert(FinancialStatementColumn).values(columns))
    line_ids: dict[str, uuid.UUID] = {}
    lines = []
    for line in draft.lines:
        line_ids[line.line_code] = uuid.uuid4()
        lines.append(
            {"id": line_ids[line.line_code], "meta_id": meta.id, "line_code": line.line_code,
             "line_code_original_value": line.line_code, "line_code_confidence": line.line_code_confidence,
             "line_code_modified": False, "line_name": line.line_name, "line_name_original_value": line.line_name,
             "line_name_confidence": line.line_name_confidence, "line_name_modified": False}
        )
    for start in range(0, len(lines), BATCH):
        await session.execute(insert(FinancialStatementLine).values(lines[start : start + BATCH]))
    values = [
        {"id": uuid.uuid4(), "line_id": line_ids[value.line_code], "column_id": column_ids[value.column_index],
         "amount": value.amount, "string_value": value.string_value, "confidence": value.confidence,
         "is_recognized": value.is_recognized, "is_modified": False}
        for value in draft.values
    ]
    for start in range(0, len(values), BATCH):
        await session.execute(insert(FinancialStatementValue).values(values[start : start + BATCH]))
    return meta


async def delete_file_statements(session: AsyncSession, file_id: uuid.UUID) -> None:
    documents = select(Document.id).where(Document.stored_file_id == file_id).scalar_subquery()
    metas = select(FinancialStatementMeta.id).where(FinancialStatementMeta.document_id.in_(documents)).scalar_subquery()
    lines = select(FinancialStatementLine.id).where(FinancialStatementLine.meta_id.in_(metas)).scalar_subquery()
    await session.execute(delete(FinancialStatementValue).where(FinancialStatementValue.line_id.in_(lines)))
    await session.execute(delete(FinancialStatementLine).where(FinancialStatementLine.meta_id.in_(metas)))
    await session.execute(delete(FinancialStatementColumn).where(FinancialStatementColumn.meta_id.in_(metas)))
    await session.execute(delete(FinancialStatementMeta).where(FinancialStatementMeta.document_id.in_(documents)))


async def statement_amounts(
    session: AsyncSession, meta_id: uuid.UUID
) -> tuple[list[int], list[tuple[str, dict[int, Decimal | None]]]]:
    columns = (
        await session.execute(
            select(FinancialStatementColumn.id, FinancialStatementColumn.column_index)
            .where(FinancialStatementColumn.meta_id == meta_id)
            .order_by(FinancialStatementColumn.column_index)
        )
    ).all()
    index_by_id = {column_id: index for column_id, index in columns}
    rows = (
        await session.execute(
            select(FinancialStatementLine.line_code, FinancialStatementValue.column_id, FinancialStatementValue.amount)
            .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]] = {}
    for code, column_id, amount in rows:
        by_code.setdefault(code or "", {})[index_by_id[column_id]] = amount
    return [index for _, index in columns], list(by_code.items())


async def value_flags(session: AsyncSession, meta_id: uuid.UUID) -> list[tuple[bool, bool | None, Decimal | None]]:
    rows = await session.execute(
        select(FinancialStatementValue.is_modified, FinancialStatementValue.is_recognized, FinancialStatementValue.confidence)
        .join(FinancialStatementLine, FinancialStatementLine.id == FinancialStatementValue.line_id)
        .where(FinancialStatementLine.meta_id == meta_id)
    )
    return [(bool(modified), recognized, confidence) for modified, recognized, confidence in rows]

Примечание к порядку строк для подстрок §7: statement_amounts сортирует по line_code — пятизначные коды идут сразу за своим четырёхзначным родителем (1150, 11501, 11502, 1160), этого достаточно правилу M-9 (_subline_rules).

Step 4: result_mapping.py

NO_DOCUMENTS = "в результате нет документов"; в _load_result и map_file проверка if not result.documents: raise MappingFatal(NO_DOCUMENTS) вместо проверки _osv_documents. Docstring модуля (пункт про MappingFatal) — «нет документов».
_clear_file: перед удалением документов вызвать await statements_repo.delete_file_statements(session, file_id).
В map_file после цикла по листам ОСВ и до session.add_all(documents) — сбор форм; после вставки строк ОСВ — вставка форм. Код ветки (вставить в map_file, statement_docs — [d for d in result.documents if isinstance(d, StatementDocument)]):

    rules = load_check_rules()
    forms: list[tuple[Document, StatementDraft]] = []
    seen_forms: set[DocumentType] = set()
    other_locators: dict[DocumentType, list[str]] = {}
    for doc in statement_docs:
        doc_type = DocumentType(doc.document_type)
        if doc_type in (DocumentType.DISCLOSURE, DocumentType.OTHER):
            other_locators.setdefault(doc_type, []).append(doc.locator)
            continue
        if doc_type in seen_forms:
            warning = f"{doc.locator}: вторая форма {doc_type.value} в файле пропущена (OQ-77)"
            warnings.append(warning)
            first = next(document for document, _ in forms if document.document_type is doc_type)
            first.status = DocumentStatus.NEEDS_CHECK
            continue
        seen_forms.add(doc_type)
        draft = build_statement(doc)
        warnings.extend(draft.warnings)
        check = counterparty_check(doc.header.inn, doc.header.organization_name, owner.inn, owner_names)
        document = Document(
            id=uuid.uuid4(),
            stored_file_id=file.id,
            legal_entity_id=file.legal_entity_id,
            document_type=doc_type,
            osv_account_code=None,
            period_year=draft.period_year,
            period_quarter=draft.period_quarter,
            is_cumulative=None,
            reporting_date=draft.period_date,
            locator=doc.locator,
            status=DocumentStatus.NEEDS_CHECK,  # уточняется после вставки значений (нужны проверки по суммам)
            period_check=DocumentCheckResult.UNKNOWN,
            counterparty_check=check,
            created_at=now,
            updated_at=now,
        )
        forms.append((document, draft))
    for doc_type, locators in other_locators.items():
        documents.append(
            Document(
                id=uuid.uuid4(), stored_file_id=file.id, legal_entity_id=file.legal_entity_id, document_type=doc_type,
                locator=",".join(locators), status=DocumentStatus.RECOGNIZED, period_check=DocumentCheckResult.UNKNOWN,
                created_at=now, updated_at=now,
            )
        )
    documents.extend(document for document, _ in forms)

После session.add_all(documents), flush и вставки строк ОСВ:

    for document, draft in forms:
        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)
        report = run_checks(document.document_type.value, amounts, column_indexes, rules)
        if document.status is not DocumentStatus.NEEDS_CHECK or not any(
            w.startswith(f"{draft.locator}: вторая форма") for w in warnings
        ):
            document.status = statement_status(
                job_done=job_done,
                checks_passed=not report.failed,
                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 statements_repo.value_flags(session, meta.id)),
                has_lines=bool(draft.lines),
                warnings=draft.warnings,
            )

Проще и надёжнее хранить признак «у формы был повтор» в словаре duplicated: set[DocumentType] и передавать в statement_status warnings=[*draft.warnings, *(["повтор формы"] if document.document_type in duplicated else [])] — так и сделать вместо разбора строк предупреждений (код выше — иллюстрация места вставки). job.group_schema по-прежнему строится только из листов ОСВ (sheets). Импорты: StatementDocument из src.contract.result; build_statement, statement_status, values_ok, StatementDraft из src.domain.statement_mapping; run_checks, load_check_rules из src.domain.statement_checks; from src.repositories import statements as statements_repo.

Если в map_file листов ОСВ нет, sheets пуст — цикл ОСВ не выполняется, entries пуст, вставка строк ОСВ пропускается (проверить, что insert(...).values([]) не вызывается).

Step 5: Тесты — cd api && .venv/bin/pytest tests/test_mapping_statements.py tests/test_result_mapping.py tests/test_mapping_golden.py -q -p no:cacheprovider && cd .. && make lint → PASS. test_mapping_golden (ОСВ) не должен измениться: снимок st04_mixed_book в expected_osv_parse не лежит.
Step 6: Commit

git add api/src/repositories/statements.py api/src/services/result_mapping.py api/tests/test_mapping_statements.py api/tests/test_result_mapping.py
git commit -m "feat(statements): маппинг форм ББ/ОФР в map_file — meta, колонки, строки, значения, статус (D-1…D-3, D-6)"

Task 5: API формы — `GET /documents/{id}/statement`

Files:

Create: api/src/schemas/statements.py, api/src/services/statements.py, api/src/api/endpoints/files/statements.py
Modify: api/src/repositories/statements.py (чтение формы, пара для межформенной), api/src/api/endpoints/files/__init__.py (роутер), api/src/api/dependencies/files.py (StatementServiceDep), api/tests/test_openapi_routes.py
Test: api/tests/test_api_statements.py (новый)

Interfaces:

Consumes: run_checks, cross_form_check, load_check_rules, FailedCheck (Task 1); edition_for (Task 2); statement_amounts (Task 4); files_repo.get_document (видимость).
Produces:
- Схемы (src/schemas/statements.py): HeaderFieldOut(value: str | None, original_value: str | None, confidence: Decimal | None = None, verified: bool | None, modified: bool | None); StatementHeaderOut(inn: HeaderFieldOut, report_date: HeaderFieldOut, units: HeaderFieldOut, organization_name, okud, okpo, okopf, okfs, address: str | None); ColumnOut(id, column_index, label, period_date, is_current); ValueOut(id, column_id, column_index, amount: Decimal | None, string_value, confidence, is_recognized, is_modified, original_amount, modified_by, modified_at); LineOut(id, line_code, line_code_confidence, line_name, line_name_confidence, values: list[ValueOut]); FailedCheckOut(column_index: int, relation: str, expected: Decimal, actual: Decimal); AcceptedOut(by: uuid.UUID, at: datetime, comment: str); ChecksOut(failed: list[FailedCheckOut], soft_failed: list[FailedCheckOut], cross_form: FailedCheckOut | None, accepted: AcceptedOut | None); StatementDocumentOut(id, document_type, status, period_year, period_quarter, reporting_date, counterparty_check, locator, file_id); StatementOut(document: StatementDocumentOut, edition: str | None, header: StatementHeaderOut, columns: list[ColumnOut], lines: list[LineOut], checks: ChecksOut).
- StatementService(session, audit: RequestAudit, scope: VisibilityScope): async def get(document_id) -> StatementOut; служебные async def _visible_statement(document_id) -> tuple[Document, FinancialStatementMeta] (404 / 409 DOCUMENT_NOT_STATEMENT), async def build_out(document, meta) -> StatementOut, async def checks_report(document, meta) -> tuple[ChecksReport, FailedCheck | None].
- Коды ошибок (константы модуля): DOCUMENT_NOT_STATEMENT, DOCUMENT_READONLY, UNKNOWN_CELL, DUPLICATE_CELL, NOTHING_TO_ACCEPT, INVALID_INN (используются в Task 6).
- repositories/statements.py: async def get_meta(session, document_id) -> FinancialStatementMeta | None; async def load_form(session, meta_id) -> tuple[list[FinancialStatementColumn], list[tuple[FinancialStatementLine, list[FinancialStatementValue]]]] (строки по line_code, значения по column_index); async def pair_amounts(session, document: Document) -> tuple[tuple[Decimal | None, Decimal | None], Decimal | None] — для межформенной: (1370 на отчётную дату, 1370 на 31.12 прошлого года) из ББ и 2400 текущей колонки ОФР (D-5; пара — документ другого типа формы того же ЮЛ, года и квартала, файл не SUPERSEDED, подтверждённый первым, затем последний по created_at).
Step 1: Падающие тесты — api/tests/test_api_statements.py (часть GET; Task 6 и 7 допишут правку и подтверждение):

"""API формы ББ/ОФР (spec documents-api «Форма ББ/ОФР для экрана проверки», «Правка формы», «Акцепт проверок формы»,
«Подтверждение документа»; statement-checks; design D-5, D-7…D-9, D-14)."""

import json
from pathlib import Path

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from src.models.directories import LegalEntity, RegionalOffice
from src.models.enums import DocumentType
from src.models.files import Document
from tests.conftest import AuthClientFactory
from tests.factories import make_company_group, make_legal_entity
from tests.fixtures_helpers import GOLDEN_STATEMENTS, map_golden_snapshot

pytestmark = [pytest.mark.integration, pytest.mark.fixtures]

KA_RF = "ka_rf@armka.local"
RKA = "rka@armka.local"  # только DOCUMENT_VIEW
ROSTOV = "Ростовский региональный филиал"
MEZH_BALANCE = GOLDEN_STATEMENTS / "mezheninovskaya" / "2025-Q3" / "1_Форма1_бух_09.json"
MEZH_INCOME = GOLDEN_STATEMENTS / "mezheninovskaya" / "2025-Q3" / "1_Форма2_бух_09.json"


async def rostov_entity(session: AsyncSession, path: Path, name: str = "ООО Межениновская") -> LegalEntity:
    office = (await session.execute(select(RegionalOffice).where(RegionalOffice.name == ROSTOV))).scalar_one()
    group = await make_company_group(session, name=f"ГК {name}", office=office)
    data = json.loads(path.read_text("utf-8"))
    inn = next((d["header"].get("inn") for d in data["documents"] if d.get("header", {}).get("inn")), None)
    return await make_legal_entity(session, name=name, inn=inn, group=group)


async def form(session: AsyncSession, fixtures_dir: Path, path: Path = MEZH_BALANCE) -> Document:
    entity = await rostov_entity(session, fixtures_dir / path)
    documents = await map_golden_snapshot(session, entity=entity, path=fixtures_dir / path, uploaded_by=None)
    return next(d for d in documents if d.document_type in (DocumentType.BALANCE, DocumentType.INCOME))


async def test_get_statement(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    response = await client.get(f"/documents/{document.id}/statement")
    assert response.status_code == 200, response.text
    body = response.json()
    assert body["edition"] == "fsbu4"
    assert body["document"]["document_type"] == "BALANCE" and body["document"]["period_quarter"] == "Q3"
    assert len(body["lines"]) == 37
    codes = [line["line_code"] for line in body["lines"]]
    assert codes == sorted(codes)
    columns = len(body["columns"])
    assert all(len(line["values"]) == columns for line in body["lines"])
    assert body["checks"]["failed"] == [] and body["checks"]["accepted"] is None
    assert set(body["header"]["inn"]) == {"value", "original_value", "confidence", "verified", "modified"}


async def test_statement_of_osv_is_conflict(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    from tests.factories import make_document, make_stored_file
    from src.models.enums import FileStatus, RecognitionStatus

    entity = await rostov_entity(seeded, fixtures_dir / MEZH_BALANCE, name="ООО ОСВ")
    file, job = await make_stored_file(seeded, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done)
    osv = await make_document(seeded, file=file, job=job)
    client = await auth_client(KA_RF)
    response = await client.get(f"/documents/{osv.id}/statement")
    assert response.status_code == 409 and response.json()["detail"]["code"] == "DOCUMENT_NOT_STATEMENT"


async def test_statement_hidden_is_404(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    document = await form(seeded, fixtures_dir)
    client = await auth_client("client_manager@armka.local")  # другой филиал
    assert (await client.get(f"/documents/{document.id}/statement")).status_code == 404


async def test_cross_form_check(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    entity = await rostov_entity(seeded, fixtures_dir / MEZH_BALANCE)
    balance = next(
        d for d in await map_golden_snapshot(seeded, entity=entity, path=fixtures_dir / MEZH_BALANCE, uploaded_by=None)
    )
    await map_golden_snapshot(seeded, entity=entity, path=fixtures_dir / MEZH_INCOME, uploaded_by=None)
    client = await auth_client(KA_RF)
    body = (await client.get(f"/documents/{balance.id}/statement")).json()
    cross = body["checks"]["cross_form"]
    assert cross is None or cross["relation"] == "1370 − 1370(начало года) = 2400"

Доменные ошибки отдаются как {"detail": {"code", "message", …}} (CLAUDE.md ядра, «Ошибки»): код — response.json()["detail"]["code"].

Step 2: Убедиться, что падают — cd api && .venv/bin/pytest tests/test_api_statements.py -q -p no:cacheprovider.
Step 3: Репозиторий — дописать в src/repositories/statements.py:

async def get_meta(session: AsyncSession, document_id: uuid.UUID) -> FinancialStatementMeta | None:
    stmt = select(FinancialStatementMeta).where(FinancialStatementMeta.document_id == document_id)
    return (await session.execute(stmt)).scalar_one_or_none()


async def load_form(
    session: AsyncSession, meta_id: uuid.UUID
) -> tuple[list[FinancialStatementColumn], list[tuple[FinancialStatementLine, list[FinancialStatementValue]]]]:
    columns = list(
        (
            await session.execute(
                select(FinancialStatementColumn)
                .where(FinancialStatementColumn.meta_id == meta_id)
                .order_by(FinancialStatementColumn.column_index)
            )
        ).scalars()
    )
    order = {column.id: column.column_index for column in columns}
    lines = list(
        (
            await session.execute(
                select(FinancialStatementLine)
                .where(FinancialStatementLine.meta_id == meta_id)
                .order_by(FinancialStatementLine.line_code)
            )
        ).scalars()
    )
    values = (
        await session.execute(
            select(FinancialStatementValue).where(FinancialStatementValue.line_id.in_([line.id for line in lines]))
        )
    ).scalars()
    by_line: dict[uuid.UUID, list[FinancialStatementValue]] = {line.id: [] for line in lines}
    for value in values:
        by_line[value.line_id].append(value)
    for items in by_line.values():
        items.sort(key=lambda value: order[value.column_id])
    return columns, [(line, by_line[line.id]) for line in lines]

pair_amounts — реализация по Interfaces: найти пару запросом select(Document).join(StoredFile).where(Document.legal_entity_id == document.legal_entity_id, Document.period_year == document.period_year, Document.period_quarter == document.period_quarter, Document.document_type == <другой тип>, StoredFile.status != FileStatus.SUPERSEDED) .order_by((Document.status == DocumentStatus.CONFIRMED).desc(), Document.created_at.desc()).limit(1); затем у ББ взять значения строки 1370 в колонке is_current и в колонке с period_date = date(year − 1, 12, 31); у ОФР — строку 2400 в колонке is_current. Нет пары, строки или колонки — соответствующий None. Если сам document — ОФР, роли меняются.

Step 4: Схемы, сервис, маршрут

src/services/statements.py — каркас (методы правки и акцепта добавит Task 6):

"""Форма ББ/ОФР для экрана «Проверка распознавания» (spec documents-api; design D-5, D-7, D-8, D-14)."""

import uuid

from sqlalchemy.ext.asyncio import AsyncSession

from src.api.dependencies.audit import RequestAudit
from src.common.exceptions import ConflictError, NotFoundError
from src.domain.policy import VisibilityScope
from src.domain.statement_checks import ChecksReport, FailedCheck, cross_form_check, load_check_rules, run_checks
from src.domain.statement_mapping import edition_for
from src.models.enums import DocumentType
from src.models.files import Document
from src.models.statements import FinancialStatementMeta
from src.repositories import files as files_repo
from src.repositories import statements as repo
from src.schemas import statements as schemas

DOCUMENT_NOT_FOUND = "документ не найден"
DOCUMENT_NOT_STATEMENT = "DOCUMENT_NOT_STATEMENT"
DOCUMENT_READONLY = "DOCUMENT_READONLY"
UNKNOWN_CELL = "UNKNOWN_CELL"
DUPLICATE_CELL = "DUPLICATE_CELL"
NOTHING_TO_ACCEPT = "NOTHING_TO_ACCEPT"
INVALID_INN = "INVALID_INN"
FORM_TYPES = frozenset({DocumentType.BALANCE, DocumentType.INCOME})


class StatementService:
    def __init__(self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope) -> None:
        self._session = session
        self._audit = audit
        self._scope = scope

    async def _visible_statement(self, document_id: uuid.UUID) -> tuple[Document, FinancialStatementMeta]:
        document = await files_repo.get_document(self._session, self._scope, document_id)
        if document is None:
            raise NotFoundError(DOCUMENT_NOT_FOUND)
        meta = await repo.get_meta(self._session, document.id) if document.document_type in FORM_TYPES else None
        if meta is None:
            raise ConflictError(DOCUMENT_NOT_STATEMENT, "документ — не форма ББ/ОФР")
        return document, meta

    async def checks_report(self, document: Document, meta: FinancialStatementMeta) -> tuple[ChecksReport, FailedCheck | None]:
        rules = load_check_rules()
        column_indexes, amounts = await repo.statement_amounts(self._session, meta.id)
        report = run_checks(document.document_type.value, amounts, column_indexes, rules)
        balance_1370, income_2400 = await repo.pair_amounts(self._session, document)
        return report, cross_form_check(balance_1370, income_2400, rules.tolerance)

    async def get(self, document_id: uuid.UUID) -> schemas.StatementOut:
        document, meta = await self._visible_statement(document_id)
        return await self.build_out(document, meta)

    async def build_out(self, document: Document, meta: FinancialStatementMeta) -> schemas.StatementOut:
        columns, lines = await repo.load_form(self._session, meta.id)
        report, cross = await self.checks_report(document, meta)
        return schemas.statement_out(
            document, meta, columns, lines, report, cross, edition=edition_for(document.reporting_date)
        )

src/schemas/statements.py — модели из Interfaces (BaseModel, ConfigDict(from_attributes=True) где строятся из ORM) и функция statement_out(document, meta, columns, lines, report, cross, *, edition) -> StatementOut, собирающая ответ: header.inn = HeaderFieldOut(value=meta.inn, original_value=meta.inn_original_value, confidence=meta.inn_confidence, verified=meta.inn_verified, modified=meta.inn_modified), header.report_date — value = meta.report_date.isoformat() или None, original_value = meta.report_date_original_value; header.units аналогично; values[].column_index — по column_id; checks.accepted — AcceptedOut, если meta.checks_accepted_at не пуст. Суммы — Decimal (в JSON строкой, как EntryOut).

src/api/endpoints/files/statements.py — router = APIRouter(prefix="/documents"), маршрут GET /{document_id}/statement (response_model=StatementOut, dependencies=[Depends(require("DOCUMENT_VIEW"))], responses 404/409), вызывающий service.get. Подключить роутер в src/api/endpoints/files/__init__.py рядом с documents.router; StatementServiceDep в src/api/dependencies/files.py — как DocumentServiceDep. В tests/test_openapi_routes.py добавить "/api/v1/documents/{document_id}/statement": {"get", "patch"} и "/api/v1/documents/{document_id}/statement/checks/accept": {"post"} (PATCH и accept появятся в Task 6 — до тех пор временно {"get"} и без accept; Task 6 дополняет).

Step 5: Тесты — cd api && .venv/bin/pytest tests/test_api_statements.py tests/test_openapi_routes.py -q -p no:cacheprovider && cd .. && make lint → PASS.
Step 6: Commit

git add api/src/schemas/statements.py api/src/services/statements.py api/src/api/endpoints/files api/src/api/dependencies/files.py api/src/repositories/statements.py api/tests/test_api_statements.py api/tests/test_openapi_routes.py
git commit -m "feat(statements): GET /documents/{id}/statement — шапка, колонки, строки, проверки §7, редакция (D-5, D-14)"

Task 6: Правка формы и акцепт проверок

Files:

Modify: api/src/services/statements.py (patch, accept_checks, recompute_status), api/src/schemas/statements.py (StatementPatch, HeaderPatch, CellPatch, ChecksAcceptIn), api/src/api/endpoints/files/statements.py (PATCH, POST accept), api/src/repositories/statements.py (lock_cells), api/tests/test_openapi_routes.py
Test: api/tests/test_api_statements.py (дописать)

Interfaces:

Consumes: StatementService._visible_statement, build_out, checks_report (Task 5); files_repo.lock_file, files_repo.lock_document; statement_status, values_ok, quarter_of (Task 2); counterparty_check (Task 2); CounterpartyIndex.owner_names (существующий, repositories/counterparties.py); RequestAudit.record.
Produces:
- Схемы: CellPatch(line_id: uuid.UUID, column_id: uuid.UUID, amount: Decimal | None = Field(max_digits=18, decimal_places=2)); HeaderPatch(inn: str | None = None, report_date: date | None = None, units: str | None = Field(None, min_length=1, max_length=50)) — поле отсутствует ≠ null (model_fields_set); StatementPatch(header: HeaderPatch | None = None, values: list[CellPatch] = []); ChecksAcceptIn(comment: str = Field(min_length=1, max_length=2000)) (после strip).
- StatementService.patch(document_id, body: StatementPatch) -> StatementOut; StatementService.accept_checks(document_id, body: ChecksAcceptIn) -> StatementOut; StatementService.recompute_status(document, meta) -> None (D-6 с учётом акцепта; CONFIRMED не трогает); StatementService.hard_failures(document, meta) -> tuple[FailedCheck, ...] (жёсткие нарушения, пустой кортеж при акцепте) — используется Task 7.
- repo.lock_cells(session, meta_id, cells: Sequence[tuple[uuid.UUID, uuid.UUID]]) -> dict[tuple[uuid.UUID, uuid.UUID], FinancialStatementValue] — значения этого meta_id по (line_id, column_id); чужие ячейки в словарь не попадают.
Step 1: Падающие тесты — дописать в api/tests/test_api_statements.py:

from decimal import Decimal

from src.models.audit_ex import AuditEvent
from src.models.enums import AuditAction, DocumentStatus
from src.models.statements import FinancialStatementMeta, FinancialStatementValue


async def _first_cell(client, document_id) -> tuple[dict, dict]:
    body = (await client.get(f"/documents/{document_id}/statement")).json()
    line = next(line for line in body["lines"] if line["values"][0]["amount"] is not None)
    return line, line["values"][0]


async def test_patch_value_and_revert(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    line, cell = await _first_cell(client, document.id)
    original = cell["amount"]
    new = str(Decimal(original) + 20)
    response = await client.patch(
        f"/documents/{document.id}/statement",
        json={"values": [{"line_id": line["id"], "column_id": cell["column_id"], "amount": new}]},
    )
    assert response.status_code == 200, response.text
    edited = next(l for l in response.json()["lines"] if l["id"] == line["id"])["values"][0]
    assert Decimal(edited["amount"]) == Decimal(new) and Decimal(edited["original_amount"]) == Decimal(original)
    assert edited["is_modified"] is True and edited["modified_by"] is not None
    event = (
        await seeded.execute(
            select(AuditEvent).where(AuditEvent.object_id == document.id, AuditEvent.action == AuditAction.RECOGNIZED_DATA_EDIT)
        )
    ).scalars().one()
    assert event.before_value and event.after_value
    response = await client.patch(
        f"/documents/{document.id}/statement",
        json={"values": [{"line_id": line["id"], "column_id": cell["column_id"], "amount": original}]},
    )
    reverted = next(l for l in response.json()["lines"] if l["id"] == line["id"])["values"][0]
    assert reverted["is_modified"] is False and Decimal(reverted["original_amount"]) == Decimal(original)


async def test_patch_report_date_recomputes_period(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    response = await client.patch(f"/documents/{document.id}/statement", json={"header": {"report_date": "2025-06-30"}})
    body = response.json()
    assert body["document"]["period_quarter"] == "Q2" and body["header"]["report_date"]["modified"] is True
    assert body["header"]["report_date"]["original_value"] is not None  # исходник не тронут


async def test_patch_inn_validates_and_rechecks(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    bad = await client.patch(f"/documents/{document.id}/statement", json={"header": {"inn": "12345"}})
    assert bad.status_code == 422 and bad.json()["detail"]["code"] == "INVALID_INN"
    other = await client.patch(f"/documents/{document.id}/statement", json={"header": {"inn": "7700000000"}})
    assert other.json()["document"]["counterparty_check"] == "MISMATCH"
    assert other.json()["document"]["status"] == "NEEDS_CHECK"


async def test_patch_duplicate_cell_rejected(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    """Review Focus 1."""
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    line, cell = await _first_cell(client, document.id)
    item = {"line_id": line["id"], "column_id": cell["column_id"], "amount": "1"}
    response = await client.patch(f"/documents/{document.id}/statement", json={"values": [item, {**item, "amount": "2"}]})
    assert response.status_code == 422 and response.json()["detail"]["code"] == "DUPLICATE_CELL"


async def test_patch_amount_precision_rejected(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    """Review Focus 2: больше двух знаков или больше 16 цифр до запятой — 422 схемы, данные не меняются."""
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    line, cell = await _first_cell(client, document.id)
    for amount in ("1.234", "12345678901234567"):
        response = await client.patch(
            f"/documents/{document.id}/statement",
            json={"values": [{"line_id": line["id"], "column_id": cell["column_id"], "amount": amount}]},
        )
        assert response.status_code == 422


async def test_patch_unknown_cell(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    import uuid

    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    response = await client.patch(
        f"/documents/{document.id}/statement",
        json={"values": [{"line_id": str(uuid.uuid4()), "column_id": str(uuid.uuid4()), "amount": "1"}]},
    )
    assert response.status_code == 422 and response.json()["detail"]["code"] == "UNKNOWN_CELL"


async def test_patch_requires_edit_right(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    document = await form(seeded, fixtures_dir)
    client = await auth_client(RKA)
    assert (await client.patch(f"/documents/{document.id}/statement", json={"header": {"units": "руб."}})).status_code == 403


async def test_empty_patch_no_audit(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    assert (await client.patch(f"/documents/{document.id}/statement", json={})).status_code == 200
    count = (await seeded.execute(select(AuditEvent).where(AuditEvent.object_id == document.id))).scalars().all()
    assert count == []


async def _break_balance(seeded: AsyncSession, client, document_id) -> None:
    body = (await client.get(f"/documents/{document_id}/statement")).json()
    line_1600 = next(line for line in body["lines"] if line["line_code"] == "1600")
    cell = line_1600["values"][0]
    await client.patch(
        f"/documents/{document_id}/statement",
        json={"values": [{"line_id": line_1600["id"], "column_id": cell["column_id"], "amount": str(Decimal(cell["amount"] or 0) + 1000)}]},
    )


async def test_accept_and_reset(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    nothing = await client.post(f"/documents/{document.id}/statement/checks/accept", json={"comment": "ок"})
    assert nothing.status_code == 409 and nothing.json()["detail"]["code"] == "NOTHING_TO_ACCEPT"
    await _break_balance(seeded, client, document.id)
    broken = (await client.get(f"/documents/{document.id}/statement")).json()
    assert any(item["relation"].startswith("1600") for item in broken["checks"]["failed"])
    assert broken["document"]["status"] == "NEEDS_CHECK"
    accepted = await client.post(
        f"/documents/{document.id}/statement/checks/accept", json={"comment": "округление в форме клиента"}
    )
    assert accepted.status_code == 200
    assert accepted.json()["checks"]["accepted"]["comment"] == "округление в форме клиента"
    await client.patch(f"/documents/{document.id}/statement", json={"header": {"units": "руб."}})
    after = (await client.get(f"/documents/{document.id}/statement")).json()
    assert after["checks"]["accepted"] is None and after["document"]["status"] == "NEEDS_CHECK"


async def test_patch_confirmed_readonly(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    assert (await client.post(f"/documents/{document.id}/confirm")).status_code == 200
    response = await client.patch(f"/documents/{document.id}/statement", json={"header": {"units": "руб."}})
    assert response.status_code == 409 and response.json()["detail"]["code"] == "DOCUMENT_READONLY"

Step 2: Убедиться, что падают — cd api && .venv/bin/pytest tests/test_api_statements.py -q -p no:cacheprovider (новые тесты — 405/404 до реализации).
Step 3: Схемы — в src/schemas/statements.py:

class CellPatch(BaseModel):
    model_config = ConfigDict(extra="forbid")
    line_id: uuid.UUID
    column_id: uuid.UUID
    amount: Decimal | None = Field(max_digits=18, decimal_places=2)


class HeaderPatch(BaseModel):
    """Отсутствующее поле не меняется; `null` у `report_date`/`inn` — очистить значение."""

    model_config = ConfigDict(extra="forbid")
    inn: str | None = None
    report_date: datetime.date | None = None
    units: str | None = Field(None, min_length=1, max_length=50)


class StatementPatch(BaseModel):
    model_config = ConfigDict(extra="forbid")
    header: HeaderPatch | None = None
    values: list[CellPatch] = Field(default_factory=list, max_length=5000)


class ChecksAcceptIn(BaseModel):
    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
    comment: str = Field(min_length=1, max_length=2000)

Step 4: Сервис — в StatementService (src/services/statements.py):

    async def _editable(self, document_id: uuid.UUID) -> tuple[Document, FinancialStatementMeta]:
        """Видимость → блокировки «файл → документ» → проверки D-7 (порядок 409 как у подтверждения)."""
        visible, _ = await self._visible_statement(document_id)
        file = await files_repo.lock_file(self._session, visible.stored_file_id)
        document = await files_repo.lock_document(self._session, document_id)
        if file is None or document is None:
            raise NotFoundError(DOCUMENT_NOT_FOUND)
        if file.status is FileStatus.SUPERSEDED:
            raise ConflictError(FILE_SUPERSEDED, "файл документа заменён новой версией")
        if document.status is DocumentStatus.CONFIRMED:
            raise ConflictError(DOCUMENT_READONLY, "подтверждённая форма только для чтения: загрузите исправленный файл")
        if document.status not in CONFIRMABLE_STATUSES:
            raise ConflictError(DOCUMENT_NOT_CONFIRMABLE, f"форма в статусе {document.status.value} не правится")
        meta = await repo.get_meta(self._session, document.id)
        if meta is None:
            raise ConflictError(DOCUMENT_NOT_STATEMENT, "документ — не форма ББ/ОФР")
        return document, meta

    async def hard_failures(self, document: Document, meta: FinancialStatementMeta) -> tuple[FailedCheck, ...]:
        if meta.checks_accepted_at is not None:
            return ()
        report, _ = await self.checks_report(document, meta)
        return report.failed

    async def recompute_status(self, document: Document, meta: FinancialStatementMeta) -> None:
        """D-6 после правки или акцепта; `CONFIRMED` не трогает. Предупреждения маппинга файла не учитываются."""
        if document.status is DocumentStatus.CONFIRMED:
            return
        job = await self._session.get(RecognitionJob, meta.recognition_job_id)
        has_lines = bool((await repo.statement_amounts(self._session, meta.id))[1])
        document.status = statement_status(
            job_done=job is not None and job.status is RecognitionStatus.done,
            checks_passed=not await self.hard_failures(document, meta),
            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=has_lines,
            warnings=(),
        )
        document.updated_at = utc_now()

    async def patch(self, document_id: uuid.UUID, body: StatementPatch) -> StatementOut:
        document, meta = await self._editable(document_id)
        keys = [(cell.line_id, cell.column_id) for cell in body.values]
        if len(keys) != len(set(keys)):
            raise DomainValidationError(DUPLICATE_CELL, "ячейка указана в запросе дважды")
        cells = await repo.lock_cells(self._session, meta.id, keys)
        if len(cells) != len(keys):
            raise DomainValidationError(UNKNOWN_CELL, "ячейка не принадлежит форме")
        before: dict[str, object] = {}
        after: dict[str, object] = {}
        now = utc_now()
        user_id = self._audit.user_id
        for cell in body.values:
            value = cells[(cell.line_id, cell.column_id)]
            if value.amount == cell.amount:
                continue
            key = f"value:{value.id}"
            before[key], after[key] = value.amount, cell.amount
            if not value.is_modified and value.original_amount is None:
                value.original_amount = value.amount
            value.amount = cell.amount
            value.is_modified = value.amount != value.original_amount
            value.modified_by, value.modified_at = user_id, now
        header = body.header
        if header is not None:
            fields = header.model_fields_set
            if "inn" in fields and header.inn != meta.inn:
                if header.inn is not None and not (header.inn.isdigit() and len(header.inn) in (10, 12)):
                    raise DomainValidationError(INVALID_INN, "ИНН — 10 или 12 цифр")
                before["inn"], after["inn"] = meta.inn, header.inn
                meta.inn, meta.inn_modified = header.inn, True
                owner = await self._session.get(LegalEntity, document.legal_entity_id)
                index = await CounterpartyIndex.load(self._session)
                document.counterparty_check = counterparty_check(
                    meta.inn, meta.organization_name, owner.inn, index.owner_names(owner)
                )
            if "report_date" in fields and header.report_date != meta.report_date:
                before["report_date"], after["report_date"] = meta.report_date, header.report_date
                meta.report_date, meta.report_date_modified = header.report_date, True
                columns, _ = await repo.load_form(self._session, meta.id)
                period = statement_period(meta.report_date, [column.period_date for column in columns])
                document.reporting_date = period
                document.period_year = period.year if period is not None else None
                document.period_quarter = quarter_of(period) if period is not None else None
            if "units" in fields and header.units is not None and header.units != meta.units:
                before["units"], after["units"] = meta.units, header.units
                meta.units, meta.units_modified = header.units, True
        if not after:
            return await self.build_out(document, meta)
        if meta.checks_accepted_at is not None:
            before["checks_accepted"] = {"by": meta.checks_accepted_by, "at": meta.checks_accepted_at}
            after["checks_accepted"] = None
            meta.checks_accepted_by = meta.checks_accepted_at = meta.checks_accepted_comment = None
        await self._session.flush()
        await self.recompute_status(document, meta)
        await self._audit.record(
            action=AuditAction.RECOGNIZED_DATA_EDIT,
            section=AuditSection.RECOGNITION,
            object_type=OBJECT_TYPE,
            object_id=document.id,
            before_value=before,
            after_value=after,
        )
        return await self.build_out(document, meta)

    async def accept_checks(self, document_id: uuid.UUID, body: ChecksAcceptIn) -> StatementOut:
        document, meta = await self._editable(document_id)
        if not await self.hard_failures(document, meta):
            raise ConflictError(NOTHING_TO_ACCEPT, "непройденных жёстких проверок нет")
        now = utc_now()
        meta.checks_accepted_by, meta.checks_accepted_at, meta.checks_accepted_comment = self._audit.user_id, now, body.comment
        await self._session.flush()
        await self.recompute_status(document, meta)
        await self._audit.record(
            action=AuditAction.RECOGNIZED_DATA_EDIT,
            section=AuditSection.RECOGNITION,
            object_type=OBJECT_TYPE,
            object_id=document.id,
            before_value={"checks_accepted": None},
            after_value={"checks_accepted": {"by": self._audit.user_id, "at": now, "comment": body.comment}},
        )
        return await self.build_out(document, meta)

Импорты и константы: OBJECT_TYPE = "document"; FILE_SUPERSEDED, DOCUMENT_NOT_CONFIRMABLE, CONFIRMABLE_STATUSES — переиспользовать из src.services.documents (не дублировать строки); DomainValidationError из src.common.exceptions; utc_now из src.common.clock; CounterpartyIndex из src.repositories.counterparties; LegalEntity, RecognitionJob, модели и enum'ы. Если CounterpartyIndex.owner_names требует загрузки всего индекса и это дорого — допустимо собрать имена владельца напрямую (legal_entity.name + его алиасы через normalize_alias), повторив логику owner_names.

repo.lock_cells:

async def lock_cells(
    session: AsyncSession, meta_id: uuid.UUID, cells: Sequence[tuple[uuid.UUID, uuid.UUID]]
) -> dict[tuple[uuid.UUID, uuid.UUID], FinancialStatementValue]:
    if not cells:
        return {}
    line_ids = {line_id for line_id, _ in cells}
    stmt = (
        select(FinancialStatementValue)
        .join(FinancialStatementLine, FinancialStatementLine.id == FinancialStatementValue.line_id)
        .where(FinancialStatementLine.meta_id == meta_id, FinancialStatementValue.line_id.in_(line_ids))
        .with_for_update(of=FinancialStatementValue)
    )
    wanted = set(cells)
    return {
        (value.line_id, value.column_id): value
        for value in (await session.execute(stmt)).scalars()
        if (value.line_id, value.column_id) in wanted
    }

Step 5: Маршруты — в src/api/endpoints/files/statements.py: PATCH /{document_id}/statement (DOCUMENT_EDIT, response_model=StatementOut, responses 404/409/422) → service.patch; POST /{document_id}/statement/checks/accept (DOCUMENT_EDIT) → service.accept_checks. В tests/test_openapi_routes.py: "/api/v1/documents/{document_id}/statement": {"get", "patch"}, "/api/v1/documents/{document_id}/statement/checks/accept": {"post"}.
Step 6: Тесты и линт — cd api && .venv/bin/pytest tests/test_api_statements.py tests/test_openapi_routes.py -q -p no:cacheprovider && cd .. && make lint → PASS (кроме test_patch_confirmed_readonly, если подтверждение формы ещё не проверяет период — тест проходит: у формы Межениновской период есть).
Step 7: Commit

git add api/src/services/statements.py api/src/schemas/statements.py api/src/api/endpoints/files/statements.py api/src/repositories/statements.py api/tests/test_api_statements.py api/tests/test_openapi_routes.py
git commit -m "feat(statements): правка формы (значения, ИНН, дата, единицы) и акцепт проверок — аудит, пересчёт статуса (D-7, D-8)"

Task 7: Подтверждение формы и сквозной сценарий чек-листа

Files:

Modify: api/src/services/documents.py (confirm), api/src/api/endpoints/files/documents.py (описание 409)
Test: api/tests/test_api_statements.py (дописать), api/tests/test_consolidation_documents.py (форма двигает версию)

Interfaces:

Consumes: StatementService.hard_failures (Task 6), repo.get_meta (Task 5).
Produces: коды PERIOD_REQUIRED, CHECKS_NOT_PASSED (константы в services/documents.py); ConflictError(CHECKS_NOT_PASSED, …, failures=[{column_index, relation, expected, actual}]).
Step 1: Падающие тесты — дописать в api/tests/test_api_statements.py:

async def test_confirm_requires_checks(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    await _break_balance(seeded, client, document.id)
    blocked = await client.post(f"/documents/{document.id}/confirm")
    assert blocked.status_code == 409 and blocked.json()["detail"]["code"] == "CHECKS_NOT_PASSED"
    assert blocked.json()["detail"]["failures"]
    await client.post(f"/documents/{document.id}/statement/checks/accept", json={"comment": "округление"})
    confirmed = await client.post(f"/documents/{document.id}/confirm")
    assert confirmed.status_code == 200 and confirmed.json()["status"] == "CONFIRMED"
    meta = (await seeded.execute(select(FinancialStatementMeta).where(FinancialStatementMeta.document_id == document.id))).scalar_one()
    assert meta.reviewed_by is not None and meta.reviewed_at is not None


async def test_confirm_after_non_quarter_date_edit(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    """Review Focus 5: дата не конец квартала → квартал NULL, NEEDS_CHECK, PERIOD_REQUIRED; исправление даты снимает."""
    document = await form(seeded, fixtures_dir)
    client = await auth_client(KA_RF)
    body = (await client.patch(f"/documents/{document.id}/statement", json={"header": {"report_date": "2025-10-31"}})).json()
    assert body["document"]["period_quarter"] is None and body["document"]["status"] == "NEEDS_CHECK"
    blocked = await client.post(f"/documents/{document.id}/confirm")
    assert blocked.status_code == 409 and blocked.json()["detail"]["code"] == "PERIOD_REQUIRED"
    await client.patch(f"/documents/{document.id}/statement", json={"header": {"report_date": "2025-09-30"}})
    assert (await client.post(f"/documents/{document.id}/confirm")).status_code == 200


async def test_confirmed_form_in_checklist(seeded: AsyncSession, fixtures_dir: Path, auth_client: AuthClientFactory) -> None:
    from tests.factories import make_version

    document = await form(seeded, fixtures_dir)
    entity = await seeded.get(LegalEntity, document.legal_entity_id)
    version = await make_version(seeded, entities=[entity], year=2025, quarter="Q3")
    client = await auth_client(KA_RF)
    assert (await client.post(f"/documents/{document.id}/confirm")).status_code == 200
    checklist = (await client.get(f"/consolidations/versions/{version.id}/checklist")).json()
    balance_index = next(i for i, rule in enumerate(checklist["rules"]) if rule["document_type"] == "BALANCE")
    row = next(row for row in checklist["rows"] if row["legal_entity_id"] == str(entity.id))
    assert row["cells"][balance_index]["status"] == "CONFIRMED"

Сигнатуру make_version проверить в tests/factories.py (sed -n '/^async def make_version/,/^ \"\"\"/p') и подставить реальные параметры (периметр, год, квартал, статус INITIATED). В tests/test_consolidation_documents.py добавить сценарий спеки «Форма двигает версию»: версия в DOCUMENTS_PROCESSING, map_golden_snapshot формы mezheninovskaya/…/1_Форма1_бух_09.json с uploaded_by = пользователь (документ NEEDS_CHECK — ИНН владельца отличается), затем вызов on_documents_mapped как в существующем тесте для ОСВ → версия DATA_CORRECTION, привязки нет.

Step 2: Убедиться, что падают — cd api && .venv/bin/pytest tests/test_api_statements.py -q -p no:cacheprovider -k "confirm".
Step 3: Реализация — в DocumentService.confirm после проверки CONFIRMABLE_STATUSES и до смены статуса:

        if document.document_type in FORM_TYPES:
            meta = await statements_repo.get_meta(self._session, document.id)
            if meta is not None:
                if document.period_year is None or document.period_quarter is None:
                    raise ConflictError(PERIOD_REQUIRED, "у формы не определён период: поправьте дату отчёта")
                failures = await StatementService(self._session, self._audit, self._scope).hard_failures(document, meta)
                if failures:
                    raise ConflictError(
                        CHECKS_NOT_PASSED,
                        "не пройдены логические проверки формы: исправьте значения или акцептуйте с комментарием",
                        failures=[
                            {"column_index": f.column_index, "relation": f.relation, "expected": str(f.expected), "actual": str(f.actual)}
                            for f in failures
                        ],
                    )

и после установки confirmed_by/at — meta.reviewed_by, meta.reviewed_at = self._audit.user_id, now (если meta). Импорт StatementService внутри метода (from src.services.statements import StatementService # noqa: PLC0415) — services/statements.py импортирует константы из services/documents.py (цикл). FORM_TYPES — из services.statements. Описание 409 маршрута confirm дополнить кодами.

Step 4: Тесты и линт — cd api && .venv/bin/pytest tests/test_api_statements.py tests/test_api_documents.py tests/test_consolidation_documents.py -q -p no:cacheprovider && cd .. && make lint → PASS.
Step 5: Commit

git add api/src/services/documents.py api/src/api/endpoints/files/documents.py api/tests/test_api_statements.py api/tests/test_consolidation_documents.py
git commit -m "feat(statements): подтверждение формы — период и проверки §7, reviewed_by; форма в чек-листе и в переходе версии (D-9)"

Task 8: Комментарии к документу

Files:

Create: api/src/api/endpoints/files/comments.py, api/src/schemas/comments.py
Modify: api/src/services/documents.py (comments, add_comment, comments_count в file_out/document_out), api/src/repositories/files.py (comments_count_by_document), api/src/schemas/files.py (DocumentBrief.comments_count), api/src/schemas/documents.py (DocumentOut.comments_count), api/src/services/primary_files.py (передача счётчиков в file_out), api/src/api/endpoints/files/__init__.py, api/tests/test_openapi_routes.py
Test: api/tests/test_api_comments.py

Interfaces:

Produces: CommentIn(text: str = Field(min_length=1, max_length=4000)) со str_strip_whitespace; CommentAuthorOut(id, full_name); CommentOut(id, text, author: CommentAuthorOut, created_at); DocumentService.add_comment(document_id, body) -> CommentOut, DocumentService.comments(document_id) -> list[CommentOut]; files_repo.comments_count_by_document(session, document_ids) -> dict[uuid.UUID, int]; COMMENT_OBJECT_TYPE = "document".
Step 1: Падающие тесты — api/tests/test_api_comments.py:

"""Комментарии к документу (spec documents-api «Комментарии к документу», «Файлы организации и карточка файла»; D-11)."""

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from src.models.enums import DocumentStatus, FileStatus, RecognitionStatus
from tests.conftest import AuthClientFactory
from tests.factories import make_document, make_stored_file
from tests.test_api_documents import KA_RF, RKA, _rostov_entity

pytestmark = pytest.mark.integration


async def _doc(session: AsyncSession, status: DocumentStatus = DocumentStatus.CONFIRMED):
    entity = await _rostov_entity(session, name="ООО Комментарии")
    file, job = await make_stored_file(session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done)
    return file, await make_document(session, file=file, job=job, status=status)


async def test_add_and_list(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    file, document = await _doc(seeded)
    client = await auth_client(KA_RF)
    created = await client.post(f"/documents/{document.id}/comments", json={"text": "  нет подписи руководителя  "})
    assert created.status_code == 201, created.text
    assert created.json()["text"] == "нет подписи руководителя"
    assert created.json()["author"]["full_name"]
    await client.post(f"/documents/{document.id}/comments", json={"text": "второй"})
    listed = (await client.get(f"/documents/{document.id}/comments")).json()
    assert [item["text"] for item in listed] == ["нет подписи руководителя", "второй"]
    card = (await client.get(f"/files/{file.id}")).json()
    assert card["documents"][0]["comments_count"] == 2
    assert (await client.get(f"/documents/{document.id}")).json()["comments_count"] == 2


async def test_validation_and_rights(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    _, document = await _doc(seeded)
    client = await auth_client(KA_RF)
    assert (await client.post(f"/documents/{document.id}/comments", json={"text": "   "})).status_code == 422
    assert (await client.post(f"/documents/{document.id}/comments", json={"text": "x" * 4001})).status_code == 422
    reader = await auth_client(RKA)
    assert (await reader.post(f"/documents/{document.id}/comments", json={"text": "можно?"})).status_code == 403
    assert (await reader.get(f"/documents/{document.id}/comments")).status_code == 200
    stranger = await auth_client("client_manager@armka.local")
    assert (await stranger.get(f"/documents/{document.id}/comments")).status_code == 404

Step 2: Убедиться, что падают.
Step 3: Реализация. Модель Comment — src/models/audit_ex.py (поля object_type, object_id, text, author_id, created_at). add_comment: видимость _visible → Comment(object_type="document", object_id=document.id, text=body.text, author_id=self._audit.user_id, created_at=utc_now()) → flush → ответ с ФИО автора (AppUser.full_name). comments: select(Comment, AppUser.full_name).join(AppUser, AppUser.id == Comment.author_id).where(object_type == "document", object_id == document.id).order_by(Comment.created_at, Comment.id). comments_count_by_document: select(Comment.object_id, func.count()).where(Comment.object_type == "document", Comment.object_id.in_(ids)).group_by(…). file_out(file, job, documents, counts: Mapping[uuid.UUID, int] | None = None) — DocumentBrief(..., comments_count=counts.get(id, 0)); во всех местах построения карточек файла (primary_files._cards, DocumentService, загрузка) передать счётчики (grep -n "file_out(" src -r). DocumentOut.comments_count — в document_out. Маршруты: POST /documents/{document_id}/comments (status_code=201, DOCUMENT_EDIT), GET …/comments (DOCUMENT_VIEW, response_model=list[CommentOut]). В tests/test_openapi_routes.py — "/api/v1/documents/{document_id}/comments": {"get", "post"}.
Step 4: Тесты — cd api && .venv/bin/pytest tests/test_api_comments.py tests/test_api_files_read.py tests/test_api_documents.py tests/test_openapi_routes.py -q -p no:cacheprovider && cd .. && make lint → PASS.
Step 5: Commit — git add -A api/src api/tests && git commit -m "feat(documents): комментарии к документу и comments_count в карточке (D-11)" (проверить git status, что в коммит не попало лишнее).

Task 9: Отдача исходного файла потоком

Files:

Modify: api/src/core/storage.py (ObjectStorage.stream), api/src/core/clients.py (S3Storage.stream), api/tests/conftest.py (FakeStorage.stream), api/src/services/primary_files.py (content), api/src/api/endpoints/files/read.py (маршрут), api/src/api/endpoints/consolidations/export.py (вынести content_disposition в src/common/filenames.py, если его там нет), api/tests/test_openapi_routes.py
Test: api/tests/test_api_files_content.py, api/tests/test_clients.py (поток S3Storage на фейковом клиенте)

Interfaces:

Produces:
- ObjectStorage.stream(obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes] — асинхронный генератор; нет объекта — ObjectNotFoundError до первой порции (ошибка видна до отправки заголовков). Реализация S3: get_object → response.content.iter_chunked(chunk_size); в finally — response.close() и await response.release().
- PrimaryFilesService.content(file_id, disposition: Literal["attachment","inline"]) -> tuple[AsyncIterator[bytes], str, str] — поток, Content-Type, Content-Disposition; 404 FILE_CONTENT_NOT_FOUND, если объекта нет.
- content_disposition(filename: str, *, disposition: str = "attachment", ascii_fallback: str) -> str в src/common/filenames.py.
- MIME: PDF application/pdf, PNG image/png, JPG image/jpeg, XLS application/vnd.ms-excel, XLSX application/vnd.openxmlformats-officedocument.spreadsheetml.sheet, XLSB application/vnd.ms-excel.sheet.binary.macroEnabled.12.
Step 1: Падающие тесты — api/tests/test_api_files_content.py:

"""Исходный файл потоком (spec documents-api «Исходный файл»; design D-10)."""

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from src.models.files import S3Object
from tests.conftest import AuthClientFactory, FakeStorage
from tests.test_api_documents import KA_RF, _rostov_entity
from tests.factories import make_stored_file
from src.models.enums import FileStatus, FileType, RecognitionStatus

pytestmark = pytest.mark.integration
PDF = b"%PDF-1.4\n" + bytes(range(256)) * 4096  # ~1 МиБ: проверка порционной отдачи


async def _file(session: AsyncSession, storage: FakeStorage, name: str, file_type: FileType):
    entity = await _rostov_entity(session, name="ООО Файлы")
    file, _ = await make_stored_file(
        session, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done, original_name=name
    )
    file.file_type = file_type
    key = (await session.execute(select(S3Object.key).where(S3Object.id == file.s3_object_id))).scalar_one()
    storage.objects[key] = PDF
    await session.flush()
    return file


async def test_inline_pdf(app, seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    storage = FakeStorage()
    app.state.storage = storage
    file = await _file(seeded, storage, "aep_6.pdf", FileType.PDF)
    client: AsyncClient = await auth_client(KA_RF)
    response = await client.get(f"/files/{file.id}/content", params={"disposition": "inline"})
    assert response.status_code == 200
    assert response.headers["content-type"] == "application/pdf"
    assert response.headers["content-disposition"] == "inline; filename=\"file.pdf\"; filename*=UTF-8''aep_6.pdf"
    assert response.content == PDF


async def test_cyrillic_name_and_missing(app, seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    storage = FakeStorage()
    app.state.storage = storage
    file = await _file(seeded, storage, "ОСВ_66 за 2024.xls", FileType.XLS)
    client = await auth_client(KA_RF)
    response = await client.get(f"/files/{file.id}/content")
    assert response.headers["content-disposition"].startswith('attachment; filename="file.xls"; filename*=UTF-8\'\'%D0%9E')
    storage.objects.clear()
    missing = await client.get(f"/files/{file.id}/content")
    assert missing.status_code == 404 and missing.json()["detail"]["code"] == "FILE_CONTENT_NOT_FOUND"


async def test_stream_releases_on_error() -> None:
    """Review Focus 4: при обрыве потребителя ответ S3 закрывается и отпускается."""
    from src.core.clients import S3Storage

    class Content:
        async def iter_chunked(self, size: int):
            yield b"a" * size
            yield b"b" * size

    class Response:
        def __init__(self) -> None:
            self.content = Content()
            self.closed = self.released = False

        def close(self) -> None:
            self.closed = True

        async def release(self) -> None:
            self.released = True

    response = Response()

    class Client:
        async def get_object(self, bucket: str, name: str) -> Response:
            return response

    storage = S3Storage(Client(), "arm-ka")  # type: ignore[arg-type]
    stream = storage.stream("uploads/x", chunk_size=4)
    assert await stream.__anext__() == b"aaaa"
    await stream.aclose()  # потребитель оборвал чтение
    assert response.closed and response.released

Если фикстура app в conftest не допускает подмену app.state.storage после создания — сделать это так же, как в tests/test_api_files_upload.py (grep -n "state.storage" tests/test_api_files_upload.py).

Step 2: Убедиться, что падают.
Step 3: Реализация. S3Storage.stream:

    async def stream(self, obj_name: str, chunk_size: int = 1 << 20) -> AsyncIterator[bytes]:
        """Объект порциями; `NoSuchKey` → `ObjectNotFoundError` до первой порции; ответ отпускается и при обрыве."""
        try:
            response = await self.client.get_object(self.bucket, obj_name)
        except S3Error as error:
            if error.code == NO_SUCH_KEY:
                raise ObjectNotFoundError(obj_name) from error
            raise
        try:
            async for chunk in response.content.iter_chunked(chunk_size):
                yield chunk
        finally:
            response.close()
            await response.release()

Чтобы ошибка «нет объекта» возникала до заголовков ответа, сервис вызывает stream, берёт первую порцию (await anext(iterator, b"")) и отдаёт StreamingResponse из генератора «первая порция + остальные». FakeStorage.stream — генератор по self.objects[obj_name] порциями chunk_size, KeyError → ObjectNotFoundError. Маршрут GET /files/{file_id}/content (DOCUMENT_VIEW, disposition: Literal["attachment","inline"] = "attachment") возвращает StreamingResponse(iterator, media_type=..., headers={"Content-Disposition": ...}); файл ищется с видимостью (files_repo.get_file), ключ — S3Object.key по stored_file.s3_object_id; файл SUPERSEDED разрешён. В test_openapi_routes.py — "/api/v1/files/{file_id}/content": {"get"}.

Step 4: Тесты — cd api && .venv/bin/pytest tests/test_api_files_content.py tests/test_api_osv_export.py tests/test_openapi_routes.py -q -p no:cacheprovider && cd .. && make lint → PASS.
Step 5: Commit — git commit -m "feat(files): GET /files/{id}/content — исходный файл потоком из S3, RFC 5987 (D-10)" (добавить изменённые файлы явно).

Task 10: Справочник имён пользователей

Files:

Create: api/src/api/endpoints/users.py
Modify: api/src/repositories/users.py (users_brief), api/src/schemas/admin.py или новый api/src/schemas/users.py (UserBriefOut), api/src/api/router.py, api/tests/test_openapi_routes.py
Test: api/tests/test_api_users_brief.py

Interfaces:

Produces: GET /api/v1/users/brief?ids=<uuid>,<uuid> → list[UserBriefOut(id, full_name, is_active)]; users_repo.users_brief(session, ids: Sequence[uuid.UUID]) -> list[AppUser]; ошибка TOO_MANY_IDS (422, DomainValidationError), MAX_BRIEF_IDS = 200.
Step 1: Падающие тесты — api/tests/test_api_users_brief.py:

"""Краткий справочник пользователей (spec users-admin «Краткий справочник пользователей»; D-12)."""

import uuid

import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from src.models.access import AppUser
from tests.conftest import AuthClientFactory

pytestmark = pytest.mark.integration


async def test_brief_for_non_admin(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    users = (await seeded.execute(select(AppUser).order_by(AppUser.email).limit(2))).scalars().all()
    unknown = uuid.uuid4()
    client = await auth_client("ka_rf@armka.local")  # без USERS_ADMIN
    ids = ",".join([*(str(user.id) for user in users), str(unknown)])
    response = await client.get("/users/brief", params={"ids": ids})
    assert response.status_code == 200
    body = {item["id"]: item for item in response.json()}
    assert set(body) == {str(user.id) for user in users}
    assert all(set(item) == {"id", "full_name", "is_active"} for item in body.values())


async def test_brief_limits(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    client = await auth_client("ka_rf@armka.local")
    too_many = ",".join(str(uuid.uuid4()) for _ in range(201))
    response = await client.get("/users/brief", params={"ids": too_many})
    assert response.status_code == 422 and response.json()["detail"]["code"] == "TOO_MANY_IDS"
    assert (await client.get("/users/brief", params={"ids": "not-a-uuid"})).status_code == 422

Step 2: Убедиться, что падают.
Step 3: Реализация. Маршрут router = APIRouter(prefix="/users", tags=["users"]), GET /brief, зависимость Annotated[Principal, Depends(current_user)] (как me.py), параметр ids: str = Query(...); разбор — split(","), пустые пропускаются, uuid.UUID(...) с ValueError → DomainValidationError("INVALID_IDS", "…") (422), больше 200 — TOO_MANY_IDS. Репозиторий: select(AppUser).where(AppUser.id.in_(ids)). Подключить users.router в build_api_router до admin.router (пути не пересекаются: /users/brief против /admin/users). test_openapi_routes.py — путь "/api/v1/users/brief": {"get"} и префикс "/api/v1/users" в CHECKED_PREFIXES.
Step 4: Тесты и линт — PASS.
Step 5: Commit — git commit -m "feat(users): GET /users/brief — имена пользователей для всех аутентифицированных (D-12)".

Task 11: `documents_count` у ЮЛ

Files:

Modify: api/src/repositories/directories.py (documents_count_by_entity), api/src/services/legal_entities.py (page, get), api/src/schemas/directories.py (LegalEntityOut.documents_count: int = 0), endpoint списка — если он строит LegalEntityOut.model_validate(entity) напрямую, передавать счётчик
Test: api/tests/test_api_legal_entities.py (дописать)

Interfaces:

Produces: async def documents_count_by_entity(session, entity_ids: Sequence[uuid.UUID]) -> dict[uuid.UUID, int] — count(document.id) по document.legal_entity_id с join(StoredFile) и StoredFile.status != SUPERSEDED.
Step 1: Падающий тест — в api/tests/test_api_legal_entities.py:

async def test_documents_count(seeded: AsyncSession, auth_client: AuthClientFactory) -> None:
    from src.models.enums import DocumentType, FileStatus, RecognitionStatus
    from tests.factories import make_document, make_legal_entity, make_stored_file

    entity = await make_legal_entity(seeded, name="ООО Счётчик")  # монозаёмщик, видим ГО
    live, job = await make_stored_file(seeded, entity=entity, status=FileStatus.PROCESSED, job_status=RecognitionStatus.done)
    await make_document(seeded, file=live, job=job, account="60")
    await make_document(seeded, file=live, job=job, account="62")
    old, old_job = await make_stored_file(seeded, entity=entity, status=FileStatus.SUPERSEDED, job_status=RecognitionStatus.done)
    await make_document(seeded, file=old, job=old_job, account="66")
    client = await auth_client("admin@armka.local")
    listed = (await client.get("/directories/legal-entities", params={"q": "Счётчик"})).json()
    assert listed["items"][0]["documents_count"] == 2
    card = (await client.get(f"/directories/legal-entities/{entity.id}")).json()
    assert card["documents_count"] == 2

Пользователь с видимостью всех ЮЛ — подобрать из существующих тестов файла (grep -n "auth_client(" tests/test_api_legal_entities.py | head -3); make_document с разными account — чтобы не нарушить уникальность документа в файле.

Step 2–4: убедиться, что падает; реализовать подзапрос на страницу (ids страницы → словарь счётчиков) и в карточке; тесты и линт — PASS.
Step 5: Commit — git commit -m "feat(directories): documents_count в списке и карточке ЮЛ (D-13)".

Task 12: Чек-лист в CSV

Files:

Create: api/src/domain/checklist_csv.py
Modify: api/src/api/endpoints/consolidations/checklist.py (параметр format), api/src/services/consolidations.py (имя ГК для имени файла, если не доступно в ChecklistOut)
Test: api/tests/test_domain_checklist_csv.py, api/tests/test_api_consolidation_checklist.py (дописать)

Interfaces:

Produces: render_checklist_csv(checklist: ChecklistOut) -> bytes — UTF-8 с BOM, ;, \r\n; заголовок «Организация;ИНН;<правило>…»; подписи CELL_LABELS = {CONFIRMED: "Загружен", PROCESSING: "Обработка", NEEDS_CHECK: "Требуется проверка", RECOGNIZED: "Требуется проверка", ERROR: "Ошибка загрузки", MISSING: "Отсутствует"}; правило: BALANCE «Бух. баланс», INCOME «ОФР», OSV «ОСВ », иначе document_type; при missing_quarters — « (нет: Q1, Q2)». Значения с ; или " экранируются (csv.writer с delimiter=";"). checklist_filename(group_name: str, year: int, quarter: str) -> str = checklist-<ГК>-<год>-<квартал>.csv.
Step 1: Падающие тесты — api/tests/test_domain_checklist_csv.py:

"""CSV чек-листа (spec consolidations-api «Чек-лист комплекта»; D-13)."""

import uuid

from src.domain.checklist import CellStatus
from src.domain.checklist_csv import checklist_filename, render_checklist_csv
from src.models.enums import DocumentType, PeriodQuarter
from src.schemas.consolidations import ChecklistCellOut, ChecklistOut, ChecklistRowOut, ChecklistRuleOut


def test_render() -> None:
    checklist = ChecklistOut(
        rules=[
            ChecklistRuleOut(document_type=DocumentType.BALANCE, account=None, is_mandatory=True),
            ChecklistRuleOut(document_type=DocumentType.OSV, account="60", is_mandatory=True),
        ],
        rows=[
            ChecklistRowOut(
                legal_entity_id=uuid.uuid4(), legal_entity_name='ООО "Альфа; Бета"', legal_entity_inn="7700000000",
                cells=[
                    ChecklistCellOut(status=CellStatus.CONFIRMED, missing_quarters=[], document_ids=[]),
                    ChecklistCellOut(status=CellStatus.MISSING, missing_quarters=[PeriodQuarter.Q2], document_ids=[]),
                ],
            )
        ],
    )
    content = render_checklist_csv(checklist)
    assert content.startswith("".encode())
    lines = content.decode("utf-8-sig").splitlines()
    assert lines[0] == "Организация;ИНН;Бух. баланс;ОСВ 60"
    assert lines[1] == '"ООО ""Альфа; Бета""";7700000000;Загружен;Отсутствует (нет: Q2)'


def test_filename() -> None:
    assert checklist_filename("Русский бисквит", 2025, "Q3") == "checklist-Русский бисквит-2025-Q3.csv"

В tests/test_api_consolidation_checklist.py — сценарий «Чек-лист в CSV»: GET …/checklist?format=csv → 200, content-type начинается с text/csv, тело начинается с BOM, первая строка — заголовок, content-disposition содержит filename*=UTF-8''checklist-.

Step 2: Убедиться, что падают.
Step 3: Реализация — src/domain/checklist_csv.py на csv.writer(io.StringIO(), delimiter=";", lineterminator="\r\n"), результат ("" + buffer.getvalue()).encode("utf-8"). Маршрут: параметр format: Literal["json", "csv"] = "json"; при csv — Response(content=render_checklist_csv(checklist), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": content_disposition(checklist_filename(...), disposition="attachment", ascii_fallback="checklist.csv")}); у маршрута response_model=ChecklistOut сохраняется для JSON (для CSV вернуть Response — FastAPI пропустит валидацию). Имя ГК, год и квартал — из карточки версии (service.get_version или аналог; при отсутствии метода — добавить в ConsolidationService лёгкий version_title(version_id) -> tuple[str, int, str]).
Step 4: Тесты и линт — PASS.
Step 5: Commit — git commit -m "feat(consolidations): чек-лист в CSV — подписи макета, BOM, «;» (D-13)".

Task 13: README, CLAUDE.md, tasks.md, полная проверка

Files:

Modify: README.md (разделы API документов и ошибок), CLAUDE.md (блок «Первичные документы» — формы, §7, правка; карта domain/), openspec/changes/slice-3b-statements-api/tasks.md
Step 1: README — в разделе эндпоинтов документов добавить: GET/PATCH /documents/{id}/statement, POST …/statement/checks/accept, POST/GET /documents/{id}/comments, GET /files/{id}/content, GET /users/brief, documents_count, ?format=csv чек-листа; в каталоге ошибок «Ошибки API» — DOCUMENT_NOT_STATEMENT, DOCUMENT_READONLY, UNKNOWN_CELL, DUPLICATE_CELL, INVALID_INN (форма), NOTHING_TO_ACCEPT, PERIOD_REQUIRED, CHECKS_NOT_PASSED, FILE_CONTENT_NOT_FOUND, TOO_MANY_IDS, INVALID_IDS. Сообщение «в результате нет документов» вместо «нет документов ОСВ».
Step 2: CLAUDE.md — в «Первичные документы»: формы ББ/ОФР маппятся в map_file (domain/statement_mapping.py, repositories/statements.py), проверки §7 — domain/statement_checks.py + копия rules/statement_checks.yaml (меняется в обоих репозиториях), правка формы — services/statements.py (только до подтверждения; аудит RECOGNIZED_DATA_EDIT).
Step 3: tasks.md — перенести задачи 0–15 плана с отметками [x] по факту; openspec validate slice-3b-statements-api --strict → valid.
Step 4: Полная проверка — make check-full (включает dbml-check, test-integration, e2e, docker-build, web-check) → зелёный; вывод — в отчёт задачи (пойдёт в описание PR). При красном e2e из-за отсутствия профиля services — это skip, не ошибка.
Step 5: Commit — git commit -m "docs(arm-core): README, CLAUDE.md, tasks.md под Срез 3b".

Task 14: Документы workspace (bounded-PR в docs)

Files (репозиторий docs, worktree /Users/andrey.epifancev/projects/awp-ca/.wt/slice-3b/docs): context/05-domain-rules.md (§7, §9.3, §9.5), context/07-open-questions.md, context/03-architecture.md, context/09-delivery-plan.md, dev/api-contract-slice-3b.md (новый), dev/README.md (ссылка), dev/qa-guide.md.

Step 1: Issue и worktree

cd /Users/andrey.epifancev/projects/awp-ca
tea issues create --login home --repo RSHB/arm-ka-docs --title "Срез 3b (docs): §7/§9 по факту, контракт API форм и мелких дыр, OQ" \
  --description "Сопровождение change arm-core slice-3b-statements-api (#36): §7 (проверки в ядре, межформенная мягкая, акцепт), §9.3/§9.5 (формы ББ/ОФР, правка до подтверждения, условия подтверждения), dev/api-contract-slice-3b.md, OQ (аудит скачивания первички), 03-architecture (копия словаря §7), 09-delivery-plan (3b реализован)."
# номер — <D>
rm /Users/andrey.epifancev/projects/awp-ca/.wt/slice-3b/docs   # симлинк
git -C /Users/andrey.epifancev/projects/awp-ca/docs worktree add /Users/andrey.epifancev/projects/awp-ca/.wt/slice-3b/docs -b feat/<D>-slice-3b-docs main
touch /Users/andrey.epifancev/projects/awp-ca/.wt/slice-3b/docs/.bounded

Step 2: Правки

§7: «Ядро (Срез 3b) считает §7 по текущим суммам формы словарём arm-core/api/rules/statement_checks.yaml (копия словаря tabular-parser): те же жёсткие и мягкие соотношения, допуск ±1, округление 0,001 %, abs_terms, подстроки. Межформенная 1370 − 1370(начало года) = 2400 — мягкая (не блокирует подтверждение). Акцепт — с комментарием, сбрасывается любой правкой формы.»
§9.3: статус документа формы — правило D-6 спеки 3b; §9.5 п. 2 — формы маппятся (вместо «игнорируются»), период ББ/ОФР по дате отчёта или самой поздней дате колонок; новый пункт «Правка формы» — значения и ИНН/дата/единицы только до подтверждения, исправление после — новым файлом; п. 4 — условия подтверждения формы (PERIOD_REQUIRED, CHECKS_NOT_PASSED).
07-open-questions.md: новый OQ (следующий номер после последнего): «Журналировать ли скачивание первички и CSV (ТЗ 8.2 „выгрузка“)? Сейчас не пишется, как экспорт сводной ОСВ» — допущение принято; OQ-23, OQ-36, OQ-77, OQ-140, OQ-145 — дописать «Срез 3b: допущение реализовано».
03-architecture.md: раздел Среза 3b (маппинг форм в map_file, statement_checks.yaml — копия, риск расхождения).
09-delivery-plan.md: Срез 3 → 3a закрыт, 3b реализован (бэкенд и API, UI отдельно); статусы 2a и 2c — «закрыт» (сейчас «ожидает merge/ревью»).
dev/api-contract-slice-3b.md — по образцу dev/api-contract-slice-4.md: общее (коды ошибок, формат денег), эндпоинты (таблица), схемы (StatementOut, StatementPatch, CommentOut, UserBriefOut), нюансы (подтверждённая форма только для чтения; is_modified снимается возвратом к исходному; акцепт сбрасывается правкой; редакция формы; CSV — BOM и «;»), сквозной сценарий экрана «Проверка распознавания». Ссылка в dev/README.md.
dev/qa-guide.md: как проверить форму на стенде (загрузка fixtures/statements-xlsx/mezheninovskaya/..., просмотр формы, правка, акцепт, подтверждение, чек-лист).

Step 3: Проверка, коммит, PR — make validate-contracts в worktree docs; коммит docs(context): Срез 3b — §7/§9 по факту, контракт API форм, OQ (#<D>); push; tea pulls create --login home --repo RSHB/arm-ka-docs --head feat/<D>-slice-3b-docs --base main с «Closes #», списком «Rulings I made» и атрибуцией.

Task 15: Завершение — финальное ревью и PR

Step 1: git fetch origin && git merge --no-edit origin/main (force-push запрещён), make check-full, openspec validate slice-3b-statements-api --strict.
Step 2: финальное ревью всей ветки (opus): git diff origin/main...HEAD; исправление critical/important; повторный make check-full.
Step 3: PR arm-core: tea pulls create --login home --repo RSHB/arm-core --head feat/36-slice-3b-statements-api --base main --title "Срез 3b: формы ББ/ОФР в ядре и API до сводной ОСВ" с «Closes #36», ссылкой на change и PR docs, выводом make check-full, таблицей всех находок (маркер, суть, статус), «Rulings I made», deferred, атрибуцией.
Step 4: отчёт: ссылки на PR, находки, рулинги, deferred, что осталось владельцу (мерж docs и arm-core; OQ об аудите скачивания).

150 KiB Raw Blame History Unescape Escape

Срез 3b — формы ББ/ОФР в ядре и API до сводной ОСВ: план реализации

Global Constraints

Review Focus

Структура файлов

Task 0: Подготовка окружения

Task 1: Проверки §7 — словарь и чистые функции

Task 2: Чистые функции маппинга формы

Task 3: Помощники golden-тестов форм

Task 4: Маппинг форм в map_file

Task 5: API формы — GET /documents/{id}/statement

Task 6: Правка формы и акцепт проверок

Task 7: Подтверждение формы и сквозной сценарий чек-листа

Task 8: Комментарии к документу

Task 9: Отдача исходного файла потоком

Task 10: Справочник имён пользователей

Task 11: documents_count у ЮЛ

Task 12: Чек-лист в CSV

Task 13: README, CLAUDE.md, tasks.md, полная проверка

Task 14: Документы workspace (bounded-PR в docs)

Task 15: Завершение — финальное ревью и PR

150 KiB

Raw Blame History

Task 4: Маппинг форм в `map_file`

Task 5: API формы — `GET /documents/{id}/statement`

Task 11: `documents_count` у ЮЛ