fix(mapping): категории вместо текста исключений в mapping_error, вставка строк по level
M-1 (= T7.6): metrics.mapping_error виден пользователю (JobBrief.mapping_error) — ошибка схемы даёт «результат не соответствует схеме recognition-result: ошибок N, первая: <loc> <type>» без input_value и ссылок pydantic; исчерпавшая попытки ошибка БД/S3 — «маппинг не удался: ошибка БД/S3 (<ExcType>), попыток N» без [SQL: …] и параметров. Полный текст исключения — только в логе. T7.2: черновики osv_entry сортируются устойчиво по level перед вставкой пачками — родитель раньше потомка на любой границе пачки, независимо от порядка строк результата. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012vJkmBGKKqMoEMzvsXG3y2
This commit is contained in:
Andrey Epifancev
parent
24d556e41e
commit
583edf2a8d
@ -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()
|
||||
|
||||
@ -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]):
|
||||
"""«Повторный маппинг идемпотентен»: остатки прерванной вставки (документ и половина строк) удаляются, у файла
|
||||
ровно один набор документов и строк."""
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user