feat(scripts): purge_statements — очистка форм ББ/ОФР перед выкладкой v5.19 (Р-24, Р-27)

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

README.md

@@ -614,6 +614,32 @@ advisory-блокировкой.
 (физически объект не удаляется — ретеншн OQ-22, OQ-111) и пишет `FILE_DELETE` с карточкой файла в `before_value`. Нельзя удалить файл с документом, подтверждённым пользователем (`confirmed_by` задан), и с автоподтверждённым документом, привязанным к версии вне статусов «в работе» `INITIATED`/`DOCUMENTS_PROCESSING`/`DATA_CORRECTION` (`409 FILE_HAS_CONFIRMED_DOCUMENTS`); автоподтверждённые привязки к версиям «в работе» снимаются (`FILE_DELETE.after_value.unbound_version_ids`, OQ-157). Нельзя также удалить файл, который заменил другие (`409 FILE_SUPERSEDES_OTHERS`); `SUPERSEDED` файл удалить можно. Блокировки — в порядке
 «задания → файл → документы → advisory ГК → версии по id».
 
+## Выкладка v5.19: очистка форм
+
+Перед `alembic upgrade` v5.19 на контуре (после резервной копии БД) база очищается от форм ББ/ОФР — они пересоздаются
+повторной загрузкой по новым правилам (Р-24). Запускает владелец или платформенная команда по его заявке; на стендах
+разработчиков — по желанию. Из `api/`, одна транзакция:
+
+```
+.venv/bin/python -m scripts.purge_statements                          # --dry-run (по умолчанию): только отчёт
+.venv/bin/python -m scripts.purge_statements --apply                  # очистка
+.venv/bin/python -m scripts.purge_statements --apply --include-mixed  # смешанные файлы удалить целиком
+```
+
+Отчёт: число файлов только с формами, смешанных файлов (есть и ОСВ, и формы; их повторная загрузка невозможна из-за
+дедупа по `sha256`, поэтому владелец решает до `--apply`, нужен ли `--include-mixed`), документов форм, подтверждённых
+документов, привязок к неверифицированным версиям, файлов с ОСВ, заменённых удаляемыми.
+
+- Файл только с формами удаляется целиком, как `DELETE /files/{id}`, но без запрета `FILE_HAS_CONFIRMED_DOCUMENTS`:
+  привязки снимаются (статус версий не меняется), `superseded_by_file_id` ссылающихся файлов обнуляется, объектам S3
+  ставится `expires_at`. У смешанного файла удаляются только `BALANCE` / `INCOME` с формами, комментариями и привязками.
+- Отказ без изменений — только если документ упомянут в `ex_snapshot_item` или привязан к версии `VERIFIED` либо
+  `ARCHIVED` со срезом ЕХ (Р-35); отчёт называет версии. Версии в работе, на согласовании, `REJECTED`, `CANCELLED` не
+  мешают.
+- Аудит: на файл — системное `FILE_DELETE` (`user_id` пуст, описание «очистка форм перед v5.19», карточка файла в
+  `before_value`; у смешанного — `after_value.deleted_documents`). ОСВ, консолидации, справочники и журнал не трогаются.
+- Коды выхода: `0` — отчёт или выполнено, `1` — отказ, `2` — ошибка аргументов.
+
 ## Ограничения Среза 2c
 
 - Аудит обмена (смена статусов задания/файла в inbox, вахтёре, цикле маппинга) не пишется — `ASSUMPTION(OQ-55)`;

api/scripts/purge_statements.py

@@ -0,0 +1,334 @@
+"""Очистка форм ББ/ОФР перед выкладкой ядра с миграцией v5.19 (spec arm-core «Очистка форм перед выкладкой»,
+Р-24, Р-27).
+
+Формы пересоздаются повторной загрузкой файлов уже по новым правилам. Запуск из `api/`:
+
+    .venv/bin/python -m scripts.purge_statements                         # отчёт (--dry-run), база не меняется
+    .venv/bin/python -m scripts.purge_statements --apply                 # очистка, одна транзакция
+    .venv/bin/python -m scripts.purge_statements --apply --include-mixed # смешанные файлы удаляются целиком
+
+Отбор — документы `BALANCE` / `INCOME` файлов любого статуса.
+Отказ без изменений (P3-12, Р-35) — только если отобранный (или удаляемый вместе с файлом) документ упомянут в
+`ex_snapshot_item` либо привязан к верифицированной версии: `VERIFIED` или `ARCHIVED` со срезом `ex_snapshot`
+(`source_version_id` = версия). Привязки к версиям в работе, на согласовании, `REJECTED`, `CANCELLED` (и архивным
+отменённым, без среза ЕХ) снимаются, статус версий не меняется.
+Файл только с формами (нет документов `OSV`) удаляется целиком, как `DELETE /files/{id}`, но без запрета
+`FILE_HAS_CONFIRMED_DOCUMENTS`; у смешанного файла (есть и ОСВ, и формы) удаляются только `BALANCE` / `INCOME`
+(ключ `--include-mixed` — файл целиком). ОСВ вне удаляемых файлов, консолидации, справочники и аудит не трогаются.
+В ядре результатов КФО ещё нет — пересчитывать после очистки нечего.
+
+Коды выхода (P3-13): 0 — отчёт или выполнено, 1 — отказ, 2 — ошибка аргументов.
+"""
+
+import argparse
+import asyncio
+import sys
+import uuid
+from collections.abc import Callable, Sequence
+from contextlib import AbstractAsyncContextManager
+from dataclasses import dataclass, field
+from datetime import datetime
+from pathlib import Path
+
+_API_DIR = Path(__file__).resolve().parents[1]
+if str(_API_DIR) not in sys.path:  # запуск файлом: в sys.path только `api/scripts`, а нужен `api/`
+    sys.path.insert(0, str(_API_DIR))
+
+from sqlalchemy import and_, delete, exists, func, or_, select, update  # noqa: E402
+from sqlalchemy.ext.asyncio import AsyncSession  # noqa: E402
+from sqlalchemy.sql.elements import ColumnElement  # noqa: E402
+
+from src.common.clock import utc_now  # noqa: E402
+from src.db.db import session_transaction  # noqa: E402
+from src.models.audit_ex import Comment, ExSnapshot, ExSnapshotItem  # noqa: E402
+from src.models.consolidation import ConsolidationDocument, ConsolidationVersion  # noqa: E402
+from src.models.enums import (  # noqa: E402
+    AuditAction,
+    AuditSection,
+    ConsolidationStatus,
+    DocumentStatus,
+    DocumentType,
+    S3BucketRole,
+)
+from src.models.files import Document, StoredFile  # noqa: E402
+from src.repositories import files as files_repo  # noqa: E402
+from src.repositories import statements as statements_repo  # noqa: E402
+from src.services import consolidation_documents  # noqa: E402
+from src.services.audit import AuditService  # noqa: E402
+from src.services.documents import file_out  # noqa: E402
+from src.services.primary_files import OBJECT_TYPE, UPLOADS_PREFIX  # noqa: E402
+from src.services.result_mapping import RESULTS_PREFIX  # noqa: E402
+
+EXIT_OK, EXIT_REFUSED, EXIT_USAGE = 0, 1, 2
+DESCRIPTION = "очистка форм перед v5.19"
+FORM_TYPES = (DocumentType.BALANCE, DocumentType.INCOME)
+SNAPSHOT_ITEM_STATUS = "EX_SNAPSHOT_ITEM"  # «статус» блокировки, найденной по ex_snapshot_item, а не по привязке
+
+Transaction = Callable[[], AbstractAsyncContextManager[AsyncSession]]
+
+
+@dataclass(frozen=True)
+class BlockingVersion:
+    """Причина отказа: документ формы заморожен в верифицированной версии (или упомянут в срезе ЕХ)."""
+
+    version_id: uuid.UUID
+    status: str
+    document_id: uuid.UUID
+
+
+@dataclass
+class PurgePlan:
+    """Что сделает очистка; строится `plan_purge` (только чтение)."""
+
+    form_only_files: list[uuid.UUID] = field(default_factory=list)
+    mixed_files: list[uuid.UUID] = field(default_factory=list)
+    form_documents: int = 0
+    confirmed_documents: int = 0
+    unverified_bindings: int = 0
+    blocking: list[BlockingVersion] = field(default_factory=list)
+    superseded_with_osv: list[uuid.UUID] = field(default_factory=list)
+    include_mixed: bool = False
+
+
+class PurgeRefusedError(Exception):
+    """Очистка отклонена: есть документы, замороженные в верифицированных версиях."""
+
+    def __init__(self, blocking: Sequence[BlockingVersion]) -> None:
+        super().__init__("очистка отклонена")
+        self.blocking = list(blocking)
+
+
+def _frozen_version() -> ColumnElement[bool]:
+    """Условие «версия верифицирована»: `VERIFIED`, либо `ARCHIVED` со срезом ЕХ этой версии (P3-12)."""
+    snapshot = exists().where(ExSnapshot.source_version_id == ConsolidationVersion.id)
+    return or_(
+        ConsolidationVersion.status == ConsolidationStatus.VERIFIED,
+        and_(ConsolidationVersion.status == ConsolidationStatus.ARCHIVED, snapshot),
+    )
+
+
+async def _documents_of(
+    session: AsyncSession, file_ids: Sequence[uuid.UUID]
+) -> list[tuple[uuid.UUID, uuid.UUID, DocumentType]]:
+    rows = await session.execute(
+        select(Document.id, Document.stored_file_id, Document.document_type).where(
+            Document.stored_file_id.in_(list(file_ids))
+        )
+    )
+    return [(row[0], row[1], row[2]) for row in rows.all()]
+
+
+async def _find_blocking(session: AsyncSession, document_ids: Sequence[uuid.UUID]) -> list[BlockingVersion]:
+    if not document_ids:
+        return []
+    ids = list(document_ids)
+    bound = await session.execute(
+        select(ConsolidationVersion.id, ConsolidationVersion.status, ConsolidationDocument.document_id)
+        .join(ConsolidationDocument, ConsolidationDocument.consolidation_version_id == ConsolidationVersion.id)
+        .where(ConsolidationDocument.document_id.in_(ids), _frozen_version())
+    )
+    found = {BlockingVersion(version_id, status.value, document_id) for version_id, status, document_id in bound.all()}
+    items = await session.execute(
+        select(ExSnapshot.source_version_id, ExSnapshotItem.document_id)
+        .join(ExSnapshotItem, ExSnapshotItem.ex_snapshot_id == ExSnapshot.id)
+        .where(ExSnapshotItem.document_id.in_(ids))
+    )
+    found |= {BlockingVersion(version_id, SNAPSHOT_ITEM_STATUS, document_id) for version_id, document_id in items.all()}
+    return sorted(found, key=lambda item: (str(item.version_id), str(item.document_id), item.status))
+
+
+async def plan_purge(session: AsyncSession, *, include_mixed: bool) -> PurgePlan:
+    """Отбор и отчёт без изменений. «Только с формами» — нет документов `OSV`; «смешанный» — есть и ОСВ, и формы."""
+    forms = (
+        await session.execute(
+            select(Document.id, Document.stored_file_id, Document.status).where(Document.document_type.in_(FORM_TYPES))
+        )
+    ).all()
+    plan = PurgePlan(include_mixed=include_mixed, form_documents=len(forms))
+    plan.confirmed_documents = sum(1 for _, _, status in forms if status is DocumentStatus.CONFIRMED)
+    file_ids = sorted({file_id for _, file_id, _ in forms})
+    if not file_ids:
+        return plan
+    documents = await _documents_of(session, file_ids)
+    with_osv = {file_id for _, file_id, kind in documents if kind is DocumentType.OSV}
+    plan.form_only_files = [file_id for file_id in file_ids if file_id not in with_osv]
+    plan.mixed_files = [file_id for file_id in file_ids if file_id in with_osv]
+    deleted_files = set(plan.form_only_files) | (set(plan.mixed_files) if include_mixed else set())
+    affected = [
+        document_id for document_id, file_id, kind in documents if file_id in deleted_files or (kind in FORM_TYPES)
+    ]
+    plan.blocking = await _find_blocking(session, affected)
+    blocked_ids = {item.document_id for item in plan.blocking}
+    bindings = await session.execute(
+        select(ConsolidationDocument.document_id, func.count())
+        .where(ConsolidationDocument.document_id.in_(affected))
+        .group_by(ConsolidationDocument.document_id)
+    )
+    plan.unverified_bindings = sum(count for document_id, count in bindings.all() if document_id not in blocked_ids)
+    if deleted_files:
+        rows = await session.execute(
+            select(StoredFile.id).where(StoredFile.superseded_by_file_id.in_(list(deleted_files)))
+        )
+        replaced = [file_id for (file_id,) in rows.all()]
+        plan.superseded_with_osv = sorted(await _files_with_osv(session, replaced))
+    return plan
+
+
+async def _files_with_osv(session: AsyncSession, file_ids: Sequence[uuid.UUID]) -> set[uuid.UUID]:
+    rows = await session.execute(
+        select(Document.stored_file_id)
+        .where(Document.stored_file_id.in_(list(file_ids)), Document.document_type == DocumentType.OSV)
+        .distinct()
+    )
+    return {file_id for (file_id,) in rows.all()}
+
+
+async def _purge_whole_file(session: AsyncSession, audit: AuditService, file_id: uuid.UUID, now: datetime) -> None:
+    """Файл целиком — как `PrimaryFileService.delete`, но без запрета на подтверждённые документы."""
+    jobs = await files_repo.lock_jobs(session, file_id)
+    file = await files_repo.lock_file(session, file_id)
+    if file is None:
+        return
+    documents = await files_repo.lock_file_documents(session, file_id)
+    bound = await consolidation_documents.bound_versions(session, documents=documents)
+    await consolidation_documents.unbind_documents(session, document_ids=[document.id for document in documents])
+    await session.execute(
+        update(StoredFile).where(StoredFile.superseded_by_file_id == file_id).values(superseded_by_file_id=None)
+    )
+    current = next((job for job in jobs if job.id == file.current_recognition_job_id), None)
+    counts = await files_repo.comments_count_by_document(session, [document.id for document in documents])
+    before = file_out(file, current, documents, counts).model_dump(mode="json")
+    object_ids = [file.s3_object_id, *(job.result_s3_object_id for job in jobs if job.result_s3_object_id)]
+    await files_repo.delete_file_rows(session, file)
+    owned_prefixes = [
+        (S3BucketRole.uploads, f"{UPLOADS_PREFIX}/{file_id}/"),
+        (S3BucketRole.results, f"{RESULTS_PREFIX}/{file_id}/"),
+    ]
+    await files_repo.expire_s3_objects(session, object_ids, owned_prefixes=owned_prefixes, now=now)
+    await audit.record(
+        action=AuditAction.FILE_DELETE,
+        section=AuditSection.PRIMARY_DOCUMENTS,
+        object_type=OBJECT_TYPE,
+        object_id=file_id,
+        user_id=None,
+        description=DESCRIPTION,
+        before_value=before,
+        after_value={"unbound_version_ids": sorted({version_id for version_id, _ in bound})},
+    )
+
+
+async def _purge_forms_of_file(session: AsyncSession, audit: AuditService, file_id: uuid.UUID) -> None:
+    """Смешанный файл: удаляются только `BALANCE` / `INCOME`; файл, ОСВ и задания остаются."""
+    jobs = await files_repo.lock_jobs(session, file_id)
+    file = await files_repo.lock_file(session, file_id)
+    if file is None:
+        return
+    documents = await files_repo.lock_file_documents(session, file_id)
+    forms = [document for document in documents if document.document_type in FORM_TYPES]
+    if not forms:
+        return
+    form_ids = [document.id for document in forms]
+    bound = await consolidation_documents.bound_versions(session, documents=forms)
+    await consolidation_documents.unbind_documents(session, document_ids=form_ids)
+    current = next((job for job in jobs if job.id == file.current_recognition_job_id), None)
+    counts = await files_repo.comments_count_by_document(session, [document.id for document in documents])
+    before = file_out(file, current, documents, counts).model_dump(mode="json")
+    await statements_repo.delete_documents_statements(session, form_ids)
+    await session.execute(
+        delete(Comment).where(Comment.object_type == files_repo.COMMENT_OBJECT_TYPE, Comment.object_id.in_(form_ids))
+    )
+    await session.execute(delete(Document).where(Document.id.in_(form_ids)))
+    await audit.record(
+        action=AuditAction.FILE_DELETE,
+        section=AuditSection.PRIMARY_DOCUMENTS,
+        object_type=OBJECT_TYPE,
+        object_id=file_id,
+        user_id=None,
+        description=DESCRIPTION,
+        before_value=before,
+        after_value={
+            "deleted_documents": [str(document_id) for document_id in form_ids],
+            "unbound_version_ids": sorted({version_id for version_id, _ in bound}),
+        },
+    )
+
+
+async def apply_purge(session: AsyncSession, plan: PurgePlan, *, now: datetime) -> None:
+    """Выполняет очистку в транзакции сессии (коммит — за вызывающим). Порядок блокировок: файл → документы → версии.
+
+    Блокирует файлы плана (по `id`), заново проверяет отказ под блокировками и только потом меняет данные: при
+    `PurgeRefusedError` база не изменена."""
+    whole = list(plan.form_only_files) + (list(plan.mixed_files) if plan.include_mixed else [])
+    partial = [] if plan.include_mixed else list(plan.mixed_files)
+    for file_id in sorted([*whole, *partial]):
+        await files_repo.lock_jobs(session, file_id)
+        await files_repo.lock_file(session, file_id)
+        await files_repo.lock_file_documents(session, file_id)
+    fresh = await plan_purge(session, include_mixed=plan.include_mixed)
+    if fresh.blocking:
+        raise PurgeRefusedError(fresh.blocking)
+    audit = AuditService(session)
+    for file_id in sorted(whole):
+        await _purge_whole_file(session, audit, file_id, now)
+    for file_id in sorted(partial):
+        await _purge_forms_of_file(session, audit, file_id)
+
+
+def render_report(plan: PurgePlan) -> str:
+    lines = [
+        f"файлов только с формами: {len(plan.form_only_files)}",
+        f"смешанных файлов: {len(plan.mixed_files)}"
+        + (" (будут удалены целиком, --include-mixed)" if plan.include_mixed else " (удаляются только формы)"),
+        f"документов форм: {plan.form_documents}",
+        f"подтверждённых документов: {plan.confirmed_documents}",
+        f"привязок к неверифицированным версиям: {plan.unverified_bindings}",
+    ]
+    if plan.mixed_files:
+        lines.append("смешанные файлы (повторная загрузка невозможна — дедуп по sha256):")
+        lines.extend(f"  {file_id}" for file_id in plan.mixed_files)
+    if plan.superseded_with_osv:
+        lines.append("заменённые удаляемыми файлами файлы с ОСВ (остаются SUPERSEDED):")
+        lines.extend(f"  {file_id}" for file_id in plan.superseded_with_osv)
+    if plan.blocking:
+        lines.append("ОТКАЗ: формы заморожены в верифицированных версиях, решение — за владельцем:")
+        lines.extend(
+            f"  версия {item.version_id} ({item.status}), документ {item.document_id}" for item in plan.blocking
+        )
+    return "\n".join(lines)
+
+
+def _parse(argv: Sequence[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(prog="purge_statements", description="Очистка форм ББ/ОФР перед выкладкой v5.19")
+    mode = parser.add_mutually_exclusive_group()
+    mode.add_argument("--dry-run", action="store_true", help="только отчёт (по умолчанию)")
+    mode.add_argument("--apply", action="store_true", help="выполнить очистку одной транзакцией")
+    parser.add_argument("--include-mixed", action="store_true", help="смешанные файлы удалить целиком")
+    return parser.parse_args(list(argv))
+
+
+async def run(argv: Sequence[str], *, transaction: Transaction = session_transaction) -> int:
+    """Точка входа; `transaction` — фабрика транзакции (тесты подставляют свою сессию)."""
+    try:
+        args = _parse(argv)
+    except SystemExit as error:
+        return EXIT_USAGE if error.code else EXIT_OK
+    try:
+        async with transaction() as session:
+            plan = await plan_purge(session, include_mixed=args.include_mixed)
+            print(render_report(plan))
+            if not args.apply:
+                return EXIT_OK
+            if plan.blocking:
+                print("очистка отклонена, база не изменена", file=sys.stderr)
+                return EXIT_REFUSED
+            await apply_purge(session, plan, now=utc_now())
+    except PurgeRefusedError as error:
+        print(render_report(PurgePlan(blocking=error.blocking)), file=sys.stderr)
+        print("очистка отклонена, база не изменена", file=sys.stderr)
+        return EXIT_REFUSED
+    print("очистка выполнена")
+    return EXIT_OK
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(run(sys.argv[1:])))

api/src/repositories/statements.py

@@ -26,6 +26,7 @@ __all__ = [
     "add_line",
     "add_value",
     "column_ids",
+    "delete_documents_statements",
     "delete_file_statements",
     "form_lines",
     "get_meta",
@@ -110,6 +111,16 @@ async def insert_statement(
 async def delete_file_statements(session: AsyncSession, file_id: uuid.UUID) -> None:
     """`document_issue`, значения, строки, колонки и шапки форм всех документов файла (в порядке FK)."""
     documents = select(Document.id).where(Document.stored_file_id == file_id).scalar_subquery()
+    await _delete_statements(session, documents)
+
+
+async def delete_documents_statements(session: AsyncSession, document_ids: Sequence[uuid.UUID]) -> None:
+    """То же для перечисленных документов (очистка форм смешанного файла: ОСВ и прочие документы файла остаются)."""
+    if document_ids:
+        await _delete_statements(session, list(document_ids))
+
+
+async def _delete_statements(session: AsyncSession, documents: Any) -> None:
     await session.execute(delete(DocumentIssue).where(DocumentIssue.document_id.in_(documents)))
     metas = select(FinancialStatementMeta.id).where(FinancialStatementMeta.document_id.in_(documents)).scalar_subquery()
     lines = select(FinancialStatementLine.id).where(FinancialStatementLine.meta_id.in_(metas)).scalar_subquery()

api/tests/test_purge_statements.py

@@ -0,0 +1,460 @@
+"""Очистка форм ББ/ОФР перед выкладкой v5.19 `scripts/purge_statements.py` (spec arm-core «Очистка форм перед
+выкладкой»; P3-12, P3-13, Р-24, Р-27, Р-35).
+
+Скрипт получает сессию из `transaction` — тесты подставляют `db_session` (транзакция теста), боевую БД не трогают.
+"""
+
+import datetime
+import uuid
+from collections.abc import AsyncIterator, Callable
+from contextlib import AbstractAsyncContextManager, asynccontextmanager
+
+import pytest
+from sqlalchemy import func, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from scripts import purge_statements
+from src.models.audit_ex import AuditEvent, Comment, ExSnapshot, ExSnapshotItem
+from src.models.consolidation import Consolidation, ConsolidationDocument, ConsolidationVersion
+from src.models.directories import CompanyGroup, LegalEntity
+from src.models.enums import (
+    AuditAction,
+    ConsolidationStatus,
+    DocumentStatus,
+    DocumentType,
+    ExSnapshotStatus,
+    FileStatus,
+    PeriodQuarter,
+    StatementType,
+)
+from src.models.files import Document, RecognitionJob, S3Object, StoredFile
+from src.models.osv import OsvEntry
+from src.models.statements import (
+    FinancialStatementColumn,
+    FinancialStatementLine,
+    FinancialStatementMeta,
+    FinancialStatementValue,
+)
+from src.services.primary_files import PrimaryFilesService
+from tests.factories import (
+    make_company_group,
+    make_consolidation,
+    make_document,
+    make_legal_entity,
+    make_osv_entry,
+    make_regional_office,
+    make_related_group,
+    make_stored_file,
+    make_user,
+    make_version,
+)
+
+pytestmark = pytest.mark.integration
+
+EXIT_OK, EXIT_REFUSED = 0, 1
+DESCRIPTION = "очистка форм перед v5.19"
+FORM_AND_MIXED_FILES = 2  # событий FILE_DELETE: файл только с формами и смешанный
+
+
+def _transaction(session: AsyncSession) -> Callable[[], AbstractAsyncContextManager[AsyncSession]]:
+    @asynccontextmanager
+    async def factory() -> AsyncIterator[AsyncSession]:
+        yield session
+
+    return factory
+
+
+async def _run(session: AsyncSession, *argv: str) -> int:
+    return await purge_statements.run(list(argv), transaction=_transaction(session))
+
+
+async def _count(session: AsyncSession, model: type, *where: object) -> int:
+    return (await session.execute(select(func.count()).select_from(model).where(*where))).scalar_one()
+
+
+class World:
+    """ГК с ЮЛ, инициатор и счётчик номеров версий."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self.session = session
+        self.entity: LegalEntity
+        self.user_id: uuid.UUID
+        self._version_no = 1
+
+    async def build(self) -> "World":
+        session = self.session
+        office = await _office(session)
+        group = await make_company_group(session, name="ГК Очистка", office=office)
+        self.group = group
+        self.entity = await make_legal_entity(session, name="ООО Очистка", group=group)
+        self.user = await make_user(session, email="purge@armka.local", office=office)
+        self.consolidation, self.first_version = await make_consolidation(
+            session,
+            group=group,
+            year=2025,
+            quarter=PeriodQuarter.Q1,
+            members=[self.entity],
+            initiator=self.user,
+            status=ConsolidationStatus.INITIATED,
+        )
+        return self
+
+    async def version(self, status: ConsolidationStatus):  # type: ignore[no-untyped-def]
+        """Ещё одна версия консолидации в статусе `status`."""
+        self._version_no += 1
+        return await make_version(
+            self.session,
+            consolidation=self.consolidation,
+            version_no=self._version_no,
+            members=[self.entity],
+            initiator=self.user,
+            status=status,
+        )
+
+    async def form_document(
+        self,
+        file: StoredFile,
+        job: RecognitionJob,
+        document_type: DocumentType = DocumentType.BALANCE,
+        *,
+        status: DocumentStatus = DocumentStatus.RECOGNIZED,
+    ) -> Document:
+        """Документ формы с шапкой, колонкой, строкой и значением."""
+        session = self.session
+        ordinal = (
+            await session.execute(
+                select(func.coalesce(func.max(Document.file_ordinal) + 1, 0)).where(Document.stored_file_id == file.id)
+            )
+        ).scalar_one()
+        now = datetime.datetime.now(datetime.UTC)
+        document = Document(
+            stored_file_id=file.id,
+            legal_entity_id=file.legal_entity_id,
+            document_type=document_type,
+            file_ordinal=ordinal,
+            period_year=2025,
+            period_quarter=PeriodQuarter.Q1,
+            status=status,
+            confirmed_by=self.user.id if status is DocumentStatus.CONFIRMED else None,
+            created_at=now,
+            updated_at=now,
+        )
+        session.add(document)
+        await session.flush()
+        if document_type in (DocumentType.BALANCE, DocumentType.INCOME):
+            meta = FinancialStatementMeta(
+                document_id=document.id,
+                recognition_job_id=job.id,
+                statement_type=StatementType.balance if document_type is DocumentType.BALANCE else StatementType.income,
+            )
+            session.add(meta)
+            await session.flush()
+            column = FinancialStatementColumn(meta_id=meta.id, column_index=0)
+            line = FinancialStatementLine(meta_id=meta.id, line_code="1600", row_index=1)
+            session.add_all([column, line])
+            await session.flush()
+            session.add(FinancialStatementValue(line_id=line.id, column_id=column.id, amount=1))
+            await session.flush()
+        return document
+
+    async def bind(self, document: Document, version) -> None:  # type: ignore[no-untyped-def]
+        self.session.add(ConsolidationDocument(consolidation_version_id=version.id, document_id=document.id))
+        await self.session.flush()
+
+    async def comment(self, document: Document) -> None:
+        self.session.add(
+            Comment(
+                object_type="document",
+                object_id=document.id,
+                text="проверить",
+                author_id=self.user.id,
+                created_at=datetime.datetime.now(datetime.UTC),
+            )
+        )
+        await self.session.flush()
+
+    async def snapshot(self, version, *, document: Document | None = None) -> ExSnapshot:  # type: ignore[no-untyped-def]
+        """Срез ЕХ версии `version`; при `document` — с `ex_snapshot_item` на документ."""
+        related = await make_related_group(self.session, group=self.group)
+        now = datetime.datetime.now(datetime.UTC)
+        snapshot = ExSnapshot(
+            related_group_id=related.id,
+            company_group_id=self.group.id,
+            reporting_date=datetime.date(2025, 4, 1),
+            year=2025,
+            quarter=PeriodQuarter.Q1,
+            version_no=1,
+            status=ExSnapshotStatus.ACTUAL,
+            source_version_id=version.id,
+            source_version_no=version.version_no,
+            verified_at=now,
+            written_at=now,
+        )
+        self.session.add(snapshot)
+        await self.session.flush()
+        if document is not None:
+            self.session.add(
+                ExSnapshotItem(ex_snapshot_id=snapshot.id, item_type="primary_document", document_id=document.id)
+            )
+            await self.session.flush()
+        return snapshot
+
+
+async def _office(session: AsyncSession):  # type: ignore[no-untyped-def]
+    return await make_regional_office(session)
+
+
+@pytest.fixture
+async def world(seeded: AsyncSession) -> World:
+    return await World(seeded).build()
+
+
+async def _mixed_and_form_only(world: World) -> dict[str, object]:
+    """Файл только с ББ+ОФР (ББ CONFIRMED, привязан к INITIATED), смешанный файл (ОСВ + ББ), файл только с ОСВ."""
+    session = world.session
+    form_file, form_job = await make_stored_file(session, entity=world.entity, original_name="forms.pdf")
+    balance = await world.form_document(form_file, form_job, DocumentType.BALANCE, status=DocumentStatus.CONFIRMED)
+    income = await world.form_document(form_file, form_job, DocumentType.INCOME)
+    await world.bind(balance, world.first_version)
+    mixed_file, mixed_job = await make_stored_file(session, entity=world.entity, original_name="mixed.xls")
+    osv = await make_document(session, file=mixed_file, job=mixed_job)
+    mixed_balance = await world.form_document(mixed_file, mixed_job, DocumentType.BALANCE)
+    osv_file, osv_job = await make_stored_file(session, entity=world.entity, original_name="osv.xls")
+    await make_document(session, file=osv_file, job=osv_job)
+    return {
+        "form_file": form_file,
+        "form_job": form_job,
+        "balance": balance,
+        "income": income,
+        "mixed_file": mixed_file,
+        "mixed_job": mixed_job,
+        "osv": osv,
+        "mixed_balance": mixed_balance,
+        "osv_file": osv_file,
+    }
+
+
+TABLES = (
+    StoredFile,
+    Document,
+    RecognitionJob,
+    FinancialStatementMeta,
+    FinancialStatementColumn,
+    FinancialStatementLine,
+    FinancialStatementValue,
+    Comment,
+    ConsolidationDocument,
+    OsvEntry,
+    AuditEvent,
+)
+
+
+async def _counts(session: AsyncSession) -> dict[str, int]:
+    return {model.__tablename__: await _count(session, model) for model in TABLES}
+
+
+async def test_dry_run_reports_and_changes_nothing(world: World, capsys: pytest.CaptureFixture[str]) -> None:
+    await _mixed_and_form_only(world)
+    before = await _counts(world.session)
+
+    code = await _run(world.session)
+
+    out = capsys.readouterr().out
+    assert code == EXIT_OK
+    assert "файлов только с формами: 1" in out
+    assert "смешанных файлов: 1" in out
+    assert "документов форм: 3" in out
+    assert "подтверждённых документов: 1" in out
+    assert "привязок к неверифицированным версиям: 1" in out
+    assert await _counts(world.session) == before
+
+
+async def test_usage_error(world: World) -> None:
+    assert await _run(world.session, "--unknown") == purge_statements.EXIT_USAGE
+
+
+@pytest.mark.parametrize("status", [ConsolidationStatus.VERIFIED, ConsolidationStatus.ARCHIVED])
+async def test_apply_refuses_verified_binding(
+    world: World, capsys: pytest.CaptureFixture[str], status: ConsolidationStatus
+) -> None:
+    objects = await _mixed_and_form_only(world)
+    version = await world.version(status)
+    if status is ConsolidationStatus.ARCHIVED:  # архивированная версия мешает, только если заморожена в ЕХ
+        await world.snapshot(version)
+    await world.bind(objects["income"], version)  # type: ignore[arg-type]
+    before = await _counts(world.session)
+
+    code = await _run(world.session, "--apply")
+
+    out = capsys.readouterr()
+    assert code == EXIT_REFUSED
+    assert str(version.id) in out.out + out.err
+    assert status.value in out.out + out.err
+    assert await _counts(world.session) == before
+
+
+async def test_apply_refuses_ex_snapshot_item(world: World, capsys: pytest.CaptureFixture[str]) -> None:
+    objects = await _mixed_and_form_only(world)
+    version = await world.version(ConsolidationStatus.CANCELLED)
+    snapshot = await world.snapshot(version, document=objects["balance"])  # type: ignore[arg-type]
+    before = await _counts(world.session)
+
+    code = await _run(world.session, "--apply")
+
+    out = capsys.readouterr()
+    assert code == EXIT_REFUSED
+    assert str(objects["balance"].id) in out.out + out.err  # type: ignore[attr-defined]
+    assert str(snapshot.source_version_id) in out.out + out.err
+    assert await _counts(world.session) == before
+
+
+async def test_apply_ignores_unverified_versions(world: World) -> None:
+    session = world.session
+    file, job = await make_stored_file(session, entity=world.entity, original_name="forms.pdf")
+    balance = await world.form_document(file, job, DocumentType.BALANCE)
+    income = await world.form_document(file, job, DocumentType.INCOME)
+    other = await world.form_document(file, job, DocumentType.BALANCE)
+    archived = await world.form_document(file, job, DocumentType.BALANCE)
+    versions = {
+        ConsolidationStatus.VERIFICATION_RF: balance,
+        ConsolidationStatus.REJECTED: income,
+        ConsolidationStatus.CANCELLED: other,
+        ConsolidationStatus.ARCHIVED: archived,  # архивная без ex_snapshot (из CANCELLED)
+    }
+    statuses = {}
+    for status, document in versions.items():
+        version = await world.version(status)
+        await world.bind(document, version)
+        statuses[version.id] = status
+
+    code = await _run(session, "--apply")
+
+    assert code == EXIT_OK
+    assert await _count(session, ConsolidationDocument) == 0
+    assert await _count(session, Document, Document.stored_file_id == file.id) == 0
+    assert await _count(session, StoredFile, StoredFile.id == file.id) == 0
+    rows = (await session.execute(select(ConsolidationVersion.id, ConsolidationVersion.status))).all()
+    for version_id, status in rows:
+        if version_id in statuses:
+            assert status is statuses[version_id]
+
+
+async def test_apply_deletes_form_only_file(world: World) -> None:
+    session = world.session
+    file, job = await make_stored_file(session, entity=world.entity, original_name="forms.pdf")
+    balance = await world.form_document(file, job, DocumentType.BALANCE, status=DocumentStatus.CONFIRMED)
+    await world.form_document(file, job, DocumentType.INCOME)
+    await world.form_document(file, job, DocumentType.DISCLOSURE)
+    await world.comment(balance)
+    version = await world.version(ConsolidationStatus.DATA_CORRECTION)
+    await world.bind(balance, version)
+    old, _ = await make_stored_file(session, entity=world.entity, original_name="old.pdf")
+    old.status = FileStatus.SUPERSEDED
+    old.superseded_by_file_id = file.id
+    await session.flush()
+    file_id, s3_id, old_id, version_id = file.id, file.s3_object_id, old.id, version.id
+    before_status = version.status
+
+    code = await _run(session, "--apply")
+
+    assert code == EXIT_OK
+    assert await _count(session, StoredFile, StoredFile.id == file_id) == 0
+    assert await _count(session, Document, Document.stored_file_id == file_id) == 0
+    for model in (FinancialStatementMeta, FinancialStatementColumn, FinancialStatementLine, FinancialStatementValue):
+        assert await _count(session, model) == 0, model.__tablename__
+    assert await _count(session, Comment) == 0
+    assert await _count(session, RecognitionJob, RecognitionJob.file_id == file_id) == 0
+    old_row = await session.get(StoredFile, old_id, populate_existing=True)
+    assert old_row is not None
+    assert old_row.superseded_by_file_id is None
+    assert old_row.status is FileStatus.SUPERSEDED
+    s3 = await session.get(S3Object, s3_id, populate_existing=True)
+    assert s3 is not None and s3.expires_at is not None
+    assert await _count(session, ConsolidationDocument) == 0
+    reloaded = await session.get(ConsolidationVersion, version_id, populate_existing=True)
+    assert reloaded is not None and reloaded.status is before_status
+    [event] = (
+        await session.execute(
+            select(AuditEvent).where(AuditEvent.action == AuditAction.FILE_DELETE, AuditEvent.object_id == file_id)
+        )
+    ).scalars()
+    assert event.user_id is None
+    assert event.description == DESCRIPTION
+    assert event.before_value is not None
+    assert event.before_value["id"] == str(file_id)
+
+
+async def test_apply_mixed_keeps_osv_by_default(world: World) -> None:
+    session = world.session
+    objects = await _mixed_and_form_only(world)
+    mixed_file: StoredFile = objects["mixed_file"]  # type: ignore[assignment]
+    mixed_balance: Document = objects["mixed_balance"]  # type: ignore[assignment]
+    osv: Document = objects["osv"]  # type: ignore[assignment]
+    await world.comment(mixed_balance)
+    version = await world.version(ConsolidationStatus.DATA_CORRECTION)
+    await world.bind(mixed_balance, version)
+    file_id, osv_id, balance_id, job_id = mixed_file.id, osv.id, mixed_balance.id, objects["mixed_job"].id  # type: ignore[attr-defined]
+
+    code = await _run(session, "--apply")
+
+    assert code == EXIT_OK
+    assert await _count(session, StoredFile, StoredFile.id == file_id) == 1
+    assert await _count(session, Document, Document.id == osv_id) == 1
+    assert await _count(session, Document, Document.id == balance_id) == 0
+    assert await _count(session, RecognitionJob, RecognitionJob.id == job_id) == 1
+    assert await _count(session, Comment, Comment.object_id == balance_id) == 0
+    assert await _count(session, ConsolidationDocument, ConsolidationDocument.document_id == balance_id) == 0
+    events = (
+        (
+            await session.execute(
+                select(AuditEvent).where(AuditEvent.action == AuditAction.FILE_DELETE, AuditEvent.object_id == file_id)
+            )
+        )
+        .scalars()
+        .all()
+    )
+    [event] = events
+    assert event.user_id is None and event.description == DESCRIPTION
+    assert event.after_value is not None
+    assert event.after_value["deleted_documents"] == [str(balance_id)]
+
+
+async def test_apply_include_mixed_deletes_whole_file(world: World) -> None:
+    session = world.session
+    objects = await _mixed_and_form_only(world)
+    mixed_file: StoredFile = objects["mixed_file"]  # type: ignore[assignment]
+    file_id = mixed_file.id
+
+    code = await _run(session, "--apply", "--include-mixed")
+
+    assert code == EXIT_OK
+    assert await _count(session, StoredFile, StoredFile.id == file_id) == 0
+    assert await _count(session, Document, Document.stored_file_id == file_id) == 0
+    assert await _count(session, StoredFile) == 1  # остался только файл с ОСВ
+
+
+async def test_osv_and_directories_untouched(world: World) -> None:
+    session = world.session
+    await _mixed_and_form_only(world)
+    await make_osv_entry(session)
+    models = (OsvEntry, Consolidation, ConsolidationVersion, LegalEntity, CompanyGroup)
+    before = {model.__tablename__: await _count(session, model) for model in models}
+
+    assert await _run(session, "--apply") == EXIT_OK
+
+    assert {model.__tablename__: await _count(session, model) for model in models} == before
+    assert await _count(session, AuditEvent, AuditEvent.action == AuditAction.FILE_DELETE) == FORM_AND_MIXED_FILES
+
+
+async def test_reupload_after_purge_not_duplicate(world: World, seeded: AsyncSession) -> None:
+    session = world.session
+    file, job = await make_stored_file(session, entity=world.entity, original_name="forms.pdf", content_hash="ab" * 32)
+    await world.form_document(file, job, DocumentType.BALANCE)
+    await session.execute(update(StoredFile).where(StoredFile.id == file.id).values(status=FileStatus.PROCESSED))
+    await session.flush()
+    entity_id = world.entity.id
+    assert await _run(session, "--apply") == EXIT_OK
+
+    service = PrimaryFilesService.__new__(PrimaryFilesService)
+    service._session = session  # type: ignore[attr-defined]  # find_duplicate читает только сессию
+    assert await service.find_duplicate(entity_id, "ab" * 32) is None