diff --git a/README.md b/README.md index b82e3e8..f1927c5 100644 --- a/README.md +++ b/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)`; diff --git a/api/scripts/purge_statements.py b/api/scripts/purge_statements.py new file mode 100644 index 0000000..fade70e --- /dev/null +++ b/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:]))) diff --git a/api/src/repositories/statements.py b/api/src/repositories/statements.py index ac4debb..0153491 100644 --- a/api/src/repositories/statements.py +++ b/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() diff --git a/api/tests/test_purge_statements.py b/api/tests/test_purge_statements.py new file mode 100644 index 0000000..594c381 --- /dev/null +++ b/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