diff --git a/CLAUDE.md b/CLAUDE.md index c7025d0..431bd42 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -8,7 +8,7 @@ публикуется **без замены**), забрать **все** `Исходящие//*.json` (ИС кладёт по одному json на каждый распознанный документ), склеить и сохранить сырой ответ в S3 `raw/{job_id}.json` (в БД не регистрируется), очистить горячую папку (только при `SCAN_CLEANUP=true`), нормализовать в `../docs/contracts/recognition-result.schema.json` -(`statementDocument` v1.2: по одному документу на таблицу `table_structure[i]`), опубликовать `recognition.result` +(`statementDocument` v1.2: по одному документу на таблицу `table_structure[i]`), опубликовать `files.tabular_parsed` (`done` / `needs_check` / `error`). Таймаут → ошибка `TIMEOUT` по `min(reply_deadline, старт + SCAN_TIMEOUT_SECONDS=300)`, старт — запись файла во «Входящие». **Нет доступа к БД.** diff --git a/README.md b/README.md index 50ff796..997130a 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,7 @@ Адаптер внешней ИС Сканирование (Content AI) для АРМ КА. Принимает из NATS запросы на распознавание изображений (PDF/PNG/JPG), передаёт файл в ИС через «горячие папки», забирает ответ, сохраняет сырой ответ в S3, нормализует его -и публикует результат в `recognition.result`. Контракт обмена — `../docs/contracts/` (v1.2: `statementDocument`; +и публикует результат в `files.tabular_parsed`. Контракт обмена — `../docs/contracts/` (v1.2: `statementDocument`; `schema_version` в конвертах и файле результата сервис по-прежнему пишет `1.0` — `CONTRACT_VERSION`, решение №14 change'а `slice-3a-ocr-normalizer`: поля 1.1–1.2 документы PDF-пути не используют), архитектура — `../docs/context/03-architecture.md`. @@ -23,9 +23,9 @@ change'а `slice-3a-ocr-normalizer`: поля 1.1–1.2 документы PDF- ## Поток обработки ``` -recognition.image.request (NATS, queue group scan-gateway) +files.img_uploaded (NATS, queue group scan-gateway) 1. разбор и проверка запроса ──нет/не по схеме──► error SCHEMA_MISMATCH (без ack) - 2. ack: recognition.result, status=processing (до чтения S3 и до ожидания слота) + 2. ack: files.tabular_parsed, status=processing (до чтения S3 и до ожидания слота) 3. слот (SCAN_MAX_INFLIGHT одновременных заданий) 4. файл из S3 (s3.bucket/s3.key) 5. горячая папка: /. (через /...part, без замены) @@ -34,7 +34,7 @@ recognition.image.request (NATS, queue group scan-gateway) 7. склейка всех *.json каталога ──► S3 raw/{job_id}.json (до нормализации) 8. очистка горячей папки (только при SCAN_CLEANUP=true) 9. нормализатор (OcrNormalizer) ──► (done/needs_check) S3 results/{file_id}/{job_id}/result.json - 10. recognition.result: итоговый конверт (done | needs_check | error), в нём raw_s3_key + 10. files.tabular_parsed: итоговый конверт (done | needs_check | error), в нём raw_s3_key ``` Каждое задание заканчивается опубликованным конвертом: сбой превращается в `error` с кодом (таблица ниже), а не в @@ -370,8 +370,8 @@ MinIO и NATS); те же значения подхватываются из `EN Схемы и примеры — `../docs/contracts/`; их байт-в-байт копии лежат в `api/contracts/` (идентичность проверяет unit-тест `api/tests/unit/test_contract.py`, входит в `make check`), pydantic-модели — `api/src/contract/`. -- Запрос: subject `recognition.image.request`, подписка в queue group `scan-gateway`, `Nats-Msg-Id = :`. -- Ответы — subject `recognition.result`: ack `status=processing`, `Nats-Msg-Id = ack::`; итог +- Запрос: subject `files.img_uploaded`, подписка в queue group `scan-gateway`, `Nats-Msg-Id = :`. +- Ответы — subject `files.tabular_parsed`: ack `status=processing`, `Nats-Msg-Id = ack::`; итог `status = done | needs_check | error`, `Nats-Msg-Id = result::`. `service.name = "scan-gateway"`. - Ключи S3: исходник — из запроса (`s3.key`), сырой ответ (склейка файлов ИС) — `raw/{job_id}.json`, результат — `results/{file_id}/{job_id}/result.json`. Бакет один. @@ -410,8 +410,8 @@ curl -s localhost:8000/healthcheck PDF из `../fixtures/ocr` (`aep_3.pdf` -> набор `aep_3.json` + `aep_3_ofr.json`, `aep_6.pdf` -> `aep_6.json`, `aep_5.PDF` -> `aep_5.json`); любое другое содержимое даёт `INVALID_DOCUMENT`, если не задан `SCAN_EMULATOR_DEFAULT`. `original_name` в запросе определяет только расширение входящего файла. -3. Подписаться на `recognition.result` любым клиентом NATS. -4. Опубликовать в `recognition.image.request` тело по образцу `api/contracts/examples/request-image.json` +3. Подписаться на `files.tabular_parsed` любым клиентом NATS. +4. Опубликовать в `files.img_uploaded` тело по образцу `api/contracts/examples/request-image.json` (свои `job_id` и `file_id` — UUID, `original_name`, `s3.key`, `reply_deadline` в будущем, `attempt` = 1) с заголовком `Nats-Msg-Id: :1`. 5. Ожидать: ack `processing`, через `SCAN_EMULATOR_DELAY` (+ до `SCAN_SETTLE_SECONDS + SCAN_POLL_SECONDS` после первого опроса, увидевшего набор) — diff --git a/api/.env.example b/api/.env.example index 10d57bb..9005d7d 100644 --- a/api/.env.example +++ b/api/.env.example @@ -27,7 +27,7 @@ OPENBAO__NATS__NATS_URL=nats://localhost:4222 OPENBAO__NATS__NATS_SEED= OPENBAO__NATS__NATS_JWT= -# Включает подписку на recognition.image.request. false — воркер не запускается (сервис поднимется без NATS). +# Включает подписку на files.img_uploaded. false — воркер не запускается (сервис поднимется без NATS). NATS_ENABLED=true # PUBLISH_RETRIES — число повторов публикации ack/результата в NATS после первой попытки (при 3 — до 4 попыток всего); # DONE_CACHE_SIZE — размер LRU обработанных job_id (защита от повторной доставки запроса) diff --git a/api/contracts/recognition-request.schema.json b/api/contracts/recognition-request.schema.json index 780573c..9d59f2c 100644 --- a/api/contracts/recognition-request.schema.json +++ b/api/contracts/recognition-request.schema.json @@ -2,7 +2,7 @@ "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "arm-ka/recognition-request/v1", "title": "Запрос распознавания файла (arm-core → tabular-parser / scan-gateway), контракт v1.0", - "description": "Тело сообщения на subject recognition.tabular.request или recognition.image.request; subject должен соответствовать полю pipeline (tabular → recognition.tabular.request, image → recognition.image.request). Заголовок Nats-Msg-Id = : (не поле тела). Один файл (stored_file) → один job.", + "description": "Тело сообщения на subject files.tabular_uploaded или files.img_uploaded; subject должен соответствовать полю pipeline (tabular → files.tabular_uploaded, image → files.img_uploaded). Заголовок Nats-Msg-Id = : (не поле тела). Один файл (stored_file) → один job.", "$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.", "type": "object", "required": ["schema_version", "job_id", "file_id", "pipeline", "file_type", "original_name", "s3", "reply_deadline", "attempt"], diff --git a/api/contracts/recognition-result-message.schema.json b/api/contracts/recognition-result-message.schema.json index 1f20727..7c1cfb5 100644 --- a/api/contracts/recognition-result-message.schema.json +++ b/api/contracts/recognition-result-message.schema.json @@ -2,7 +2,7 @@ "$schema": "https://json-schema.org/draft/2020-12/schema", "$id": "arm-ka/recognition-result-message/v1", "title": "Сообщение результата распознавания (tabular-parser / scan-gateway → arm-core), контракт v1.0", - "description": "Тело сообщения на subject recognition.result. Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:: — для status = processing; result:: — для done, needs_check и error. Форма запроса : описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).", + "description": "Тело сообщения на subject files.tabular_parsed. Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:: — для status = processing; result:: — для done, needs_check и error. Форма запроса : описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).", "$comment": "Версионирование: schema_version = MAJOR.MINOR. MINOR (1.1) — необязательные поля и новые значения enum; MAJOR (2.0) — удаление/переименование полей, смена типов или семантики. Потребитель принимает тот же MAJOR и любой MINOR. Новый MINOR раскатывается сначала на потребителей (ядро и сервисы обновляют копии схем раньше, чем производитель начнёт слать 1.1), потому что закрытые enum отвергают незнакомые значения.", "type": "object", "required": ["schema_version", "job_id", "file_id", "status", "service"], diff --git a/api/src/app.py b/api/src/app.py index 808738c..59f13c2 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -5,7 +5,7 @@ - lifespan: регистрация SMB-сессии (только `SCAN_HOTFOLDER_MODE=smb`), S3 (`S3Storage`), NATS (`NatsGateway`), воркер (`Worker` с `OcrNormalizer`). Справочники нормализатора загружаются из `SCAN_RULES_DIR` в `create_app`: битый справочник — `ValueError`, сервис не стартует. Подключение к NATS блокирует до доступности сервера, поэтому идёт - фоновой задачей: HTTP отвечает сразу (`nats = down`), подписка `recognition.image.request` появляется после + фоновой задачей: HTTP отвечает сразу (`nats = down`), подписка `files.img_uploaded` появляется после подключения. Сбой этой задачи (например, `NatsConfigError`) — лог error с трассировкой и `nats = down`, HTTP-сервер жив. Остановка: отмена подключения, остановка воркера (задания в работе отменяются), закрытие NATS и S3. diff --git a/api/src/contract/message.py b/api/src/contract/message.py index 18b1303..7ea62fa 100644 --- a/api/src/contract/message.py +++ b/api/src/contract/message.py @@ -1,4 +1,4 @@ -"""Конверт сообщения на `recognition.result`: `recognition-result-message.schema.json` v1.0. +"""Конверт сообщения на `files.tabular_parsed`: `recognition-result-message.schema.json` v1.0. `processing` — ack «взял в работу» (до чтения файла из S3); `done | needs_check | error` — итог. Условная обязательность — буквально из `allOf` схемы: для `done | needs_check` обязательны `result_s3` и diff --git a/api/src/contract/subjects.py b/api/src/contract/subjects.py index be58d01..243b11d 100644 --- a/api/src/contract/subjects.py +++ b/api/src/contract/subjects.py @@ -12,9 +12,9 @@ from typing import Literal, get_args Pipeline = Literal["tabular", "image"] NATS_MSG_ID_HEADER = "Nats-Msg-Id" -RESULT_SUBJECT = "recognition.result" -TABULAR_REQUEST_SUBJECT = "recognition.tabular.request" -IMAGE_REQUEST_SUBJECT = "recognition.image.request" +RESULT_SUBJECT = "files.tabular_parsed" +TABULAR_REQUEST_SUBJECT = "files.tabular_uploaded" +IMAGE_REQUEST_SUBJECT = "files.img_uploaded" REQUEST_SUBJECTS: Mapping[str, str] = { "tabular": TABULAR_REQUEST_SUBJECT, @@ -23,7 +23,7 @@ REQUEST_SUBJECTS: Mapping[str, str] = { def request_subject(pipeline: str) -> str: - """Subject запроса по pipeline: `tabular` → `recognition.tabular.request`, `image` → `recognition.image.request`.""" + """Subject запроса по pipeline: `tabular` → `files.tabular_uploaded`, `image` → `files.img_uploaded`.""" try: return REQUEST_SUBJECTS[pipeline] except KeyError: @@ -36,10 +36,13 @@ def request_msg_id(job_id: uuid.UUID | str, attempt: int) -> str: def ack_msg_id(job_id: uuid.UUID | str, attempt: int) -> str: - """`Nats-Msg-Id` ack (`status = processing`) на `recognition.result`: `ack::`.""" + """`Nats-Msg-Id` ack (`status = processing`) на `files.tabular_parsed`: `ack::`.""" return f"ack:{job_id}:{attempt}" def result_msg_id(job_id: uuid.UUID | str, attempt: int) -> str: - """`Nats-Msg-Id` результата (`done | needs_check | error`) на `recognition.result`: `result::`.""" + """`Nats-Msg-Id` результата (`done | needs_check | error`) на `files.tabular_parsed`. + + Форма: `result::`. + """ return f"result:{job_id}:{attempt}" diff --git a/api/src/core/config.py b/api/src/core/config.py index d30dbce..b14165f 100644 --- a/api/src/core/config.py +++ b/api/src/core/config.py @@ -71,9 +71,9 @@ class Settings(BaseSettings): server: ServerSettings = ServerSettings() ROOT_PATH: str = "/" VERSION: str = "1.0.0" - SERVICE_NAME: str = "scan-gateway" # service.name в конверте recognition.result + SERVICE_NAME: str = "scan-gateway" # service.name в конверте files.tabular_parsed - # NATS: подписка на recognition.image.request и публикация. Выключено — воркер не запускается (dev без NATS) + # NATS: подписка на files.img_uploaded и публикация. Выключено — воркер не запускается (dev без NATS) NATS_ENABLED: bool = True PUBLISH_RETRIES: PositiveInt = 3 # повторы публикации результата/ack после первой попытки (всего до 4) DONE_CACHE_SIZE: PositiveInt = 1000 # размер LRU обработанных job_id (дедупликация повторов) diff --git a/api/src/core/nats.py b/api/src/core/nats.py index af5507d..2144880 100644 --- a/api/src/core/nats.py +++ b/api/src/core/nats.py @@ -8,7 +8,7 @@ поэтому `connect()` возвращается только при успехе (вызывающий оборачивает его в задачу или `wait_for`). После переподключения nats-py сам отправляет SUB для всех подписок с прежней queue group — подписку повторно оформлять не нужно. -- `subscribe_requests(cb)` — подписка на `recognition.image.request` в queue group `scan-gateway`: из нескольких +- `subscribe_requests(cb)` — подписка на `files.img_uploaded` в queue group `scan-gateway`: из нескольких инстансов сообщение получает ровно один. `cb` получает исходное сообщение nats-py (`msg.data`, `msg.headers`); исключения `cb` логируются, подписка живёт. - `publish(subject, payload, msg_id)` — публикация с заголовком `Nats-Msg-Id` и `flush` (подтверждение сервером) с @@ -209,7 +209,7 @@ class NatsGateway: # --- подписка --------------------------------------------------------------------------------------------- async def subscribe_requests(self, cb: MessageHandler, queue: str = REQUEST_QUEUE_GROUP) -> Any: - """Подписка на `recognition.image.request` в queue group `queue`; возвращает подписку nats-py после + """Подписка на `files.img_uploaded` в queue group `queue`; возвращает подписку nats-py после подтверждения сервером (`flush`). `cb` получает исходное `Msg` (`data`, `headers`); его исключения логируются, подписка не падает. @@ -292,7 +292,7 @@ class NatsGateway: ) from exc async def publish_result(self, message: RecognitionResultMessage, attempt: int) -> None: - """Публикует конверт в `recognition.result`: тело — `message.to_wire()` (JSON, UTF-8), `Nats-Msg-Id` — + """Публикует конверт в `files.tabular_parsed`: тело — `message.to_wire()` (JSON, UTF-8), `Nats-Msg-Id` — `ack:…` для `processing`, `result:…` для итога; `attempt` — из запроса. Ошибки — как у `publish`.""" payload = json.dumps(message.to_wire(), ensure_ascii=False).encode("utf-8") await self.publish(RESULT_SUBJECT, payload, message_msg_id(message, attempt)) diff --git a/api/src/core/worker.py b/api/src/core/worker.py index 2e21ff9..a02f996 100644 --- a/api/src/core/worker.py +++ b/api/src/core/worker.py @@ -107,7 +107,7 @@ class ObjectStorage(Protocol): class ResultPublisher(Protocol): - """Часть `NatsGateway`: публикация конверта в `recognition.result` с `Nats-Msg-Id` и повторами.""" + """Часть `NatsGateway`: публикация конверта в `files.tabular_parsed` с `Nats-Msg-Id` и повторами.""" async def publish_result(self, message: RecognitionResultMessage, attempt: int) -> None: ... @@ -181,7 +181,7 @@ class _WaitResult: class Worker: - """Обработка запросов `recognition.image.request` (см. докстринг модуля). + """Обработка запросов `files.img_uploaded` (см. докстринг модуля). Зависимости внедряются: `storage` (S3), `publisher` (NATS), `hotfolder`, `normalizer`; `clock` — часы UTC для дедлайна (R-20), `sleep` — пауза между опросами (подмены для тестов). diff --git a/api/tests/integration/test_roundtrip.py b/api/tests/integration/test_roundtrip.py index 0761271..b74aaef 100644 --- a/api/tests/integration/test_roundtrip.py +++ b/api/tests/integration/test_roundtrip.py @@ -22,12 +22,12 @@ ответа, в порядке возрастания имени, с телами байт в байт; входящий файл совпадает с объектом S3. Готовность набора — по settle (`SETTLE_SECONDS`, 0.2 с). -Общая инфраструктура: NATS и MinIO используют и другие сервисы workspace. Чужие конверты на `recognition.result` +Общая инфраструктура: NATS и MinIO используют и другие сервисы workspace. Чужие конверты на `files.tabular_parsed` отсекаются по `job_id` теста; ключи MinIO уникальны (`uploads/itest-/…`, `raw/.json`, `results/…`) и удаляются. Подписка шлюза идёт в queue group `scan-gateway`: если в этом NATS живёт другой экземпляр scan-gateway, он может забрать запрос теста (тест тогда упадёт по таймауту ожидания ack). Перед запуском занятость группы можно проверить -мониторингом NATS: `curl -s 'http://127.0.0.1:8222/subsz?subs=1'` — подписчиков `recognition.image.request` быть +мониторингом NATS: `curl -s 'http://127.0.0.1:8222/subsz?subs=1'` — подписчиков `files.img_uploaded` быть не должно. Автоматической защиты в тесте нет намеренно (она была бы гонкой). Недоступная инфраструктура — падение теста с понятным сообщением, не skip; нет репозитория fixtures — skip сценариев на реальных данных (маркер `fixtures`). """ @@ -163,7 +163,7 @@ class Envelope: @dataclass class Roundtrip: - """Запущенный шлюз (lifespan приложения), эмулятор и слушатель `recognition.result`.""" + """Запущенный шлюз (lifespan приложения), эмулятор и слушатель `files.tabular_parsed`.""" settings: Settings app: FastAPI @@ -209,7 +209,7 @@ class Roundtrip: } async def publish(self, body: dict[str, Any]) -> None: - """Публикация как в ядре: `recognition.image.request`, `Nats-Msg-Id = :`.""" + """Публикация как в ядре: `files.img_uploaded`, `Nats-Msg-Id = :`.""" payload = json.dumps(body, ensure_ascii=False).encode("utf-8") headers = {NATS_MSG_ID_HEADER: request_msg_id(body["job_id"], body["attempt"])} await self.publisher.publish(IMAGE_REQUEST_SUBJECT, payload, headers=headers) @@ -378,7 +378,7 @@ def _settings(monkeypatch: pytest.MonkeyPatch, root: Path, *, cleanup: bool) -> async def _start_gateway(stack: AsyncExitStack, app: FastAPI) -> None: - """Lifespan приложения; возврат — когда шлюз подписан на `recognition.image.request`.""" + """Lifespan приложения; возврат — когда шлюз подписан на `files.img_uploaded`.""" await stack.enter_async_context(app.router.lifespan_context(app)) [connect_task] = [t for t in asyncio.all_tasks() if t.get_name() == "nats-connect"] # задача завершается после SUB с flush: дальше подписка шлюза действует на сервере diff --git a/api/tests/unit/test_contract.py b/api/tests/unit/test_contract.py index e8d0f4a..c2691f9 100644 --- a/api/tests/unit/test_contract.py +++ b/api/tests/unit/test_contract.py @@ -629,9 +629,9 @@ def test_ack_and_result_msg_ids() -> None: def test_request_subject_by_pipeline() -> None: - assert request_subject("tabular") == "recognition.tabular.request" - assert request_subject("image") == "recognition.image.request" - assert RESULT_SUBJECT == "recognition.result" + assert request_subject("tabular") == "files.tabular_uploaded" + assert request_subject("image") == "files.img_uploaded" + assert RESULT_SUBJECT == "files.tabular_parsed" assert NATS_MSG_ID_HEADER == "Nats-Msg-Id" @@ -643,7 +643,8 @@ def test_request_subject_rejects_unknown_pipeline() -> None: def test_request_subject_matches_example_pipeline() -> None: for name in ("request-tabular.json", "request-image.json"): request = RecognitionRequest.model_validate(_example(name)) - assert request_subject(request.pipeline) == f"recognition.{request.pipeline}.request" + expected = {"tabular": "files.tabular_uploaded", "image": "files.img_uploaded"}[request.pipeline] + assert request_subject(request.pipeline) == expected def test_integer_valued_float_is_normalized_to_int() -> None: @@ -704,7 +705,7 @@ def test_is_supported_major_agrees_with_schema_pattern(version: str) -> None: def test_image_request_subject() -> None: - assert IMAGE_REQUEST_SUBJECT == "recognition.image.request" + assert IMAGE_REQUEST_SUBJECT == "files.img_uploaded" # --- конструкторы конвертов ----------------------------------------------------------------------------------------- diff --git a/api/tests/unit/test_nats.py b/api/tests/unit/test_nats.py index 06e3baf..ca359cf 100644 --- a/api/tests/unit/test_nats.py +++ b/api/tests/unit/test_nats.py @@ -253,7 +253,7 @@ async def test_subscribe_requests_uses_image_subject_and_queue_group(gateway: Na assert REQUEST_QUEUE_GROUP == "scan-gateway" [subscription] = fake.subscriptions - assert subscription["subject"] == IMAGE_REQUEST_SUBJECT == "recognition.image.request" + assert subscription["subject"] == IMAGE_REQUEST_SUBJECT == "files.img_uploaded" assert subscription["queue"] == "scan-gateway" assert fake.flush_calls == 1 # подписка подтверждена сервером до возврата diff --git a/api/tests/worker_helpers.py b/api/tests/worker_helpers.py index 63fac4f..dbdbe5e 100644 --- a/api/tests/worker_helpers.py +++ b/api/tests/worker_helpers.py @@ -279,7 +279,7 @@ class Env: class FakeMsg: def __init__(self, data: bytes) -> None: self.data = data - self.subject = "recognition.image.request" + self.subject = "files.img_uploaded" self.headers: dict[str, str] | None = None diff --git a/openspec/changes/nats-files-subjects/.openspec.yaml b/openspec/changes/nats-files-subjects/.openspec.yaml new file mode 100644 index 0000000..01660ba --- /dev/null +++ b/openspec/changes/nats-files-subjects/.openspec.yaml @@ -0,0 +1,2 @@ +schema: spec-driven +created: 2026-10-08 diff --git a/openspec/changes/nats-files-subjects/design.md b/openspec/changes/nats-files-subjects/design.md new file mode 100644 index 0000000..63d5945 --- /dev/null +++ b/openspec/changes/nats-files-subjects/design.md @@ -0,0 +1,14 @@ +## Context + +Переименование subject'ов по решению владельца 08.10.2026; схема обмена (два subject'а запроса по `pipeline`, один +общий subject ответа) не меняется. Подробности и риски — design сквозного change'а `nats-files-subjects` в docs. + +## Decisions + +Решения R-1…R-5 приняты в сквозном change'е (docs, `openspec/changes/nats-files-subjects/design.md`) и здесь не +дублируются. + +## Decisions made without discussion + +Своих решений у этого change'а нет: маршрут (R-1), версия контракта (R-2), DBML (R-3), константы вместо настроек (R-4), +историю не переписываем (R-5) — см. design docs. diff --git a/openspec/changes/nats-files-subjects/proposal.md b/openspec/changes/nats-files-subjects/proposal.md new file mode 100644 index 0000000..110508f --- /dev/null +++ b/openspec/changes/nats-files-subjects/proposal.md @@ -0,0 +1,31 @@ +## Why + +Решение владельца 08.10.2026: subject'ы NATS — «правильно как в старом АРМ». На проде заведены `files.tabular_uploaded`, +`files.img_uploaded`, `files.tabular_parsed`, `files.img_parsed`; используем три, `files.img_parsed` не используем, +старый АРМ гасится. Сквозной change — `nats-files-subjects` в RSHB/arm-ka-docs (issue #74, design с решениями R-1…R-5). +Issue этого репозитория — RSHB/arm-scan-gateway#22. + +## What Changes + +- `recognition.tabular.request` → `files.tabular_uploaded`, `recognition.image.request` → `files.img_uploaded`, + `recognition.result` → `files.tabular_parsed` (константы `api/src/contract/subjects.py`, докстринги, тесты, README, + `CLAUDE.md`). +- После изменения сервис подписан на `files.img_uploaded` в queue group `scan-gateway`, ack и результат публикует в `files.tabular_parsed` (общий subject ответа, `files.img_parsed` не используется). +- Копии схем `api/contracts/recognition-request.schema.json` и `recognition-result-message.schema.json` — побайтно из + `docs/contracts/` (изменены только описания subject'ов). + +## Non-goals + +- Тела сообщений, `Nats-Msg-Id`, queue group'ы и `schema_version` не меняются. +- Subject'ы не выносятся в настройки (OQ-83). + +## Capabilities + +### New Capabilities + +### Modified Capabilities +- `image-recognition-gateway`: имена subject'ов запроса и ответа. + +## Impact + +Выкатывается одновременно с остальными сервисами: сервис со старыми именами не увидит сообщений с новыми. diff --git a/openspec/changes/nats-files-subjects/specs/image-recognition-gateway/spec.md b/openspec/changes/nats-files-subjects/specs/image-recognition-gateway/spec.md new file mode 100644 index 0000000..990b0b0 --- /dev/null +++ b/openspec/changes/nats-files-subjects/specs/image-recognition-gateway/spec.md @@ -0,0 +1,63 @@ +## MODIFIED Requirements + +### Requirement: Подтверждение запроса до обработки +Сервис SHALL быть подписан на `files.img_uploaded` в queue group `scan-gateway`. Получив запрос, проходящий схему, +сервис SHALL до чтения файла из S3 опубликовать в `files.tabular_parsed` конверт `status = processing` +(`service.name = "scan-gateway"`, `Nats-Msg-Id = ack::`). Повторный запрос с `job_id`, который уже +обрабатывается или обработан этим процессом, SHALL получить повторный ack, и второй файл во входящий каталог MUST NOT записываться. + +#### Scenario: Ack раньше файла +- **WHEN** получен запрос для `aep_5.pdf` +- **THEN** конверт `processing` опубликован раньше, чем файл появился во входящем каталоге + +#### Scenario: Повторный запрос +- **WHEN** тот же запрос доставлен повторно во время ожидания ответа ИС +- **THEN** публикуется ещё один ack, во входящем каталоге остаётся один файл `.pdf` + +### Requirement: Результат по контракту +Сервис SHALL публиковать в `files.tabular_parsed` конверт по `recognition-result-message.schema.json` с +`service.name = "scan-gateway"`, `Nats-Msg-Id = result::`; при `status ∈ {done, needs_check}` — файл +результата (`recognition-result.schema.json`, `schema_version = "1.0"` — `CONTRACT_VERSION` сервиса, решение №14; схема допускает любой MINOR) в `results/{file_id}/{job_id}/result.json` и `result_s3`; +при неудаче публикации SHALL повторить её до трёх раз. Нормализатор получает склеенный ответ и SHALL считать файл +успешным, если `error.code == ""` (поле `error` реального успешного ответа — `{"code": "", "message": ""}`, а не `null`; +отсутствующий или `null` `error` ошибкой не считается), и ошибочным, если `error.code` непуст, `parse_error` не `null` +либо корень ответа файла не является объектом JSON (ошибка этого файла — `INVALID_FORMAT`); успех MUST NOT определяться +проверкой «`error` не `null`». Если ошибочны **все** файлы — `status = error`, `error.code` и `error.message` берутся у +первого ошибочного файла по имени (для `parse_error` и не-объекта — `INVALID_FORMAT`); если код ИС этого файла не +входит в перечень кодов контракта, `error.code = UNKNOWN_ERROR`, а исходный код SHALL быть указан в `error.message`. +Если есть хотя бы один успешный файл — документы собираются из успешных файлов по требованию «Нормализация ответа ИС», +а статус определяется по требованию «Статус результата нормализации». В обоих случаях `metrics.warnings` SHALL +содержать по одной строке `<имя файла>: <код>: <сообщение>` на каждый ошибочный файл, где `<код>` — исходный код ИС как +получен (для `parse_error` и не-объекта — `INVALID_FORMAT`); если предупреждений нет, поле `warnings` SHALL быть опущено. + +#### Scenario: Успешный ответ ИС в Срезе 0 +- **WHEN** ответ из одного файла с `error = {"code": "", "message": ""}` и непустым `data_ocr` (по образцу `aep_6.json`) +- **THEN** начиная со Среза 3a конверт больше не `NOT_IMPLEMENTED`: поведение — по сценарию «Успешный ответ ИС» ниже + +#### Scenario: Успешный ответ ИС +- **WHEN** ответ из одного файла с `error = {"code": "", "message": ""}` и непустым `data_ocr` (`aep_6.json`) +- **THEN** конверт `status = needs_check`, `raw_s3_key = raw/.json`, `result_s3` указывает на файл результата с двумя документами (`BALANCE`, `INCOME`) + +#### Scenario: Ошибка ИС +- **WHEN** единственный файл ответа `.json` содержит `error.code = INVALID_DOCUMENT` +- **THEN** конверт `status = error`, `error.code = INVALID_DOCUMENT`, `error.message` из ответа, `raw_s3_key` заполнен, `metrics.warnings` содержит одну строку `.json: INVALID_DOCUMENT: <сообщение>` + +#### Scenario: Ошибка одного из документов +- **WHEN** ответ состоит из `.json` (успех, `aep_3.json`) и `_ofr.json` (`INVALID_DOCUMENT`, `aep_3_ofr.json`) +- **THEN** конверт `status = needs_check`, документы собраны из `.json`, `metrics.warnings` содержит строку `_ofr.json: INVALID_DOCUMENT: <сообщение>` и предупреждения нормализации + +#### Scenario: Все документы ошибочны +- **WHEN** оба файла ответа ошибочны (первый по имени — `INVALID_DOCUMENT`, второй — не JSON) +- **THEN** конверт `status = error`, `error.code = INVALID_DOCUMENT` (код первого ошибочного файла по имени), `metrics.warnings` содержит две строки; файл не JSON в warnings несёт код `INVALID_FORMAT` + +#### Scenario: Все файлы не JSON +- **WHEN** единственный файл ответа не разбирается как JSON +- **THEN** конверт `status = error`, `error.code = INVALID_FORMAT`, `metrics.warnings` содержит одну строку `.json: INVALID_FORMAT: <сообщение разбора>` + +#### Scenario: Корень ответа не объект +- **WHEN** файл ответа — корректный JSON, но корень — массив +- **THEN** этот файл ошибочен с кодом `INVALID_FORMAT`; остальные файлы оцениваются по общим правилам + +#### Scenario: Код ошибки ИС вне перечня контракта +- **WHEN** единственный файл ответа содержит `error.code = SOME_NEW_CODE` +- **THEN** конверт `status = error`, `error.code = UNKNOWN_ERROR`, `error.message` содержит `SOME_NEW_CODE`, строка в `metrics.warnings` — `.json: SOME_NEW_CODE: <сообщение>` diff --git a/openspec/changes/nats-files-subjects/tasks.md b/openspec/changes/nats-files-subjects/tasks.md new file mode 100644 index 0000000..0ea8448 --- /dev/null +++ b/openspec/changes/nats-files-subjects/tasks.md @@ -0,0 +1,11 @@ +## 1. Переименование + +- [x] 1.1 `api/src/contract/subjects.py`: `files.tabular_uploaded`, `files.img_uploaded`, `files.tabular_parsed` +- [x] 1.2 Докстринги, логи, тесты, README, `CLAUDE.md` — новые имена +- [x] 1.3 Копии схем `api/contracts/` побайтно из `docs/contracts/` +- [x] 1.4 Дельта спеки `image-recognition-gateway` + +## 2. Проверка + +- [x] 2.1 `make check` +- [x] 2.2 Интеграционные тесты NATS