From 110b2ce61d482f8a3e8ec026058278f382bb0196 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Fri, 6 Nov 2026 13:27:02 +0400 Subject: [PATCH] =?UTF-8?q?feat(#190):=20=D1=80=D0=B5=D0=B3=D0=B8=D1=81?= =?UTF-8?q?=D1=82=D1=80=D0=B0=D1=86=D0=B8=D1=8F=20=D1=81=D1=8B=D1=80=D0=BE?= =?UTF-8?q?=D0=B3=D0=BE=20=D0=BE=D1=82=D0=B2=D0=B5=D1=82=D0=B0=20=D0=98?= =?UTF-8?q?=D0=A1=20=D0=A1=D0=BA=D0=B0=D0=BD=D0=B8=D1=80=D0=BE=D0=B2=D0=B0?= =?UTF-8?q?=D0=BD=D0=B8=D0=B5=20(s3=5Fbucket=5Frole.raw)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _register_raw_object (по образцу _register_result_object) регистрирует raw_s3_key в s3_object с bucket_role=raw; бакет берётся у файла задания (stored_file -> s3_object), не у message.result_s3 (отсутствует при status=error) — именно error чаще всего и даёт сырой ответ без результата. Регистрация не зависит от message.status (R-2). Поле RecognitionJob.raw_s3_object_id уже существовало (Task 1) — решение v5.13 «не регистрируется» отменено v5.25 (OQ-179), комментарий contract/message.py обновлён. Тесты — tests/test_recognition_inbox.py (там же тестируются _apply_result и _register_result_object через handle_message, а не в test_recognition_exchange.py, который покрывает только outbox/вахтёр). Co-Authored-By: Claude Sonnet 5 --- api/src/contract/message.py | 2 +- api/src/services/recognition_exchange.py | 65 ++++++++++++- api/tests/test_recognition_inbox.py | 112 +++++++++++++++++++++++ 3 files changed, 177 insertions(+), 2 deletions(-) diff --git a/api/src/contract/message.py b/api/src/contract/message.py index a371cab..aaef408 100644 --- a/api/src/contract/message.py +++ b/api/src/contract/message.py @@ -52,7 +52,7 @@ class _MessageBase(ContractModel): result_s3: OptionalNotNull[S3Ref] = None # results/{file_id}/{job_id}/result.json result_schema_version: OptionalNotNull[ResultSchemaVersion] = None # schema_version файла результата error: OptionalNotNull[MessageError] = None - raw_s3_key: str | None = None # scan-gateway: raw/{job_id}.json; в БД не регистрируется + raw_s3_key: str | None = None # scan-gateway: raw/{job_id}.json @model_validator(mode="after") def _check_status_requirements(self) -> Self: diff --git a/api/src/services/recognition_exchange.py b/api/src/services/recognition_exchange.py index dbf4a24..325b1cb 100644 --- a/api/src/services/recognition_exchange.py +++ b/api/src/services/recognition_exchange.py @@ -848,11 +848,72 @@ async def _register_result_object(session: AsyncSession, message: ResultMessage, return existing +async def _register_raw_object( + session: AsyncSession, job: RecognitionJob, message: ResultMessage, now: datetime +) -> uuid.UUID | None: + """Сырой ответ ИС Сканирование в реестре `s3_object` (`bucket_role = raw`, `created_by = service.name`) — + v5.25: решение «не регистрировать» (OQ-179) отменено, сырой ответ нужен для эталонного прогона переразметки. + + Регистрируется независимо от `message.status` (R-2): сырой ответ полезен и у `error` — именно там + `message.result_s3` обычно отсутствует. Бакет — у ТОГО ЖЕ файла, что у задания (`stored_file` → `s3_object`), + не `message.result_s3` (может быть `None` при `status = error`). Если объект файла задания почему-то не + найден в реестре (не должно случаться в штатной работе), регистрация сырого ответа тихо пропускается — не + прерывает приём результата, только warning (тот же защитный стиль, что у `_failure_dimensions`). + + Уникальность `(bucket, key)` — как у `_register_result_object`: конфликт — переиспользуем существующую строку. + """ + assert message.raw_s3_key is not None + bucket = ( + await session.execute( + select(S3Object.bucket) + .join(StoredFile, StoredFile.s3_object_id == S3Object.id) + .where(StoredFile.id == job.file_id) + ) + ).scalar_one_or_none() + if bucket is None: + logger.warning( + "Сырой ответ ИС Сканирование не зарегистрирован: объект файла задания не найден в s3_object. " + "job_id={job_id}, file_id={file_id}, raw_s3_key={raw_s3_key}", + job_id=job.id, + file_id=job.file_id, + raw_s3_key=message.raw_s3_key, + ) + return None + key = message.raw_s3_key + stmt = ( + pg_insert(S3Object) + .values( + bucket_role=S3BucketRole.raw, + bucket=bucket, + key=key, + created_by=message.service.name.value, + created_at=now, + ) + .on_conflict_do_nothing(index_elements=[S3Object.bucket, S3Object.key]) + .returning(S3Object.id) + ) + inserted = (await session.execute(stmt)).scalar_one_or_none() + if inserted is not None: + return inserted + existing = ( + await session.execute(select(S3Object.id).where(S3Object.bucket == bucket, S3Object.key == key)) + ).scalar_one() + logger.warning( + "Сырой ответ ИС Сканирование уже зарегистрирован в s3_object, строка переиспользована. job_id={job_id}, " + "bucket={bucket}, key={key}", + job_id=job.id, + bucket=bucket, + key=key, + ) + return existing + + async def _apply_result( session: AsyncSession, job: RecognitionJob, message: ResultMessage, msg_id: str, now: datetime ) -> bool: """Результат (`done | needs_check | error`): дубликат и финальный статус — без изменений (`False`); иначе — - регистрация файла результата, обновление задания и статуса файла (`True`, дальше — commit).""" + регистрация файла результата и сырого ответа (если он есть, независимо от статуса — R-2), обновление задания + и статуса файла (`True`, дальше — commit).""" if job.result_msg_id == msg_id: logger.debug( "Повторная доставка результата (дубликат по result_msg_id), изменений нет. job_id={job_id}, " @@ -873,6 +934,8 @@ async def _apply_result( return False result_s3_object_id = await _register_result_object(session, message, now) if message.result_s3 else None + if message.raw_s3_key is not None: + job.raw_s3_object_id = await _register_raw_object(session, job, message, now) job.status = RESULT_STATUSES[message.status] job.service_name = message.service.name.value job.service_version = message.service.version diff --git a/api/tests/test_recognition_inbox.py b/api/tests/test_recognition_inbox.py index 65037bb..cfa321d 100644 --- a/api/tests/test_recognition_inbox.py +++ b/api/tests/test_recognition_inbox.py @@ -91,6 +91,10 @@ def _result_key(file_id: uuid.UUID, job_id: uuid.UUID) -> str: return f"results/{file_id}/{job_id}/result.json" +def _raw_key(job_id: uuid.UUID) -> str: + return f"raw/{job_id}.json" + + def envelope( job_id: uuid.UUID, file_id: uuid.UUID, status: str, *, version: str = "1.0", **extra: Any ) -> dict[str, Any]: @@ -731,6 +735,114 @@ async def test_existing_s3_object_is_reused(session_factory, jobs: CommittedJobs assert len(await _s3_rows(session_factory, _result_key(file_id, job_id))) == 1 +# --- сырой ответ ИС Сканирование (s3_bucket_role.raw, v5.25, задача 11, R-2) -------------------------------------- + + +@pytest.mark.integration +async def test_done_result_with_raw_s3_key_registers_raw_object(session_factory, jobs: CommittedJobs): + """Сценарий «Успешный результат» с `raw_s3_key`: сырой ответ регистрируется отдельной строкой `s3_object` + (`bucket_role = raw`), на ТОМ ЖЕ бакете, что у загрузки файла, `recognition_job.raw_s3_object_id` — на неё.""" + job_id = await jobs.make_queued_job(status=RecognitionStatus.processing) + file_id = await _file_id(jobs, job_id) + body = envelope(job_id, file_id, "done", raw_s3_key=_raw_key(job_id)) + + await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1))) + + job = await _row(session_factory, job_id) + assert job["raw_s3_object_id"] is not None + assert job["raw_s3_object_id"] != job["result_s3_object_id"] + (raw,) = await _s3_rows(session_factory, _raw_key(job_id)) + assert raw["id"] == job["raw_s3_object_id"] + assert raw["bucket_role"] is S3BucketRole.raw + assert raw["bucket"] == BUCKET # тот же бакет, что у stored_file файла — не из message.result_s3 + assert raw["created_by"] == "tabular-parser" + + +@pytest.mark.integration +async def test_result_without_raw_s3_key_leaves_raw_object_id_none(session_factory, jobs: CommittedJobs): + """Результат без `raw_s3_key` (например, Excel через tabular-parser) не создаёт raw-объект и не трогает + `raw_s3_object_id`.""" + job_id = await jobs.make_queued_job(status=RecognitionStatus.processing) + file_id = await _file_id(jobs, job_id) + total_before = await _s3_total(session_factory) + + await _handle(session_factory, fake_msg(envelope(job_id, file_id, "done"), result_msg_id(job_id, 1))) + + job = await _row(session_factory, job_id) + assert job["raw_s3_object_id"] is None + assert await _s3_total(session_factory) == total_before + 1 # только результат, без raw + + +@pytest.mark.integration +async def test_service_error_result_with_raw_s3_key_registers_raw_object(session_factory, jobs: CommittedJobs): + """Ruling R-2: регистрация сырого ответа не зависит от `message.status` — при `status=error` `result_s3` + обычно отсутствует (sites §14.5), но сырой ответ сервиса всё равно регистрируется; бакет берётся у файла + задания (`stored_file` → `s3_object`), не у `message.result_s3` (его и нет).""" + job_id = await jobs.make_queued_job(status=RecognitionStatus.processing) + file_id = await _file_id(jobs, job_id) + body = envelope( + job_id, file_id, "error", error={"code": "NOT_IMPLEMENTED", "message": "x"}, raw_s3_key=_raw_key(job_id) + ) + + await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1))) + + job = await _row(session_factory, job_id) + assert job["result_s3_object_id"] is None # error: message.result_s3 отсутствует + assert job["raw_s3_object_id"] is not None + (raw,) = await _s3_rows(session_factory, _raw_key(job_id)) + assert raw["id"] == job["raw_s3_object_id"] + assert raw["bucket_role"] is S3BucketRole.raw + assert raw["bucket"] == BUCKET + + +@pytest.mark.integration +async def test_existing_raw_object_is_reused(session_factory, jobs: CommittedJobs): + """Ключ сырого ответа уже зарегистрирован — строка переиспользуется, дубль (bucket, key) не вставляется (тот + же паттерн идемпотентности, что у `_register_result_object`/`test_existing_s3_object_is_reused`).""" + job_id = await jobs.make_queued_job(status=RecognitionStatus.processing) + file_id = await _file_id(jobs, job_id) + async with session_factory() as session, session.begin(): + existing = S3Object( + bucket_role=S3BucketRole.raw, + bucket=BUCKET, + key=_raw_key(job_id), + created_by="tabular-parser", + created_at=datetime.now(UTC), + ) + session.add(existing) + await session.flush() + existing_id = existing.id + jobs.track_s3_object(existing_id) + + body = envelope(job_id, file_id, "done", raw_s3_key=_raw_key(job_id)) + await _handle(session_factory, fake_msg(body, result_msg_id(job_id, 1))) + + assert (await _row(session_factory, job_id))["raw_s3_object_id"] == existing_id + assert len(await _s3_rows(session_factory, _raw_key(job_id))) == 1 + + +@pytest.mark.integration +async def test_register_raw_object_without_stored_file_row_is_skipped(session_factory, logs: list[Any]): + """Защитное поведение `_register_raw_object` (не должно случаться в штатной работе: `stored_file.s3_object_id` + NOT NULL FK — у файла задания всегда есть строка в `s3_object`): если бакет всё же не находится (здесь — + `job.file_id` не ссылается ни на одну строку `stored_file`), регистрация тихо пропускается — `None` и warning, + без исключения внутрь `_apply_result` (транзакция результата не должна падать из-за этого).""" + job_id = uuid.uuid4() + file_id = uuid.uuid4() # не существует в stored_file + body = envelope( + job_id, file_id, "error", error={"code": "NOT_IMPLEMENTED", "message": "x"}, raw_s3_key=_raw_key(job_id) + ) + message = rx._parse_message(fake_msg(body, result_msg_id(job_id, 1))) + assert message is not None + job = RecognitionJob(id=job_id, file_id=file_id) + + async with session_factory() as session: + result = await rx._register_raw_object(session, job, message, datetime.now(UTC)) + + assert result is None + assert len(_logged(logs, "не зарегистрирован", "WARNING")) == 1 + + # --- статусы stored_file (Срез 2c: spec primary-files «Жизненный цикл статуса файла», D-5, R-4, R-5) -------------