diff --git a/api/src/app.py b/api/src/app.py index 3d59e9f..8f72817 100644 --- a/api/src/app.py +++ b/api/src/app.py @@ -7,8 +7,9 @@ middleware (gzip, proxy headers, CORS по `cors_options`; при DEBUG=False п middleware не подключается), healthcheck (uptime, version, nats: ok|down|off), раздача собранного SPA. Документация API (`/docs`, `/redoc`, `/openapi.json`) — только при DEBUG (OQ-66). Приложение собирает `create_app(settings)`. Недоступность S3 или NATS при старте не роняет и не задерживает приложение: ошибка логируется, healthcheck показывает -состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор и -вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) — и останавливает их до закрытия NATS. Сообщения +состояние. При NATS_ENABLED=true lifespan запускает фоновые задачи обмена `recognition_job` — outbox-публикатор, +вахтёр сроков (`src/services/recognition_exchange.py`, design D-6) и, если S3 подключено, цикл маппинга результатов +(`src/services/result_mapping.py`, Срез 2c D-1) — и останавливает их до закрытия NATS. Сообщения `recognition.result` (ack и результаты сервисов) обрабатывает inbox `handle_message` того же модуля с фабрикой сессий приложения (inbox меняет задание и статус файла; маппинг результата в доменные таблицы — отдельный цикл, Срез 2c D-1). Остановка: фоновые задачи обмена → NATS → S3 → SMB → пул соединений БД; каждый шаг независим — ошибка одного @@ -120,9 +121,10 @@ def _create_nats() -> NatsConnection | None: return NatsConnection(settings.OPENBAO.NATS, on_connected=_subscribe_results) -def _start_exchange(nats: NatsConnection) -> ExchangeTasks: - """Фоновые задачи обмена: публикатор outbox и вахтёр сроков (только вместе с NATS, т.е. при NATS_ENABLED=true).""" - return ExchangeTasks.start(async_session_maker, nats, settings) +def _start_exchange(nats: NatsConnection, storage: S3Storage | None) -> ExchangeTasks: + """Фоновые задачи обмена: публикатор outbox, вахтёр сроков и цикл маппинга результатов (последний — только при + подключённом S3); все — только вместе с NATS, т.е. при NATS_ENABLED=true.""" + return ExchangeTasks.start(async_session_maker, nats, settings, storage) def _smb_configured() -> bool: @@ -182,7 +184,7 @@ async def lifespan(app: FastAPI): app.state.nats = _create_nats() if app.state.nats is not None: await app.state.nats.start() # подключение в фоне: старт приложения не ждёт NATS - app.state.exchange = _start_exchange(app.state.nats) # публикатор сам ждёт соединения + app.state.exchange = _start_exchange(app.state.nats, app.state.storage) # публикатор ждёт NATS сам logger.info("Приложение запущено") yield diff --git a/api/src/services/recognition_exchange.py b/api/src/services/recognition_exchange.py index 78e18a5..dec11f8 100644 --- a/api/src/services/recognition_exchange.py +++ b/api/src/services/recognition_exchange.py @@ -76,9 +76,11 @@ from src.contract import ( ) from src.core.config import Settings from src.core.nats import NatsPublishError +from src.core.storage import ObjectStorage from src.models.enums import FileStatus, RecognitionPipeline, RecognitionStatus, S3BucketRole from src.models.files import RecognitionJob, S3Object, StoredFile from src.services.file_status import set_file_status_for_job, set_file_status_for_jobs +from src.services.result_mapping import run_mapping_loop WATCHDOG_PERIOD_SECONDS = 30 # период вахтёра сроков (spec «Таймаут ожидания результата») PUBLISH_BATCH_SIZE = 50 # заданий за один проход публикатора (design D-6, LIMIT 50) @@ -434,15 +436,20 @@ async def run_watchdog_loop( class ExchangeTasks: - """Фоновые задачи обмена в lifespan (design D-6): публикатор и вахтёр. Запускаются только при `NATS_ENABLED=true` - (вызывающий код создаёт их вместе с `NatsConnection`) и останавливаются до закрытия NATS.""" + """Фоновые задачи обмена в lifespan (design D-6; Срез 2c D-1): публикатор, вахтёр и — при подключённом S3 — цикл + маппинга результатов (`services/result_mapping.py`). Запускаются только при `NATS_ENABLED=true` (вызывающий код + создаёт их вместе с `NatsConnection`) и останавливаются до закрытия NATS.""" def __init__(self, tasks: Sequence[asyncio.Task[None]]) -> None: self.tasks = tuple(tasks) @classmethod def start( - cls, session_factory: async_sessionmaker[AsyncSession], nats: RequestPublisher, settings: Settings + cls, + session_factory: async_sessionmaker[AsyncSession], + nats: RequestPublisher, + settings: Settings, + storage: ObjectStorage | None = None, ) -> "ExchangeTasks": publisher = asyncio.create_task( run_publisher_loop( @@ -458,12 +465,23 @@ class ExchangeTasks: run_watchdog_loop(session_factory, period_seconds=WATCHDOG_PERIOD_SECONDS), name="recognition-deadline-watchdog", ) + tasks = [publisher, watchdog] + if storage is not None: + tasks.append( + asyncio.create_task( + run_mapping_loop(session_factory, storage, settings), name="recognition-result-mapper" + ) + ) + else: + logger.warning("S3 не подключено: маппинг результатов не запущен") logger.info( - "Фоновые задачи обмена запущены: публикатор (период {poll} с), вахтёр сроков (период {watchdog} с)", + "Фоновые задачи обмена запущены: публикатор (период {poll} с), вахтёр сроков (период {watchdog} с), " + "маппинг результатов: {mapping}", poll=settings.OUTBOX_POLL_SECONDS, watchdog=WATCHDOG_PERIOD_SECONDS, + mapping=f"период {settings.MAPPING_POLL_SECONDS} с" if storage is not None else "нет", ) - return cls([publisher, watchdog]) + return cls(tasks) async def stop(self) -> None: """Отменяет задачи и дожидается их завершения; исключения наружу не выходят.""" diff --git a/api/src/services/result_mapping.py b/api/src/services/result_mapping.py new file mode 100644 index 0000000..63cb0a2 --- /dev/null +++ b/api/src/services/result_mapping.py @@ -0,0 +1,573 @@ +"""Цикл маппинга результата распознавания в `document` + `osv_entry` (spec osv-result-mapping; design D-1, D-2, D-3, +D-7, D-9; рулинги R-3, R-4, R-5; docs/context/05-domain-rules.md §1.3, §1.5, §3 п. 4, §9.2, §9.3). + +Inbox файл результата не читает (OQ-67, OQ-69): признак «результат ещё не смапплен» — `stored_file.status = +PROCESSING` у файла, чьё ТЕКУЩЕЕ задание (`current_recognition_job_id`) уже `done`/`needs_check`. Один проход +`process_once` обрабатывает одного кандидата в три шага: + +1. короткая транзакция: `pick_candidate` (`FOR UPDATE OF recognition_job SKIP LOCKED`) и проверка ключа результата + (бакет хранилища, папка `results/{file_id}/{job_id}/`, OQ-70); +2. БЕЗ транзакции: чтение объекта из S3 и валидация `RecognitionResult` v1.x — блокировки не держатся на время сети; +3. транзакция маппинга: задание и файл перечитываются под блокировкой (порядок «задание → файл», как у inbox, + вахтёра и публикатора, — без взаимоблокировок); файл уже не `PROCESSING` или задание уже не текущее (результат + старого задания, Review Focus 1) — ничего не меняется. Иначе `map_file`: очистка документов и строк файла, + документы и строки, `group_schema`, предупреждения, `PROCESSED`. + +Ошибки (D-2): неповторяемые (`MappingFatal`: ключ вне папки, объекта нет, JSON не проходит схему, нет документов ОСВ) +— сразу файл `ERROR` и `metrics.mapping_error`; прочие — откат транзакции маппинга, в отдельной транзакции +`metrics.mapping_attempts + 1`, на `MAPPING_MAX_ATTEMPTS` — `ERROR`. Статус файла — только через +`services/file_status.py` (только текущее задание). `audit_event` не пишется: обмен и маппинг не аудируются +(ASSUMPTION(OQ-55)); `FILE_REPLACE` автозамены — отдельно. +""" + +import asyncio +import dataclasses +import uuid +from collections.abc import Awaitable, Callable, Mapping, Sequence +from dataclasses import dataclass +from datetime import UTC, datetime +from decimal import Decimal +from typing import Any + +from loguru import logger +from pydantic import ValidationError +from sqlalchemy import Integer, delete, func, insert, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.contract.result import OsvDocument, RecognitionResult +from src.core.config import Settings +from src.core.storage import ObjectNotFoundError, ObjectStorage +from src.domain.osv_mapping import ( + EntryDraft, + build_entries, + counterparty_check, + document_fields, + document_status, + opening_matches, +) +from src.models.accounting import ChartOfAccount +from src.models.directories import LegalEntity +from src.models.enums import ( + DocumentCheckResult, + DocumentType, + FileStatus, + OsvDataType, + OsvGroupType, + PeriodQuarter, + RecognitionStatus, +) +from src.models.files import Document, RecognitionJob, S3Object, StoredFile +from src.models.osv import OsvEntry +from src.repositories.counterparties import CounterpartyIndex +from src.services.file_status import set_file_status_for_job + +__all__ = [ + "ENTRY_BATCH_SIZE", + "MAPPING_ATTEMPTS_KEY", + "MAPPING_ERROR_KEY", + "MAPPING_WARNINGS_KEY", + "Candidate", + "MappingFatal", + "map_file", + "pick_candidate", + "process_once", + "run_mapping_loop", + "validate_result_key", +] + +MAPPING_ATTEMPTS_KEY = "mapping_attempts" # счётчик неудачных попыток в `recognition_job.metrics` (D-2) +MAPPING_ERROR_KEY = "mapping_error" # причина `ERROR` файла по маппингу — для карточки файла +MAPPING_WARNINGS_KEY = "mapping_warnings" # предупреждения маппинга файла, список строк (D-7, R-6) +ENTRY_BATCH_SIZE = 1000 # строк `osv_entry` на один INSERT: 23 колонки × 1000 < 32767 параметров asyncpg +MAPPING_ERROR_MAX_LENGTH = 2000 +RESULTS_PREFIX = "results" +MAPPABLE_JOB_STATUSES = (RecognitionStatus.done, RecognitionStatus.needs_check) +CUMULATIVE_CHECK_QUARTERS = frozenset({PeriodQuarter.Q2, PeriodQuarter.Q3, PeriodQuarter.Q4}) +RESOLVED_DATA_TYPES = frozenset({OsvDataType.BU, OsvDataType.SUM}) # резолв контрагента — §3 п. 4, R-3 +ROOT_LEVEL = 0 +_FORBIDDEN_KEY_SEGMENTS = frozenset({"", ".", ".."}) + +Clock = Callable[[], datetime] +SleepFn = Callable[[float], Awaitable[None]] + + +class MappingFatal(Exception): + """Ошибка без повтора (ключ, объект, схема, нет документов): файл сразу `ERROR` (D-2).""" + + +@dataclass(frozen=True) +class Candidate: + """Задание с результатом, ожидающим маппинга, и ключ файла результата.""" + + job_id: uuid.UUID + file_id: uuid.UUID + bucket: str + key: str + + +def _utcnow() -> datetime: + return datetime.now(UTC) + + +def _short(text: str) -> str: + return text if len(text) <= MAPPING_ERROR_MAX_LENGTH else text[: MAPPING_ERROR_MAX_LENGTH - 1] + "…" + + +def _metrics(job: RecognitionJob) -> dict[str, Any]: + """Копия `metrics` задания (не объект — `{}`): JSONB меняется только присваиванием нового dict.""" + return dict(job.metrics) if isinstance(job.metrics, dict) else {} + + +def _attempts(metrics: Mapping[str, Any]) -> int: + value = metrics.get(MAPPING_ATTEMPTS_KEY) + return value if isinstance(value, int) and not isinstance(value, bool) else 0 + + +# --- шаг 1: кандидат и ключ ----------------------------------------------------------------------------------------- + + +async def pick_candidate(session: AsyncSession, *, max_attempts: int) -> Candidate | None: + """Следующее задание для маппинга (D-1): `done`/`needs_check`, текущее для файла в `PROCESSING`, попыток меньше + `max_attempts`, со ссылкой на файл результата; старейшее по `finished_at`. Строка задания блокируется, занятые + другим проходом пропускаются (несколько инстансов ядра).""" + job = RecognitionJob + attempts = func.coalesce(job.metrics[MAPPING_ATTEMPTS_KEY].astext.cast(Integer), 0) + stmt = ( + select(job.id, job.file_id, S3Object.bucket, S3Object.key) + .join(StoredFile, (StoredFile.id == job.file_id) & (StoredFile.current_recognition_job_id == job.id)) + .join(S3Object, S3Object.id == job.result_s3_object_id) + .where( + job.status.in_(MAPPABLE_JOB_STATUSES), + StoredFile.status == FileStatus.PROCESSING, + attempts < max_attempts, + ) + .order_by(job.finished_at.asc().nulls_last(), job.id) + .limit(1) + .with_for_update(of=job, skip_locked=True) + ) + row = (await session.execute(stmt)).one_or_none() + if row is None: + return None + return Candidate(job_id=row.id, file_id=row.file_id, bucket=row.bucket, key=row.key) + + +def validate_result_key(candidate: Candidate, storage_bucket: str) -> None: + """OQ-70: результат читается только из бакета хранилища и только из папки своего задания + `results/{file_id}/{job_id}/<имя>`; сегменты `.`/`..`/пустые запрещены. Иначе — `MappingFatal`.""" + if candidate.bucket != storage_bucket: + raise MappingFatal( + f"файл результата в чужом бакете: {candidate.bucket}/{candidate.key} (ожидался бакет {storage_bucket})" + ) + prefix = f"{RESULTS_PREFIX}/{candidate.file_id}/{candidate.job_id}/" + rest = candidate.key[len(prefix) :] if candidate.key.startswith(prefix) else None + if rest is None or any(segment in _FORBIDDEN_KEY_SEGMENTS for segment in rest.split("/")): + raise MappingFatal(f"ключ файла результата вне папки задания: {candidate.key} (ожидалось {prefix}<имя>)") + + +# --- шаг 2: чтение результата --------------------------------------------------------------------------------------- + + +def _osv_documents(result: RecognitionResult) -> list[OsvDocument]: + return [document for document in result.documents if isinstance(document, OsvDocument)] + + +async def _load_result(storage: ObjectStorage, candidate: Candidate) -> RecognitionResult: + """Объект результата из S3 и его валидация по контракту; неповторяемые ошибки — `MappingFatal`.""" + try: + content = await storage.get_bytes(candidate.key) + except ObjectNotFoundError as exc: + raise MappingFatal(f"файл результата не найден в S3: {candidate.bucket}/{candidate.key}") from exc + try: + result = RecognitionResult.model_validate_json(content) + except ValidationError as exc: + raise MappingFatal(f"файл результата не соответствует схеме recognition-result: {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}" + ) + if not _osv_documents(result): + raise MappingFatal("в результате нет документов ОСВ") + return result + + +# --- шаг 3: маппинг файла ------------------------------------------------------------------------------------------- + + +def _with_counterparty(entry: EntryDraft, index: CounterpartyIndex) -> EntryDraft: + """R-3: `counterparty_id` — только строкам `COUNTERPARTY` с `data_type ∈ {BU, SUM}` (§3 п. 4, без policy).""" + if entry.group_type is not OsvGroupType.COUNTERPARTY or entry.data_type not in RESOLVED_DATA_TYPES: + return entry + return dataclasses.replace(entry, counterparty_id=index.resolve(entry.counterparty_inn, entry.counterparty_name)) + + +def _entry_values( + entry: EntryDraft, *, document_id: uuid.UUID, job_id: uuid.UUID, owner_id: uuid.UUID +) -> dict[str, Any]: + """Строка `osv_entry` (DBML v5.15) из черновика; суммы — `Decimal` из контракта без изменений.""" + opening_debit, opening_credit, turnover_debit, turnover_credit, closing_debit, closing_credit = entry.sums + return { + "id": entry.id, + "recognition_job_id": job_id, + "document_id": document_id, + "owner_legal_entity_id": owner_id, + "parent_entry_id": entry.parent_id, + "group_type": entry.group_type, + "group_name": entry.group_name, + "row_number": entry.row_number, + "level": entry.level, + "row_value": entry.row_value, + "data_type": entry.data_type, + "account_code": entry.account_code, + "account_id": entry.account_id, + "account_root": entry.account_root, + "counterparty_name": entry.counterparty_name, + "counterparty_inn": entry.counterparty_inn, + "counterparty_id": entry.counterparty_id, + "opening_debit": opening_debit, + "opening_credit": opening_credit, + "turnover_debit": turnover_debit, + "turnover_credit": turnover_credit, + "closing_debit": closing_debit, + "closing_credit": closing_credit, + "is_modified": False, + } + + +async def _clear_file(session: AsyncSession, file_id: uuid.UUID) -> None: + """D-3: строки и документы файла от прерванной попытки (файл в `PROCESSING` правок аналитика не содержит).""" + documents = select(Document.id).where(Document.stored_file_id == file_id).scalar_subquery() + await session.execute(delete(OsvEntry).where(OsvEntry.document_id.in_(documents))) + await session.execute(delete(Document).where(Document.stored_file_id == file_id)) + + +async def _q1_root_opening( + session: AsyncSession, *, legal_entity_id: uuid.UUID, account_code: str, year: int +) -> tuple[Decimal | None, Decimal | None] | None: + """D-9: сальдо на начало корня документа ОСВ Q1 того же ЮЛ, счёта и года из файла не `SUPERSEDED` (при нескольких — + последний созданный); корень — строка уровня 0 без родителя с минимальным `row_number`. Нет Q1/корня — `None`.""" + document_id = ( + await session.execute( + select(Document.id) + .join(StoredFile, StoredFile.id == Document.stored_file_id) + .where( + Document.legal_entity_id == legal_entity_id, + Document.document_type == DocumentType.OSV, + Document.osv_account_code == account_code, + Document.period_year == year, + Document.period_quarter == PeriodQuarter.Q1, + StoredFile.status != FileStatus.SUPERSEDED, + ) + .order_by(Document.created_at.desc().nulls_last(), Document.id) + .limit(1) + ) + ).scalar_one_or_none() + if document_id is None: + return None + root = ( + await session.execute( + select(OsvEntry.opening_debit, OsvEntry.opening_credit) + .where( + OsvEntry.document_id == document_id, + OsvEntry.level == ROOT_LEVEL, + OsvEntry.parent_entry_id.is_(None), + ) + .order_by(OsvEntry.row_number) + .limit(1) + ) + ).one_or_none() + return (root.opening_debit, root.opening_credit) if root is not None else None + + +def _group_schema(sheet: OsvDocument) -> dict[str, Any]: + return { + "locator": sheet.locator, + "groups": [{"type": group.type, "name": group.name, "has_inn": group.has_inn} for group in sheet.group_schema], + } + + +def _split_duplicates(sheets: Sequence[OsvDocument]) -> tuple[list[OsvDocument], dict[str, list[str]], list[str]]: + """D-7: первый лист каждого счёта — документ; повтор счёта в файле — только предупреждение, которое относится и к + первому листу (он получает `NEEDS_CHECK`). Возвращает листы-документы, предупреждения по `locator` первого листа и + все предупреждения о повторах.""" + first_locator: dict[str, str] = {} + kept: list[OsvDocument] = [] + by_locator: dict[str, list[str]] = {} + warnings: list[str] = [] + for sheet in sheets: + code = document_fields(sheet).osv_account_code + if code in first_locator: + warning = f"sheet={sheet.locator}: счёт {code} уже есть в листе {first_locator[code]}" + warnings.append(warning) + by_locator.setdefault(first_locator[code], []).append(warning) + continue + first_locator[code] = sheet.locator + kept.append(sheet) + return kept, by_locator, warnings + + +async def map_file( # noqa: PLR0913 - все входы маппинга файла именованные + session: AsyncSession, + *, + job: RecognitionJob, + file: StoredFile, + result: RecognitionResult, + chart: Mapping[str, uuid.UUID], + index: CounterpartyIndex, + now: datetime, +) -> list[str]: + """Маппинг результата в транзакции вызывающего (D-3, D-6…D-9); возвращает предупреждения маппинга файла. + + Очистка прежних документов/строк файла → документ на каждый лист ОСВ (повтор счёта — предупреждение) → строки + пакетами по `ENTRY_BATCH_SIZE` → `job.group_schema`, `metrics.mapping_warnings` → файл `PROCESSED` (только из + `PROCESSING` и только для текущего задания). Без commit. + """ + sheets = _osv_documents(result) + if not sheets: + raise MappingFatal("в результате нет документов ОСВ") + owner = await session.get(LegalEntity, file.legal_entity_id) + if owner is None: + raise MappingFatal(f"ЮЛ-владелец файла не найден: {file.legal_entity_id}") + owner_names = index.owner_names(owner) + job_done = job.status is RecognitionStatus.done + + await _clear_file(session, file.id) + kept, duplicate_warnings, warnings = _split_duplicates(sheets) + documents: list[Document] = [] + entries: list[dict[str, Any]] = [] + for sheet in kept: + fields = document_fields(sheet) + mapped = build_entries(sheet, chart) + sheet_warnings: list[str] = [] # предупреждения листа сверх повторов счёта (те уже в `warnings`) + if mapped.orphans: + sheet_warnings.append( + f"sheet={sheet.locator}: строки без родителя в результате: {', '.join(map(str, mapped.orphans))}" + ) + is_cumulative = fields.is_cumulative + if ( + fields.period_quarter in CUMULATIVE_CHECK_QUARTERS + and is_cumulative is False + and fields.period_year is not None + and mapped.root is not None + ): + q1_opening = await _q1_root_opening( + session, + legal_entity_id=file.legal_entity_id, + account_code=fields.osv_account_code, + year=fields.period_year, + ) + # ASSUMPTION(OQ-02): квартальная ОСВ с сальдо на начало, равным Q1, — нарастающая + if q1_opening is not None and opening_matches(mapped.root, q1_opening): + is_cumulative = True + sheet_warnings.append( + f"sheet={sheet.locator}: период {sheet.header.period_text}: нарастающая по данным Q1" + ) + check = counterparty_check(sheet.header, owner.inn, owner_names) + document = Document( + id=uuid.uuid4(), + stored_file_id=file.id, + legal_entity_id=file.legal_entity_id, + document_type=DocumentType.OSV, + osv_account_code=fields.osv_account_code, + period_year=fields.period_year, + period_quarter=fields.period_quarter, + is_cumulative=is_cumulative, + locator=fields.locator, + status=document_status( + job_done=job_done, + checks=sheet.checks, + counterparty=check, + warnings=[*duplicate_warnings.get(sheet.locator, ()), *sheet_warnings], + ), + period_check=DocumentCheckResult.UNKNOWN, # до версии консолидации (Срез 4) + counterparty_check=check, + created_at=now, + updated_at=now, + ) + documents.append(document) + warnings.extend(sheet_warnings) + entries.extend( + _entry_values( + _with_counterparty(entry, index), + document_id=document.id, + job_id=job.id, + owner_id=file.legal_entity_id, + ) + for entry in mapped.entries + ) + + session.add_all(documents) + await session.flush() + for start in range(0, len(entries), ENTRY_BATCH_SIZE): + await session.execute(insert(OsvEntry).values(entries[start : start + ENTRY_BATCH_SIZE])) + + job.group_schema = [_group_schema(sheet) for sheet in sheets] + job.metrics = {**_metrics(job), MAPPING_WARNINGS_KEY: warnings} + await session.flush() + if not await set_file_status_for_job( + session, job, FileStatus.PROCESSED, now=now, only_from={FileStatus.PROCESSING} + ): + raise RuntimeError(f"файл {file.id} не переведён в PROCESSED: не PROCESSING или задание {job.id} не текущее") + await session.refresh(file, ["status", "updated_at"]) + return warnings + + +# --- проход и цикл -------------------------------------------------------------------------------------------------- + + +async def _lock_job(session: AsyncSession, job_id: uuid.UUID) -> RecognitionJob | None: + stmt = ( + select(RecognitionJob) + .where(RecognitionJob.id == job_id) + .with_for_update() + .execution_options(populate_existing=True) + ) + return (await session.execute(stmt)).scalar_one_or_none() + + +async def _lock_file(session: AsyncSession, file_id: uuid.UUID) -> StoredFile | None: + stmt = ( + select(StoredFile).where(StoredFile.id == file_id).with_for_update().execution_options(populate_existing=True) + ) + 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` (транзакция вызывающего).""" + 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( + "Маппинг результата невозможен: файл переведён в ERROR. error_category=mapping_fatal, job_id={job_id}, " + "file_id={file_id}, file_changed={changed}, error={error}", + job_id=candidate.job_id, + file_id=candidate.file_id, + changed=changed, + error=message, + ) + + +async def _register_failure( + session_factory: async_sessionmaker[AsyncSession], + candidate: Candidate, + exc: Exception, + *, + max_attempts: int, + now: datetime, +) -> None: + """D-2: после отката — `metrics.mapping_attempts + 1` отдельной транзакцией; на `max_attempts` — файл `ERROR`.""" + async with session_factory() as session, session.begin(): + job = await _lock_job(session, candidate.job_id) + if job is None: + return + metrics = _metrics(job) + attempts = _attempts(metrics) + 1 + metrics[MAPPING_ATTEMPTS_KEY] = attempts + exhausted = attempts >= max_attempts + if exhausted: + metrics[MAPPING_ERROR_KEY] = _short( + f"маппинг не удался после {attempts} попыток: {type(exc).__name__}: {exc}" + ) + job.metrics = metrics + await session.flush() + if exhausted: + await set_file_status_for_job(session, job, FileStatus.ERROR, now=now) + logger.opt(exception=exc).error( + "Ошибка маппинга результата, транзакция откачена. error_category=mapping_error, job_id={job_id}, " + "file_id={file_id}, attempts={attempts}, file_error={exhausted}", + job_id=candidate.job_id, + file_id=candidate.file_id, + attempts=attempts, + exhausted=exhausted, + ) + + +async def _map_candidate( + session_factory: async_sessionmaker[AsyncSession], candidate: Candidate, result: RecognitionResult, clock: Clock +) -> None: + """Шаг 3: транзакция маппинга. Задание и файл перечитываются под блокировкой (порядок «задание → файл»); файл уже + не ждёт результата этого задания (его опередили, задание не текущее — Review Focus 1) — ничего не меняется.""" + async with session_factory() as session, session.begin(): + job = await _lock_job(session, candidate.job_id) + file = await _lock_file(session, candidate.file_id) + if ( + job is None + or file is None + or job.status not in MAPPABLE_JOB_STATUSES + or file.status is not FileStatus.PROCESSING + or file.current_recognition_job_id != job.id + ): + logger.info( + "Маппинг пропущен: файл уже не ждёт результата этого задания. job_id={job_id}, file_id={file_id}", + job_id=candidate.job_id, + file_id=candidate.file_id, + ) + return + rows = await session.execute(select(ChartOfAccount.code, ChartOfAccount.id)) + chart = {code: account_id for code, account_id in rows} + index = await CounterpartyIndex.load(session) + warnings = await map_file(session, job=job, file=file, result=result, chart=chart, index=index, now=clock()) + logger.info( + "Результат распознавания смапплен: файл PROCESSED. job_id={job_id}, file_id={file_id}, warnings={warnings}", + job_id=candidate.job_id, + file_id=candidate.file_id, + warnings=len(warnings), + ) + + +async def process_once( + session_factory: async_sessionmaker[AsyncSession], + storage: ObjectStorage, + settings: Settings, + *, + now: Clock | None = None, +) -> bool: + """Один проход цикла маппинга (модель шагов — в docstring модуля). `False` — кандидатов нет; `True` — кандидат + обработан (смапплен, отклонён, отложен до следующей попытки или пропущен, потому что его опередили).""" + clock = now or _utcnow + async with session_factory() as session, session.begin(): + candidate = await pick_candidate(session, max_attempts=settings.MAPPING_MAX_ATTEMPTS) + if candidate is None: + return False + try: + validate_result_key(candidate, storage.bucket) + except MappingFatal as exc: + await _fail(session, candidate, str(exc), clock()) + return True + + try: + result = await _load_result(storage, candidate) # шаг 2 — вне транзакции: блокировки не ждут S3 + 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()) + except Exception as exc: # сбой S3 или БД: транзакция маппинга откачена, повтор в следующем проходе + await _register_failure( + session_factory, candidate, exc, max_attempts=settings.MAPPING_MAX_ATTEMPTS, now=clock() + ) + return True + + +async def run_mapping_loop( + session_factory: async_sessionmaker[AsyncSession], + storage: ObjectStorage, + settings: Settings, + *, + sleep: SleepFn = asyncio.sleep, +) -> None: + """Бесконечный цикл маппинга: после обработанного кандидата — следующий сразу, без кандидатов — пауза + `MAPPING_POLL_SECONDS`; ошибка прохода логируется, цикл продолжается.""" + while True: + processed = False + try: + processed = await process_once(session_factory, storage, settings) + except Exception as exc: + logger.exception( + "Ошибка прохода цикла маппинга. error_category=mapping_loop_error, error_type={error_type}, " + "error={error}", + error_type=type(exc).__name__, + error=repr(exc), + ) + await sleep(0 if processed else settings.MAPPING_POLL_SECONDS) diff --git a/api/tests/test_app.py b/api/tests/test_app.py index 63bb893..c739343 100644 --- a/api/tests/test_app.py +++ b/api/tests/test_app.py @@ -122,7 +122,7 @@ def fake_engine(monkeypatch: pytest.MonkeyPatch) -> _FakeEngine: @pytest.fixture(autouse=True) def exchange_loops(monkeypatch: pytest.MonkeyPatch) -> list[str]: - """Циклы публикатора и вахтёра подменены ожиданием: быстрые тесты lifespan не ходят в БД.""" + """Циклы публикатора, вахтёра и маппинга подменены ожиданием: быстрые тесты lifespan не ходят в БД.""" started: list[str] = [] async def publisher(session_factory: Any, nats: Any, **kwargs: Any) -> None: @@ -133,8 +133,13 @@ def exchange_loops(monkeypatch: pytest.MonkeyPatch) -> list[str]: started.append("watchdog") await asyncio.Event().wait() + async def mapper(session_factory: Any, storage: Any, settings: Any, **kwargs: Any) -> None: + started.append("mapper") + await asyncio.Event().wait() + monkeypatch.setattr(recognition_exchange, "run_publisher_loop", publisher) monkeypatch.setattr(recognition_exchange, "run_watchdog_loop", watchdog) + monkeypatch.setattr(recognition_exchange, "run_mapping_loop", mapper) return started @@ -308,6 +313,54 @@ async def test_lifespan_starts_exchange_tasks_and_stops_them_before_nats( assert fake_nats.stopped +@pytest.mark.usefixtures("_no_smb") +async def test_lifespan_starts_mapping_loop_with_storage(monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str]): + """S3 подключено: третья фоновая задача — цикл маппинга результатов (Срез 2c, design D-1) с хранилищем + приложения; без S3 — только публикатор и вахтёр и предупреждение в лог.""" + storage = SimpleNamespace(client=_FakeS3Client(), bucket="arm-ka") + received: list[Any] = [] + + async def connect_s3() -> SimpleNamespace: + return storage + + async def mapper(session_factory: Any, mapper_storage: Any, settings: Any, **kwargs: Any) -> None: + received.append(mapper_storage) + exchange_loops.append("mapper") + await asyncio.Event().wait() + + monkeypatch.setattr(recognition_exchange, "run_mapping_loop", mapper) + monkeypatch.setattr(app_module, "_connect_s3", connect_s3) + monkeypatch.setattr(app_module, "_create_nats", _RecordingNats) + + async with app_module.lifespan(app_module.app): + exchange = app_module.app.state.exchange + assert exchange is not None + assert sorted(task.get_name() for task in exchange.tasks) == [ + "recognition-deadline-watchdog", + "recognition-outbox-publisher", + "recognition-result-mapper", + ] + await asyncio.sleep(0) + assert sorted(exchange_loops) == ["mapper", "publisher", "watchdog"] + assert received == [storage] + assert all(task.done() for task in exchange.tasks) + + async def unavailable() -> None: + return None + + exchange_loops.clear() + monkeypatch.setattr(app_module, "_connect_s3", unavailable) + records: list[Any] = [] + handler_id = logger.add(lambda message: records.append(message.record), level="WARNING") + try: + async with app_module.lifespan(app_module.app): + await asyncio.sleep(0) + assert sorted(exchange_loops) == ["publisher", "watchdog"] + finally: + logger.remove(handler_id) + assert any("маппинг результатов не запущен" in record["message"] for record in records) + + @pytest.mark.usefixtures("_no_smb") async def test_lifespan_without_nats_starts_no_exchange_tasks( monkeypatch: pytest.MonkeyPatch, exchange_loops: list[str] diff --git a/api/tests/test_exchange_lifespan_nats.py b/api/tests/test_exchange_lifespan_nats.py index af7e739..eea74f7 100644 --- a/api/tests/test_exchange_lifespan_nats.py +++ b/api/tests/test_exchange_lifespan_nats.py @@ -44,7 +44,7 @@ app_module = importlib.import_module("src.app") READY_TIMEOUT = 5.0 WAIT_TIMEOUT = 10.0 NATS_MONITOR_PORT = 8222 # мониторинг NATS из docs/dev/docker-compose.yml -EXCHANGE_TASKS = 2 # публикатор и вахтёр +EXCHANGE_TASKS = 2 # публикатор и вахтёр; S3 не подключено — цикла маппинга нет # --- инфраструктура --------------------------------------------------------------------------------------------- diff --git a/api/tests/test_result_mapping.py b/api/tests/test_result_mapping.py new file mode 100644 index 0000000..14e1854 --- /dev/null +++ b/api/tests/test_result_mapping.py @@ -0,0 +1,690 @@ +"""Цикл маппинга результата распознавания в `document`/`osv_entry` (spec osv-result-mapping; design D-1, D-2, D-3, D-7, +D-9; рулинги R-3, R-4, R-5; Review Focus 1, 2). + +Интеграционные тесты — на отдельной временной БД (`committed_sessionmaker`: цикл работает в собственных транзакциях), +результат лежит в `FakeStorage` под `results/{file_id}/{job_id}/result.json`. Синтетические результаты строятся из +примера контракта `contracts/examples/result-osv-minimal.json`; сценарий «Результат смапплен» и замер скорости — на +golden-снимках парсера (маркер `fixtures`). +""" + +import asyncio +import copy +import json +import time +import uuid +from collections.abc import Callable +from datetime import UTC, datetime +from decimal import Decimal +from pathlib import Path +from typing import Any + +import pytest +from loguru import logger +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.core.config import Settings, get_settings +from src.models.accounting import ChartOfAccount +from src.models.directories import LegalEntity +from src.models.enums import ( + DocumentCheckResult, + DocumentStatus, + DocumentType, + FileStatus, + OsvDataType, + OsvGroupType, + PeriodQuarter, + RecognitionStatus, +) +from src.models.files import Document, RecognitionJob, StoredFile +from src.models.osv import OsvEntry +from src.services import result_mapping as rm +from src.services.result_mapping import ( + MAPPING_ATTEMPTS_KEY, + MAPPING_ERROR_KEY, + MAPPING_WARNINGS_KEY, + Candidate, + MappingFatal, + process_once, + run_mapping_loop, + validate_result_key, +) +from tests.conftest import FakeStorage +from tests.factories import make_document, make_legal_entity, make_result_object, make_stored_file + +API_DIR = Path(__file__).resolve().parents[1] +MINIMAL_RESULT = API_DIR / "contracts" / "examples" / "result-osv-minimal.json" +GOLDEN_OSV_66 = Path("golden") / "expected_osv_parse" / "rb" / "2024-FY" / "ОСВ_66 за 2024.json" +GOLDEN_OSV_60 = Path("golden") / "expected_osv_parse" / "rb" / "2024-FY" / "ОСВ_60 за 2024.json" +BUCKET = "arm-ka" +OWNER_NAME = "АО «Русский бисквит»" +PERF_BUDGET_SECONDS = 2.0 # цель плана: маппинг ~1000 строк ОСВ_60 за 2024 быстрее 2 с + +pytestmark = pytest.mark.integration + + +# --- данные --------------------------------------------------------------------------------------------------------- + + +def _minimal() -> dict[str, Any]: + return json.loads(MINIMAL_RESULT.read_text("utf-8")) + + +def _golden(fixtures_dir: Path, path: Path) -> dict[str, Any]: + data = json.loads((fixtures_dir / path).read_text("utf-8")) + # у снимка нет job_id/file_id (подставляются) и версии сервиса (в контракте она обязательна) + data["service"] = {"version": "golden", **data["service"]} + return data + + +def _osv(result: dict[str, Any]) -> dict[str, Any]: + return result["documents"][0] + + +def _root(rows: list[dict[str, Any]]) -> dict[str, Any]: + return next(row for row in rows if row["level"] == 0 and row["parent_row_number"] is None) + + +def _settings() -> Settings: + return get_settings() + + +def _now() -> datetime: + return datetime.now(UTC) + + +async def _prepare( # noqa: PLR0913 - каждое поле подготовки задаётся тестом явно + maker: async_sessionmaker[AsyncSession], + storage: FakeStorage, + result: dict[str, Any], + *, + entity: LegalEntity | None = None, + owner_name: str = OWNER_NAME, + owner_inn: str | None = None, + job_status: RecognitionStatus = RecognitionStatus.done, + file_status: FileStatus = FileStatus.PROCESSING, + key: str | None = None, +) -> tuple[uuid.UUID, uuid.UUID, uuid.UUID]: + """ЮЛ-владелец, файл `PROCESSING` с текущим заданием `done`, результат в `storage`; `(file_id, job_id, entity_id)`. + + `job_id`/`file_id` результата подставляются; `key` — подменить ключ `s3_object` результата (OQ-70).""" + async with maker() as session, session.begin(): + owner = entity or await make_legal_entity(session, name=owner_name, inn=owner_inn) + file, job = await make_stored_file(session, entity=owner, status=file_status, job_status=job_status) + job.finished_at = _now() + job.metrics = copy.deepcopy(result.get("metrics")) # как inbox: метрики сервиса из конверта + payload = copy.deepcopy(result) | {"job_id": str(job.id), "file_id": str(file.id)} + s3 = await make_result_object(session, job=job, storage=storage, result=payload) + if key is not None: + storage.objects[key] = storage.objects[s3.key] + s3.key = key + return file.id, job.id, owner.id + + +async def _file(maker: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> StoredFile: + async with maker() as session: + file = await session.get(StoredFile, file_id) + assert file is not None + return file + + +async def _job(maker: async_sessionmaker[AsyncSession], job_id: uuid.UUID) -> RecognitionJob: + async with maker() as session: + job = await session.get(RecognitionJob, job_id) + assert job is not None + return job + + +async def _documents(maker: async_sessionmaker[AsyncSession], file_id: uuid.UUID) -> list[Document]: + async with maker() as session: + rows = await session.execute(select(Document).where(Document.stored_file_id == file_id)) + return list(rows.scalars()) + + +async def _entries(maker: async_sessionmaker[AsyncSession], document_id: uuid.UUID) -> list[OsvEntry]: + async with maker() as session: + rows = await session.execute( + select(OsvEntry).where(OsvEntry.document_id == document_id).order_by(OsvEntry.row_number) + ) + return list(rows.scalars()) + + +# --- проверка ключа результата (без БД) ---------------------------------------------------------------------------- + + +def _candidate(key: str, bucket: str = BUCKET) -> Candidate: + return Candidate(job_id=JOB_ID, file_id=FILE_ID, bucket=bucket, key=key) + + +JOB_ID = uuid.UUID("3f2a9c1e-7b4d-4e0a-9d51-2c8e6a1b0f01") +FILE_ID = uuid.UUID("a1c4e7d2-5b36-4f89-8e12-9d0c3b7a4e02") + + +@pytest.mark.parametrize( + "key", + [f"results/{FILE_ID}/{JOB_ID}/result.json", f"results/{FILE_ID}/{JOB_ID}/sub/result.json"], +) +def test_validate_result_key_accepts_job_folder(key: str) -> None: + validate_result_key(_candidate(key), BUCKET) + + +@pytest.mark.parametrize( + ("key", "bucket"), + [ + (f"results/{uuid.uuid4()}/{JOB_ID}/result.json", BUCKET), # чужой файл + (f"results/{FILE_ID}/{uuid.uuid4()}/result.json", BUCKET), # чужое задание + (f"uploads/{FILE_ID}/{JOB_ID}/result.json", BUCKET), # не папка результатов + (f"results/{FILE_ID}/{JOB_ID}/", BUCKET), # нет имени объекта + (f"results/{FILE_ID}/{JOB_ID}/../{uuid.uuid4()}/result.json", BUCKET), # выход из папки + (f"results/{FILE_ID}/{JOB_ID}/result.json", "other-bucket"), # чужой бакет + ], +) +def test_validate_result_key_rejects_outside_job_folder(key: str, bucket: str) -> None: + with pytest.raises(MappingFatal): + validate_result_key(_candidate(key, bucket), BUCKET) + + +# --- сценарии спеки ------------------------------------------------------------------------------------------------ + + +@pytest.mark.fixtures +async def test_golden_osv_66_mapped( # noqa: PLR0915 - один сценарий спеки целиком + committed_sessionmaker: async_sessionmaker[AsyncSession], fixtures_dir: Path +): + """«Результат смапплен», «Дерево и счета»: документ 66/2024/Q4/нарастающая, строки без TOTAL, `group_schema`.""" + maker, storage = committed_sessionmaker, FakeStorage() + golden = _golden(fixtures_dir, GOLDEN_OSV_66) + file_id, job_id, entity_id = await _prepare(maker, storage, golden) + + assert await process_once(maker, storage, _settings()) is True + + file = await _file(maker, file_id) + assert file.status is FileStatus.PROCESSED + assert file.updated_at is not None + [document] = await _documents(maker, file_id) + assert document.document_type is DocumentType.OSV + assert document.osv_account_code == "66" + assert document.period_year == 2024 # noqa: PLR2004 - год из шапки снимка + assert document.period_quarter is PeriodQuarter.Q4 + assert document.is_cumulative is True + assert document.locator == "sheet:Лист_1" + assert document.legal_entity_id == entity_id + assert document.status is DocumentStatus.RECOGNIZED # done, проверки чистые, шапка = владелец по имени + assert document.counterparty_check is DocumentCheckResult.MATCH + assert document.period_check is DocumentCheckResult.UNKNOWN + + rows = _osv(golden)["rows"] + entries = await _entries(maker, document.id) + assert len(entries) == len(rows) - 1 # всё, кроме TOTAL + total_rows = {row["row_number"] for row in rows if row["group_type"] == "TOTAL"} + assert total_rows and not total_rows & {entry.row_number for entry in entries} # в enum DBML TOTAL нет + by_row = {entry.row_number: entry for entry in entries} + for row in rows: + if row["group_type"] == "TOTAL": + continue + entry = by_row[row["row_number"]] + assert entry.recognition_job_id == job_id + assert entry.owner_legal_entity_id == entity_id + assert entry.is_modified is False + assert entry.level == row["level"] + assert entry.row_value == row["row_value"] + for name in ("opening_debit", "opening_credit", "turnover_debit", "turnover_credit"): + expected = row.get(name) + assert getattr(entry, name) == (Decimal(expected) if expected is not None else None) + if row["parent_row_number"] is None: + assert entry.parent_entry_id is None + else: + assert entry.parent_entry_id == by_row[row["parent_row_number"]].id + + async with maker() as session: + chart_6603 = (await session.execute(select(ChartOfAccount.id).where(ChartOfAccount.code == "66.03"))).scalar() + [yug] = [e for e in entries if e.account_code == "66.03" and e.group_type is OsvGroupType.COUNTERPARTY] + assert yug.counterparty_name == "Юг-Финанс / ПЕКАРНЯ СЧАСТЛИВОЧНАЯ" + parent = next(e for e in entries if e.id == yug.parent_entry_id) + assert parent.account_code == "66.03" + assert parent.group_type is OsvGroupType.ACCOUNT + assert yug.account_root == "66" + assert yug.account_id == chart_6603 is not None + + job = await _job(maker, job_id) + assert job.group_schema == [ + { + "locator": "sheet:Лист_1", + "groups": [ + {"type": "ACCOUNT", "name": "Счет", "has_inn": None}, + {"type": "COUNTERPARTY", "name": "Контрагенты", "has_inn": None}, + ], + } + ] + assert job.metrics[MAPPING_WARNINGS_KEY] == [] + assert job.metrics["rows"] == 9 # noqa: PLR2004 - метрики сервиса сохранены рядом с ключами маппинга + + +@pytest.mark.fixtures +async def test_mapping_performance_osv_60(committed_sessionmaker: async_sessionmaker[AsyncSession], fixtures_dir: Path): + """Цель плана: маппинг самой длинной ОСВ (`ОСВ_60 за 2024`, ~1000 строк) — быстрее 2 с на один проход цикла.""" + maker, storage = committed_sessionmaker, FakeStorage() + golden = _golden(fixtures_dir, GOLDEN_OSV_60) + file_id, _, _ = await _prepare(maker, storage, golden) + + started = time.perf_counter() + assert await process_once(maker, storage, _settings()) is True + elapsed = time.perf_counter() - started + + [document] = await _documents(maker, file_id) + rows = _osv(golden)["rows"] + entries = await _entries(maker, document.id) + total_rows = sum(1 for row in rows if row["group_type"] == "TOTAL") + assert len(entries) == len(rows) - total_rows + logger.info("Маппинг ОСВ_60 за 2024: {rows} строк за {elapsed:.3f} с", rows=len(entries), elapsed=elapsed) + print(f"\nPERF: ОСВ_60 за 2024 — {len(entries)} строк за {elapsed:.3f} с") # noqa: T201 - замер в отчёт + assert elapsed < PERF_BUDGET_SECONDS + + +async def test_result_key_outside_job_folder_is_error(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """«Ключ результата вне папки задания»: файл `ERROR` с ошибкой маппинга, документов нет, повтора нет.""" + maker, storage = committed_sessionmaker, FakeStorage() + file_id, job_id, _ = await _prepare( + maker, storage, _minimal(), key=f"results/{uuid.uuid4()}/{uuid.uuid4()}/result.json" + ) + + assert await process_once(maker, storage, _settings()) is True + + file = await _file(maker, file_id) + assert file.status is FileStatus.ERROR + job = await _job(maker, job_id) + assert "results/" in job.metrics[MAPPING_ERROR_KEY] + assert MAPPING_ATTEMPTS_KEY not in job.metrics # неповторяемая ошибка — без счётчика попыток + assert await _documents(maker, file_id) == [] + assert await process_once(maker, storage, _settings()) is False # повтора нет + + +async def test_missing_result_object_is_error(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """Объекта результата нет в S3 — неповторяемая ошибка (D-2): файл `ERROR`.""" + maker, storage = committed_sessionmaker, FakeStorage() + file_id, job_id, _ = await _prepare(maker, storage, _minimal()) + storage.objects.clear() + + assert await process_once(maker, storage, _settings()) is True + + assert (await _file(maker, file_id)).status is FileStatus.ERROR + assert "не найден" in (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY] + + +@pytest.mark.parametrize( + ("documents", "status", "message"), + [ + ([], "needs_check", "схем"), # контракт требует documents ≥ 1 — результат не проходит схему + ( + [{"document_type": "BALANCE", "locator": "pages:1", "header": {}, "columns": [], "lines": []}], + "needs_check", + "нет документов ОСВ", + ), + ], +) +async def test_result_without_osv_documents_is_error( + committed_sessionmaker: async_sessionmaker[AsyncSession], documents: list[Any], status: str, message: str +): + """Review Focus 2: результат без документов ОСВ → файл `ERROR` «нет документов ОСВ», не исключение цикла.""" + maker, storage = committed_sessionmaker, FakeStorage() + result = _minimal() | {"status": status, "documents": documents} + file_id, job_id, _ = await _prepare(maker, storage, result, job_status=RecognitionStatus.needs_check) + + assert await process_once(maker, storage, _settings()) is True + + assert (await _file(maker, file_id)).status is FileStatus.ERROR + assert message in (await _job(maker, job_id)).metrics[MAPPING_ERROR_KEY] + assert await _documents(maker, file_id) == [] + + +async def test_mapping_failure_retries_then_error( + committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch +): + """«Сбой маппинга с повтором»: откат (документов нет), `mapping_attempts` 1, 2 — файл `PROCESSING`; третья + неудача — `ERROR` с `metrics.mapping_error`.""" + maker, storage = committed_sessionmaker, FakeStorage() + file_id, job_id, _ = await _prepare(maker, storage, _minimal()) + real_map_file = rm.map_file + calls: list[int] = [] + + async def failing_map_file(session: AsyncSession, **kwargs: Any) -> list[str]: + await real_map_file(session, **kwargs) # вставка выполнена — и откатывается вместе с транзакцией + calls.append(1) + raise RuntimeError("подставная ошибка БД при вставке строк") + + monkeypatch.setattr(rm, "map_file", failing_map_file) + + for attempt in (1, 2): + assert await process_once(maker, storage, _settings()) is True + job = await _job(maker, job_id) + assert job.metrics[MAPPING_ATTEMPTS_KEY] == attempt + assert MAPPING_ERROR_KEY not in job.metrics + assert (await _file(maker, file_id)).status is FileStatus.PROCESSING + assert await _documents(maker, file_id) == [] + + 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] + 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 + + +async def test_repeated_mapping_is_idempotent(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """«Повторный маппинг идемпотентен»: остатки прерванной вставки (документ и половина строк) удаляются, у файла + ровно один набор документов и строк.""" + maker, storage = committed_sessionmaker, FakeStorage() + result = _minimal() + rows = _osv(result)["rows"] + rows.extend( + rows[-1] | {"row_number": number, "counterparty_name": f"ООО Лютик-{number}", "counterparty_inn": None} + for number in range(3, 7) + ) + file_id, job_id, entity_id = await _prepare(maker, storage, result) + async with maker() as session, session.begin(): + file = await session.get(StoredFile, file_id) + job = await session.get(RecognitionJob, job_id) + assert file is not None and job is not None + stale = await make_document(session, file=file, job=job, account="60", status=DocumentStatus.NEEDS_CHECK) + for row in rows[: len(rows) // 2]: + session.add( + OsvEntry( + recognition_job_id=job_id, + document_id=stale.id, + owner_legal_entity_id=entity_id, + row_number=row["row_number"], + level=row["level"], + row_value=row["row_value"], + ) + ) + + assert await process_once(maker, storage, _settings()) is True + + [document] = await _documents(maker, file_id) + assert document.id != stale.id + entries = await _entries(maker, document.id) + assert [entry.row_number for entry in entries] == [row["row_number"] for row in rows] + async with maker() as session: + total = ( + await session.execute( + select(func.count()) + .select_from(OsvEntry) + .join(Document, Document.id == OsvEntry.document_id) + .where(Document.stored_file_id == file_id) + ) + ).scalar_one() + assert total == len(rows) + + +async def test_two_sheets_of_same_account(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """«Два листа одного счёта»: один документ (первый лист), `NEEDS_CHECK`, предупреждение называет оба `locator`.""" + maker, storage = committed_sessionmaker, FakeStorage() + result = _minimal() + first = _osv(result) + second = copy.deepcopy(first) | {"locator": "sheet:Лист_2"} + second["header"]["account_code_raw"] = "60" + result["documents"] = [first, second] + file_id, job_id, _ = await _prepare(maker, storage, result, owner_inn="7700000000") + + assert await process_once(maker, storage, _settings()) is True + + [document] = await _documents(maker, file_id) + assert document.locator == "sheet:Лист_1" + assert document.osv_account_code == "60" + assert document.status is DocumentStatus.NEEDS_CHECK + assert len(await _entries(maker, document.id)) == len(first["rows"]) + job = await _job(maker, job_id) + assert job.metrics[MAPPING_WARNINGS_KEY] == ["sheet=sheet:Лист_2: счёт 60 уже есть в листе sheet:Лист_1"] + assert (await _file(maker, file_id)).status is FileStatus.PROCESSED + + +async def test_header_of_another_organization(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """«Шапка другой организации»: файл ЮЛ «Пекарня Счастливочная», в шапке `АО "Русский бисквит"` → `MISMATCH`, + `NEEDS_CHECK`.""" + maker, storage = committed_sessionmaker, FakeStorage() + result = _minimal() + _osv(result)["header"] |= {"organization_name": 'АО "Русский бисквит"', "organization_inn": None} + file_id, _, _ = await _prepare(maker, storage, result, owner_name="ООО «Пекарня Счастливочная»") + + assert await process_once(maker, storage, _settings()) is True + + [document] = await _documents(maker, file_id) + assert document.counterparty_check is DocumentCheckResult.MISMATCH + assert document.status is DocumentStatus.NEEDS_CHECK + + +async def _q1_document( + maker: async_sessionmaker[AsyncSession], + entity_id: uuid.UUID, + *, + opening_credit: Decimal, + file_status: FileStatus, +) -> None: + """Документ ОСВ 60 за 2024 Q1 другого файла того же ЮЛ с корнем (level 0) и строкой-потомком.""" + async with maker() as session, session.begin(): + entity = await session.get(LegalEntity, entity_id) + assert entity is not None + file, job = await make_stored_file( + session, entity=entity, status=file_status, job_status=RecognitionStatus.done + ) + document = await make_document(session, file=file, job=job, account="60", year=2024, quarter=PeriodQuarter.Q1) + common = {"recognition_job_id": job.id, "document_id": document.id, "owner_legal_entity_id": entity_id} + root = OsvEntry( + **common, row_number=5, level=0, row_value="60", opening_debit=None, opening_credit=opening_credit + ) + session.add(root) + await session.flush() + # строка уровня 1 с меньшим номером не должна считаться корнем + session.add(OsvEntry(**common, row_number=1, level=1, parent_entry_id=root.id, opening_credit=Decimal(1))) + + +@pytest.mark.parametrize( + ("q1_opening", "q1_file_status", "expected"), + [ + (Decimal("1000.00"), FileStatus.PROCESSED, True), # сальдо на начало совпало с Q1 + (Decimal("1000.01"), FileStatus.PROCESSED, True), # в пределах ±0.01 + (Decimal("999.00"), FileStatus.PROCESSED, False), # не совпало — по тексту + (Decimal("1000.00"), FileStatus.SUPERSEDED, False), # Q1 из заменённого файла не учитывается + ], +) +async def test_cumulative_by_q1_data( + committed_sessionmaker: async_sessionmaker[AsyncSession], + q1_opening: Decimal, + q1_file_status: FileStatus, + expected: bool, +): + """«Нарастающая по данным»: ОСВ 60 «за 3 квартал 2024» с `opening_*` корня = Q1 → `is_cumulative = true` и + предупреждение «нарастающая по данным Q1» (ASSUMPTION(OQ-02)).""" + maker, storage = committed_sessionmaker, FakeStorage() + async with maker() as session, session.begin(): + entity = await make_legal_entity(session, name=OWNER_NAME, inn="7700000000") + await _q1_document(maker, entity.id, opening_credit=q1_opening, file_status=q1_file_status) + result = _minimal() + _osv(result)["header"] |= { + "period_text": "3 квартал 2024 г.", + "period_year": 2024, + "period_quarter": "Q3", + "is_cumulative": False, + } + assert _root(_osv(result)["rows"])["opening_credit"] == "1000.00" + file_id, job_id, _ = await _prepare(maker, storage, result, entity=entity) + + assert await process_once(maker, storage, _settings()) is True + + [document] = await _documents(maker, file_id) + assert document.period_quarter is PeriodQuarter.Q3 + assert document.is_cumulative is expected + warnings = (await _job(maker, job_id)).metrics[MAPPING_WARNINGS_KEY] + if expected: + assert len(warnings) == 1 + assert "нарастающая по данным Q1" in warnings[0] + assert "sheet:Лист_1" in warnings[0] + assert document.status is DocumentStatus.NEEDS_CHECK + else: + assert warnings == [] + assert document.status is DocumentStatus.RECOGNIZED + + +@pytest.mark.parametrize( + ("job_status", "expected"), + [(RecognitionStatus.done, DocumentStatus.RECOGNIZED), (RecognitionStatus.needs_check, DocumentStatus.NEEDS_CHECK)], +) +async def test_clean_result( + committed_sessionmaker: async_sessionmaker[AsyncSession], + job_status: RecognitionStatus, + expected: DocumentStatus, +): + """«Чистый результат»: задание `done`, шапка = владелец (ИНН), нарушений нет → `RECOGNIZED`, `MATCH`, `UNKNOWN`; + задание `needs_check` → `NEEDS_CHECK`.""" + maker, storage = committed_sessionmaker, FakeStorage() + file_id, _, _ = await _prepare(maker, storage, _minimal(), owner_inn="7700000000", job_status=job_status) + + assert await process_once(maker, storage, _settings()) is True + + [document] = await _documents(maker, file_id) + assert document.status is expected + assert document.counterparty_check is DocumentCheckResult.MATCH + assert document.period_check is DocumentCheckResult.UNKNOWN + assert document.osv_account_code == "60" + assert document.period_quarter is PeriodQuarter.Q1 + assert document.is_cumulative is False + assert (await _file(maker, file_id)).status is FileStatus.PROCESSED + + +async def test_counterparty_resolved_in_entries(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """R-3: `counterparty_id` — только у `COUNTERPARTY` с `data_type ∈ {BU, SUM}`; `NU` хранится как пришёл, без + резолва; строка-счёт без контрагента.""" + maker, storage = committed_sessionmaker, FakeStorage() + async with maker() as session, session.begin(): + lyutik = await make_legal_entity(session, name="ООО «Лютик»", inn="7711111111") + result = _minimal() + rows = _osv(result)["rows"] + rows.append(rows[1] | {"row_number": 3, "data_type": "NU"}) + file_id, _, _ = await _prepare(maker, storage, result) + + assert await process_once(maker, storage, _settings()) is True + + [document] = await _documents(maker, file_id) + account, bu, nu = await _entries(maker, document.id) + assert account.group_type is OsvGroupType.ACCOUNT + assert account.counterparty_id is None + assert bu.counterparty_id == lyutik.id + assert bu.counterparty_inn == "7711111111" + assert nu.data_type is OsvDataType.NU + assert nu.counterparty_id is None + + +async def test_result_of_non_current_job_is_ignored(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """Review Focus 1: результат задания, которое уже не текущее для файла, не выбирается и файл не трогает.""" + maker, storage = committed_sessionmaker, FakeStorage() + file_id, old_job_id, _ = await _prepare(maker, storage, _minimal()) + async with maker() as session, session.begin(): + file = await session.get(StoredFile, file_id) + assert file is not None + new_job = RecognitionJob( + file_id=file_id, + pipeline=(await session.get(RecognitionJob, old_job_id)).pipeline, # type: ignore[union-attr] + request_subject="recognition.tabular.request", + schema_version="1.0", + status=RecognitionStatus.processing, + attempt=2, + ) + session.add(new_job) + await session.flush() + file.current_recognition_job_id = new_job.id + before = await _file(maker, file_id) + + assert await process_once(maker, storage, _settings()) is False + + after = await _file(maker, file_id) + assert (after.status, after.updated_at) == (FileStatus.PROCESSING, before.updated_at) + assert await _documents(maker, file_id) == [] + assert (await _job(maker, old_job_id)).group_schema is None + + +async def test_job_replaced_while_reading_result( + committed_sessionmaker: async_sessionmaker[AsyncSession], monkeypatch: pytest.MonkeyPatch +): + """Review Focus 1, гонка: пока цикл читал S3, у файла сменилось текущее задание → транзакция маппинга ничего не + меняет (задание и файл перечитываются под блокировкой).""" + maker, storage = committed_sessionmaker, FakeStorage() + file_id, job_id, _ = await _prepare(maker, storage, _minimal()) + real_get_bytes = storage.get_bytes + + async def get_bytes_and_replace_job(key: str) -> bytes: + async with maker() as session, session.begin(): + file = await session.get(StoredFile, file_id) + job = await session.get(RecognitionJob, job_id) + assert file is not None and job is not None + new_job = RecognitionJob( + file_id=file_id, + pipeline=job.pipeline, + request_subject=job.request_subject, + schema_version="1.0", + status=RecognitionStatus.queued, + attempt=2, + ) + session.add(new_job) + await session.flush() + file.current_recognition_job_id = new_job.id + return await real_get_bytes(key) + + monkeypatch.setattr(storage, "get_bytes", get_bytes_and_replace_job) + + assert await process_once(maker, storage, _settings()) is True + + file = await _file(maker, file_id) + assert file.status is FileStatus.PROCESSING + assert await _documents(maker, file_id) == [] + job = await _job(maker, job_id) + assert job.group_schema is None + assert MAPPING_ATTEMPTS_KEY not in (job.metrics or {}) + + +async def test_run_mapping_loop_processes_two_candidates(committed_sessionmaker: async_sessionmaker[AsyncSession]): + """`run_mapping_loop`: два кандидата подряд без паузы (`sleep(0)`), затем пауза `MAPPING_POLL_SECONDS`.""" + maker, storage = committed_sessionmaker, FakeStorage() + first, _, _ = await _prepare(maker, storage, _minimal()) + second, _, _ = await _prepare(maker, storage, _minimal()) + settings = _settings() + sleeps: list[float] = [] + + async def fake_sleep(seconds: float) -> None: + sleeps.append(seconds) + if len(sleeps) == 3: # noqa: PLR2004 - два кандидата и один пустой проход + raise asyncio.CancelledError + + with pytest.raises(asyncio.CancelledError): + await run_mapping_loop(maker, storage, settings, sleep=fake_sleep) + + assert sleeps == [0, 0, settings.MAPPING_POLL_SECONDS] + assert (await _file(maker, first)).status is FileStatus.PROCESSED + assert (await _file(maker, second)).status is FileStatus.PROCESSED + + +async def test_run_mapping_loop_survives_errors(monkeypatch: pytest.MonkeyPatch): + """Ошибка прохода логируется, цикл не умирает и ждёт `MAPPING_POLL_SECONDS`.""" + outcomes: list[Callable[[], bool]] = [lambda: (_ for _ in ()).throw(RuntimeError("сбой")), lambda: True] + sleeps: list[float] = [] + + async def fake_process_once(*args: Any, **kwargs: Any) -> bool: + return outcomes.pop(0)() + + async def fake_sleep(seconds: float) -> None: + sleeps.append(seconds) + if not outcomes: + raise asyncio.CancelledError + + monkeypatch.setattr(rm, "process_once", fake_process_once) + settings = _settings() + records: list[Any] = [] + handler_id = logger.add(lambda message: records.append(message.record), level="ERROR") + try: + with pytest.raises(asyncio.CancelledError): + await run_mapping_loop(None, FakeStorage(), settings, sleep=fake_sleep) # type: ignore[arg-type] + finally: + logger.remove(handler_id) + + assert sleeps == [settings.MAPPING_POLL_SECONDS, 0] + assert len(records) == 1 diff --git a/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md b/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md index c94a3f2..fa82bfd 100644 --- a/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md +++ b/docs/superpowers/plans/2026-10-05-slice-2c-osv-ingest.md @@ -228,9 +228,9 @@ async def run_mapping_loop(session_factory, storage, settings, *, sleep=asyncio. ``` `ExchangeTasks.start(session_factory, nats, settings, storage)` — третья задача `recognition-result-mapper` только при `storage is not None` (иначе warning «S3 не подключено: маппинг результатов не запущен»). -- [ ] **Step 1: Тесты** (integration, `committed_sessionmaker` + `FakeStorage`; результат — golden JSON `ОСВ_66 за 2024.json` с подставленными `job_id`/`file_id`): сценарии спеки `osv-result-mapping` «Результат смапплен» (документ 66/2024/Q4/cumulative, `locator`, 9 строк (10 строк снимка минус TOTAL), `group_schema`), «Ключ вне папки» → `ERROR`, «Сбой с повтором» (monkeypatch `map_file` бросает 2 раза → attempts, 3-й → `ERROR`), «Идемпотентность» (вставить половину строк вручную → `process_once` → ровно один набор), «Два листа одного счёта» (JSON с двумя документами `60`) → один документ `NEEDS_CHECK` + warning, «Шапка другой организации» → `MISMATCH`/`NEEDS_CHECK`, «Нарастающая по данным» (сначала документ Q1 через фабрики с корнем, затем Q3 с теми же opening) → `is_cumulative = True`, «Чистый результат» → `RECOGNIZED`; результат без документов (`status: error`-подобный JSON с `documents: []` не валиден по контракту → fatal) → `ERROR` (Review Focus 2); результат для задания, которое уже не текущее → не трогает файл (Review Focus 1); `run_mapping_loop` с заглушкой sleep обрабатывает два кандидата подряд. -- [ ] **Step 2–4:** падает → реализация → `pytest -m "integration or fixtures" tests/test_result_mapping.py -q`, `make check`. -- [ ] **Step 5: Commit** `feat(mapping): фоновый цикл маппинга результата в document/osv_entry (OQ-69, OQ-70)`. +- [x] **Step 1: Тесты** (integration, `committed_sessionmaker` + `FakeStorage`; результат — golden JSON `ОСВ_66 за 2024.json` с подставленными `job_id`/`file_id`): сценарии спеки `osv-result-mapping` «Результат смапплен» (документ 66/2024/Q4/cumulative, `locator`, 9 строк (10 строк снимка минус TOTAL), `group_schema`), «Ключ вне папки» → `ERROR`, «Сбой с повтором» (monkeypatch `map_file` бросает 2 раза → attempts, 3-й → `ERROR`), «Идемпотентность» (вставить половину строк вручную → `process_once` → ровно один набор), «Два листа одного счёта» (JSON с двумя документами `60`) → один документ `NEEDS_CHECK` + warning, «Шапка другой организации» → `MISMATCH`/`NEEDS_CHECK`, «Нарастающая по данным» (сначала документ Q1 через фабрики с корнем, затем Q3 с теми же opening) → `is_cumulative = True`, «Чистый результат» → `RECOGNIZED`; результат без документов (`status: error`-подобный JSON с `documents: []` не валиден по контракту → fatal) → `ERROR` (Review Focus 2); результат для задания, которое уже не текущее → не трогает файл (Review Focus 1); `run_mapping_loop` с заглушкой sleep обрабатывает два кандидата подряд. +- [x] **Step 2–4:** падает → реализация → `pytest -m "integration or fixtures" tests/test_result_mapping.py -q`, `make check`. +- [x] **Step 5: Commit** `feat(mapping): фоновый цикл маппинга результата в document/osv_entry (OQ-69, OQ-70)`. --- diff --git a/openspec/changes/slice-2c-osv-ingest/tasks.md b/openspec/changes/slice-2c-osv-ingest/tasks.md index fe6bd91..be4c2bc 100644 --- a/openspec/changes/slice-2c-osv-ingest/tasks.md +++ b/openspec/changes/slice-2c-osv-ingest/tasks.md @@ -35,7 +35,7 @@ `repositories/counterparties.py:CounterpartyIndex` + `resolve(row) -> uuid | None` (ИНН → имя → алиас, >1 → None). Unit-тесты: нормализация (`60.1`, `76.7.1`, `76.АВ`, `МЦ.04` → root None), дерево и TOTAL на `result-osv-minimal.json` и golden `ОСВ_66 за 2024`, резолв (ИНН/имя/алиас/неоднозначно), проверка шапки (ИНН, имя, алиас, пусто), статус документа. -- [ ] 3.2 **Цикл маппинга.** `services/result_mapping.py`: `pick_candidate` (FOR UPDATE SKIP LOCKED), проверка ключа (OQ-70), чтение +- [x] 3.2 **Цикл маппинга.** `services/result_mapping.py`: `pick_candidate` (FOR UPDATE SKIP LOCKED), проверка ключа (OQ-70), чтение S3 вне транзакции, `map_file` (одна транзакция: очистка прежних строк файла, `document`×N с обработкой повтора счёта, пакетная вставка `osv_entry`, `group_schema`, `is_cumulative` по Q1 (D-9), `metrics.mapping_warnings`, `PROCESSED`), повторы `mapping_attempts` → `ERROR`, `run_mapping_loop` в lifespan (`ExchangeTasks`). Интеграционные тесты: сценарии спеки