353b562dfc
Пара «новый документ — прежний» с counterparty_check = MISMATCH у любого из двух в автозамене не участвует: сканы разных организаций в одной карточке больше не вытесняют друг друга. UNKNOWN заменяется как прежде. Если файл оставлен только из-за MISMATCH, в mapping_warnings нового файла — «автозамена не выполнена» с именем и версией оставленного файла. Спека primary-files дополнена. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1202 lines
65 KiB
Python
1202 lines
65 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 и валидация результата по MAJOR (`1.x` или `2.0`) — блокировки не
|
||
держатся на время сети;
|
||
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.contract.result_v2 import DocumentWarning, OsvDocumentV2, RecognitionResultV2, StatementDocumentV2
|
||
from src.contract.versions import parse_result
|
||
from src.core.config import Settings
|
||
from src.core.storage import ObjectNotFoundError, ObjectStorage
|
||
from src.domain.osv_mapping import (
|
||
EntryDraft,
|
||
build_entries,
|
||
checks_clean,
|
||
counterparty_check,
|
||
document_fields,
|
||
document_status,
|
||
opening_matches,
|
||
)
|
||
from src.domain.statement_issues import IssueDraft, mapping_warning
|
||
from src.domain.statement_mapping import StatementDraft, build_statement, statement_status, values_ok
|
||
from src.domain.statement_status import duplicate_period_forms, has_error_warning, osv_status_v2
|
||
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,
|
||
DocumentIssueKind,
|
||
DocumentStatus,
|
||
DocumentType,
|
||
FileStatus,
|
||
IssueSeverity,
|
||
IssueSource,
|
||
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
|
||
from src.services.statement_intake_v2 import (
|
||
FILE_WARNINGS_KEY,
|
||
FormV2,
|
||
OtherV2,
|
||
file_warnings,
|
||
insert_forms_v2,
|
||
map_statement_docs_v2,
|
||
)
|
||
from src.services.statement_recompute import checks_on_intake_1x
|
||
from src.services.statements_engine import StatementsEngine, default_engine
|
||
|
||
__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 | RecognitionResultV2) -> list[OsvDocument]:
|
||
return [document for document in result.documents if isinstance(document, OsvDocument)]
|
||
|
||
|
||
async def _load_result(storage: ObjectStorage, candidate: Candidate) -> RecognitionResult | RecognitionResultV2:
|
||
"""Объект результата из S3 и его валидация по контракту; неповторяемые ошибки — `MappingFatal`.
|
||
|
||
Модель выбирается по MAJOR `schema_version` (`contract.versions.parse_result`): 1 — `RecognitionResult` (маппер
|
||
`1.x`), 2 — `RecognitionResultV2` (маппер `2.0`); иной MAJOR — итог проверки схемы (P3-46). Нечитаемый JSON и
|
||
не-JSON байты — тоже неповторяемая ошибка схемы (T5.2), без текста исключения в `mapping_error`."""
|
||
try:
|
||
content = await storage.get_bytes(candidate.key)
|
||
except ObjectNotFoundError as exc:
|
||
raise MappingFatal(f"файл результата не найден в S3: {candidate.bucket}/{candidate.key}") from exc
|
||
try:
|
||
result = parse_result(content)
|
||
except ValidationError as exc:
|
||
raise MappingFatal(_schema_error_summary(exc)) from exc
|
||
except ValueError as exc: # JSONDecodeError, UnicodeDecodeError: pydantic ValidationError перехвачен выше
|
||
raise MappingFatal("результат не соответствует схеме recognition-result: файл не является JSON") 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 _osv_warning_issue(warning: DocumentWarning) -> IssueDraft:
|
||
"""Предупреждение листа ОСВ `2.0` (`OSV_*`, `info`) → `document_issue` вида `warning`, источник `service`.
|
||
У ОСВ нет строк и колонок формы: `line_refs` не сохраняются, `column_index` и `field` — как пришли."""
|
||
return IssueDraft(
|
||
kind=DocumentIssueKind.warning,
|
||
code=warning.code,
|
||
severity=IssueSeverity(warning.severity),
|
||
message=warning.message,
|
||
source=IssueSource.service,
|
||
column_index=warning.column_index,
|
||
field=warning.field,
|
||
)
|
||
|
||
|
||
def _osv_status(
|
||
sheet: OsvDocument,
|
||
*,
|
||
job_done: bool,
|
||
file_error: bool,
|
||
counterparty: DocumentCheckResult,
|
||
mapping_warnings: Sequence[str],
|
||
) -> tuple[DocumentStatus, list[IssueDraft]]:
|
||
"""Статус документа ОСВ и записи `document_issue` сервиса: `2.0` — по Р-28 (`osv_status_v2`, статус листа и
|
||
предупреждения листа), `1.x` — по §9.3 (`document_status`, статус задания); у `1.x` записей сервиса нет."""
|
||
if isinstance(sheet, OsvDocumentV2):
|
||
status = osv_status_v2(
|
||
osv_done=sheet.status == "done",
|
||
checks_ok=checks_clean(sheet.checks),
|
||
file_error_warning=file_error,
|
||
counterparty=counterparty,
|
||
sheet_mapping_warnings=bool(mapping_warnings),
|
||
)
|
||
return status, [_osv_warning_issue(warning) for warning in sheet.warnings]
|
||
return document_status(
|
||
job_done=job_done, checks=sheet.checks, counterparty=counterparty, warnings=mapping_warnings
|
||
), []
|
||
|
||
|
||
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],
|
||
*,
|
||
ordinals: Mapping[int, int],
|
||
file: StoredFile,
|
||
owner: LegalEntity,
|
||
owner_names: frozenset[str],
|
||
now: datetime,
|
||
) -> tuple[list[tuple[Document, StatementDraft]], list[Document], list[str], dict[uuid.UUID, str]]:
|
||
"""Документы форм ББ/ОФР и расшифровок (spec «Документ формы ББ/ОФР», «Повтор формы в файле и прочие документы»;
|
||
D-2, D-3; P3-4, P3-6). Возвращает формы (документ + черновик; статус уточняется после вставки значений),
|
||
документы `DISCLOSURE`/`OTHER` (по одному на тип, `locator` через запятую, `RECOGNIZED`, `file_ordinal` — индекс
|
||
первого такого документа в `documents[]`), предупреждения маппинга и текст «другая форма» по `id` документов форм
|
||
одного типа и периода (Р-10, Р-18: хранятся все, каждый `NEEDS_CHECK`). `ordinals` — `id(документ результата)` →
|
||
индекс в `result.documents`."""
|
||
forms: list[tuple[Document, StatementDraft]] = []
|
||
other_locators: dict[DocumentType, list[str]] = {}
|
||
other_ordinals: dict[DocumentType, int] = {}
|
||
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)
|
||
other_ordinals.setdefault(doc_type, ordinals[id(doc)])
|
||
continue
|
||
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,
|
||
file_ordinal=ordinals[id(doc)],
|
||
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))
|
||
duplicated = duplicate_period_forms(
|
||
(document.id, document.document_type, document.period_year, document.period_quarter) for document, _ in forms
|
||
)
|
||
repeats: dict[uuid.UUID, str] = {}
|
||
for document, _ in forms:
|
||
if document.id in duplicated:
|
||
text = f"в файле есть другая форма {document.document_type.value} за тот же период"
|
||
repeats[document.id] = text
|
||
warnings.append(f"{document.locator}: {text}")
|
||
others = [
|
||
Document(
|
||
id=uuid.uuid4(),
|
||
stored_file_id=file.id,
|
||
legal_entity_id=file.legal_entity_id,
|
||
document_type=doc_type,
|
||
file_ordinal=other_ordinals[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, repeats
|
||
|
||
|
||
async def _insert_statements( # noqa: PLR0913 - все входы именованные
|
||
session: AsyncSession,
|
||
*,
|
||
job: RecognitionJob,
|
||
forms: Sequence[tuple[Document, StatementDraft]],
|
||
repeats: Mapping[uuid.UUID, str],
|
||
job_done: bool,
|
||
now: datetime,
|
||
engine: StatementsEngine | None = None,
|
||
) -> None:
|
||
"""Шапка, колонки, строки и значения каждой формы; предупреждения черновика и повтор формы — `MAPPING_WARNING`
|
||
документа; статус документа по D-6 (акцепта при маппинге нет)."""
|
||
if not forms:
|
||
return
|
||
engine = engine or default_engine()
|
||
for document, draft in forms:
|
||
meta = await statements_repo.insert_statement(session, document_id=document.id, job_id=job.id, draft=draft)
|
||
warnings = [*draft.warnings, *([repeats[document.id]] if document.id in repeats else [])]
|
||
await statements_repo.add_issues(session, document.id, [mapping_warning(text) for text in warnings], now=now)
|
||
checks_passed = await checks_on_intake_1x(session, document=document, meta=meta, engine=engine, now=now)
|
||
document.status = statement_status(
|
||
job_done=job_done,
|
||
checks_passed=checks_passed,
|
||
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=warnings,
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class _FormsPlan:
|
||
"""Документы форм/раскрытий результата до вставки: общие поля и черновики по MAJOR."""
|
||
|
||
documents: list[Document]
|
||
warnings: list[str]
|
||
repeats: dict[uuid.UUID, str]
|
||
forms_v1: list[tuple[Document, StatementDraft]]
|
||
forms_v2: list[FormV2]
|
||
others_v2: list[OtherV2]
|
||
|
||
|
||
def _plan_forms( # noqa: PLR0913 - все входы именованные
|
||
result: RecognitionResult | RecognitionResultV2,
|
||
*,
|
||
job: RecognitionJob,
|
||
ordinals: Mapping[int, int],
|
||
file: StoredFile,
|
||
owner: LegalEntity,
|
||
owner_names: frozenset[str],
|
||
now: datetime,
|
||
) -> _FormsPlan:
|
||
"""Выбор маппера форм по типу результата: `2.0` (и `job.statements_version` сервиса) или `1.x`."""
|
||
if isinstance(result, RecognitionResultV2):
|
||
job.statements_version = result.service.statements_version
|
||
forms_v2, others_v2, warnings, repeats = map_statement_docs_v2(
|
||
[doc for doc in result.documents if isinstance(doc, StatementDocumentV2)],
|
||
ordinals=ordinals,
|
||
file=file,
|
||
owner=owner,
|
||
owner_names=owner_names,
|
||
now=now,
|
||
)
|
||
documents = [*(other.document for other in others_v2), *(form.document for form in forms_v2)]
|
||
return _FormsPlan(documents, warnings, repeats, [], forms_v2, others_v2)
|
||
forms_v1, other_documents, warnings, repeats = _map_statement_docs(
|
||
[doc for doc in result.documents if isinstance(doc, StatementDocument)],
|
||
ordinals=ordinals,
|
||
file=file,
|
||
owner=owner,
|
||
owner_names=owner_names,
|
||
now=now,
|
||
)
|
||
documents = [*other_documents, *(document for document, _ in forms_v1)]
|
||
return _FormsPlan(documents, warnings, repeats, forms_v1, [], [])
|
||
|
||
|
||
async def _insert_forms( # noqa: PLR0913 - все входы именованные
|
||
session: AsyncSession,
|
||
plan: _FormsPlan,
|
||
*,
|
||
job: RecognitionJob,
|
||
result: RecognitionResult | RecognitionResultV2,
|
||
job_done: bool,
|
||
now: datetime,
|
||
engine: StatementsEngine | None,
|
||
) -> None:
|
||
"""Вставка форм: `2.0` — как пришли (`insert_forms_v2`), `1.x` — `_insert_statements`."""
|
||
if isinstance(result, RecognitionResultV2):
|
||
await insert_forms_v2(
|
||
session,
|
||
job=job,
|
||
result=result,
|
||
forms=plan.forms_v2,
|
||
others=plan.others_v2,
|
||
repeats=plan.repeats,
|
||
engine=engine or default_engine(),
|
||
now=now,
|
||
)
|
||
return
|
||
await _insert_statements(
|
||
session, job=job, forms=plan.forms_v1, repeats=plan.repeats, job_done=job_done, now=now, engine=engine
|
||
)
|
||
|
||
|
||
async def map_file( # noqa: PLR0913 - все входы маппинга файла именованные
|
||
session: AsyncSession,
|
||
*,
|
||
job: RecognitionJob,
|
||
file: StoredFile,
|
||
result: RecognitionResult | RecognitionResultV2,
|
||
chart: Mapping[str, uuid.UUID],
|
||
index: CounterpartyIndex,
|
||
now: datetime,
|
||
engine: StatementsEngine | None = None,
|
||
) -> list[str]:
|
||
"""Маппинг результата в транзакции вызывающего (D-3, D-6…D-10); возвращает предупреждения маппинга файла.
|
||
|
||
Маппер выбирается по типу результата: `RecognitionResultV2` — формы по контракту `2.0`
|
||
(`services/statement_intake_v2`, `job.statements_version` = `service.statements_version`), `RecognitionResult` —
|
||
`1.x` (`_map_statement_docs`).
|
||
|
||
Очистка прежних документов/строк файла → документ на каждый лист ОСВ (повтор счёта — предупреждение) и на каждую
|
||
форму ББ/ОФР (`_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)
|
||
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
|
||
|
||
ordinals = {id(doc): index for index, doc in enumerate(result.documents)} # v5.19: номер документа в файле
|
||
await _clear_file(session, file.id)
|
||
kept, duplicate_warnings, warnings = _split_duplicates(sheets)
|
||
documents: list[Document] = []
|
||
sheet_issues: list[tuple[Document, list[IssueDraft]]] = []
|
||
file_error = isinstance(result, RecognitionResultV2) and has_error_warning(file_warnings(result))
|
||
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
|
||
)
|
||
mapping_warnings = [*duplicate_warnings.get(sheet.locator, ()), *sheet_warnings]
|
||
status, service_drafts = _osv_status(
|
||
sheet,
|
||
job_done=job_done,
|
||
file_error=file_error,
|
||
counterparty=check,
|
||
mapping_warnings=mapping_warnings,
|
||
)
|
||
document = Document(
|
||
id=uuid.uuid4(),
|
||
stored_file_id=file.id,
|
||
legal_entity_id=file.legal_entity_id,
|
||
document_type=DocumentType.OSV,
|
||
file_ordinal=ordinals[id(sheet)],
|
||
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=status,
|
||
period_check=DocumentCheckResult.UNKNOWN, # до версии консолидации (Срез 4)
|
||
counterparty_check=check,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
documents.append(document)
|
||
sheet_issues.append((document, [*service_drafts, *(mapping_warning(text) for text in mapping_warnings)]))
|
||
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
|
||
)
|
||
|
||
plan = _plan_forms(result, job=job, ordinals=ordinals, file=file, owner=owner, owner_names=owner_names, now=now)
|
||
documents.extend(plan.documents)
|
||
warnings.extend(plan.warnings)
|
||
|
||
session.add_all(documents)
|
||
await session.flush()
|
||
for document, drafts in sheet_issues: # P3-5: предупреждения листа ОСВ (сервис 2.0 и маппинг) — в `document_issue`
|
||
await statements_repo.add_issues(session, document.id, drafts, now=now)
|
||
# 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_forms(session, plan, job=job, result=result, job_done=job_done, now=now, engine=engine)
|
||
|
||
job.group_schema = [_group_schema(sheet) for sheet in sheets]
|
||
job.metrics = _mapped_metrics(job, result, 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)
|
||
|
||
|
||
def _mapped_metrics(
|
||
job: RecognitionJob, result: RecognitionResult | RecognitionResultV2, warnings: list[str]
|
||
) -> dict[str, Any]:
|
||
"""`metrics` задания после маппинга: предупреждения маппинга и, для результата `2.0`, предупреждения файла
|
||
результата отдельным ключом — единый источник «ошибки файла» для пересчёта после правки (P3-53, R-11)."""
|
||
metrics = {**_metrics(job), MAPPING_WARNINGS_KEY: warnings}
|
||
if isinstance(result, RecognitionResultV2):
|
||
metrics[FILE_WARNINGS_KEY] = file_warnings(result)
|
||
return metrics
|
||
|
||
|
||
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 _replaceable(document: Document, keys: set[DocumentKey]) -> bool:
|
||
"""Документ участвует в автозамене: ключ среди ключей нового файла и шапка не другой организации (AURORA-1644,
|
||
§9.5 п. 3: пара с `MISMATCH` у любой стороны не заменяется; `UNKNOWN` — как прежде)."""
|
||
return _document_key(document) in keys and document.counterparty_check is not DocumentCheckResult.MISMATCH
|
||
|
||
|
||
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
|
||
|
||
|
||
def _skip_warning(old: StoredFile, documents: Sequence[Document]) -> str:
|
||
"""Предупреждение нового файла: файл `old` с документами того же ключа оставлен, потому что у нового или прежнего
|
||
документа организация в шапке не совпадает с карточкой (AURORA-1644)."""
|
||
return (
|
||
f"автозамена не выполнена: файл «{old.original_name}» (версия {old.version_no}) оставлен рядом с этим — "
|
||
f"у нового или прежнего документа организация в шапке не совпадает с организацией карточки: "
|
||
f"{', '.join(sorted(map(_describe_document, documents)))}"
|
||
)
|
||
|
||
|
||
async def _skip_warnings(
|
||
session: AsyncSession,
|
||
candidates: Sequence[Document],
|
||
all_keys: set[DocumentKey],
|
||
replaced_file_ids: set[uuid.UUID],
|
||
) -> list[str]:
|
||
"""Предупреждения о файлах с документами того же ключа, которые автозамена оставила из-за `MISMATCH`
|
||
(AURORA-1644): по одному на файл, в порядке `id`."""
|
||
skipped: dict[uuid.UUID, list[Document]] = {}
|
||
for document in candidates:
|
||
if document.stored_file_id not in replaced_file_ids and _document_key(document) in all_keys:
|
||
skipped.setdefault(document.stored_file_id, []).append(document)
|
||
if not skipped:
|
||
return []
|
||
kept = await session.execute(select(StoredFile).where(StoredFile.id.in_(sorted(skipped))).order_by(StoredFile.id))
|
||
return [_skip_warning(old, skipped[old.id]) for old in kept.scalars()]
|
||
|
||
|
||
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; файл без
|
||
консолидации (`consolidation_id` пуст) наследует её у первого по `id` заменённого файла, где она задана (v5.18,
|
||
§9.5 п. 3). На каждый заменённый файл — одно системное событие `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` (если
|
||
передан) добавляется по одному предупреждению на заменённый файл: имя файла, выпавшие документы (тип, счёт,
|
||
период, статус) и версии консолидации, с которых сняты привязки этих документов. Поведение замены не меняется.
|
||
|
||
AURORA-1644: пара «новый — прежний» с `counterparty_check = MISMATCH` у любого из двух в замене не участвует
|
||
(`_replaceable`) — документы другой организации, загруженные в карточку, не вытесняют друг друга. Файл, оставленный
|
||
только из-за этого, — предупреждение в `warnings` (`_skip_warning`); `UNKNOWN` заменяется как прежде.
|
||
|
||
Блокировки: advisory-блокировка ЮЛ сериализует автозамены одного ЮЛ (два новых файла одного документа, смапленные
|
||
параллельно, иначе не увидели бы друг друга и оба остались бы действующими); заменяемые файлы — `FOR UPDATE` в
|
||
порядке `id` (сериализация с подтверждением документа, которое перечитывает файл под `FOR UPDATE`); затем
|
||
advisory ГК и версии консолидации по `id` (D-9, R-2).
|
||
"""
|
||
all_keys = {key for key in map(_document_key, new_documents) if key is not None}
|
||
if not all_keys:
|
||
return []
|
||
keys = { # ключи, которыми новый файл может заменять: без документов другой организации (AURORA-1644)
|
||
key
|
||
for document in new_documents
|
||
if (key := _document_key(document)) is not None
|
||
and document.counterparty_check is not DocumentCheckResult.MISMATCH
|
||
}
|
||
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 all_keys})),
|
||
Document.period_year.in_(sorted({key[2] for key in all_keys})),
|
||
)
|
||
|
||
candidates = (await session.execute(matching(select(Document)))).scalars().all()
|
||
file_ids = sorted({document.stored_file_id for document in candidates if _replaceable(document, keys)})
|
||
if warnings is not None:
|
||
warnings.extend(await _skip_warnings(session, candidates, all_keys, set(file_ids)))
|
||
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 _replaceable(document, 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
|
||
if new_file.consolidation_id is None: # v5.18, §9.5 п. 3: файл без контекста наследует консоль заменённого
|
||
new_file.consolidation_id = next(
|
||
(old.consolidation_id for old in locked if old.consolidation_id is not None), None
|
||
)
|
||
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 | RecognitionResultV2,
|
||
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)
|