975 lines
53 KiB
Python
975 lines
53 KiB
Python
"""Цикл маппинга результата распознавания в `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`: очистка документов и строк файла,
|
||
документы и строки (листы ОСВ, затем формы ББ/ОФР Среза 3b и расшифровки), `group_schema` (только листы ОСВ),
|
||
предупреждения, `PROCESSED`, автозамена прежних версий (`supersede_previous`, D-10, аудит `FILE_REPLACE`) →
|
||
автоподтверждение чистых документов (`services/auto_confirm.py`, OQ-157) → переход версий по `NEEDS_CHECK` (D-6).
|
||
|
||
Ошибки (D-2): неповторяемые (`MappingFatal`: ключ вне папки, объекта нет, JSON не проходит схему, нет документов)
|
||
— сразу файл `ERROR` и `metrics.mapping_error`; прочие — откат транзакции маппинга, в отдельной транзакции
|
||
`metrics.mapping_attempts + 1`, на `MAPPING_MAX_ATTEMPTS` — `ERROR`; следующая попытка — после паузы цикла (R-15).
|
||
`mapping_error` виден пользователю: в нём категория (схема — число ошибок и путь первой; БД/S3 — тип исключения и число
|
||
попыток), без текста исключений pydantic/SQLAlchemy; полный текст — только в логе (M-1).
|
||
Статус файла — через `services/file_status.py` (только текущее задание). `audit_event` не пишется: обмен и маппинг
|
||
не аудируются (ASSUMPTION(OQ-55)); `FILE_REPLACE` автозамены — системное событие (`user_id = NULL`).
|
||
"""
|
||
|
||
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, StatementDocument
|
||
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.domain.statement_checks import load_check_rules, run_checks
|
||
from src.domain.statement_mapping import StatementDraft, build_statement, statement_status, values_ok
|
||
from src.models.accounting import ChartOfAccount
|
||
from src.models.consolidation import Consolidation, ConsolidationVersion
|
||
from src.models.directories import CompanyGroup, 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 import statements as statements_repo
|
||
from src.repositories.counterparties import CounterpartyIndex
|
||
from src.services import consolidation_documents
|
||
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",
|
||
"NO_DOCUMENTS",
|
||
"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
|
||
NO_DOCUMENTS = "в результате нет документов"
|
||
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 _schema_error_summary(exc: ValidationError) -> str:
|
||
"""M-1: для `metrics.mapping_error` (видно пользователю) — число ошибок, путь и тип первой; без `input_value` и
|
||
ссылок pydantic (полный текст — в логе через `__cause__` у `MappingFatal`)."""
|
||
errors = exc.errors(include_url=False, include_input=False, include_context=False)
|
||
first = errors[0] if errors else None
|
||
where = ".".join(map(str, first["loc"])) if first and first["loc"] else "<корень>"
|
||
kind = first["type"] if first else "?"
|
||
return f"результат не соответствует схеме recognition-result: ошибок {exc.error_count()}, первая: {where} {kind}"
|
||
|
||
|
||
def _osv_documents(result: RecognitionResult) -> list[OsvDocument]:
|
||
return [document for document in result.documents if isinstance(document, OsvDocument)]
|
||
|
||
|
||
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(_schema_error_summary(exc)) from exc
|
||
if result.job_id != candidate.job_id or result.file_id != candidate.file_id:
|
||
raise MappingFatal(
|
||
f"файл результата относится к другому заданию: job_id={result.job_id}, file_id={result.file_id}"
|
||
)
|
||
if not result.documents:
|
||
raise MappingFatal(NO_DOCUMENTS)
|
||
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
|
||
resolution = index.resolve(entry.counterparty_inn, entry.counterparty_name)
|
||
if resolution is None:
|
||
return entry
|
||
return dataclasses.replace(entry, counterparty_id=resolution.entity_id, counterparty_match_kind=resolution.kind)
|
||
|
||
|
||
def _entry_values(
|
||
entry: EntryDraft, *, document_id: uuid.UUID, job_id: uuid.UUID, owner_id: uuid.UUID
|
||
) -> dict[str, Any]:
|
||
"""Строка `osv_entry` (DBML v5.16) из черновика; суммы — `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,
|
||
"counterparty_match_kind": entry.counterparty_match_kind,
|
||
"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 statements_repo.delete_file_statements(session, file_id)
|
||
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
|
||
|
||
|
||
def _map_statement_docs( # noqa: PLR0913 - все входы именованные
|
||
statement_docs: Sequence[StatementDocument],
|
||
*,
|
||
file: StoredFile,
|
||
owner: LegalEntity,
|
||
owner_names: frozenset[str],
|
||
now: datetime,
|
||
) -> tuple[list[tuple[Document, StatementDraft]], list[Document], list[str], set[DocumentType]]:
|
||
"""Документы форм ББ/ОФР и расшифровок (spec «Документ формы ББ/ОФР», «Повтор формы в файле и прочие документы»;
|
||
D-2, D-3). Возвращает формы (документ + черновик; статус уточняется после вставки значений), документы
|
||
`DISCLOSURE`/`OTHER` (по одному на тип, `locator` через запятую, `RECOGNIZED`), предупреждения и типы форм,
|
||
у которых был повтор в файле (R-1: первый документ такого типа — `NEEDS_CHECK`)."""
|
||
forms: list[tuple[Document, StatementDraft]] = []
|
||
seen_forms: set[DocumentType] = set()
|
||
duplicated: set[DocumentType] = set()
|
||
other_locators: dict[DocumentType, list[str]] = {}
|
||
warnings: list[str] = []
|
||
for doc in statement_docs:
|
||
doc_type = DocumentType(doc.document_type)
|
||
if doc_type in (DocumentType.DISCLOSURE, DocumentType.OTHER):
|
||
other_locators.setdefault(doc_type, []).append(doc.locator)
|
||
continue
|
||
if doc_type in seen_forms:
|
||
warnings.append(f"{doc.locator}: вторая форма {doc_type.value} в файле пропущена (OQ-77)")
|
||
duplicated.add(doc_type)
|
||
continue
|
||
seen_forms.add(doc_type)
|
||
draft = build_statement(doc)
|
||
warnings.extend(draft.warnings)
|
||
check = counterparty_check(doc.header.inn, doc.header.organization_name, owner.inn, owner_names)
|
||
document = Document(
|
||
id=uuid.uuid4(),
|
||
stored_file_id=file.id,
|
||
legal_entity_id=file.legal_entity_id,
|
||
document_type=doc_type,
|
||
osv_account_code=None,
|
||
period_year=draft.period_year,
|
||
period_quarter=draft.period_quarter,
|
||
is_cumulative=None,
|
||
reporting_date=draft.period_date,
|
||
locator=doc.locator,
|
||
status=DocumentStatus.NEEDS_CHECK, # уточняется после вставки значений (нужны проверки по суммам)
|
||
period_check=DocumentCheckResult.UNKNOWN,
|
||
counterparty_check=check,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
forms.append((document, draft))
|
||
others = [
|
||
Document(
|
||
id=uuid.uuid4(),
|
||
stored_file_id=file.id,
|
||
legal_entity_id=file.legal_entity_id,
|
||
document_type=doc_type,
|
||
locator=",".join(locators),
|
||
status=DocumentStatus.RECOGNIZED,
|
||
period_check=DocumentCheckResult.UNKNOWN,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
for doc_type, locators in other_locators.items()
|
||
]
|
||
return forms, others, warnings, duplicated
|
||
|
||
|
||
async def _insert_statements(
|
||
session: AsyncSession,
|
||
*,
|
||
job: RecognitionJob,
|
||
forms: Sequence[tuple[Document, StatementDraft]],
|
||
duplicated: set[DocumentType],
|
||
job_done: bool,
|
||
) -> None:
|
||
"""Шапка, колонки, строки и значения каждой формы; статус документа по D-6 (акцепта при маппинге нет)."""
|
||
if not forms:
|
||
return
|
||
rules = load_check_rules()
|
||
for document, draft in forms:
|
||
meta = await statements_repo.insert_statement(session, document_id=document.id, job_id=job.id, draft=draft)
|
||
column_indexes, amounts = await statements_repo.statement_amounts(session, meta.id)
|
||
report = run_checks(document.document_type.value, amounts, column_indexes, rules)
|
||
document.status = statement_status(
|
||
job_done=job_done,
|
||
checks_passed=not report.failed,
|
||
counterparty=document.counterparty_check or DocumentCheckResult.UNKNOWN,
|
||
period_known=document.period_year is not None and document.period_quarter is not None,
|
||
values_ok=values_ok(await statements_repo.value_flags(session, meta.id)),
|
||
has_lines=bool(draft.lines),
|
||
warnings=[*draft.warnings, *(["повтор формы"] if document.document_type in duplicated else [])],
|
||
)
|
||
|
||
|
||
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); возвращает предупреждения маппинга файла.
|
||
|
||
Очистка прежних документов/строк файла → документ на каждый лист ОСВ (повтор счёта — предупреждение) и на каждую
|
||
форму ББ/ОФР (`_map_statement_docs`) → строки ОСВ пакетами по `ENTRY_BATCH_SIZE` → вставка форм и их статус →
|
||
`job.group_schema`, `metrics.mapping_warnings` → файл `PROCESSED` (только из
|
||
`PROCESSING` и только для текущего задания) → автозамена прежних версий (`supersede_previous`; её предупреждение
|
||
о выпавших документах дописывается в `metrics.mapping_warnings`, OQ-154). Без commit.
|
||
"""
|
||
if not result.documents:
|
||
raise MappingFatal(NO_DOCUMENTS)
|
||
sheets = _osv_documents(result)
|
||
statement_docs = [doc for doc in result.documents if isinstance(doc, StatementDocument)]
|
||
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.organization_inn, sheet.header.organization_name, 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
|
||
)
|
||
|
||
forms, other_documents, form_warnings, duplicated = _map_statement_docs(
|
||
statement_docs, file=file, owner=owner, owner_names=owner_names, now=now
|
||
)
|
||
warnings.extend(form_warnings)
|
||
documents.extend(other_documents)
|
||
documents.extend(document for document, _ in forms)
|
||
|
||
session.add_all(documents)
|
||
await session.flush()
|
||
# T7.2: родитель раньше потомка на любой границе пачки (self-FK `parent_entry_id`) — не полагаемся на порядок строк
|
||
# результата; сортировка устойчивая, внутри уровня порядок строк сохраняется
|
||
entries.sort(key=lambda values: values["level"])
|
||
for start in range(0, len(entries), ENTRY_BATCH_SIZE):
|
||
await session.execute(insert(OsvEntry).values(entries[start : start + ENTRY_BATCH_SIZE]))
|
||
|
||
await _insert_statements(session, job=job, forms=forms, duplicated=duplicated, job_done=job_done)
|
||
|
||
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 await _supersede_and_warn(session, job=job, file=file, documents=documents, warnings=warnings, now=now)
|
||
|
||
|
||
async def _supersede_and_warn( # noqa: PLR0913 - входы автозамены именованные
|
||
session: AsyncSession,
|
||
*,
|
||
job: RecognitionJob,
|
||
file: StoredFile,
|
||
documents: Sequence[Document],
|
||
warnings: list[str],
|
||
now: datetime,
|
||
) -> list[str]:
|
||
"""Автозамена нового файла; её предупреждение о выпавших документах (OQ-154) дописывается в
|
||
`metrics.mapping_warnings` задания. Возвращает полный список предупреждений маппинга.
|
||
|
||
Предупреждение — про заменённые файлы, а не про документы нового: их статусы уже посчитаны и от него не меняются.
|
||
"""
|
||
replace_warnings: list[str] = []
|
||
await supersede_previous(
|
||
session,
|
||
new_file=file,
|
||
new_documents=documents,
|
||
uploader_id=file.uploaded_by,
|
||
now=now,
|
||
warnings=replace_warnings,
|
||
)
|
||
if not replace_warnings:
|
||
return warnings
|
||
# новый список, а не `extend`: предыдущая версия `metrics` ссылается на тот же список, и SQLAlchemy не увидел бы
|
||
# изменения JSONB (старое и новое значение сравнялись бы)
|
||
warnings = [*warnings, *replace_warnings]
|
||
job.metrics = {**_metrics(job), MAPPING_WARNINGS_KEY: warnings}
|
||
await session.flush()
|
||
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})"
|
||
|
||
|
||
def _describe_dropped(document: Document) -> str:
|
||
"""Выпавший из замены документ для предупреждения (OQ-154): тип, счёт, период и статус (в т. ч. `CONFIRMED`)."""
|
||
period = (
|
||
f"{document.period_year} {document.period_quarter.value}"
|
||
if document.period_year is not None and document.period_quarter is not None
|
||
else "период не определён"
|
||
)
|
||
parts = [document.document_type.value, document.osv_account_code, period]
|
||
return f"{' '.join(part for part in parts if part)} ({document.status.value})"
|
||
|
||
|
||
async def _version_labels(session: AsyncSession, version_ids: Sequence[uuid.UUID]) -> list[str]:
|
||
"""Человекочитаемые подписи версий консолидации («ГК» год квартал, версия N) в порядке `version_ids`."""
|
||
if not version_ids:
|
||
return []
|
||
rows = await session.execute(
|
||
select(
|
||
ConsolidationVersion.id,
|
||
ConsolidationVersion.version_no,
|
||
Consolidation.year,
|
||
Consolidation.quarter,
|
||
CompanyGroup.name,
|
||
)
|
||
.join(Consolidation, Consolidation.id == ConsolidationVersion.consolidation_id)
|
||
.join(CompanyGroup, CompanyGroup.id == Consolidation.company_group_id)
|
||
.where(ConsolidationVersion.id.in_(version_ids))
|
||
)
|
||
labels = {
|
||
version_id: f"«{group_name}» {year} {quarter.value}, версия {version_no}"
|
||
for version_id, version_no, year, quarter, group_name in rows
|
||
}
|
||
return [labels.get(version_id, str(version_id)) for version_id in version_ids]
|
||
|
||
|
||
def _replace_warning(old: StoredFile, dropped: Sequence[Document], version_labels: Sequence[str]) -> str:
|
||
"""Предупреждение нового файла о документах заменённого файла `old`, которых в новом файле нет (OQ-154)."""
|
||
text = (
|
||
f"автозамена: файл «{old.original_name}» (версия {old.version_no}) заменён этим файлом целиком, "
|
||
f"но в новом файле нет документов прежнего: {', '.join(sorted(map(_describe_dropped, dropped)))}"
|
||
)
|
||
if version_labels:
|
||
text += f"; привязки сняты с версий консолидации: {', '.join(version_labels)}"
|
||
return text
|
||
|
||
|
||
async def supersede_previous( # noqa: PLR0913, PLR0915 - автозамена одной транзакцией: входы и шаги именованные
|
||
session: AsyncSession,
|
||
*,
|
||
new_file: StoredFile,
|
||
new_documents: Sequence[Document],
|
||
uploader_id: uuid.UUID | None,
|
||
now: datetime,
|
||
warnings: list[str] | None = None,
|
||
) -> 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.
|
||
|
||
Привязки документов заменённых файлов к версиям консолидации «в работе» снимаются
|
||
(`consolidation_documents.on_documents_superseded`); id версий — в `after_value.unbound_version_ids` события
|
||
заменённого файла (spec audit-trail), статус версий не меняется.
|
||
|
||
OQ-154: замена остаётся на уровне файла, но если у заменённого файла есть документы, ключа которых нет среди
|
||
`new_documents` (например, подтверждённый ББ из книги «ББ + ОФР», заменённой отдельным ОФР), в `warnings` (если
|
||
передан) добавляется по одному предупреждению на заменённый файл: имя файла, выпавшие документы (тип, счёт,
|
||
период, статус) и версии консолидации, с которых сняты привязки этих документов. Поведение замены не меняется.
|
||
|
||
Блокировки: advisory-блокировка ЮЛ сериализует автозамены одного ЮЛ (два новых файла одного документа, смапленные
|
||
параллельно, иначе не увидели бы друг друга и оба остались бы действующими); заменяемые файлы — `FOR UPDATE` в
|
||
порядке `id` (сериализация с подтверждением документа, которое перечитывает файл под `FOR UPDATE`); затем
|
||
advisory ГК и версии консолидации по `id` (D-9, R-2).
|
||
"""
|
||
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()
|
||
# привязки к версиям «в работе» снимаются у ВСЕХ документов заменённого файла (spec primary-files), а не только у
|
||
# совпавших по ключу: подхват берёт документы лишь незаменённых файлов (design №33)
|
||
file_documents = await session.execute(
|
||
select(Document).where(Document.stored_file_id.in_(ids)).execution_options(populate_existing=True)
|
||
)
|
||
documents_by_file: dict[uuid.UUID, list[Document]] = {}
|
||
for document in file_documents.scalars():
|
||
documents_by_file.setdefault(document.stored_file_id, []).append(document)
|
||
unbound: dict[uuid.UUID, list[uuid.UUID]] = {}
|
||
dropped_by_file: dict[uuid.UUID, list[Document]] = {} # документы без ключа среди новых (OQ-154)
|
||
dropped_versions: dict[uuid.UUID, list[uuid.UUID]] = {} # версии, где сняты привязки именно выпавших документов
|
||
for old in locked: # по возрастанию id; advisory ГК и версии берутся первым вызовом (порядок D-9)
|
||
old_documents = documents_by_file.get(old.id, [])
|
||
dropped = [document for document in old_documents if _document_key(document) not in keys]
|
||
dropped_by_file[old.id] = dropped
|
||
# выпавшие снимаются отдельным первым вызовом: он возвращает версии именно их привязок (после блокировки
|
||
# версий, без окна между чтением и удалением); второй вызов добирает остальные документы файла
|
||
dropped_versions[old.id] = (
|
||
await consolidation_documents.on_documents_superseded(session, documents=dropped, now=now)
|
||
if dropped
|
||
else []
|
||
)
|
||
rest_versions = await consolidation_documents.on_documents_superseded(session, documents=old_documents, now=now)
|
||
unbound[old.id] = sorted({*dropped_versions[old.id], *rest_versions})
|
||
|
||
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,
|
||
"unbound_version_ids": unbound[old.id],
|
||
},
|
||
description=(
|
||
f"заменён файлом {new_file.original_name} (загрузил {uploader}); "
|
||
f"документы: {', '.join(map(_describe_document, documents))}; "
|
||
f"confirmed={'true' if confirmed else 'false'}"
|
||
),
|
||
)
|
||
if warnings is not None and dropped_by_file[old.id]:
|
||
labels = await _version_labels(session, dropped_versions[old.id])
|
||
warnings.append(_replace_warning(old, dropped_by_file[old.id], labels))
|
||
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, *, cause: BaseException | None = None
|
||
) -> None:
|
||
"""Неповторяемая ошибка: `metrics.mapping_error` и файл текущего задания → `ERROR` (транзакция вызывающего).
|
||
`cause` — исходное исключение (например, `ValidationError`): его полный текст пишется только в лог (M-1)."""
|
||
job = await _lock_job(session, candidate.job_id)
|
||
if job is None:
|
||
return
|
||
job.metrics = {**_metrics(job), MAPPING_ERROR_KEY: _short(message)}
|
||
await session.flush()
|
||
changed = await set_file_status_for_job(session, job, FileStatus.ERROR, now=now)
|
||
logger.opt(exception=cause).error(
|
||
"Маппинг результата невозможен: файл переведён в ERROR. error_category=mapping_fatal, job_id={job_id}, "
|
||
"file_id={file_id}, file_changed={changed}, error={error}",
|
||
job_id=candidate.job_id,
|
||
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:
|
||
# M-1: только категория и тип исключения — текст SQLAlchemy/asyncpg ([SQL: …], параметры) остаётся в логе
|
||
metrics[MAPPING_ERROR_KEY] = _short(
|
||
f"маппинг не удался: ошибка БД/S3 ({type(exc).__name__}), попыток {attempts}"
|
||
)
|
||
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)
|
||
now = clock()
|
||
warnings = await map_file(session, job=job, file=file, result=result, chart=chart, index=index, now=now)
|
||
# После `map_file`, а значит после автозамены (advisory ЮЛ → старые файлы → advisory ГК → версии): версии
|
||
# блокируются последними, порядок «задание → файл → документы → версии» (D-9) не нарушается. Сначала
|
||
# автоподтверждение чистых документов (привязка и переход версии, OQ-157, design D-1), затем переход по
|
||
# `NEEDS_CHECK` (D-6) — он уже видит версию после автоподтверждения и второго перехода не делает.
|
||
documents = list((await session.execute(select(Document).where(Document.stored_file_id == file.id))).scalars())
|
||
# отложенный импорт: цикл documents → result_mapping → auto_confirm → statements → documents
|
||
from src.services.auto_confirm import auto_confirm # noqa: PLC0415
|
||
|
||
await auto_confirm(session, file=file, documents=documents, now=now)
|
||
await consolidation_documents.on_documents_mapped(session, file=file, documents=documents, now=now)
|
||
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(), cause=exc.__cause__)
|
||
except Exception as exc: # сбой S3 или БД: транзакция маппинга откачена, повтор после паузы цикла (R-15)
|
||
return await _register_failure(
|
||
session_factory, candidate, exc, max_attempts=settings.MAPPING_MAX_ATTEMPTS, now=clock()
|
||
)
|
||
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)
|