feat(mapping): автоподтверждение чистых документов в цикле маппинга (#64, OQ-157)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
b48d9e6f4d
commit
3830e718f4
@ -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}",
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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],
|
||||
):
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user