diff --git a/api/src/services/result_mapping.py b/api/src/services/result_mapping.py index 1edd5d2..b890f93 100644 --- a/api/src/services/result_mapping.py +++ b/api/src/services/result_mapping.py @@ -12,7 +12,8 @@ PROCESSING` у файла, чьё ТЕКУЩЕЕ задание (`current_recogn вахтёра и публикатора, — без взаимоблокировок); файл уже не `PROCESSING` или задание уже не текущее (результат старого задания, Review Focus 1) — ничего не меняется. Иначе `map_file`: очистка документов и строк файла, документы и строки (листы ОСВ, затем формы ББ/ОФР Среза 3b и расшифровки), `group_schema` (только листы ОСВ), - предупреждения, `PROCESSED`, автозамена прежних версий (`supersede_previous`, D-10, аудит `FILE_REPLACE`). + предупреждения, `PROCESSED`, автозамена прежних версий (`supersede_previous`, D-10, аудит `FILE_REPLACE`) → + автоподтверждение чистых документов (`services/auto_confirm.py`, OQ-157) → переход версий по `NEEDS_CHECK` (D-6). Ошибки (D-2): неповторяемые (`MappingFatal`: ключ вне папки, объекта нет, JSON не проходит схему, нет документов) — сразу файл `ERROR` и `metrics.mapping_error`; прочие — откат транзакции маппинга, в отдельной транзакции @@ -895,10 +896,15 @@ async def _map_candidate( index = await CounterpartyIndex.load(session) now = clock() warnings = await map_file(session, job=job, file=file, result=result, chart=chart, index=index, now=now) - # Переход версий (D-6) — ПОСЛЕ `map_file`, а значит после автозамены (`supersede_previous` берёт advisory ЮЛ → - # старые файлы → advisory ГК → версии): версии блокируются последними, порядок «задание → файл → документы → - # версии» (D-9) не нарушается. До автозамены хук держал бы версии и затем ждал блокировки старых файлов. + # После `map_file`, а значит после автозамены (advisory ЮЛ → старые файлы → advisory ГК → версии): версии + # блокируются последними, порядок «задание → файл → документы → версии» (D-9) не нарушается. Сначала + # автоподтверждение чистых документов (привязка и переход версии, OQ-157, design D-1), затем переход по + # `NEEDS_CHECK` (D-6) — он уже видит версию после автоподтверждения и второго перехода не делает. documents = list((await session.execute(select(Document).where(Document.stored_file_id == file.id))).scalars()) + # отложенный импорт: цикл documents → result_mapping → auto_confirm → statements → documents + from src.services.auto_confirm import auto_confirm # noqa: PLC0415 + + await auto_confirm(session, file=file, documents=documents, now=now) await consolidation_documents.on_documents_mapped(session, file=file, documents=documents, now=now) logger.info( "Результат распознавания смапплен: файл PROCESSED. job_id={job_id}, file_id={file_id}, warnings={warnings}", diff --git a/api/tests/e2e/test_osv_ingest_e2e.py b/api/tests/e2e/test_osv_ingest_e2e.py index 170df42..e28a4ec 100644 --- a/api/tests/e2e/test_osv_ingest_e2e.py +++ b/api/tests/e2e/test_osv_ingest_e2e.py @@ -141,13 +141,13 @@ async def test_real_osv_is_parsed_end_to_end( assert (brief["period_year"], brief["period_quarter"], brief["is_cumulative"]) == (2024, "Q4", True) # MATCH достигается именно по имени: ИНН в шапке ОСВ нет (D-8), имя ЮЛ совпадает с шапкой assert brief["counterparty_check"] == "MATCH" - assert brief["status"] == "RECOGNIZED" + assert brief["status"] == "CONFIRMED" # чистая ОСВ подтверждена автоматически (OQ-157) document = (await client.get(f"/documents/{brief['id']}")).json() assert document["file_id"] == str(file_id) assert document["legal_entity_id"] == str(entity.id) assert (document["period_year"], document["period_quarter"]) == (2024, "Q4") - assert (document["counterparty_check"], document["status"]) == ("MATCH", "RECOGNIZED") + assert (document["counterparty_check"], document["status"]) == ("MATCH", "CONFIRMED") entries = (await client.get(f"/documents/{brief['id']}/entries")).json() assert entries["total"] == ENTRY_ROWS diff --git a/api/tests/test_result_mapping.py b/api/tests/test_result_mapping.py index ae199c5..84d28ed 100644 --- a/api/tests/test_result_mapping.py +++ b/api/tests/test_result_mapping.py @@ -49,6 +49,7 @@ from src.models.enums import ( from src.models.files import Document, RecognitionJob, StoredFile from src.models.osv import OsvEntry from src.repositories.counterparties import CounterpartyIndex +from src.services import consolidation_documents from src.services import result_mapping as rm from src.services.recognition_exchange import expire_deadlines, handle_message from src.services.result_mapping import ( @@ -236,7 +237,7 @@ async def test_golden_osv_66_mapped( # noqa: PLR0915 - один сценари assert document.is_cumulative is True assert document.locator == "sheet:Лист_1" assert document.legal_entity_id == entity_id - assert document.status is DocumentStatus.RECOGNIZED # done, проверки чистые, шапка = владелец по имени + assert document.status is DocumentStatus.CONFIRMED # done, чистые проверки, шапка = владелец → автоподтверждение assert document.counterparty_check is DocumentCheckResult.MATCH assert document.period_check is DocumentCheckResult.UNKNOWN @@ -617,19 +618,20 @@ async def test_cumulative_by_q1_data( assert document.status is DocumentStatus.NEEDS_CHECK else: assert warnings == [] - assert document.status is DocumentStatus.RECOGNIZED + assert document.status is DocumentStatus.CONFIRMED # чистый → автоподтверждение (OQ-157) @pytest.mark.parametrize( ("job_status", "expected"), - [(RecognitionStatus.done, DocumentStatus.RECOGNIZED), (RecognitionStatus.needs_check, DocumentStatus.NEEDS_CHECK)], + [(RecognitionStatus.done, DocumentStatus.CONFIRMED), (RecognitionStatus.needs_check, DocumentStatus.NEEDS_CHECK)], ) async def test_clean_result( committed_sessionmaker: async_sessionmaker[AsyncSession], job_status: RecognitionStatus, expected: DocumentStatus, ): - """«Чистый результат»: задание `done`, шапка = владелец (ИНН), нарушений нет → `RECOGNIZED`, `MATCH`, `UNKNOWN`; + """«Чистый результат»: задание `done`, шапка = владелец (ИНН), нарушений нет → `CONFIRMED` (автоподтверждение, + OQ-157), `MATCH`, `UNKNOWN`; задание `needs_check` → `NEEDS_CHECK`.""" maker, storage = committed_sessionmaker, FakeStorage() file_id, _, _ = await _prepare(maker, storage, _minimal(), owner_inn="7700000000", job_status=job_status) @@ -1409,13 +1411,14 @@ async def _load_entity(maker: async_sessionmaker[AsyncSession], entity_id: uuid. @pytest.mark.parametrize( ("job_status", "moved"), - [(RecognitionStatus.needs_check, True), (RecognitionStatus.done, False)], # NEEDS_CHECK / RECOGNIZED (чистый) + [(RecognitionStatus.needs_check, True), (RecognitionStatus.done, False)], # NEEDS_CHECK / CONFIRMED (чистый) ) async def test_mapping_moves_documents_processing_version_only_for_needs_check( committed_sessionmaker: async_sessionmaker[AsyncSession], job_status: RecognitionStatus, moved: bool ): """«Предупреждение маппинга двигает статус» и «Документ RECOGNIZED»: ОСВ 60 за 2025 Q1 (квартальная) подходит версии - 2025 Q3; `NEEDS_CHECK` → `DATA_CORRECTION`, запись истории с `changed_by` = загрузивший; привязок нет.""" + 2025 Q3; `NEEDS_CHECK` → `DATA_CORRECTION`, запись истории с `changed_by` = загрузивший, привязок нет; чистый + документ → `CONFIRMED`, привязка и тот же переход (OQ-157).""" maker, storage = committed_sessionmaker, FakeStorage() entity_id, version_id, user_id = await _version_in(maker) file_id, _, _ = await _prepare( @@ -1430,22 +1433,14 @@ async def test_mapping_moves_documents_processing_version_only_for_needs_check( assert await process_once(maker, storage, _settings()) is True [document] = await _documents(maker, file_id) - assert document.status is (DocumentStatus.NEEDS_CHECK if moved else DocumentStatus.RECOGNIZED) + assert document.status is (DocumentStatus.NEEDS_CHECK if moved else DocumentStatus.CONFIRMED) async with maker() as session: version = await session.get(ConsolidationVersion, version_id) assert version is not None - history = list( - ( - await session.execute( - select(ConsolidationStatusHistory) - .where(ConsolidationStatusHistory.version_id == version_id) - .order_by(ConsolidationStatusHistory.changed_at) - ) - ).scalars() - ) + history = await _history_of(maker, version_id) bindings = (await session.execute(select(func.count()).select_from(ConsolidationDocument))).scalar_one() - assert bindings == 0 if moved: + assert bindings == 0 assert version.status is ConsolidationStatus.DATA_CORRECTION last = history[-1] assert (last.from_status, last.to_status, last.changed_by) == ( @@ -1454,8 +1449,16 @@ async def test_mapping_moves_documents_processing_version_only_for_needs_check( user_id, ) else: - assert version.status is ConsolidationStatus.DOCUMENTS_PROCESSING - assert len(history) == FACTORY_HISTORY # создание и переход фабрики в DOCUMENTS_PROCESSING; новых записей нет + # чистый документ подтверждён системой (OQ-157): привязан, версия перешла от загрузившего с комментарием + assert bindings == 1 + assert version.status is ConsolidationStatus.DATA_CORRECTION + last = history[-1] + assert (last.from_status, last.to_status, last.changed_by, last.comment) == ( + ConsolidationStatus.DOCUMENTS_PROCESSING, + ConsolidationStatus.DATA_CORRECTION, + user_id, + "автоподтверждение документа OSV 60", + ) async def test_mapping_does_not_move_initiated_version(committed_sessionmaker: async_sessionmaker[AsyncSession]): @@ -1479,6 +1482,142 @@ async def test_mapping_does_not_move_initiated_version(committed_sessionmaker: a assert version.status is ConsolidationStatus.INITIATED +async def _history_of( + maker: async_sessionmaker[AsyncSession], version_id: uuid.UUID +) -> list[ConsolidationStatusHistory]: + async with maker() as session: + stmt = ( + select(ConsolidationStatusHistory) + .where(ConsolidationStatusHistory.version_id == version_id) + .order_by(ConsolidationStatusHistory.changed_at) + ) + return list((await session.execute(stmt)).scalars()) + + +async def _bound_ids(maker: async_sessionmaker[AsyncSession], version_id: uuid.UUID) -> set[uuid.UUID]: + async with maker() as session: + stmt = select(ConsolidationDocument.document_id).where( + ConsolidationDocument.consolidation_version_id == version_id + ) + return set((await session.execute(stmt)).scalars()) + + +async def test_clean_document_is_auto_confirmed_bound_and_moves_version( + committed_sessionmaker: async_sessionmaker[AsyncSession], +): + """osv-result-mapping «Привязка и переход версии»: чистая ОСВ 60 за 2025 Q1 → `CONFIRMED` системой, привязана к + версии 2025 Q3 (`DOCUMENTS_PROCESSING`), переход в `DATA_CORRECTION` от загрузившего с комментарием.""" + maker, storage = committed_sessionmaker, FakeStorage() + entity_id, version_id, user_id = await _version_in(maker) + file_id, _, _ = await _prepare( + maker, storage, _minimal(), entity=await _load_entity(maker, entity_id), uploaded_by=user_id + ) + + assert await process_once(maker, storage, _settings()) is True + + [document] = await _documents(maker, file_id) + assert (document.status, document.confirmed_by) == (DocumentStatus.CONFIRMED, None) + assert document.confirmed_at is not None + assert await _bound_ids(maker, version_id) == {document.id} + last = (await _history_of(maker, version_id))[-1] + assert (last.to_status, last.changed_by, last.comment) == ( + ConsolidationStatus.DATA_CORRECTION, + user_id, + "автоподтверждение документа OSV 60", + ) + async with maker() as session: + events = ( + ( + await session.execute( + select(AuditEvent).where( + AuditEvent.action == AuditAction.DOCUMENT_CONFIRM, AuditEvent.object_id == document.id + ) + ) + ) + .scalars() + .all() + ) + assert [(event.user_id, event.description) for event in events] == [(None, "автоподтверждение: без замечаний")] + + +async def test_mixed_file_confirms_clean_and_moves_version_once( + committed_sessionmaker: async_sessionmaker[AsyncSession], +): + """Review Focus 1: в файле чистый и `NEEDS_CHECK` документы — версия получает ровно один переход.""" + maker, storage = committed_sessionmaker, FakeStorage() + entity_id, version_id, user_id = await _version_in(maker) + result = _minimal() + second = copy.deepcopy(result["documents"][0]) + second["header"]["account_code_raw"] = "62" + second["locator"] = "sheet:Лист_2" + second["checks"] = {"total_row_matches_root": False} # нарушение проверок → NEEDS_CHECK + result["documents"].append(second) + file_id, _, _ = await _prepare( + maker, storage, result, entity=await _load_entity(maker, entity_id), uploaded_by=user_id + ) + + assert await process_once(maker, storage, _settings()) is True + + statuses = sorted((d.osv_account_code, d.status) for d in await _documents(maker, file_id)) + assert statuses == [("60", DocumentStatus.CONFIRMED), ("62", DocumentStatus.NEEDS_CHECK)] + history = await _history_of(maker, version_id) + assert [h.to_status for h in history].count(ConsolidationStatus.DATA_CORRECTION) == 1 + + +async def test_supersede_rebinds_auto_confirmed_document(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """Review Focus 2: новая чистая версия файла того же ключа заменяет привязку старой.""" + maker, storage = committed_sessionmaker, FakeStorage() + entity_id, version_id, user_id = await _version_in(maker) + entity = await _load_entity(maker, entity_id) + first_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity, uploaded_by=user_id) + assert await process_once(maker, storage, _settings()) is True + [first] = await _documents(maker, first_id) + assert await _bound_ids(maker, version_id) == {first.id} + + second_id, _, _ = await _prepare(maker, storage, _minimal(), entity=entity, uploaded_by=user_id) + assert await process_once(maker, storage, _settings()) is True + + [second] = await _documents(maker, second_id) + assert second.status is DocumentStatus.CONFIRMED + assert await _bound_ids(maker, version_id) == {second.id} + assert (await _file(maker, first_id)).status is FileStatus.SUPERSEDED + + +async def test_auto_confirm_rolls_back_with_mapping( + committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch +): + """Review Focus 3: сбой после автоподтверждения откатывает и его; повтор — одно событие `DOCUMENT_CONFIRM`.""" + maker, storage = committed_sessionmaker, FakeStorage() + entity_id, _, user_id = await _version_in(maker) + file_id, job_id, _ = await _prepare( + maker, storage, _minimal(), entity=await _load_entity(maker, entity_id), uploaded_by=user_id + ) + original = consolidation_documents.on_documents_mapped + calls = {"n": 0} + + async def flaky(*args: Any, **kwargs: Any) -> list[uuid.UUID]: + calls["n"] += 1 + if calls["n"] == 1: + raise RuntimeError("сбой после автоподтверждения") + return await original(*args, **kwargs) + + monkeypatch.setattr(consolidation_documents, "on_documents_mapped", flaky) + + assert await process_once(maker, storage, _settings()) is False # повторяемая неудача (R-15) + assert await _documents(maker, file_id) == [] # маппинг вместе с автоподтверждением откатился целиком + assert await process_once(maker, storage, _settings()) is True + + [document] = await _documents(maker, file_id) + assert document.status is DocumentStatus.CONFIRMED + async with maker() as session: + count = ( + await session.execute( + select(func.count()).select_from(AuditEvent).where(AuditEvent.action == AuditAction.DOCUMENT_CONFIRM) + ) + ).scalar_one() + assert count == 1 + + async def test_mapping_without_uploader_skips_transition_without_error( committed_sessionmaker: async_sessionmaker[AsyncSession], ):