feat(#190): регистрация сырого ответа ИС Сканирование (s3_bucket_role.raw)

_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 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev 2026-11-06 13:27:02 +04:00
parent fa36602487
commit 110b2ce61d
3 changed files with 177 additions and 2 deletions

View File

@ -52,7 +52,7 @@ class _MessageBase(ContractModel):
result_s3: OptionalNotNull[S3Ref] = None # results/{file_id}/{job_id}/result.json result_s3: OptionalNotNull[S3Ref] = None # results/{file_id}/{job_id}/result.json
result_schema_version: OptionalNotNull[ResultSchemaVersion] = None # schema_version файла результата result_schema_version: OptionalNotNull[ResultSchemaVersion] = None # schema_version файла результата
error: OptionalNotNull[MessageError] = None 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") @model_validator(mode="after")
def _check_status_requirements(self) -> Self: def _check_status_requirements(self) -> Self:

View File

@ -848,11 +848,72 @@ async def _register_result_object(session: AsyncSession, message: ResultMessage,
return existing 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( async def _apply_result(
session: AsyncSession, job: RecognitionJob, message: ResultMessage, msg_id: str, now: datetime session: AsyncSession, job: RecognitionJob, message: ResultMessage, msg_id: str, now: datetime
) -> bool: ) -> bool:
"""Результат (`done | needs_check | error`): дубликат и финальный статус — без изменений (`False`); иначе — """Результат (`done | needs_check | error`): дубликат и финальный статус — без изменений (`False`); иначе —
регистрация файла результата, обновление задания и статуса файла (`True`, дальше — commit).""" регистрация файла результата и сырого ответа (если он есть, независимо от статуса — R-2), обновление задания
и статуса файла (`True`, дальше — commit)."""
if job.result_msg_id == msg_id: if job.result_msg_id == msg_id:
logger.debug( logger.debug(
"Повторная доставка результата (дубликат по result_msg_id), изменений нет. job_id={job_id}, " "Повторная доставка результата (дубликат по result_msg_id), изменений нет. job_id={job_id}, "
@ -873,6 +934,8 @@ async def _apply_result(
return False return False
result_s3_object_id = await _register_result_object(session, message, now) if message.result_s3 else None 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.status = RESULT_STATUSES[message.status]
job.service_name = message.service.name.value job.service_name = message.service.name.value
job.service_version = message.service.version job.service_version = message.service.version

View File

@ -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" 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( def envelope(
job_id: uuid.UUID, file_id: uuid.UUID, status: str, *, version: str = "1.0", **extra: Any job_id: uuid.UUID, file_id: uuid.UUID, status: str, *, version: str = "1.0", **extra: Any
) -> dict[str, 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 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) ------------- # --- статусы stored_file (Срез 2c: spec primary-files «Жизненный цикл статуса файла», D-5, R-4, R-5) -------------