diff --git a/api/src/app.py b/api/src/app.py index a34d826..23a872d 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -3,11 +3,13 @@ - `create_app(settings)` строит горячую папку (`create_hotfolder`, без обращения к ФС/шаре) и `HealthProvider`; `GET /healthcheck` отдаёт `uptime`, `version`, `nats` (`ok`/`down`/`disabled`), `hotfolder` (`ok`/`down`). - lifespan: регистрация SMB-сессии (только `SCAN_HOTFOLDER_MODE=smb`), S3 (`S3Storage`), NATS (`NatsGateway`), воркер - (`Worker` с `NullNormalizer`). Подключение к NATS блокирует до доступности сервера, поэтому идёт фоновой задачей: - HTTP отвечает сразу (`nats = down`), подписка `recognition.image.request` появляется после подключения. Сбой этой - задачи (например, `NatsConfigError`) — лог error с трассировкой и `nats = down`, HTTP-сервер жив. + (`Worker` с `OcrNormalizer`). Справочники нормализатора загружаются из `SCAN_RULES_DIR` в `create_app`: битый + справочник — `ValueError`, сервис не стартует. Подключение к NATS блокирует до доступности сервера, поэтому идёт + фоновой задачей: HTTP отвечает сразу (`nats = down`), подписка `recognition.image.request` появляется после + подключения. Сбой этой задачи (например, `NatsConfigError`) — лог error с трассировкой и `nats = down`, + HTTP-сервер жив. Остановка: отмена подключения, остановка воркера (задания в работе отменяются), закрытие NATS и S3. -- Объекты lifespan доступны в `app.state`: `worker`, `nats`, `storage`, `health` (тесты, задачи 13–14). +- Объекты lifespan доступны в `app.state`: `worker`, `nats`, `storage`, `normalizer`, `health`. - `NATS_ENABLED=false` — подключения нет (`nats = disabled`), воркер запросов не получает (dev без NATS). """ @@ -29,7 +31,8 @@ from src.core.health import HealthProvider from src.core.nats import NatsGateway from src.core.worker import Worker from src.gateway import create_hotfolder -from src.normalizer import NullNormalizer +from src.normalizer import OcrNormalizer +from src.normalizer.rules import load_rules async def run_nats(gateway: NatsGateway, worker: Worker) -> None: @@ -60,6 +63,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: version = settings.VERSION up_time = datetime.now(UTC) hotfolder = create_hotfolder(settings) + normalizer = OcrNormalizer(load_rules(settings.SCAN_RULES_DIR)) # битый справочник — сервис не стартует (D-7) health = HealthProvider(nats_enabled=settings.NATS_ENABLED, hotfolder=hotfolder) @asynccontextmanager @@ -76,7 +80,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: logger.exception("Ошибка при регистрации сессии SMB") storage = S3Storage.from_settings(settings.OPENBAO.MINIO) gateway = NatsGateway(settings) - worker = Worker(settings, storage=storage, publisher=gateway, hotfolder=hotfolder, normalizer=NullNormalizer()) + worker = Worker(settings, storage=storage, publisher=gateway, hotfolder=hotfolder, normalizer=normalizer) health.nats = gateway app.state.storage = storage app.state.nats = gateway @@ -99,6 +103,7 @@ def create_app(settings: Settings | None = None) -> FastAPI: app = FastAPI(title="scan-gateway", version=version, root_path=settings.ROOT_PATH, lifespan=lifespan) app.state.health = health + app.state.normalizer = normalizer @app.get("/healthcheck", status_code=200) async def healthcheck(): diff --git a/api/src/core/worker.py b/api/src/core/worker.py index aaad8f4..51d370b 100644 --- a/api/src/core/worker.py +++ b/api/src/core/worker.py @@ -1,7 +1,7 @@ """Воркер scan-gateway: запрос NATS → ack → S3 → горячая папка → набор ответов ИС → склейка → S3 `raw/` → -нормализатор → конверт. +очистка горячей папки (под флагом) → нормализатор → конверт. -Spec image-recognition-gateway; design D-2, D-3, D-4, D-5, D-7; решения R-14..R-20, R-24, R-25, R-30. +Spec image-recognition-gateway; design D-1, D-2, D-3, D-4, D-5, D-7; решения R-14..R-20, R-24, R-25, R-30. - `on_request(msg)` — колбэк подписки NATS. Только разбирает и проверяет тело, регистрирует `job_id` и запускает задачу задания; ничего не ждёт (nats-py вызывает колбэки одной подписки последовательно, T11.4). @@ -14,6 +14,11 @@ Spec image-recognition-gateway; design D-2, D-3, D-4, D-5, D-7; решения R - Ошибки — результат, не молчание (таблица D-7): каждое задание заканчивается опубликованным конвертом; исключение задания наружу не выходит (кроме отмены при остановке); неудачная публикация — лог error (ядро закроет задание по `reply_deadline`). +- Входящий файл — `.` (D-1): расширение берётся из `original_name`; нет допустимого расширения — + `RECOGNITION_FAILED` без записи в горячую папку. +- Очистка (D-2): сразу после сохранения сырого ответа в `raw/` (любой исход разбора) воркер вызывает + `HotFolder.cleanup`; удаляет она только при `SCAN_CLEANUP=true`. При `TIMEOUT`, сбое чтения S3 или записи в горячую + папку очистки нет; ошибка очистки — одна строка лога, исход задания не меняется. - Методы горячей папки синхронные (R-9) — вызываются через `asyncio.to_thread`; нормализатор — тоже в потоке. """ @@ -161,9 +166,10 @@ class _Rejection: @dataclass class _JobContext: - """Состояние задания, нужное для конверта при непредвиденной ошибке: сырой ответ уже в S3.""" + """Состояние задания для конверта при непредвиденной ошибке и для очистки: сырой ответ уже в S3, имя файла.""" raw_s3_key: str | None = None + inbox_name: str | None = None @dataclass(frozen=True) @@ -445,6 +451,7 @@ class Worker: if isinstance(sent, RecognitionResultMessage): return sent name, started = sent + context.inbox_name = name job_id = str(request.job_id) # 3. Опрос исходящего каталога до min(reply_deadline, старт + SCAN_TIMEOUT_SECONDS) (D-3, R-20) deadline = min(request.reply_deadline, started + timedelta(seconds=self._settings.SCAN_TIMEOUT_SECONDS)) @@ -511,14 +518,15 @@ class Worker: f"исходный файл {source.bucket}/{source.key} не прочитан из S3: {_describe(exc)}", ) - # 2. Во входящий каталог как _ (R-5, R-15) + # 2. Во входящий каталог как . (D-1; расширение — из original_name) try: name = inbox_name(job_id, request.original_name) except ValueError: return self._error( request, ErrorCode.RECOGNITION_FAILED, - f"недопустимое имя файла {_fragment(request.original_name)}: файл не передан в ИС Сканирование", + f"у файла {_fragment(request.original_name)} нет допустимого расширения: " + "файл не передан в ИС Сканирование", ) try: await asyncio.to_thread(self._hotfolder.put, name, data) @@ -557,6 +565,8 @@ class Worker: request, ErrorCode.UNKNOWN_ERROR, f"сырой ответ ИС не сохранён в S3 ({key}): {_describe(exc)}" ) context.raw_s3_key = key + # набор прочитан и сохранён — следы задания в горячей папке больше не нужны (D-2) + await self._cleanup(job_id, context.inbox_name) # 5. Нормализатор получает склеенный ответ (D-5); его исключение — UNKNOWN_ERROR с сохранённым сырым ответом try: @@ -609,6 +619,23 @@ class Worker: metrics=metrics, ) + async def _cleanup(self, job_id: str, name: str | None) -> None: + """Очистка горячей папки после прочтения набора (D-2): удаляет только при `SCAN_CLEANUP` (решает горячая папка); + ошибка — одна строка лога, исход задания не меняется.""" + if name is None: + return + try: + removed = await asyncio.to_thread(self._hotfolder.cleanup, job_id, name) + except Exception as exc: # noqa: BLE001 — любая ошибка удаления не влияет на конверт + logger.warning( + "Очистка горячей папки не удалась, задание продолжается. job_id={job_id}, error={error}", + job_id=job_id, + error=_describe(exc), + ) + return + if removed: + logger.info("Горячая папка очищена. job_id={job_id}, name={name}", job_id=job_id, name=name) + async def _wait_answer(self, job_id: str, deadline: datetime) -> _WaitResult: """Опрос `poll(job_id)` каждые `SCAN_POLL_SECONDS` до дедлайна (последний опрос — в момент дедлайна). diff --git a/api/tests/unit/conftest.py b/api/tests/unit/conftest.py index 32697eb..3de0c1c 100644 --- a/api/tests/unit/conftest.py +++ b/api/tests/unit/conftest.py @@ -13,7 +13,7 @@ from src.core.nats import NatsGateway from src.core.worker import Worker from src.gateway.hotfolder import LocalHotFolder from src.normalizer import NullNormalizer -from tests.worker_helpers import SUCCESS_ANSWER, Env, EnvFactory, FakeNatsClient, FakeStorage, SpyHotFolder +from tests.worker_helpers import PDF_BYTES, SUCCESS_ANSWER, Env, EnvFactory, FakeNatsClient, FakeStorage, SpyHotFolder @pytest.fixture @@ -38,6 +38,7 @@ async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]: fixtures = tmp_path / "fixtures" fixtures.mkdir(exist_ok=True) (fixtures / "ok_doc.json").write_text(json.dumps(SUCCESS_ANSWER, ensure_ascii=False), encoding="utf-8") + (fixtures / "ok_doc.pdf").write_bytes(PDF_BYTES) # пара ответа по sha256 содержимого (эмулятор, Task 14) settings = get_settings().model_copy( update={ "SCAN_HOTFOLDER_PATH": str(root), @@ -59,6 +60,7 @@ async def make_env(tmp_path: Path) -> AsyncIterator[EnvFactory]: settings.SCAN_OUTPUT_DIR, settings.SCAN_TMP_DIR, settle_seconds=settings.SCAN_SETTLE_SECONDS, + cleanup=settings.SCAN_CLEANUP, ) hotfolder = SpyHotFolder(local, events) kwargs: dict[str, Any] = {} diff --git a/api/tests/unit/test_app.py b/api/tests/unit/test_app.py index 41587c9..c9ad7da 100644 --- a/api/tests/unit/test_app.py +++ b/api/tests/unit/test_app.py @@ -15,6 +15,7 @@ from src.app import create_app from src.core.config import get_settings from src.core.nats import NatsGateway from src.core.worker import Worker +from src.normalizer import OcrNormalizer pytestmark = pytest.mark.usefixtures("log_lines") @@ -215,3 +216,15 @@ async def test_unhandled_error_has_json_500(local_env: Path) -> None: assert response.status_code == 500 assert response.json() == {"detail": "внутренняя ошибка сервера"} + + +def test_create_app_fails_on_broken_rules(tmp_path: Path) -> None: + (tmp_path / "statement_lines.yaml").write_text("editions: []\n", encoding="utf-8") + settings = get_settings().model_copy(update={"SCAN_RULES_DIR": str(tmp_path), "SCAN_HOTFOLDER_PATH": str(tmp_path)}) + with pytest.raises(ValueError, match="statement_lines.yaml"): + create_app(settings) + + +def test_create_app_uses_ocr_normalizer() -> None: + app = create_app(get_settings().model_copy(update={"SCAN_HOTFOLDER_PATH": "."})) + assert isinstance(app.state.normalizer, OcrNormalizer) diff --git a/api/tests/unit/test_worker.py b/api/tests/unit/test_worker.py index 69be5e8..7c32076 100644 --- a/api/tests/unit/test_worker.py +++ b/api/tests/unit/test_worker.py @@ -27,7 +27,8 @@ from src.core.clients import JSON_CONTENT_TYPE from src.core.worker import MAX_ERROR_MESSAGE_CHARS, MAX_WARNING_CHARS, MAX_WARNINGS from src.gateway.hotfolder import AnswerFile from src.gateway.merge import dump_merged, merge_answers -from src.normalizer import NOT_IMPLEMENTED_MESSAGE, NOT_OBJECT_MESSAGE, NormalizeOutcome +from src.normalizer import NOT_IMPLEMENTED_MESSAGE, NOT_OBJECT_MESSAGE, NormalizeOutcome, OcrNormalizer +from src.normalizer.rules import load_rules from tests.worker_helpers import ( API_DIR, BUCKET, @@ -112,16 +113,22 @@ async def test_ack_is_published_before_s3_read_and_before_file_appears(make_env: async with emulator(env): await env.run(request) job_id = request["job_id"] - name = f"{job_id}_ok_doc.pdf" + name = f"{job_id}.pdf" kinds = [event for event in env.events if event[1] in (f"ack:{job_id}:1", request["s3"]["key"], name)] - assert kinds == [("publish", f"ack:{job_id}:1"), ("s3_get", request["s3"]["key"]), ("put", name)] + # очистка вызывается всегда после записи (шпион её журналирует; удаляет `HotFolder.cleanup` только при SCAN_CLEANUP) + assert kinds == [ + ("publish", f"ack:{job_id}:1"), + ("s3_get", request["s3"]["key"]), + ("put", name), + ("cleanup", name), + ] async def test_file_in_inbox_is_byte_for_byte_s3_object(make_env: EnvFactory) -> None: env = await make_env() request = env.request(original_name="aep_5.pdf") await env.deliver(request) - path = env.local.input_dir / f"{request['job_id']}_aep_5.pdf" + path = env.local.input_dir / f"{request['job_id']}.pdf" await eventually(path.exists) assert path.read_bytes() == PDF_BYTES await env.worker.stop() @@ -137,8 +144,8 @@ async def test_repeated_request_gets_second_ack_and_one_file(make_env: EnvFactor await eventually(lambda: len(env.acks(job_id)) == 2) await asyncio.sleep(0.05) assert env.acks(job_id) == [f"ack:{job_id}:1", f"ack:{job_id}:2"] - assert env.inbox() == [f"{job_id}_aep_5.pdf"] - assert env.hotfolder.puts == [f"{job_id}_aep_5.pdf"] + assert env.inbox() == [f"{job_id}.pdf"] + assert env.hotfolder.puts == [f"{job_id}.pdf"] assert len(env.storage.get_calls) == 1 @@ -152,7 +159,7 @@ async def test_repeated_request_after_result_gets_ack_without_reprocessing(make_ job_id = request["job_id"] assert env.acks(job_id) == [f"ack:{job_id}:1", f"ack:{job_id}:1"] assert len(env.results(job_id)) == 1 - assert env.hotfolder.puts == [f"{job_id}_ok_doc.pdf"] + assert env.hotfolder.puts == [f"{job_id}.pdf"] async def test_done_cache_is_bounded_lru(make_env: EnvFactory) -> None: @@ -201,7 +208,7 @@ async def test_success_answer_gives_not_implemented_with_raw_key(make_env: EnvFa assert _raw(env, job_id) == merged_raw(job_id, (f"{job_id}.json", answer, None)) # сырой ответ сохранён раньше публикации результата assert env.events.index(("s3_put", f"raw/{job_id}.json")) < env.events.index(("publish", f"result:{job_id}:3")) - assert env.inbox() == [f"{job_id}_ok_doc.pdf.done"] + assert env.inbox() == [f"{job_id}.pdf.done"] @pytest.mark.fixtures @@ -212,6 +219,7 @@ async def test_real_fixture_error_is_passed_through(make_env: EnvFactory, tmp_pa real.mkdir() for source in REAL_FIXTURES.glob("*.json"): shutil.copy(source, real / source.name) + (real / "aep_5.pdf").write_bytes(PDF_BYTES) # пара ответа `aep_5.json` по sha256 содержимого запроса (Task 14) expected = json.loads((real / "aep_5.json").read_text(encoding="utf-8"))["error"] env = await make_env() request = env.request(original_name="aep_5.PDF") @@ -226,11 +234,11 @@ async def test_real_fixture_error_is_passed_through(make_env: EnvFactory, tmp_pa async def test_unknown_name_gets_emulator_invalid_document(make_env: EnvFactory) -> None: env = await make_env() - request = env.request(original_name="unknown.pdf") + request = env.request(original_name="unknown.pdf", data=b"%PDF unknown content") # пары по содержимому нет async with emulator(env): body = await env.run(request) assert_error(body, ErrorCode.INVALID_DOCUMENT, raw=True) - assert "unknown.pdf" in body["error"]["message"] + assert "нет пары по содержимому" in body["error"]["message"] @pytest.mark.parametrize("status", [ResultStatus.DONE, ResultStatus.NEEDS_CHECK]) @@ -556,8 +564,8 @@ async def test_timeout_by_scan_timeout_keeps_file_and_never_resends(make_env: En elapsed = asyncio.get_running_loop().time() - started assert_error(body, ErrorCode.TIMEOUT, raw=False) assert 0.15 < elapsed < 2 - assert env.inbox() == [f"{request['job_id']}_aep_5.pdf"] # входящий файл остаётся - assert env.hotfolder.puts == [f"{request['job_id']}_aep_5.pdf"] # повторной отправки нет + assert env.inbox() == [f"{request['job_id']}.pdf"] # входящий файл остаётся + assert env.hotfolder.puts == [f"{request['job_id']}.pdf"] # повторной отправки нет assert env.hotfolder.polls >= 5 @@ -613,7 +621,7 @@ async def test_unusable_original_name_is_recognition_failed(make_env: EnvFactory env = await make_env() body = await env.run(env.request(original_name="..")) assert_error(body, ErrorCode.RECOGNITION_FAILED, raw=False) - assert "недопустимое имя файла" in body["error"]["message"] + assert "расширени" in body["error"]["message"] assert env.hotfolder.puts == [] @@ -881,3 +889,77 @@ async def test_service_info_and_handle_returns_published_message(make_env: EnvFa assert message.error is not None and message.error.code is ErrorCode.FILE_READ_ERROR [(_, body)] = env.results(str(request.job_id)) assert body == message.to_wire() + + +# --- Срез 3a: имя ., очистка под флагом, OcrNormalizer (D-1, D-2, решения №2, №3, №26) ----------------- +class TestSlice3a: + async def test_no_extension_is_recognition_failed(self, make_env: EnvFactory) -> None: + env = await make_env() + body = env.request(original_name="scan") + result = await env.run(body) + assert result["status"] == "error" and result["error"]["code"] == "RECOGNITION_FAILED" + assert "расширени" in result["error"]["message"] + assert env.inbox() == [] + + async def test_cleanup_after_answer_when_enabled(self, make_env: EnvFactory) -> None: + env = await make_env(SCAN_CLEANUP=True) + body = env.request() + async with emulator(env): + result = await env.run(body) + assert result["status"] == "error" and result["error"]["code"] == "NOT_IMPLEMENTED" # NullNormalizer + assert env.inbox() == [] # и .pdf, и .pdf.done удалены + assert not env.local.answer_dir(body["job_id"]).exists() + assert ("cleanup", f"{body['job_id']}.pdf") in env.events + + async def test_no_cleanup_when_disabled(self, make_env: EnvFactory) -> None: + env = await make_env(SCAN_CLEANUP=False) + body = env.request() + async with emulator(env): + await env.run(body) + assert env.inbox() == [f"{body['job_id']}.pdf.done"] + assert env.local.answer_dir(body["job_id"]).is_dir() + + async def test_no_cleanup_on_timeout(self, make_env: EnvFactory) -> None: + env = await make_env(SCAN_CLEANUP=True) + body = env.request(deadline_s=0.2) + result = await env.run(body) + assert result["error"]["code"] == "TIMEOUT" + assert env.inbox() == [f"{body['job_id']}.pdf"] + assert ("cleanup", f"{body['job_id']}.pdf") not in env.events + + async def test_cleanup_error_does_not_change_result( + self, make_env: EnvFactory, monkeypatch: pytest.MonkeyPatch + ) -> None: + env = await make_env(SCAN_CLEANUP=True) + + def deny(job_id: str, name: str) -> bool: + raise PermissionError("нет прав") + + monkeypatch.setattr(env.hotfolder.inner, "cleanup", deny) + body = env.request() + async with emulator(env): + result = await env.run(body) + assert result["status"] == "error" and result["error"]["code"] == "NOT_IMPLEMENTED" + assert "metrics" not in result or "warnings" not in result.get("metrics", {}) + + async def test_existing_input_waits_for_answer(self, make_env: EnvFactory) -> None: + """Review Focus 4: входящий файл уже лежит (повтор задания), ответа нет — TIMEOUT, файл не перезаписан.""" + env = await make_env() + body = env.request(deadline_s=0.3) + env.local.put(f"{body['job_id']}.pdf", b"already there") + result = await env.run(body) + assert result["error"]["code"] == "TIMEOUT" + assert (env.local.input_dir / f"{body['job_id']}.pdf").read_bytes() == b"already there" + assert env.inbox() == [f"{body['job_id']}.pdf"] + + async def test_ocr_normalizer_produces_documents(self, make_env: EnvFactory) -> None: + env = await make_env(normalizer=OcrNormalizer(load_rules(API_DIR / "rules"))) + body = env.request() + async with emulator(env): + result = await env.run(body) + assert result["status"] == "needs_check" # SUCCESS_ANSWER: confidence < 100 в шапке + key = result["result_s3"]["key"] + stored = json.loads(env.storage.objects[(BUCKET, key)]) + assert stored["documents"][0]["document_type"] == "BALANCE" + assert stored["documents"][0]["lines"][0]["line_code"] == "1600" + assert result["raw_s3_key"] == f"raw/{body['job_id']}.json" diff --git a/api/tests/unit/test_worker_concurrency.py b/api/tests/unit/test_worker_concurrency.py index 58fe82d..3f896a4 100644 --- a/api/tests/unit/test_worker_concurrency.py +++ b/api/tests/unit/test_worker_concurrency.py @@ -76,8 +76,8 @@ class SteppedSleep: def job_of(name: str) -> str: - """`job_id` из имени файла входящего каталога `_`.""" - return name.split("_", 1)[0] + """`job_id` из имени файла входящего каталога `.`.""" + return name.split(".", 1)[0] def answer(env: Env, job_id: str) -> None: @@ -152,7 +152,7 @@ async def test_five_jobs_with_four_slots(make_env: EnvFactory) -> None: waiting_request = requests[jobs.index(waiting)] assert max(index_of(env, ("publish", f"ack:{job}:1")) for job in jobs) < freed_result assert freed_result < index_of(env, ("s3_get", source_key(waiting_request))) - assert freed_result < index_of(env, ("put", f"{waiting}_ok_doc.pdf")) + assert freed_result < index_of(env, ("put", f"{waiting}.pdf")) # Все пять заданий заканчиваются конвертами await finish_all(env, poll_sleep, jobs) @@ -188,7 +188,7 @@ async def test_single_slot_serialises_jobs(make_env: EnvFactory) -> None: ] expected: list[tuple[str, str]] = [] for job in order: - expected += [("put", f"{job}_ok_doc.pdf"), ("publish", f"result:{job}:1")] + expected += [("put", f"{job}.pdf"), ("publish", f"result:{job}:1")] assert files_and_results == expected await finish_all(env, poll_sleep, jobs)