From 1b50d652af26903329a65f2573b0374e5532c941 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Tue, 13 Oct 2026 13:27:16 +0400 Subject: [PATCH] =?UTF-8?q?=D0=9F=D1=80=D0=B0=D0=B2=D0=BA=D0=B8=20=D1=84?= =?UTF-8?q?=D0=B8=D0=BD=D0=B0=D0=BB=D1=8C=D0=BD=D0=BE=D0=B3=D0=BE=20=D1=80?= =?UTF-8?q?=D0=B5=D0=B2=D1=8C=D1=8E:=20=D1=83=D0=BC=D0=BE=D0=BB=D1=87?= =?UTF-8?q?=D0=B0=D0=BD=D0=B8=D1=8F=20subject'=D0=BE=D0=B2=20=D0=BD=D0=B5?= =?UTF-8?q?=20=D0=BD=D0=B0=D0=B7=D1=8B=D0=B2=D0=B0=D0=B5=D0=BC=20=D1=84?= =?UTF-8?q?=D0=B0=D0=BA=D1=82=D0=B8=D1=87=D0=B5=D1=81=D0=BA=D0=B8=D0=BC?= =?UTF-8?q?=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit M-3: docstring воркера, T5.4: docstring и сообщение roundtrip-теста, M-5: CLAUDE.md. Co-Authored-By: Claude Opus 5.5 --- CLAUDE.md | 2 +- api/src/core/worker.py | 6 ++++-- api/tests/integration/test_roundtrip.py | 12 ++++++------ 3 files changed, 11 insertions(+), 9 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 389a2b5..a17c556 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -4,7 +4,7 @@ ## Что это Stateless-сервис разбора XLS/XLSX/XLSB: ОСВ по счетам (§1 правил) и ББ/ОФР из Excel (§12). Вход — NATS -`files.tabular_uploaded` + файл в S3 `uploads/`; выход — JSON по `../docs/contracts/recognition-result.schema.json` +`files.tabular_uploaded` (по умолчанию; задаётся OPENBAO__NATS__NATS_*_SUBJECT) + файл в S3 `uploads/`; выход — JSON по `../docs/contracts/recognition-result.schema.json` в S3 `results/` + NATS `files.tabular_parsed`. **Нет доступа к БД.** Не знает про организации, периметры, план счетов — возвращает структуру файла; нормализация и резолв — в arm-core. diff --git a/api/src/core/worker.py b/api/src/core/worker.py index 81652a2..5af63ee 100644 --- a/api/src/core/worker.py +++ b/api/src/core/worker.py @@ -1,4 +1,5 @@ -"""Воркер распознавания: запрос `files.tabular_uploaded` → ack → S3 → парсер → S3 → `files.tabular_parsed`. +"""Воркер распознавания: запрос на subject запросов → ack → S3 → парсер → S3 → subject ответа +(по умолчанию `files.tabular_uploaded` / `files.tabular_parsed`). Design D-2, D-4, D-5, D-7; spec tabular-recognition-service «Подтверждение запроса до обработки», «Результат по контракту», «Ошибки транспорта превращаются в результат», «Парсер — заменяемый интерфейс». @@ -200,7 +201,8 @@ class RecognitionWorker: # --- приём запроса ---------------------------------------------------------------------------------------------- async def on_request(self, msg: InboundMessage) -> None: - """Обработчик подписки на `files.tabular_uploaded`; исключения наружу не выходят.""" + """Обработчик подписки на subject запросов (по умолчанию `files.tabular_uploaded`); + исключения наружу не выходят.""" if not self._accepting: logger.warning("Воркер останавливается: запрос не принят (без ack, ядро переотправит)") return diff --git a/api/tests/integration/test_roundtrip.py b/api/tests/integration/test_roundtrip.py index ae47aaa..9bf7ed1 100644 --- a/api/tests/integration/test_roundtrip.py +++ b/api/tests/integration/test_roundtrip.py @@ -3,8 +3,8 @@ Приложение поднимается в процессе теста через lifespan (`src.app`) с `NATS_ENABLED=True` — тот же код, что в проде: `_connect_s3`, `NatsConnection` с подпиской в queue group `tabular-parser`, `RecognitionWorker` с парсером по умолчанию (`get_parser(settings)` → `OsvParser`); исходник — битый zip, поэтому результат — ошибка `UNREADABLE`. -Тест играет роль ядра: кладёт исходник в MinIO `uploads/…`, публикует запрос в `files.tabular_uploaded` с -`Nats-Msg-Id = :` и собирает конверты из `files.tabular_parsed`. +Тест играет роль ядра: кладёт исходник в MinIO `uploads/…`, публикует запрос на subject запросов прогона с +`Nats-Msg-Id = :` и собирает конверты с subject'а ответа прогона. Spec tabular-recognition-service: «Ack раньше результата», «Повторный запрос во время обработки» (здесь — после обработки: битый файл отвергается мгновенно; повтор во время обработки покрыт unit-тестами воркера), ошибка разбора @@ -14,7 +14,7 @@ Spec tabular-recognition-service: «Ack раньше результата», « Изоляция: subject'ы прогона уникальны — контейнер стенда сообщения теста не получает (OQ-83); `job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может -работать другой сервис — сообщения `files.tabular_parsed` фильтруются по своему `job_id`. Ожидания — по событиям с +работать другой сервис — сообщения с subject'а ответа фильтруются по своему `job_id`. Ожидания — по событиям с таймаутом; единственная фиксированная пауза — явное окно тишины `QUIET_WINDOW` для отрицательных проверок. """ @@ -124,7 +124,7 @@ def _assert_ack(msg: Msg, request: dict[str, Any]) -> None: class Core: - """Тестовый клиент в роли ядра: публикует запросы и собирает `files.tabular_parsed` по своим `job_id`.""" + """Тестовый клиент в роли ядра: публикует запросы и собирает сообщения с subject'а ответа по своим `job_id`.""" def __init__(self, client: NATS) -> None: self.client = client @@ -151,7 +151,7 @@ class Core: return await asyncio.wait_for(self._inboxes[request["job_id"]].get(), MESSAGE_TIMEOUT) except TimeoutError: pytest.fail( - f"нет сообщения files.tabular_parsed для job_id={request['job_id']} за {MESSAGE_TIMEOUT} с " + f"нет сообщения на subject ответа прогона для job_id={request['job_id']} за {MESSAGE_TIMEOUT} с " "(сервис не ответил; если на шине есть другой инстанс tabular-parser, запрос мог уйти ему)", pytrace=False, ) @@ -241,7 +241,7 @@ async def service(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[None]: @pytest.fixture async def core(service: None) -> AsyncIterator[Core]: - """Клиент «ядра», подписанный на `files.tabular_parsed` до первой публикации запроса.""" + """Клиент «ядра», подписанный на subject ответа прогона до первой публикации запроса.""" client = NATS() await client.connect(get_settings().OPENBAO.NATS.NATS_URL, connect_timeout=2, max_reconnect_attempts=1) core = Core(client)