diff --git a/api/src/services/result_mapping.py b/api/src/services/result_mapping.py index 21439c0..dcb1010 100644 --- a/api/src/services/result_mapping.py +++ b/api/src/services/result_mapping.py @@ -17,6 +17,8 @@ PROCESSING` у файла, чьё ТЕКУЩЕЕ задание (`current_recogn Ошибки (D-2): неповторяемые (`MappingFatal`: ключ вне папки, объекта нет, JSON не проходит схему, нет документов ОСВ) — сразу файл `ERROR` и `metrics.mapping_error`; прочие — откат транзакции маппинга, в отдельной транзакции `metrics.mapping_attempts + 1`, на `MAPPING_MAX_ATTEMPTS` — `ERROR`; следующая попытка — после паузы цикла (R-15). +`mapping_error` виден пользователю: в нём категория (схема — число ошибок и путь первой; БД/S3 — тип исключения и число +попыток), без текста исключений pydantic/SQLAlchemy; полный текст — только в логе (M-1). Статус файла — через `services/file_status.py` (только текущее задание). `audit_event` не пишется: обмен и маппинг не аудируются (ASSUMPTION(OQ-55)); `FILE_REPLACE` автозамены — системное событие (`user_id = NULL`). """ @@ -175,6 +177,16 @@ def validate_result_key(candidate: Candidate, storage_bucket: str) -> None: # --- шаг 2: чтение результата --------------------------------------------------------------------------------------- +def _schema_error_summary(exc: ValidationError) -> str: + """M-1: для `metrics.mapping_error` (видно пользователю) — число ошибок, путь и тип первой; без `input_value` и + ссылок pydantic (полный текст — в логе через `__cause__` у `MappingFatal`).""" + errors = exc.errors(include_url=False, include_input=False, include_context=False) + first = errors[0] if errors else None + where = ".".join(map(str, first["loc"])) if first and first["loc"] else "<корень>" + kind = first["type"] if first else "?" + return f"результат не соответствует схеме recognition-result: ошибок {exc.error_count()}, первая: {where} {kind}" + + def _osv_documents(result: RecognitionResult) -> list[OsvDocument]: return [document for document in result.documents if isinstance(document, OsvDocument)] @@ -188,7 +200,7 @@ async def _load_result(storage: ObjectStorage, candidate: Candidate) -> Recognit try: result = RecognitionResult.model_validate_json(content) except ValidationError as exc: - raise MappingFatal(f"файл результата не соответствует схеме recognition-result: {exc}") from exc + raise MappingFatal(_schema_error_summary(exc)) from exc if result.job_id != candidate.job_id or result.file_id != candidate.file_id: raise MappingFatal( f"файл результата относится к другому заданию: job_id={result.job_id}, file_id={result.file_id}" @@ -405,6 +417,9 @@ async def map_file( # noqa: PLR0913 - все входы маппинга фай session.add_all(documents) await session.flush() + # T7.2: родитель раньше потомка на любой границе пачки (self-FK `parent_entry_id`) — не полагаемся на порядок строк + # результата; сортировка устойчивая, внутри уровня порядок строк сохраняется + entries.sort(key=lambda values: values["level"]) for start in range(0, len(entries), ENTRY_BATCH_SIZE): await session.execute(insert(OsvEntry).values(entries[start : start + ENTRY_BATCH_SIZE])) @@ -558,15 +573,18 @@ async def _lock_file(session: AsyncSession, file_id: uuid.UUID) -> StoredFile | return (await session.execute(stmt)).scalar_one_or_none() -async def _fail(session: AsyncSession, candidate: Candidate, message: str, now: datetime) -> None: - """Неповторяемая ошибка: `metrics.mapping_error` и файл текущего задания → `ERROR` (транзакция вызывающего).""" +async def _fail( + session: AsyncSession, candidate: Candidate, message: str, now: datetime, *, cause: BaseException | None = None +) -> None: + """Неповторяемая ошибка: `metrics.mapping_error` и файл текущего задания → `ERROR` (транзакция вызывающего). + `cause` — исходное исключение (например, `ValidationError`): его полный текст пишется только в лог (M-1).""" job = await _lock_job(session, candidate.job_id) if job is None: return job.metrics = {**_metrics(job), MAPPING_ERROR_KEY: _short(message)} await session.flush() changed = await set_file_status_for_job(session, job, FileStatus.ERROR, now=now) - logger.error( + logger.opt(exception=cause).error( "Маппинг результата невозможен: файл переведён в ERROR. error_category=mapping_fatal, job_id={job_id}, " "file_id={file_id}, file_changed={changed}, error={error}", job_id=candidate.job_id, @@ -595,8 +613,9 @@ async def _register_failure( metrics[MAPPING_ATTEMPTS_KEY] = attempts exhausted = attempts >= max_attempts if exhausted: + # M-1: только категория и тип исключения — текст SQLAlchemy/asyncpg ([SQL: …], параметры) остаётся в логе metrics[MAPPING_ERROR_KEY] = _short( - f"маппинг не удался после {attempts} попыток: {type(exc).__name__}: {exc}" + f"маппинг не удался: ошибка БД/S3 ({type(exc).__name__}), попыток {attempts}" ) job.metrics = metrics await session.flush() @@ -674,7 +693,7 @@ async def process_once( await _map_candidate(session_factory, candidate, result, clock) except MappingFatal as exc: async with session_factory() as session, session.begin(): - await _fail(session, candidate, str(exc), clock()) + await _fail(session, candidate, str(exc), clock(), cause=exc.__cause__) except Exception as exc: # сбой S3 или БД: транзакция маппинга откачена, повтор после паузы цикла (R-15) return await _register_failure( session_factory, candidate, exc, max_attempts=settings.MAPPING_MAX_ATTEMPTS, now=clock() diff --git a/api/tests/test_result_mapping.py b/api/tests/test_result_mapping.py index a678d3d..8cf3882 100644 --- a/api/tests/test_result_mapping.py +++ b/api/tests/test_result_mapping.py @@ -21,6 +21,7 @@ from typing import Any import pytest from loguru import logger from sqlalchemy import func, select +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from src.contract import ack_msg_id, result_msg_id @@ -378,12 +379,78 @@ async def test_mapping_failure_retries_then_error( assert await process_once(maker, storage, _settings()) is True job = await _job(maker, job_id) assert job.metrics[MAPPING_ATTEMPTS_KEY] == 3 # noqa: PLR2004 - MAPPING_MAX_ATTEMPTS по умолчанию - assert "подставная ошибка БД" in job.metrics[MAPPING_ERROR_KEY] + # M-1: пользователю — категория и число попыток, текст исключения остаётся только в логе + assert job.metrics[MAPPING_ERROR_KEY] == "маппинг не удался: ошибка БД/S3 (RuntimeError), попыток 3" assert (await _file(maker, file_id)).status is FileStatus.ERROR assert len(calls) == 3 # noqa: PLR2004 - три попытки assert await process_once(maker, storage, _settings()) is False +_INTERNALS = ("[SQL:", "input_value", "http", "sqlalche.me", "parameters") + + +async def test_schema_error_hides_input_values(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """M-1: ошибка схемы результата — в `mapping_error` число ошибок и путь/тип первой, без `input_value` и ссылок + pydantic (полный текст — только в логе).""" + maker, storage = committed_sessionmaker, FakeStorage() + result = _minimal() + _osv(result)["rows"][1]["opening_debit"] = "секрет http://example.invalid" + file_id, job_id, _ = await _prepare(maker, storage, result) + + assert await process_once(maker, storage, _settings()) is True + + assert (await _file(maker, file_id)).status is FileStatus.ERROR + message = (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY] + assert message.startswith("результат не соответствует схеме recognition-result: ошибок ") + assert "первая: documents.0." in message + assert "opening_debit" in message + assert "секрет" not in message + assert not [marker for marker in _INTERNALS if marker in message] + + +async def test_retryable_db_error_hides_sql( + committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch +): + """M-1: исчерпавшая попытки ошибка БД — в `mapping_error` категория, тип исключения и число попыток, без + `[SQL: …]`, параметров и ссылок SQLAlchemy.""" + maker, storage = committed_sessionmaker, FakeStorage() + file_id, job_id, _ = await _prepare(maker, storage, _minimal()) + + async def failing_map_file(session: AsyncSession, **kwargs: Any) -> list[str]: + raise IntegrityError("INSERT INTO osv_entry (id) VALUES ($1)", {"id": "секрет"}, Exception("fk violation")) + + monkeypatch.setattr(rm, "map_file", failing_map_file) + settings = _settings().model_copy(update={"MAPPING_MAX_ATTEMPTS": 1}) + + assert await process_once(maker, storage, settings) is True + + assert (await _file(maker, file_id)).status is FileStatus.ERROR + message = (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY] + assert message == "маппинг не удался: ошибка БД/S3 (IntegrityError), попыток 1" + assert not [marker for marker in _INTERNALS if marker in message] + + +async def test_child_before_parent_across_batches( + committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch +): + """T7.2: строки вставляются родитель раньше потомка независимо от порядка строк в результате — потомок перед + родителем на границе пачки не нарушает self-FK `parent_entry_id`.""" + maker, storage = committed_sessionmaker, FakeStorage() + result = _minimal() + rows = _osv(result)["rows"] + rows.reverse() # потомок (level 1) раньше корня (level 0) + monkeypatch.setattr(rm, "ENTRY_BATCH_SIZE", 1) # каждая строка — отдельная пачка + file_id, _, _ = await _prepare(maker, storage, result) + + assert await process_once(maker, storage, _settings()) is True + + assert (await _file(maker, file_id)).status is FileStatus.PROCESSED + [document] = await _documents(maker, file_id) + entries = await _entries(maker, document.id) + root, child = entries + assert child.parent_entry_id == root.id + + async def test_repeated_mapping_is_idempotent(committed_sessionmaker: async_sessionmaker[AsyncSession]): """«Повторный маппинг идемпотентен»: остатки прерванной вставки (документ и половина строк) удаляются, у файла ровно один набор документов и строк."""