nats: subject'ы как в старом АРМ — files.*_uploaded, files.tabular_parsed (#28)
Решение владельца 08.10.2026: recognition.tabular.request → files.tabular_uploaded, recognition.image.request → files.img_uploaded, recognition.result → files.tabular_parsed (общий subject ack и результатов; files.img_parsed не используется). Тела сообщений, Nats-Msg-Id, queue group'ы и schema_version не меняются. Константы api/src/contract/subjects.py, докстринги, тесты, README, CLAUDE.md; копии схем api/contracts/ из docs/contracts/ побайтно; change nats-files-subjects с дельтой спеки. Сквозной change — RSHB/arm-ka-docs#74. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
parent
49f725d238
commit
817487b976
@ -4,8 +4,8 @@
|
||||
|
||||
## Что это
|
||||
Stateless-сервис разбора XLS/XLSX/XLSB: ОСВ по счетам (§1 правил) и ББ/ОФР из Excel (§12). Вход — NATS
|
||||
`recognition.tabular.request` + файл в S3 `uploads/`; выход — JSON по `../docs/contracts/recognition-result.schema.json`
|
||||
в S3 `results/` + NATS `recognition.result`. **Нет доступа к БД.** Не знает про организации, периметры, план счетов —
|
||||
`files.tabular_uploaded` + файл в S3 `uploads/`; выход — JSON по `../docs/contracts/recognition-result.schema.json`
|
||||
в S3 `results/` + NATS `files.tabular_parsed`. **Нет доступа к БД.** Не знает про организации, периметры, план счетов —
|
||||
возвращает структуру файла; нормализация и резолв — в arm-core.
|
||||
|
||||
## Главное правило
|
||||
@ -30,7 +30,7 @@ Stateless-сервис разбора XLS/XLSX/XLSB: ОСВ по счетам (
|
||||
`core/config.py` (pydantic-settings, `SERVICE_NAME` — только `service.name` конвертов (queue group — константа в `core/nats.py`), `OPENBAO__MINIO__*`
|
||||
включая `MINIO_SECURE`, `OPENBAO__NATS__*`, `DEBUG`, `ENV_FILE`, `NATS_ENABLED`, `WORKER_CONCURRENCY`, `PUBLISH_RETRIES`, `DONE_CACHE_SIZE`), `core/clients.py` (S3), `core/logger.py`,
|
||||
`core/exception_handlers.py` (без SQLAlchemy), `common/exceptions.py`, `pytest.ini`, `requirements*.txt`, `raisa_config.yml`,
|
||||
`setup/setup.sh` + `pip.conf` (без npm-шага). Шаблонный `core/nats_callback.py` (consumer `recognition.result` — зона
|
||||
`setup/setup.sh` + `pip.conf` (без npm-шага). Шаблонный `core/nats_callback.py` (consumer `files.tabular_parsed` — зона
|
||||
ядра) удалён задачей 4.3: его место занял воркер `core/worker.py`.
|
||||
|
||||
**Вырезано (Срез 0):** `web/` целиком и `setup/.npmrc`; из `app.py` — mount SPA (`SPAStaticFiles`, `os.makedirs("web")`),
|
||||
@ -47,7 +47,7 @@ CORS, регистрация SMB-сессии, импорт `raisa_fastapi_prote
|
||||
- `api/rules/` — README о назначении; сами словари `*.yaml` (в документах `rules/*.yaml` относительно `api/`)
|
||||
появились в Срезе 2a — 4.1;
|
||||
- `api/src/core/nats.py` и `api/src/core/worker.py` — подписка queue group `tabular-parser` на
|
||||
`recognition.tabular.request` → ack → S3 `uploads/` → парсер → S3 `results/` → `recognition.result` — 4.2–4.3;
|
||||
`files.tabular_uploaded` → ack → S3 `uploads/` → парсер → S3 `results/` → `files.tabular_parsed` — 4.2–4.3;
|
||||
- `api/pyproject.toml` (ruff, mypy) и `Makefile` — 2.1; `Dockerfile`, `api/.env.example`, `api/.env.test` — 2.2.
|
||||
|
||||
## Добавлено в Срезе 2a (парсер ОСВ; `openspec/changes/slice-2a-osv-parser/`)
|
||||
|
||||
@ -30,16 +30,16 @@ Stateless-сервис распознавания табличных файло
|
||||
|
||||
## Как обрабатывается запрос
|
||||
|
||||
1. `arm-core` публикует запрос в NATS, subject `recognition.tabular.request` (схема `recognition-request`), исходный
|
||||
1. `arm-core` публикует запрос в NATS, subject `files.tabular_uploaded` (схема `recognition-request`), исходный
|
||||
файл лежит в S3 под ключом `uploads/{file_id}/{job_id}/<имя>`.
|
||||
2. Сервис подписан на этот subject в queue group `tabular-parser` — несколько экземпляров делят нагрузку между собой.
|
||||
3. Запрос валидируется по схеме. Если `job_id` ещё не обработан этим процессом (process-local LRU), сервис **до чтения
|
||||
файла** публикует в `recognition.result` конверт `status = processing` (ack) с заголовком
|
||||
файла** публикует в `files.tabular_parsed` конверт `status = processing` (ack) с заголовком
|
||||
`Nats-Msg-Id = ack:<job_id>:<attempt>`. Повторный запрос с тем же `job_id` получает повторный ack и второй раз не
|
||||
обрабатывается.
|
||||
4. Файл читается из S3 (`s3.bucket` / `s3.key`) и передаётся парсеру («байты + тип файла → документы, метрики или ошибка»).
|
||||
5. Результат (схема `recognition-result`) записывается в S3 под ключом `results/{file_id}/{job_id}/result.json`.
|
||||
6. В `recognition.result` публикуется конверт результата (схема `recognition-result-message`; статусы
|
||||
6. В `files.tabular_parsed` публикуется конверт результата (схема `recognition-result-message`; статусы
|
||||
`processing | done | needs_check | error`) с заголовком `Nats-Msg-Id = result:<job_id>:<attempt>`. Публикация
|
||||
при неудаче повторяется до `PUBLISH_RETRIES` раз (после первой попытки, с паузой 2 с).
|
||||
|
||||
|
||||
@ -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 = <job_id>:<attempt> (не поле тела). Один файл (stored_file) → один job.",
|
||||
"description": "Тело сообщения на subject files.tabular_uploaded или files.img_uploaded; subject должен соответствовать полю pipeline (tabular → files.tabular_uploaded, image → files.img_uploaded). Заголовок Nats-Msg-Id = <job_id>:<attempt> (не поле тела). Один файл (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"],
|
||||
|
||||
@ -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:<job_id>:<attempt> — для status = processing; result:<job_id>:<attempt> — для done, needs_check и error. Форма запроса <job_id>:<attempt> описана в схеме recognition-request. status = processing — ack «взял в работу», публикуется до чтения файла из S3. group_schema в конверт не входит: источник — файл результата (recognition-result).",
|
||||
"description": "Тело сообщения на subject files.tabular_parsed. Заголовок Nats-Msg-Id (не поле тела) на этом subject имеет две формы: ack:<job_id>:<attempt> — для status = processing; result:<job_id>:<attempt> — для done, needs_check и error. Форма запроса <job_id>:<attempt> описана в схеме 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"],
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
|
||||
Что здесь есть: настройки, подключение S3 (MinIO), NATS (`src/core/nats.py`: nkeys JWT, пустые seed/jwt — без
|
||||
аутентификации, фоновое подключение — старт не ждёт NATS), воркер `src/core/worker.py` (подписка на
|
||||
`recognition.tabular.request` после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck`
|
||||
`files.tabular_uploaded` после подключения к NATS), middleware (gzip, proxy headers), `/healthcheck`
|
||||
(`nats = ok | down | off`, R-17), обработчики исключений. Сервис без БД и фронта (DECISION-4).
|
||||
|
||||
При `NATS_ENABLED=false` ни NATS, ни воркер не запускаются. Остановка: воркер перестаёт принимать запросы и
|
||||
|
||||
@ -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` и
|
||||
|
||||
@ -1,6 +1,6 @@
|
||||
"""Запрос распознавания файла: `recognition-request.schema.json` v1.0.
|
||||
|
||||
tabular-parser принимает этот запрос из `recognition.tabular.request` (производитель — arm-core); один файл — один job.
|
||||
tabular-parser принимает этот запрос из `files.tabular_uploaded` (производитель — arm-core); один файл — один job.
|
||||
Модель намеренно совпадает с копией в arm-core (контракт — `../docs/contracts`, обзор —
|
||||
`docs/context/03-architecture.md`). Subject — по `pipeline` (`request_subject`); заголовок
|
||||
`Nats-Msg-Id = <job_id>:<attempt>` (`request_msg_id`) — не поле тела.
|
||||
|
||||
@ -16,9 +16,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,
|
||||
@ -27,7 +27,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:
|
||||
@ -40,10 +40,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:<job_id>:<attempt>`."""
|
||||
"""`Nats-Msg-Id` ack (`status = processing`) на `files.tabular_parsed`: `ack:<job_id>:<attempt>`."""
|
||||
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:<job_id>:<attempt>`."""
|
||||
"""`Nats-Msg-Id` результата (`done | needs_check | error`) на `files.tabular_parsed`.
|
||||
|
||||
Форма: `result:<job_id>:<attempt>`.
|
||||
"""
|
||||
return f"result:{job_id}:{attempt}"
|
||||
|
||||
@ -10,7 +10,7 @@
|
||||
завершился исключением, supervisor повторяет попытку с паузой 1 → 2 → 4 → … ≤ 30 с. После подключения один раз
|
||||
вызывается `on_connected(conn)` (подписки воркера); если он упал, созданные в этой попытке подписки снимаются и
|
||||
вызов повторяется с той же паузой. Подписки после разрыва nats-py восстанавливает сам.
|
||||
- `subscribe_requests(cb)` — подписка на `recognition.tabular.request` в queue group `tabular-parser`: из нескольких
|
||||
- `subscribe_requests(cb)` — подписка на `files.tabular_uploaded` в queue group `tabular-parser`: из нескольких
|
||||
инстанций сообщение получает одна. После `subscribe` — `flush`: сервер обработал SUB, ошибки соединения проявляются
|
||||
здесь. Отказ в правах (`Permissions Violation`) nats-py 2.14 так не проявляет: соединение остаётся открытым,
|
||||
`subscribe`/`flush`/`publish` проходят, сервер лишь присылает `-ERR` в `error_cb` — он логируется отдельно уровнем
|
||||
@ -278,7 +278,7 @@ class NatsConnection:
|
||||
async def subscribe_requests(
|
||||
self, cb: MessageHandler, *, queue: str = REQUEST_QUEUE_GROUP, subject: str = TABULAR_REQUEST_SUBJECT
|
||||
) -> Any:
|
||||
"""Подписка на запросы (`recognition.tabular.request`) в queue group; исключения `cb` логируются."""
|
||||
"""Подписка на запросы (`files.tabular_uploaded`) в queue group; исключения `cb` логируются."""
|
||||
if not self.is_connected:
|
||||
raise NatsNotConnectedError(f"Нет соединения с NATS: подписка на {subject} невозможна")
|
||||
|
||||
|
||||
@ -1,4 +1,4 @@
|
||||
"""Воркер распознавания: запрос `recognition.tabular.request` → ack → S3 → парсер → S3 → `recognition.result`.
|
||||
"""Воркер распознавания: запрос `files.tabular_uploaded` → ack → S3 → парсер → S3 → `files.tabular_parsed`.
|
||||
|
||||
Design D-2, D-4, D-5, D-7; spec tabular-recognition-service «Подтверждение запроса до обработки», «Результат по
|
||||
контракту», «Ошибки транспорта превращаются в результат», «Парсер — заменяемый интерфейс».
|
||||
@ -198,7 +198,7 @@ class RecognitionWorker:
|
||||
|
||||
# --- приём запроса ----------------------------------------------------------------------------------------------
|
||||
async def on_request(self, msg: InboundMessage) -> None:
|
||||
"""Обработчик подписки на `recognition.tabular.request`; исключения наружу не выходят."""
|
||||
"""Обработчик подписки на `files.tabular_uploaded`; исключения наружу не выходят."""
|
||||
if not self._accepting:
|
||||
logger.warning("Воркер останавливается: запрос не принят (без ack, ядро переотправит)")
|
||||
return
|
||||
|
||||
@ -2,7 +2,7 @@
|
||||
spec «Соединение с NATS по правилам шаблона, без JetStream»; design D-6).
|
||||
|
||||
`.env.test` содержит `NATS_ENABLED=false` (для приложения) — тесты строят соединение сами. Subject'ы уникальны на
|
||||
тест (`test.tabular.<uuid>`): общие `recognition.*` не трогаем, на шине может работать другой сервис.
|
||||
тест (`test.tabular.<uuid>`): общие `files.*` не трогаем, на шине может работать другой сервис.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
|
||||
@ -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/…`, публикует запрос в `recognition.tabular.request` с
|
||||
`Nats-Msg-Id = <job_id>:<attempt>` и собирает конверты из `recognition.result`.
|
||||
Тест играет роль ядра: кладёт исходник в MinIO `uploads/…`, публикует запрос в `files.tabular_uploaded` с
|
||||
`Nats-Msg-Id = <job_id>:<attempt>` и собирает конверты из `files.tabular_parsed`.
|
||||
|
||||
Spec tabular-recognition-service: «Ack раньше результата», «Повторный запрос во время обработки» (здесь — после
|
||||
обработки: битый файл отвергается мгновенно; повтор во время обработки покрыт unit-тестами воркера), ошибка разбора
|
||||
@ -13,7 +13,7 @@ Spec tabular-recognition-service: «Ack раньше результата», «
|
||||
«Реальные формы через воркер» (книга НПЗ → `done`, документы BALANCE и INCOME, `result_schema_version 1.2`).
|
||||
|
||||
Изоляция: `job_id`/`file_id`/ключ S3 уникальны на прогон (uuid4), исходник удаляется после теста; на шине может
|
||||
работать другой сервис — сообщения `recognition.result` фильтруются по своему `job_id`. Ожидания — по событиям с
|
||||
работать другой сервис — сообщения `files.tabular_parsed` фильтруются по своему `job_id`. Ожидания — по событиям с
|
||||
таймаутом; единственная фиксированная пауза — явное окно тишины `QUIET_WINDOW` для отрицательных проверок.
|
||||
"""
|
||||
|
||||
@ -117,7 +117,7 @@ def _assert_ack(msg: Msg, request: dict[str, Any]) -> None:
|
||||
|
||||
|
||||
class Core:
|
||||
"""Тестовый клиент в роли ядра: публикует запросы и собирает `recognition.result` по своим `job_id`."""
|
||||
"""Тестовый клиент в роли ядра: публикует запросы и собирает `files.tabular_parsed` по своим `job_id`."""
|
||||
|
||||
def __init__(self, client: NATS) -> None:
|
||||
self.client = client
|
||||
@ -144,7 +144,7 @@ class Core:
|
||||
return await asyncio.wait_for(self._inboxes[request["job_id"]].get(), MESSAGE_TIMEOUT)
|
||||
except TimeoutError:
|
||||
pytest.fail(
|
||||
f"нет сообщения recognition.result для job_id={request['job_id']} за {MESSAGE_TIMEOUT} с "
|
||||
f"нет сообщения files.tabular_parsed для job_id={request['job_id']} за {MESSAGE_TIMEOUT} с "
|
||||
"(сервис не ответил; если на шине есть другой инстанс tabular-parser, запрос мог уйти ему)",
|
||||
pytrace=False,
|
||||
)
|
||||
@ -232,7 +232,7 @@ async def service(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[None]:
|
||||
|
||||
@pytest.fixture
|
||||
async def core(service: None) -> AsyncIterator[Core]:
|
||||
"""Клиент «ядра», подписанный на `recognition.result` до первой публикации запроса."""
|
||||
"""Клиент «ядра», подписанный на `files.tabular_parsed` до первой публикации запроса."""
|
||||
client = NATS()
|
||||
await client.connect(get_settings().OPENBAO.NATS.NATS_URL, connect_timeout=2, max_reconnect_attempts=1)
|
||||
core = Core(client)
|
||||
|
||||
@ -87,7 +87,7 @@ class FakeNatsConnection:
|
||||
self.is_connected = False
|
||||
|
||||
async def subscribe_requests(self, cb: Callable[[Any], Awaitable[None]]) -> None:
|
||||
self.handlers.append(("recognition.tabular.request", cb))
|
||||
self.handlers.append(("files.tabular_uploaded", cb))
|
||||
|
||||
async def publish(self, subject: str, payload: bytes, msg_id: str) -> None:
|
||||
self.published.append((subject, json.loads(payload), msg_id))
|
||||
@ -171,7 +171,7 @@ async def test_nats_enabled_subscribes_worker_and_reports_ok(
|
||||
assert isinstance(worker, RecognitionWorker)
|
||||
|
||||
# Запрос, доставленный подписке, проходит через воркер с парсером по умолчанию (`get_parser(settings)` →
|
||||
# OsvParser): байты «xlsx» — не книга, ack и результат UNSUPPORTED_FORMAT в recognition.result
|
||||
# OsvParser): байты «xlsx» — не книга, ack и результат UNSUPPORTED_FORMAT в files.tabular_parsed
|
||||
_subject, handler = connection.handlers[0]
|
||||
message = SimpleNamespace(
|
||||
data=json.dumps(REQUEST).encode(), headers={NATS_MSG_ID_HEADER: f"{JOB_ID}:1"}, subject=_subject
|
||||
|
||||
@ -639,9 +639,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"
|
||||
|
||||
|
||||
@ -653,7 +653,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:
|
||||
|
||||
@ -309,7 +309,7 @@ async def test_on_closed_logs_error_when_not_stopping_and_info_on_stop(log_recor
|
||||
async def test_permission_violation_is_logged_as_permission_error_not_reconnect(log_records: list[Any]):
|
||||
"""nats-py 2.14 при Permissions Violation не рвёт соединение, только вызывает error_cb (текст в нижнем регистре)."""
|
||||
conn = NatsConnection(_nats_settings(), client_factory=_FakeWorld(), sleep=_RecordingSleep())
|
||||
error = Exception('nats: permissions violation for subscription to "recognition.tabular.request"')
|
||||
error = Exception('nats: permissions violation for subscription to "files.tabular_uploaded"')
|
||||
|
||||
repeats = 2 # повтор не глушится лимитом ошибок подключения
|
||||
for _ in range(repeats):
|
||||
@ -318,7 +318,7 @@ async def test_permission_violation_is_logged_as_permission_error_not_reconnect(
|
||||
permission = [r for r in log_records if "error_category=nats_permission" in r["message"]]
|
||||
assert len(permission) == repeats
|
||||
assert all(r["level"].name == "ERROR" for r in permission)
|
||||
assert all("subject=recognition.tabular.request" in r["message"] for r in permission)
|
||||
assert all("subject=files.tabular_uploaded" in r["message"] for r in permission)
|
||||
assert not any("повторяем подключение" in r["message"] for r in log_records)
|
||||
assert conn._error_streak == 0 # серия ошибок подключения не затронута
|
||||
|
||||
|
||||
@ -156,7 +156,7 @@ def _msg(body: Any, *, raw: bytes | None = None, msg_id: str | None = None) -> S
|
||||
if msg_id is None and isinstance(body, dict) and "job_id" in body:
|
||||
msg_id = f"{body['job_id']}:{body.get('attempt')}"
|
||||
headers = {NATS_MSG_ID_HEADER: msg_id} if msg_id else None
|
||||
return SimpleNamespace(subject="recognition.tabular.request", data=data, headers=headers)
|
||||
return SimpleNamespace(subject="files.tabular_uploaded", data=data, headers=headers)
|
||||
|
||||
|
||||
def _request(**overrides: Any) -> dict[str, Any]:
|
||||
|
||||
2
openspec/changes/nats-files-subjects/.openspec.yaml
Normal file
2
openspec/changes/nats-files-subjects/.openspec.yaml
Normal file
@ -0,0 +1,2 @@
|
||||
schema: spec-driven
|
||||
created: 2026-10-08
|
||||
14
openspec/changes/nats-files-subjects/design.md
Normal file
14
openspec/changes/nats-files-subjects/design.md
Normal file
@ -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.
|
||||
31
openspec/changes/nats-files-subjects/proposal.md
Normal file
31
openspec/changes/nats-files-subjects/proposal.md
Normal file
@ -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-tabular-parser#28.
|
||||
|
||||
## 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.tabular_uploaded` в queue group `tabular-parser`, ack и результат публикует в `files.tabular_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
|
||||
- `tabular-recognition-service`: имена subject'ов запроса и ответа.
|
||||
|
||||
## Impact
|
||||
|
||||
Выкатывается одновременно с остальными сервисами: сервис со старыми именами не увидит сообщений с новыми.
|
||||
@ -0,0 +1,38 @@
|
||||
## MODIFIED Requirements
|
||||
|
||||
### Requirement: Подтверждение запроса до обработки
|
||||
Сервис SHALL быть подписан на `files.tabular_uploaded` в queue group `tabular-parser`. Получив запрос, проходящий
|
||||
`recognition-request.schema.json`, сервис SHALL до чтения файла из S3 опубликовать в `files.tabular_parsed` конверт
|
||||
`status = processing` с `job_id`, `file_id`, `service {name: "tabular-parser", version}` и заголовком
|
||||
`Nats-Msg-Id = ack:<job_id>:<attempt>`. Повторный запрос с `job_id`, который уже обрабатывается или обработан этим
|
||||
процессом, SHALL получить повторный ack и MUST NOT обрабатываться второй раз.
|
||||
|
||||
#### Scenario: Ack раньше результата
|
||||
- **WHEN** опубликован запрос из `docs/contracts/examples/request-tabular.json`
|
||||
- **THEN** первым сообщением сервиса в `files.tabular_parsed` является конверт `processing`, и только затем — результат
|
||||
|
||||
#### Scenario: Повторный запрос во время обработки
|
||||
- **WHEN** тот же запрос доставлен второй раз, пока первый обрабатывается
|
||||
- **THEN** публикуется второй ack `processing`, файл из S3 повторно не читается, результат публикуется один раз
|
||||
|
||||
### Requirement: Результат по контракту
|
||||
После подтверждения сервис SHALL прочитать объект `s3.key` из бакета `s3.bucket`, вызвать парсер, записать файл результата
|
||||
(по `recognition-result.schema.json`) в `results/{file_id}/{job_id}/result.json` и опубликовать конверт результата (по
|
||||
`recognition-result-message.schema.json`) с заголовком `Nats-Msg-Id = result:<job_id>:<attempt>`. При неудаче публикации
|
||||
сервис SHALL повторить её до трёх раз с паузой, затем залогировать ошибку.
|
||||
|
||||
#### Scenario: Запрос с заглушкой парсера (Срез 0)
|
||||
- **WHEN** воркеру явно передана заглушка `NullParser` (только тесты воркера; реализация по умолчанию — реальный парсер),
|
||||
запрос подтверждён и объект существует в `uploads/`
|
||||
- **THEN** в `files.tabular_parsed` появляется конверт `status = error`, `error.code = NOT_IMPLEMENTED`, `job_id`/`file_id`
|
||||
из запроса, `metrics.duration_ms ≥ 0`; файл результата может отсутствовать
|
||||
|
||||
#### Scenario: Запрос с реальным парсером
|
||||
- **WHEN** запрос подтверждён и объект в `uploads/` существует, но не является книгой Excel (сигнатура не XLS/XLSX/XLSB)
|
||||
- **THEN** в `files.tabular_parsed` появляется конверт `status = error`, `error.code = UNSUPPORTED_FORMAT` (повреждённая
|
||||
книга — `UNREADABLE`), `job_id`/`file_id` из запроса, `metrics.duration_ms ≥ 0`; файл результата может отсутствовать;
|
||||
корректная ОСВ в том же потоке даёт `done` (см. «Парсер — заменяемый интерфейс»)
|
||||
|
||||
#### Scenario: Повтор публикации
|
||||
- **WHEN** первая публикация конверта результата завершается ошибкой соединения, вторая — успешно
|
||||
- **THEN** конверт опубликован один раз с тем же `Nats-Msg-Id`, обработка не повторяется
|
||||
11
openspec/changes/nats-files-subjects/tasks.md
Normal file
11
openspec/changes/nats-files-subjects/tasks.md
Normal file
@ -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 Дельта спеки `tabular-recognition-service`
|
||||
|
||||
## 2. Проверка
|
||||
|
||||
- [x] 2.1 `make check`
|
||||
- [x] 2.2 Интеграционные тесты NATS
|
||||
@ -1,7 +1,7 @@
|
||||
schema: spec-driven
|
||||
|
||||
context: |
|
||||
Репозиторий tabular-parser — часть системы АРМ КА (ДКБ, Россельхозбанк): stateless-сервис разбора XLS/XLSX/XLSB (ОСВ по счетам, ББ/ОФР из Excel). Без доступа к БД: вход — NATS recognition.tabular.request + файл в S3, выход — JSON по ../docs/contracts/recognition-result.schema.json. Структура ищется по содержимому (05-domain-rules.md §1.0, §12), синонимы — rules/*.yaml.
|
||||
Репозиторий tabular-parser — часть системы АРМ КА (ДКБ, Россельхозбанк): stateless-сервис разбора XLS/XLSX/XLSB (ОСВ по счетам, ББ/ОФР из Excel). Без доступа к БД: вход — NATS files.tabular_uploaded + файл в S3, выход — JSON по ../docs/contracts/recognition-result.schema.json. Структура ищется по содержимому (05-domain-rules.md §1.0, §12), синонимы — rules/*.yaml.
|
||||
Workspace из нескольких репозиториев; общий контекст — ../CLAUDE.md и ../docs/ (репозиторий arm-ka-docs).
|
||||
Источники истины по приоритету: ../docs/model/data-model-v5.17.dbml → ../docs/model/entities-v5.17.md →
|
||||
../docs/context/05-domain-rules.md → страницы ТЗ в ../docs/source/tz (черновик, старые имена v4).
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user