"""Форма ББ/ОФР для экрана «Проверка распознавания» (spec documents-api; design D-5, D-7, D-8, D-14). Видимость — через ЮЛ-владельца (`files_repo.get_document`): чужой документ — тот же 404, что несуществующий. Нарушения §7 хранятся в `document_issue` и пересчитываются после каждой правки общим кодом `arm_statements`. """ import re import uuid from dataclasses import dataclass from decimal import Decimal from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from src.api.dependencies.audit import RequestAudit from src.common.clock import utc_now from src.common.exceptions import ConflictError, DomainValidationError, NotFoundError from src.domain.osv_mapping import counterparty_check from src.domain.policy import VisibilityScope from src.domain.statement_aggregate import CodeAggregate, checks_input from src.domain.statement_checks import FailedCheck, cross_form_check, subline_parents from src.domain.statement_issues import COLUMN_CLEARS, HEADER_CLEARS from src.domain.statement_lines import TemplateLine, load_line_catalog from src.domain.statement_mapping import edition_for, quarter_of, statement_period from src.models.directories import LegalEntity from src.models.enums import ( AuditAction, AuditSection, DocumentIssueKind, DocumentStatus, DocumentType, FileStatus, IssueSource, StatementLineStatus, ) from src.models.files import Document, RecognitionJob from src.models.statements import ( DocumentIssue, FinancialStatementColumn, FinancialStatementLine, FinancialStatementMeta, FinancialStatementValue, ) from src.repositories import files as files_repo from src.repositories import statements as repo from src.repositories.counterparties import CounterpartyIndex from src.schemas import statements as schemas from src.services.documents import ( CHECKS_NOT_PASSED, CONFIRMABLE_STATUSES, DOCUMENT_NOT_CONFIRMABLE, FILE_SUPERSEDED, PERIOD_REQUIRED, ) from src.services.statement_line_edits import apply_line_edits from src.services.statement_recompute import recompute_form from src.services.statements_engine import PeriodKind, StatementsEngine, default_engine DOCUMENT_NOT_FOUND = "документ не найден" # одинаково для чужого и несуществующего DOCUMENT_NOT_STATEMENT = "DOCUMENT_NOT_STATEMENT" DOCUMENT_READONLY = "DOCUMENT_READONLY" LINE_EXISTS = "строка уже есть в форме — передайте line_id" UNKNOWN_CELL = "UNKNOWN_CELL" UNKNOWN_COLUMN = "UNKNOWN_COLUMN" DUPLICATE_COLUMN = "DUPLICATE_COLUMN" DUPLICATE_CELL = "DUPLICATE_CELL" NOTHING_TO_ACCEPT = "NOTHING_TO_ACCEPT" INVALID_INN = "INVALID_INN" LINES_UNRESOLVED = "LINES_UNRESOLVED" DUPLICATE_FORM = "DUPLICATE_FORM" OBJECT_TYPE = "document" ABSOLUTE = "absolute" INN_RE = re.compile(r"\d{10}|\d{12}", re.ASCII) FORM_TYPES = frozenset({DocumentType.BALANCE, DocumentType.INCOME}) async def form_aggregate( session: AsyncSession, document: Document, meta: FinancialStatementMeta ) -> tuple[list[FinancialStatementColumn], dict[str, CodeAggregate]]: """Колонки и агрегат формы по коду со справочными наименованиями редакции документа (P3-14).""" catalog = load_line_catalog() edition = edition_for(document.reporting_date) form = document.document_type.value return await repo.form_aggregate(session, meta.id, lambda code: catalog.reference_name(edition, form, code)) async def statement_hard_failures( session: AsyncSession, document: Document, meta: FinancialStatementMeta ) -> tuple[DocumentIssue, ...]: """Нарушенные жёсткие проверки §7 — записи `document_issue` вида `relation`; при акцепте — пустой кортеж (D-8 Среза 3b). Форма без записей (до первого пересчёта) нарушений не показывает.""" if meta.checks_accepted_at is not None: return () return tuple(issue for issue in await repo.issues(session, document.id) if issue.kind is DocumentIssueKind.relation) @dataclass(frozen=True) class FormBlocker: """Причина отказа в подтверждении формы: код 409, сообщение и поля `detail`.""" code: str message: str extra: dict[str, Any] async def form_blocker(session: AsyncSession, document: Document, meta: FinancialStatementMeta) -> FormBlocker | None: """Первое препятствие подтверждению формы ББ/ОФР или `None` (общий путь ручного и автоподтверждения). Порядок: `PERIOD_REQUIRED` → `LINES_UNRESOLVED` (строки `unreadable`/`missing` с непустой ячейкой) → `DUPLICATE_FORM` (уже подтверждена форма того же файла, типа и периода) → `CHECKS_NOT_PASSED` (жёсткие §7 без акцепта). """ if document.period_year is None or document.period_quarter is None: return FormBlocker(PERIOD_REQUIRED, "у формы не определён период: поправьте дату отчёта", {}) lines = await repo.unresolved_lines(session, meta.id) if lines: return FormBlocker( LINES_UNRESOLVED, "в форме есть нераспознанные строки со значениями: назначьте код или отметьте строку расшифровкой", { "lines": [ {"line_id": str(line_id), "row_index": row_index, "line_code_text": code_text} for line_id, row_index, code_text in lines ] }, ) twin = await files_repo.confirmed_twin(session, document) if twin is not None: return FormBlocker( DUPLICATE_FORM, "форма этого типа и периода в файле уже подтверждена", {"document_id": str(twin)} ) failures = await statement_hard_failures(session, document, meta) if failures: return FormBlocker( CHECKS_NOT_PASSED, "не пройдены логические проверки формы: исправьте значения или акцептуйте с комментарием", { "failures": [ { "column_index": failure.column_index, "relation": failure.relation, "expected": None if failure.expected is None else str(failure.expected), "actual": None if failure.actual is None else str(failure.actual), } for failure in failures ] }, ) return None @dataclass class _Edit: """Накопитель одного PATCH: `before/after` для единственного события аудита и признак правки данных (сброс акцепта — только ею; подтверждения акцепт не сбрасывают, Р-25).""" before: dict[str, Any] after: dict[str, Any] data_edited: bool = False def record(self, key: str, old: Any, new: Any, *, data: bool = True) -> None: # noqa: ANN401 - значения разных типов self.before[key], self.after[key] = old, new self.data_edited = self.data_edited or data class StatementService: def __init__( self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope, engine: StatementsEngine | None = None ) -> None: self._session = session self._audit = audit self._scope = scope self._engine = engine if engine is not None else default_engine() async def _visible_statement(self, document_id: uuid.UUID) -> tuple[Document, FinancialStatementMeta]: """Документ формы в пределах видимости: 404 — нет или чужой; 409 `DOCUMENT_NOT_STATEMENT` — не форма.""" document = await files_repo.get_document(self._session, self._scope, document_id) if document is None: raise NotFoundError(DOCUMENT_NOT_FOUND) meta = await repo.get_meta(self._session, document.id) if document.document_type in FORM_TYPES else None if meta is None: raise ConflictError(DOCUMENT_NOT_STATEMENT, "документ — не форма ББ/ОФР") return document, meta async def _editable(self, document_id: uuid.UUID) -> tuple[Document, FinancialStatementMeta]: """Видимость → блокировки «файл → документ» → проверки D-7 (порядок 409 как у подтверждения).""" visible, _ = await self._visible_statement(document_id) file = await files_repo.lock_file(self._session, visible.stored_file_id) document = await files_repo.lock_document(self._session, document_id) if file is None or document is None: raise NotFoundError(DOCUMENT_NOT_FOUND) if file.status is FileStatus.SUPERSEDED: raise ConflictError(FILE_SUPERSEDED, "файл документа заменён новой версией") if document.status is DocumentStatus.CONFIRMED: raise ConflictError(DOCUMENT_READONLY, "подтверждённая форма только для чтения: загрузите новый файл") if document.status not in CONFIRMABLE_STATUSES: raise ConflictError(DOCUMENT_NOT_CONFIRMABLE, f"форма в статусе {document.status.value} не правится") # читаем под блокировкой: `_visible_statement` загрузил шапку до неё, а параллельный запрос мог её изменить # (например, поставить акцепт) — без `refresh` в сессии остался бы устаревший экземпляр meta = await repo.get_meta(self._session, document.id, refresh=True) if meta is None: raise ConflictError(DOCUMENT_NOT_STATEMENT, "документ — не форма ББ/ОФР") return document, meta async def checks_report( self, document: Document, columns: list[FinancialStatementColumn], aggregates: dict[str, CodeAggregate] ) -> tuple[FailedCheck | None, dict[str, str]]: """Мягкая межформенная проверка (D-5) и родители подстрок (`parent_code` ответа; AURORA-1527). Нарушения §7 сюда не входят: ответ читает их из `document_issue`. Родители — по агрегату формы по коду, в том же порядке, что и проверка «сумма подстрок = родитель» пакета.""" _, amounts = checks_input(columns, aggregates) parents = subline_parents([code for code, _ in amounts]) balance_1370, income_2400 = await repo.pair_amounts(self._session, document) return cross_form_check(balance_1370, income_2400, self._engine.tolerance()), parents async def get(self, document_id: uuid.UUID) -> schemas.StatementOut: document, meta = await self._visible_statement(document_id) return await self.build_out(document, meta) async def build_out(self, document: Document, meta: FinancialStatementMeta) -> schemas.StatementOut: """Ответ формы по текущему состоянию БД, достроенный по справочнику редакции (design D-3; используется и ответами правки/акцепта). В БД ничего не пишет.""" columns, lines = await repo.load_form(self._session, meta.id) _, aggregates = await form_aggregate(self._session, document, meta) cross, parents = await self.checks_report(document, columns, aggregates) edition = edition_for(document.reporting_date) catalog = load_line_catalog() return schemas.statement_out( document, meta, columns, lines, cross, edition=edition, template=catalog.template_lines(edition, document.document_type.value), parents=parents, aggregates=aggregates, issues=await repo.issues(self._session, document.id), catalog=catalog, ) async def hard_failures(self, document: Document, meta: FinancialStatementMeta) -> tuple[DocumentIssue, ...]: """Нарушенные жёсткие проверки §7; при акцепте — пустой кортеж (используется и подтверждением формы).""" return await statement_hard_failures(self._session, document, meta) async def recompute(self, document: Document, meta: FinancialStatementMeta) -> None: """Единая точка пересчёта после правки или акцепта (spec arm-core «Пересчёт после правки»; D-6): общий код `arm_statements`, запись результата — `recompute_form`. `CONFIRMED` не трогает.""" if document.status is DocumentStatus.CONFIRMED: return job = await self._session.get(RecognitionJob, meta.recognition_job_id) await recompute_form(self._session, document=document, meta=meta, engine=self._engine, job=job) async def _apply_cells( self, document: Document, meta: FinancialStatementMeta, body: schemas.StatementPatch, edit: _Edit ) -> None: """Значения `body.values` (D-7 Среза 3b; design D-4 statement-template-lines): сумма — `original_amount` при первой правке, `is_modified`, `modified_by/at`; `confirmed` — подтверждение без правки значения (Р-25). Ячейка, которой нет в БД, создаётся при первой непустой правке суммы; строка по `line_code` — из справочника редакции документа. Вся адресация проверяется до первой вставки.""" if not body.values: return keys = [(cell.line_id or cell.line_code, cell.column_id) for cell in body.values] if len(keys) != len(set(keys)): raise DomainValidationError(DUPLICATE_CELL, "ячейка указана в запросе дважды") targets = await self._resolve_cells(document, meta, body) existing = await repo.lock_cells( self._session, meta.id, [(line.id, cell.column_id) for cell, line, _ in targets if line is not None] ) for cell, line, _ in targets: if cell.confirmed is not None and line is not None and (line.id, cell.column_id) not in existing: raise DomainValidationError(UNKNOWN_CELL, "подтверждать нечего: значения в этой ячейке нет") created: dict[str, FinancialStatementLine] = {} now = utc_now() for cell, line, item in targets: value = existing.get((line.id, cell.column_id)) if line is not None else None if cell.confirmed is not None: if value is not None and line is not None: self._apply_confirmation(value, line, cell.confirmed, edit, now) continue if value is None: if cell.amount is None: continue # пустая сумма в несуществующую ячейку — не изменение line = line or created.get(item.code if item else "") # noqa: PLW2901 - строка, созданная выше if line is None and item is not None: line = created[item.code] = await repo.add_line(self._session, meta.id, item.code, item.name) # noqa: PLW2901 if line is None: raise DomainValidationError(UNKNOWN_CELL, "ячейка не принадлежит форме") value = await repo.add_value(self._session, line.id, cell.column_id) if line is not None: self._apply_amount(value, line, cell.amount, edit, now) def _apply_amount( self, value: FinancialStatementValue, line: FinancialStatementLine, amount: Decimal | None, edit: _Edit, now: Any, # noqa: ANN401 - datetime ) -> None: """Новая сумма ячейки: `original_amount` при первой правке, `is_modified`, `modified_by/at`.""" if value.amount == amount: return edit.record(f"value:{value.id}", value.amount, amount) if not value.is_modified and value.original_amount is None: value.original_amount = value.amount value.amount = amount value.is_modified = value.amount != value.original_amount value.modified_by, value.modified_at = self._audit.user_id, now def _apply_confirmation( self, value: FinancialStatementValue, line: FinancialStatementLine, confirmed: bool, # noqa: FBT001 - значение поля запроса edit: _Edit, now: Any, # noqa: ANN401 - datetime ) -> None: """Подтверждение ячейки (`is_confirmed`, `confirmed_by/at`) без правки значения; `is_modified` не ставится. Снятие возвращает `needs_review` по уверенности распознавания. Акцепт проверок не сбрасывает (Р-25).""" if value.is_confirmed == confirmed: return edit.record(f"value:{value.id}:confirmed", value.is_confirmed, confirmed, data=False) value.is_confirmed = confirmed value.confirmed_by, value.confirmed_at = (self._audit.user_id, now) if confirmed else (None, None) async def _resolve_cells( self, document: Document, meta: FinancialStatementMeta, body: schemas.StatementPatch ) -> list[tuple[schemas.CellPatch, FinancialStatementLine | None, TemplateLine | None]]: """Адресация ячеек правки: строка формы (`line_id`) либо строка справочника редакции (`line_code`). Ничего не вставляет — любая ошибка адресации (422) происходит до первой записи.""" columns = await repo.column_ids(self._session, meta.id) lines = await repo.form_lines(self._session, meta.id) by_id = {line.id: line for line in lines} codes = {line.line_code for line in lines if line.line_status is StatementLineStatus.ok} edition = edition_for(document.reporting_date) catalog_lines = load_line_catalog().template_lines(edition, document.document_type.value) template = {item.code: item for item in catalog_lines} targets: list[tuple[schemas.CellPatch, FinancialStatementLine | None, TemplateLine | None]] = [] for cell in body.values: if cell.column_id not in columns: raise DomainValidationError(UNKNOWN_CELL, "ячейка не принадлежит форме") if cell.line_id is not None: line = by_id.get(cell.line_id) if line is None: raise DomainValidationError(UNKNOWN_CELL, "ячейка не принадлежит форме") targets.append((cell, line, None)) continue if cell.line_code in codes: raise DomainValidationError(UNKNOWN_CELL, LINE_EXISTS) item = template.get(cell.line_code or "") if item is None: raise DomainValidationError(UNKNOWN_CELL, "строки с таким кодом нет в форме этой редакции") targets.append((cell, None, item)) return targets async def _apply_columns( self, meta: FinancialStatementMeta, columns: list[schemas.ColumnPatch], edit: _Edit ) -> bool: """Даты колонок (spec arm-core «PATCH формы»): новая дата — `period_kind = absolute`, снимаются предупреждения колонки; дата, равная текущей, у колонки с `COLUMN_WITHOUT_DATE` / `COLUMN_DATE_UNCERTAIN` — подтверждение, иначе запрос не меняет ничего. Возвращает `True`, если даты колонок изменились (нужен пересчёт `is_current` и периода).""" if not columns: return False by_id = {column.id: column for column in (await repo.load_form(self._session, meta.id))[0]} changed = False for patch in columns: column = by_id[patch.column_id] cleared = await repo.delete_issues( self._session, meta.document_id, codes=COLUMN_CLEARS, kind=DocumentIssueKind.warning, source=IssueSource.service, column_id=column.id, ) if patch.period_date != column.period_date: edit.record(f"column:{column.id}", column.period_date, patch.period_date) column.period_date, column.period_kind = patch.period_date, ABSOLUTE changed = True elif cleared: edit.record(f"column:{column.id}:date_confirmed", False, True, data=False) column.period_kind = ABSOLUTE return changed async def _check_columns(self, meta: FinancialStatementMeta, columns: list[schemas.ColumnPatch]) -> None: """Адресация `columns[]` до первой записи (P3-10): колонка этой формы, без повторов.""" if not columns: return known = await repo.column_ids(self._session, meta.id) if any(patch.column_id not in known for patch in columns): raise DomainValidationError(UNKNOWN_COLUMN, "колонка не принадлежит форме") if len({patch.column_id for patch in columns}) != len(columns): raise DomainValidationError(DUPLICATE_COLUMN, "колонка указана в запросе дважды") async def _clear_issues(self, document: Document, field: str) -> None: """Снять предупреждения сервиса, которые снимает правка (подтверждение) поля шапки.""" await repo.delete_issues( self._session, document.id, codes=HEADER_CLEARS[field], kind=DocumentIssueKind.warning, source=IssueSource.service, ) async def _apply_header( self, document: Document, meta: FinancialStatementMeta, header: schemas.HeaderPatch, edit: _Edit ) -> bool: """Поля шапки (D-7): `*_modified = true`, `*_original_value` не трогаем; пересчёт контрагента. Дата отчёта, равная текущей, у поля из `needs_review_fields` — подтверждение даты. Возвращает `True`, если дата отчёта изменилась (нужен пересчёт периода).""" fields = header.model_fields_set who_changed = False if "inn" in fields and header.inn != meta.inn: edit.record("inn", meta.inn, header.inn) meta.inn, meta.inn_modified = header.inn, True await self._clear_issues(document, "inn") who_changed = True name = header.organization_name if "organization_name" in fields and name is not None and name != meta.organization_name: edit.record("organization_name", meta.organization_name, name) meta.organization_name, meta.organization_name_modified = name, True await self._clear_issues(document, "organization_name") who_changed = True if who_changed: owner = await self._session.get(LegalEntity, document.legal_entity_id) if owner is None: raise NotFoundError(DOCUMENT_NOT_FOUND) index = await CounterpartyIndex.load(self._session) document.counterparty_check = counterparty_check( meta.inn, meta.organization_name, owner.inn, index.owner_names(owner) ) date_changed = False review = [str(item) for item in meta.needs_review_fields] if isinstance(meta.needs_review_fields, list) else [] if "report_date" in fields: if header.report_date != meta.report_date: edit.record("report_date", meta.report_date, header.report_date) meta.report_date, meta.report_date_modified = header.report_date, True date_changed = True await self._clear_issues(document, "report_date") if "report_date" in review: meta.needs_review_fields = [item for item in review if item != "report_date"] elif header.report_date is not None and "report_date" in review: edit.record( "needs_review_fields", review, [item for item in review if item != "report_date"], data=False ) meta.needs_review_fields = [item for item in review if item != "report_date"] await self._clear_issues(document, "report_date") if "units" in fields and header.units is not None and header.units != meta.units: edit.record("units", meta.units, header.units) meta.units, meta.units_modified = header.units, True return date_changed async def _resolve_relative_columns(self, meta: FinancialStatementMeta, edit: _Edit) -> bool: """Правка `header.report_date` (в том числе `null`): `period_date` колонок с `period_kind` `relative:*` пересчитывается от новой даты отчёта до пересчёта формы; ручные (`absolute`) даты не трогаются. `True` — даты колонок изменились (нужен пересчёт `is_current` и периода).""" changed = False for column in (await repo.load_form(self._session, meta.id))[0]: if column.period_kind is None or not str(column.period_kind).startswith("relative:"): continue resolved = self._engine.resolve_relative(PeriodKind(str(column.period_kind)), meta.report_date) if resolved != column.period_date: edit.record(f"column:{column.id}", column.period_date, resolved, data=False) column.period_date = resolved changed = True return changed async def _refresh_period(self, document: Document, meta: FinancialStatementMeta, *, columns_changed: bool) -> None: """Период документа — по §9.5 п. 2 (дата отчёта, иначе самая поздняя дата колонки); `is_current` колонок — колонка с самой поздней датой (меньший `column_index` при равенстве), как при приёме.""" columns, _ = await repo.load_form(self._session, meta.id) if columns_changed: latest = max((c.period_date for c in columns if c.period_date is not None), default=None) first = min((c.column_index for c in columns if c.period_date == latest), default=None) for column in columns: column.is_current = None if latest is None else column.column_index == first period = statement_period(meta.report_date, [column.period_date for column in columns]) document.reporting_date = period document.period_year = period.year if period is not None else None document.period_quarter = quarter_of(period) if period is not None else None async def patch(self, document_id: uuid.UUID, body: schemas.StatementPatch) -> schemas.StatementOut: """Правка строк, значений, дат колонок и шапки (D-7): одна транзакция, один `audit_event`, один пересчёт. Порядок частей — `lines` → `values` → `columns` → `header` (P3-10); адресация проверяется до первой записи. Акцепт сбрасывает только правка данных; подтверждения (ячейки, даты) — нет (Р-25).""" document, meta = await self._editable(document_id) header = body.header if header is not None and "inn" in header.model_fields_set and not INN_RE.fullmatch(header.inn or ""): raise DomainValidationError(INVALID_INN, "ИНН — 10 или 12 цифр") await self._check_columns(meta, body.columns) edit = _Edit({}, {}) if await apply_line_edits( self._session, document=document, meta=meta, items=body.lines, user_id=self._audit.user_id, before=edit.before, after=edit.after, ): edit.data_edited = True await self._apply_cells(document, meta, body, edit) columns_changed = await self._apply_columns(meta, body.columns, edit) date_changed = False if header is not None: date_changed = await self._apply_header(document, meta, header, edit) if date_changed: columns_changed = await self._resolve_relative_columns(meta, edit) or columns_changed if not edit.after: return await self.build_out(document, meta) if columns_changed or date_changed: await self._refresh_period(document, meta, columns_changed=columns_changed) if edit.data_edited and meta.checks_accepted_at is not None: edit.record( "checks_accepted", {"by": meta.checks_accepted_by, "at": meta.checks_accepted_at, "comment": meta.checks_accepted_comment}, None, ) meta.checks_accepted_by = meta.checks_accepted_at = meta.checks_accepted_comment = None await self._session.flush() await self.recompute(document, meta) await self._audit.record( action=AuditAction.RECOGNIZED_DATA_EDIT, section=AuditSection.RECOGNITION, object_type=OBJECT_TYPE, object_id=document.id, before_value=edit.before, after_value=edit.after, ) return await self.build_out(document, meta) async def accept_checks(self, document_id: uuid.UUID, body: schemas.ChecksAcceptIn) -> schemas.StatementOut: """Акцепт непройденных жёстких проверок (D-8): поля акцепта в шапке, пересчёт статуса, аудит.""" document, meta = await self._editable(document_id) if not await self.hard_failures(document, meta): raise ConflictError(NOTHING_TO_ACCEPT, "непройденных жёстких проверок нет") now = utc_now() by = self._audit.user_id meta.checks_accepted_by, meta.checks_accepted_at, meta.checks_accepted_comment = by, now, body.comment await self._session.flush() await self.recompute(document, meta) await self._audit.record( action=AuditAction.RECOGNIZED_DATA_EDIT, section=AuditSection.RECOGNITION, object_type=OBJECT_TYPE, object_id=document.id, before_value={"checks_accepted": None}, after_value={"checks_accepted": {"by": by, "at": now, "comment": body.comment}}, ) return await self.build_out(document, meta)