"""Форма ББ/ОФР для экрана «Проверка распознавания» (spec documents-api; design D-5, D-7, D-8, D-14). Видимость — через ЮЛ-владельца (`files_repo.get_document`): чужой документ — тот же 404, что несуществующий. Проверки §7 считаются на лету и не хранятся. """ import re import uuid 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_checks import ChecksReport, FailedCheck, cross_form_check, load_check_rules, run_checks from src.domain.statement_mapping import edition_for, quarter_of, statement_period, statement_status, values_ok from src.models.directories import LegalEntity from src.models.enums import ( AuditAction, AuditSection, DocumentCheckResult, DocumentStatus, DocumentType, FileStatus, RecognitionStatus, ) from src.models.files import Document, RecognitionJob from src.models.statements import FinancialStatementMeta 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 CONFIRMABLE_STATUSES, DOCUMENT_NOT_CONFIRMABLE, FILE_SUPERSEDED DOCUMENT_NOT_FOUND = "документ не найден" # одинаково для чужого и несуществующего DOCUMENT_NOT_STATEMENT = "DOCUMENT_NOT_STATEMENT" DOCUMENT_READONLY = "DOCUMENT_READONLY" UNKNOWN_CELL = "UNKNOWN_CELL" DUPLICATE_CELL = "DUPLICATE_CELL" NOTHING_TO_ACCEPT = "NOTHING_TO_ACCEPT" INVALID_INN = "INVALID_INN" OBJECT_TYPE = "document" INN_RE = re.compile(r"\d{10}|\d{12}", re.ASCII) FORM_TYPES = frozenset({DocumentType.BALANCE, DocumentType.INCOME}) async def statement_hard_failures( session: AsyncSession, document: Document, meta: FinancialStatementMeta ) -> tuple[FailedCheck, ...]: """Нарушенные жёсткие проверки §7 по текущим суммам формы; при акцепте — пустой кортеж (D-8 Среза 3b).""" if meta.checks_accepted_at is not None: return () column_indexes, amounts = await repo.statement_amounts(session, meta.id) report = run_checks(document.document_type.value, amounts, column_indexes, load_check_rules()) return tuple(report.failed) async def form_confirmable(session: AsyncSession, document: Document) -> bool: """Условия подтверждения формы ББ/ОФР без исключений (§7, D-9 Среза 3b; автоподтверждение — design D-3, OQ-157): период определён и жёстких нарушений §7 нет или они акцептованы. Не-форма и форма без шапки (созданная до Среза 3b) — `True`: условий у них нет. Ручной путь (`DocumentService._check_statement`) проверяет то же, но отвечает 409.""" if document.document_type not in FORM_TYPES: return True meta = await repo.get_meta(session, document.id, refresh=True) if meta is None: return True if document.period_year is None or document.period_quarter is None: return False return not await statement_hard_failures(session, document, meta) class StatementService: def __init__(self, session: AsyncSession, audit: RequestAudit, scope: VisibilityScope) -> None: self._session = session self._audit = audit self._scope = scope 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, meta: FinancialStatementMeta ) -> tuple[ChecksReport, FailedCheck | None]: """Проверки §7 по текущим суммам формы и мягкая межформенная проверка (D-5).""" rules = load_check_rules() column_indexes, amounts = await repo.statement_amounts(self._session, meta.id) report = run_checks(document.document_type.value, amounts, column_indexes, rules) balance_1370, income_2400 = await repo.pair_amounts(self._session, document) return report, cross_form_check(balance_1370, income_2400, rules.tolerance) 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: """Ответ формы по текущему состоянию БД (используется и ответами правки/акцепта).""" columns, lines = await repo.load_form(self._session, meta.id) report, cross = await self.checks_report(document, meta) return schemas.statement_out( document, meta, columns, lines, report, cross, edition=edition_for(document.reporting_date) ) async def hard_failures(self, document: Document, meta: FinancialStatementMeta) -> tuple[FailedCheck, ...]: """Нарушенные жёсткие проверки §7; при акцепте — пустой кортеж (используется и подтверждением формы).""" return await statement_hard_failures(self._session, document, meta) async def recompute_status(self, document: Document, meta: FinancialStatementMeta) -> None: """D-6 после правки или акцепта; `CONFIRMED` не трогает. Предупреждения маппинга файла не хранятся на уровне документа, поэтому передаются пустыми (`warnings=()`). """ if document.status is DocumentStatus.CONFIRMED: return job = await self._session.get(RecognitionJob, meta.recognition_job_id) document.status = statement_status( job_done=job is not None and job.status is RecognitionStatus.done, checks_passed=not await self.hard_failures(document, meta), 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 repo.value_flags(self._session, meta.id)), has_lines=await repo.has_lines(self._session, meta.id), warnings=(), ) document.updated_at = utc_now() async def _apply_cells( self, meta: FinancialStatementMeta, body: schemas.StatementPatch, before: dict[str, Any], after: dict[str, Any] ) -> None: """Значения `body.values` (D-7): `original_amount` при первой правке, `is_modified`, `modified_by/at`.""" keys = [(cell.line_id, cell.column_id) for cell in body.values] if len(keys) != len(set(keys)): raise DomainValidationError(DUPLICATE_CELL, "ячейка указана в запросе дважды") cells = await repo.lock_cells(self._session, meta.id, keys) if len(cells) != len(set(keys)): raise DomainValidationError(UNKNOWN_CELL, "ячейка не принадлежит форме") now = utc_now() for cell in body.values: value = cells[(cell.line_id, cell.column_id)] if value.amount == cell.amount: continue key = f"value:{value.id}" before[key], after[key] = value.amount, cell.amount if not value.is_modified and value.original_amount is None: value.original_amount = value.amount value.amount = cell.amount value.is_modified = value.amount != value.original_amount value.modified_by, value.modified_at = self._audit.user_id, now async def _apply_header( self, document: Document, meta: FinancialStatementMeta, header: schemas.HeaderPatch, before: dict[str, Any], after: dict[str, Any], ) -> None: """Поля шапки (D-7): `*_modified = true`, `*_original_value` не трогаем; пересчёт контрагента и периода.""" fields = header.model_fields_set if "inn" in fields and header.inn != meta.inn: before["inn"], after["inn"] = meta.inn, header.inn meta.inn, meta.inn_modified = header.inn, True 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) ) if "report_date" in fields and header.report_date != meta.report_date: before["report_date"], after["report_date"] = meta.report_date, header.report_date meta.report_date, meta.report_date_modified = header.report_date, True columns, _ = await repo.load_form(self._session, meta.id) 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 if "units" in fields and header.units is not None and header.units != meta.units: before["units"], after["units"] = meta.units, header.units meta.units, meta.units_modified = header.units, True async def patch(self, document_id: uuid.UUID, body: schemas.StatementPatch) -> schemas.StatementOut: """Правка значений и шапки (D-7): одна транзакция, один `audit_event`, сброс акцепта, пересчёт статуса.""" 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 цифр") before: dict[str, Any] = {} after: dict[str, Any] = {} await self._apply_cells(meta, body, before, after) if header is not None: await self._apply_header(document, meta, header, before, after) if not after: return await self.build_out(document, meta) if meta.checks_accepted_at is not None: before["checks_accepted"] = { "by": meta.checks_accepted_by, "at": meta.checks_accepted_at, "comment": meta.checks_accepted_comment, } after["checks_accepted"] = None meta.checks_accepted_by = meta.checks_accepted_at = meta.checks_accepted_comment = None await self._session.flush() await self.recompute_status(document, meta) await self._audit.record( action=AuditAction.RECOGNIZED_DATA_EDIT, section=AuditSection.RECOGNITION, object_type=OBJECT_TYPE, object_id=document.id, before_value=before, after_value=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_status(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)