"""Цикл маппинга результата распознавания в `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`, автозамена прежних версий (`supersede_previous`, D-10, аудит `FILE_REPLACE`). Ошибки (D-2): неповторяемые (`MappingFatal`: ключ вне папки, объекта нет, JSON не проходит схему, нет документов ОСВ) — сразу файл `ERROR` и `metrics.mapping_error`; прочие — откат транзакции маппинга, в отдельной транзакции `metrics.mapping_attempts + 1`, на `MAPPING_MAX_ATTEMPTS` — `ERROR`; следующая попытка — после паузы цикла (R-15). Статус файла — через `services/file_status.py` (только текущее задание). `audit_event` не пишется: обмен и маппинг не аудируются (ASSUMPTION(OQ-55)); `FILE_REPLACE` автозамены — системное событие (`user_id = NULL`). """ 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, Select, delete, func, insert, select, update 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 ( AuditAction, AuditSection, DocumentCheckResult, DocumentStatus, 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.audit import AuditService 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", "supersede_previous", "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 STORED_FILE_OBJECT_TYPE = "stored_file" # `audit_event.object_type` события FILE_REPLACE SUPERSEDE_LOCK_PREFIX = "supersede" # пространство ключей advisory-блокировки автозамены в пределах ЮЛ _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-10); возвращает предупреждения маппинга файла. Очистка прежних документов/строк файла → документ на каждый лист ОСВ (повтор счёта — предупреждение) → строки пакетами по `ENTRY_BATCH_SIZE` → `job.group_schema`, `metrics.mapping_warnings` → файл `PROCESSED` (только из `PROCESSING` и только для текущего задания) → автозамена прежних версий (`supersede_previous`). Без 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"]) await supersede_previous(session, new_file=file, new_documents=documents, uploader_id=file.uploaded_by, now=now) return warnings # --- автозамена прежних версий -------------------------------------------------------------------------------------- DocumentKey = tuple[DocumentType, str | None, int, PeriodQuarter] def _document_key(document: Document) -> DocumentKey | None: """Ключ совпадения документов для автозамены (D-10); документ без года или квартала в замене не участвует.""" if document.period_year is None or document.period_quarter is None: return None return (document.document_type, document.osv_account_code, document.period_year, document.period_quarter) def _describe_document(document: Document) -> str: period = f"{document.period_year} {document.period_quarter.value if document.period_quarter else ''}".strip() return f"{document.document_type.value} {document.osv_account_code or ''} {period} ({document.status.value})" async def supersede_previous( session: AsyncSession, *, new_file: StoredFile, new_documents: Sequence[Document], uploader_id: uuid.UUID | None, now: datetime, ) -> list[StoredFile]: """D-10, spec primary-files «Автозамена прежних версий»: файлы того же ЮЛ (не `SUPERSEDED`, не сам новый) с документом тех же `document_type`, `osv_account_code`, `period_year`, `period_quarter` → `SUPERSEDED`, `superseded_by_file_id = new_file.id`; у нового файла `version_no` = максимум заменённых + 1. На каждый заменённый файл — одно системное событие `FILE_REPLACE` (`user_id = NULL`, инициатор загрузки — в `description`, `confirmed=true`, если среди заменённых документов есть `CONFIRMED`; решение владельца 05.10.2026). Документы и строки заменённых файлов не трогаются (история). Возвращает заменённые файлы; в транзакции вызывающего, без commit. Блокировки: advisory-блокировка ЮЛ сериализует автозамены одного ЮЛ (два новых файла одного документа, смапленные параллельно, иначе не увидели бы друг друга и оба остались бы действующими); заменяемые файлы — `FOR UPDATE` в порядке `id` (сериализация с подтверждением документа, которое перечитывает файл под `FOR UPDATE`). """ keys = {key for key in map(_document_key, new_documents) if key is not None} if not keys: return [] lock_key = func.hashtextextended(f"{SUPERSEDE_LOCK_PREFIX}:{new_file.legal_entity_id}", 0) await session.execute(select(func.pg_advisory_xact_lock(lock_key))) def matching(stmt: Select[tuple[Document]]) -> Select[tuple[Document]]: return stmt.join(StoredFile, StoredFile.id == Document.stored_file_id).where( StoredFile.legal_entity_id == new_file.legal_entity_id, StoredFile.id != new_file.id, StoredFile.status != FileStatus.SUPERSEDED, Document.document_type.in_(sorted({key[0] for key in keys})), Document.period_year.in_(sorted({key[2] for key in keys})), ) candidates = (await session.execute(matching(select(Document)))).scalars().all() file_ids = sorted({document.stored_file_id for document in candidates if _document_key(document) in keys}) if not file_ids: return [] locked = ( ( await session.execute( select(StoredFile) .where(StoredFile.id.in_(file_ids), StoredFile.status != FileStatus.SUPERSEDED) .order_by(StoredFile.id) .with_for_update() .execution_options(populate_existing=True) ) ) .scalars() .all() ) if not locked: return [] ids = [old.id for old in locked] # документы перечитываются ПОСЛЕ блокировки файлов: подтверждение, зафиксированное до неё, попадёт в `confirmed` fresh = await session.execute( matching(select(Document)).where(Document.stored_file_id.in_(ids)).execution_options(populate_existing=True) ) replaced_documents: dict[uuid.UUID, list[Document]] = {} for document in fresh.scalars(): if _document_key(document) in keys: replaced_documents.setdefault(document.stored_file_id, []).append(document) await session.execute( update(StoredFile) .where(StoredFile.id.in_(ids), StoredFile.status != FileStatus.SUPERSEDED) .values(status=FileStatus.SUPERSEDED, superseded_by_file_id=new_file.id, updated_at=now) .execution_options(synchronize_session=False) ) new_file.version_no = max(old.version_no for old in locked) + 1 new_file.updated_at = now await session.flush() audit = AuditService(session) uploader = f"user:{uploader_id}" if uploader_id is not None else "неизвестен" for old in locked: await session.refresh(old, ["status", "superseded_by_file_id", "updated_at"]) documents = sorted(replaced_documents.get(old.id, []), key=_describe_document) confirmed = any(document.status is DocumentStatus.CONFIRMED for document in documents) await audit.record( action=AuditAction.FILE_REPLACE, section=AuditSection.PRIMARY_DOCUMENTS, object_type=STORED_FILE_OBJECT_TYPE, object_id=old.id, user_id=None, after_value={"file_id": new_file.id, "version_no": new_file.version_no}, description=( f"заменён файлом {new_file.original_name} (загрузил {uploader}); " f"документы: {', '.join(map(_describe_document, documents))}; " f"confirmed={'true' if confirmed else 'false'}" ), ) logger.info( "Автозамена: прежние версии файла заменены. file_id={file_id}, version_no={version_no}, replaced={replaced}", file_id=new_file.id, version_no=new_file.version_no, replaced=[str(old_id) for old_id in ids], ) return list(locked) # --- проход и цикл -------------------------------------------------------------------------------------------------- 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, ) -> bool: """D-2: после отката — `metrics.mapping_attempts + 1` отдельной транзакцией; на `max_attempts` — файл `ERROR`. Возвращает `True`, если попытки исчерпаны (файл переведён в `ERROR` и покинул очередь).""" async with session_factory() as session, session.begin(): job = await _lock_job(session, candidate.job_id) if job is None: return False 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, ) return 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 модуля). Результат — выполнена ли полезная работа, после которой цикл может не спать: `True` — кандидат смапплен, отклонён неповторяемой ошибкой или исчерпал попытки (файл `ERROR`), либо пропущен, потому что его опередили; `False` — кандидатов нет или повторяемая неудача (R-15: пауза `MAPPING_POLL_SECONDS` перед следующей попыткой, иначе кратковременный сбой S3/БД сжёг бы все попытки за миллисекунды).""" 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 или БД: транзакция маппинга откачена, повтор после паузы цикла (R-15) return 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: """Бесконечный цикл маппинга: после полезной работы (`process_once` → `True`) — следующий кандидат сразу; без кандидатов или после повторяемой неудачи — пауза `MAPPING_POLL_SECONDS` (R-15); ошибка прохода логируется, цикл продолжается.""" 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)