feat(mapping): фоновый цикл маппинга результата в document/osv_entry (OQ-69, OQ-70)
services/result_mapping.py: pick_candidate (FOR UPDATE OF job SKIP LOCKED, файл PROCESSING и задание текущее),
проверка ключа results/{file_id}/{job_id}/ и бакета, чтение S3 вне транзакции, map_file одной транзакцией
(очистка документов/строк файла, документ на лист ОСВ, повтор счёта — предупреждение и NEEDS_CHECK первого листа,
строки пакетами по 1000, резолв контрагента COUNTERPARTY+BU/SUM (R-3), нарастающая по Q1 (D-9), group_schema,
metrics.mapping_warnings, PROCESSED через set_file_status_for_job). Неповторяемые ошибки — сразу ERROR с
metrics.mapping_error; прочие — откат и mapping_attempts, на MAPPING_MAX_ATTEMPTS — ERROR. run_mapping_loop —
третья задача ExchangeTasks при подключённом S3 (иначе предупреждение).
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
63343d3cba
commit
e704b028ec
@ -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
|
||||
|
||||
@ -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:
|
||||
"""Отменяет задачи и дожидается их завершения; исключения наружу не выходят."""
|
||||
|
||||
573
api/src/services/result_mapping.py
Normal file
573
api/src/services/result_mapping.py
Normal file
@ -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)
|
||||
@ -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]
|
||||
|
||||
@ -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 не подключено — цикла маппинга нет
|
||||
|
||||
|
||||
# --- инфраструктура ---------------------------------------------------------------------------------------------
|
||||
|
||||
690
api/tests/test_result_mapping.py
Normal file
690
api/tests/test_result_mapping.py
Normal file
@ -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
|
||||
@ -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)`.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@ -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`). Интеграционные тесты: сценарии спеки
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user